2 # Copyright (C) 2006-2022 CEA/DEN, EDF R&D
4 # This library is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU Lesser General Public
6 # License as published by the Free Software Foundation; either
7 # version 2.1 of the License, or (at your option) any later version.
9 # This library is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 # Lesser General Public License for more details.
14 # You should have received a copy of the GNU Lesser General Public
15 # License along with this library; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
31 class TestExec(unittest.TestCase):
34 SALOMERuntime.RuntimeSALOME_setRuntime(1)
35 self.l = loader.YACSLoader()
36 self.e = pilot.ExecutorSwig()
37 self.p = self.l.load("samples/aschema.xml")
38 self.workdir = tempfile.mkdtemp(suffix=".yacstest")
41 def test1_StepByStep(self):
42 # --- execution step by step
44 print("================= Start of STEPBYSTEP ===================")
45 self.e.setExecMode(1) # YACS::STEPBYSTEP
47 run1 = threading.Thread(None, self.e.RunPy, "stepbystep", (self.p,0))
49 time.sleep(0.1) # let the thread be initialised
56 bp = self.e.getTasksToLoad()
57 print("nexts possible steps = ", bp)
59 tte= bp[-1:] # only one node at each step, the last one in the list
60 r = self.e.setStepsToExecute(tte)
61 self.e.resumeCurrentBreakPoint()
62 tocont = self.e.isNotFinished()
66 print("toContinue = ", tocont)
69 self.e.resumeCurrentBreakPoint()
71 self.assertEqual(106, self.p.getChildByName('node48').getEffectiveState())
72 self.assertEqual(999, self.p.getChildByName('node13').getEffectiveState())
73 self.assertEqual(888, self.p.getChildByName('node14').getEffectiveState())
74 self.assertEqual(777, self.p.getChildByName('c1').getEffectiveState())
75 self.assertEqual(777, self.p.getChildByName('c1.c1.n2').getEffectiveState())
76 self.assertEqual(106, self.p.getChildByName('c0.c1.n1').getEffectiveState())
77 self.assertEqual(999, self.p.getChildByName('c0.n2').getEffectiveState())
78 self.assertEqual(888, self.p.getChildByName('node62').getEffectiveState())
79 print("================= End of STEPBYSTEP =====================")
82 def test2_StopToBreakpoint(self):
83 # --- start execution, set a breakpoint before node48, then continue
85 print("================= Start of BREAKPOINT ===================")
87 self.e.setListOfBreakPoints(brp)
88 self.e.setExecMode(2) # YACS::STOPBEFORENODES
89 self.run2 = threading.Thread(None, self.e.RunPy, "breakpoint", (self.p,0))
94 print("================= reach BREAKPOINT ======================")
95 # --- resume from breakpoint
96 print("=========== BREAKPOINT, start RESUME ====================")
98 self.e.setExecMode(0) # YACS::CONTINUE
99 self.e.resumeCurrentBreakPoint()
102 #self.e.displayDot(p)
104 self.assertEqual(106, self.p.getChildByName('node48').getEffectiveState())
105 self.assertEqual(999, self.p.getChildByName('node13').getEffectiveState())
106 self.assertEqual(888, self.p.getChildByName('node14').getEffectiveState())
107 self.assertEqual(777, self.p.getChildByName('c1').getEffectiveState())
108 self.assertEqual(777, self.p.getChildByName('c1.c1.n2').getEffectiveState())
109 self.assertEqual(106, self.p.getChildByName('c0.c1.n1').getEffectiveState())
110 self.assertEqual(999, self.p.getChildByName('c0.n2').getEffectiveState())
111 self.assertEqual(888, self.p.getChildByName('node62').getEffectiveState())
112 print("================= End of RESUME =========================")
115 def test3_RunWithoutBreakpoints(self):
116 # --- start execution, run without breakpoints
119 print("================= Start of CONTINUE =====================")
120 self.e.setExecMode(0) # YACS::CONTINUE
121 run3 = threading.Thread(None, self.e.RunPy, "continue", (self.p,0))
125 #self.e.displayDot(p)
127 self.assertEqual(106, self.p.getChildByName('node48').getEffectiveState())
128 self.assertEqual(999, self.p.getChildByName('node13').getEffectiveState())
129 self.assertEqual(888, self.p.getChildByName('node14').getEffectiveState())
130 self.assertEqual(777, self.p.getChildByName('c1').getEffectiveState())
131 self.assertEqual(777, self.p.getChildByName('c1.c1.n2').getEffectiveState())
132 self.assertEqual(106, self.p.getChildByName('c0.c1.n1').getEffectiveState())
133 self.assertEqual(999, self.p.getChildByName('c0.n2').getEffectiveState())
134 self.assertEqual(888, self.p.getChildByName('node62').getEffectiveState())
135 print("================= End of CONTINUE =======================")
138 def test4_StopOnError(self):
139 # --- stop execution on first error and save state
142 print("================= Start of STOPONERROR ==================")
143 self.e.setStopOnError()
144 run4 = threading.Thread(None, self.e.RunPy, "continue", (self.p,0))
148 self.e.saveState(os.path.join(self.workdir, "dumpErrorASchema.xml"))
149 self.e.setStopOnError(False)
150 self.e.resumeCurrentBreakPoint()
154 #self.e.displayDot(self.p)
155 s13 = self.p.getChildByName('node13').getEffectiveState()
156 s43 = self.p.getChildByName('node43').getEffectiveState()
157 self.assertTrue((s13==999) or (s43==999))
158 print("================= End of STOPONERROR =====================")
161 def test5_PartialExec(self):
162 # --- stop execution after breakpoint
165 print("================= Start of PARTIALEXEC ===================")
167 self.e.setListOfBreakPoints(brp)
168 self.e.setExecMode(2) # YACS::STOPBEFORENODES
169 #self.e.displayDot(self.p)
170 run5 = threading.Thread(None, self.e.RunPy, "breakpoint", (self.p,0))
174 #self.e.displayDot(self.p)
175 self.e.saveState(os.path.join(self.workdir, 'dumpPartialASchema.xml'))
176 #self.e.displayDot(self.p)
177 self.e.stopExecution()
179 #self.e.displayDot(self.p)
180 self.assertEqual(106, self.p.getChildByName('node34').getEffectiveState())
181 self.assertEqual(101, self.p.getChildByName('node35').getEffectiveState())
182 print("================= reach BREAKPOINT PARTIAL EXEC ==========")
187 if __name__ == '__main__':
188 dir_test = tempfile.mkdtemp(suffix=".yacstest")
189 file_test = os.path.join(dir_test,"UnitTestsResult")
190 with open(file_test, 'a') as f:
191 f.write(" --- TEST src/yacsloader: testExec.py\n")
192 suite = unittest.makeSuite(TestExec)
193 result = unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
195 sys.exit(not result.wasSuccessful())