-#!/usr/bin/env python
-# Copyright (C) 2006-2014 CEA/DEN, EDF R&D
+#!/usr/bin/env python3
+# Copyright (C) 2006-2022 CEA/DEN, EDF R&D
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
import time
import unittest
import threading
+import tempfile
+import os
import SALOMERuntime
import loader
import pilot
+import salome
class TestResume(unittest.TestCase):
self.l = loader.YACSLoader()
self.e = pilot.ExecutorSwig()
self.p = self.l.load("samples/bloc2.xml")
+ workdir = tempfile.mkdtemp(suffix=".yacstest")
+ self.statefile = os.path.join(workdir, 'dumpPartialBloc2.xml')
pass
+ def tearDown(self):
+ salome.salome_init()
+ cm = salome.lcc.getContainerManager()
+ cm.ShutdownContainers()
+ pass
+
def test1_PartialExec(self):
# --- stop execution after breakpoint
time.sleep(1)
- print "================= Start of PARTIALEXEC ==================="
+ print("================= Start of PARTIALEXEC ===================")
brp=['b1.b2.node1']
self.e.setListOfBreakPoints(brp)
self.e.setExecMode(2) # YACS::STOPBEFORENODES
time.sleep(0.1)
self.e.waitPause()
#self.e.displayDot(self.p)
- self.e.saveState('dumpPartialBloc2.xml')
+ self.e.saveState(self.statefile)
#self.e.displayDot(self.p)
self.e.stopExecution()
#self.e.displayDot(self.p)
self.assertEqual(101, self.p.getChildByName('b1.b2.node1').getEffectiveState())
self.assertEqual(106, self.p.getChildByName('b1.node1').getEffectiveState())
- print "================= reach BREAKPOINT PARTIAL EXEC =========="
- pass
+ print("================= reach BREAKPOINT PARTIAL EXEC ==========")
- def test2_ExecFromLoadState(self):
# --- reload state from previous partial execution then exec
time.sleep(1)
- print "================= Start of EXECLOADEDSTATE ==============="
+ print("================= Start of EXECLOADEDSTATE ===============")
sp = loader.stateParser()
sl = loader.stateLoader(sp,self.p)
- sl.parse('dumpPartialBloc2.xml')
+ sl.parse(self.statefile)
#self.e.displayDot(self.p)
self.e.setExecMode(0) # YACS::CONTINUE
- run2 = threading.Thread(None, self.e.RunPy, "loadState", (self.p,0,1,1))
+ run2 = threading.Thread(None, self.e.RunPy, "loadState", (self.p,0,True,True))
run2.start()
time.sleep(0.1)
self.e.waitPause()
self.assertEqual(106, self.p.getChildByName('b1.b2.node1').getEffectiveState())
self.assertEqual(106, self.p.getChildByName('b1.b2.node2').getEffectiveState())
self.assertEqual(106, self.p.getChildByName('b1.b2.loop1.node1').getEffectiveState())
- print "================= End of EXECLOADEDSTATE ================="
+ print("================= End of EXECLOADEDSTATE =================")
pass
-import os
-U = os.getenv('USER')
-f=open("/tmp/" + U + "/UnitTestsResult", 'a')
-f.write(" --- TEST src/yacsloader: testResume.py\n")
-suite = unittest.makeSuite(TestResume)
-result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
-f.close()
-sys.exit(not result.wasSuccessful())
+if __name__ == '__main__':
+ dir_test = tempfile.mkdtemp(suffix=".yacstest")
+ file_test = os.path.join(dir_test,"UnitTestsResult")
+ with open(file_test, 'a') as f:
+ f.write(" --- TEST src/yacsloader: testResume.py\n")
+ suite = unittest.makeSuite(TestResume)
+ result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
+ sys.exit(not result.wasSuccessful())