2 # Copyright (C) 2006-2016 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
29 class TestExec(unittest.TestCase):
32 SALOMERuntime.RuntimeSALOME_setRuntime(1)
33 self.l = loader.YACSLoader()
34 self.e = pilot.ExecutorSwig()
35 self.p = self.l.load("samples/aschema.xml")
38 def test1_StepByStep(self):
39 # --- execution step by step
41 print("================= Start of STEPBYSTEP ===================")
42 self.e.setExecMode(1) # YACS::STEPBYSTEP
44 run1 = threading.Thread(None, self.e.RunPy, "stepbystep", (self.p,0))
46 time.sleep(0.1) # let the thread be initialised
53 bp = self.e.getTasksToLoad()
54 print("nexts possible steps = ", bp)
56 tte= bp[-1:] # only one node at each step, the last one in the list
57 r = self.e.setStepsToExecute(tte)
58 self.e.resumeCurrentBreakPoint()
59 tocont = self.e.isNotFinished()
63 print("toContinue = ", tocont)
66 self.e.resumeCurrentBreakPoint()
68 self.assertEqual(106, self.p.getChildByName('node48').getEffectiveState())
69 self.assertEqual(999, self.p.getChildByName('node13').getEffectiveState())
70 self.assertEqual(888, self.p.getChildByName('node14').getEffectiveState())
71 self.assertEqual(777, self.p.getChildByName('c1').getEffectiveState())
72 self.assertEqual(777, self.p.getChildByName('c1.c1.n2').getEffectiveState())
73 self.assertEqual(106, self.p.getChildByName('c0.c1.n1').getEffectiveState())
74 self.assertEqual(999, self.p.getChildByName('c0.n2').getEffectiveState())
75 self.assertEqual(888, self.p.getChildByName('node62').getEffectiveState())
76 print("================= End of STEPBYSTEP =====================")
79 def test2_StopToBreakpoint(self):
80 # --- start execution, set a breakpoint before node48, then continue
82 print("================= Start of BREAKPOINT ===================")
84 self.e.setListOfBreakPoints(brp)
85 self.e.setExecMode(2) # YACS::STOPBEFORENODES
86 self.run2 = threading.Thread(None, self.e.RunPy, "breakpoint", (self.p,0))
91 print("================= reach BREAKPOINT ======================")
92 # --- resume from breakpoint
93 print("=========== BREAKPOINT, start RESUME ====================")
95 self.e.setExecMode(0) # YACS::CONTINUE
96 self.e.resumeCurrentBreakPoint()
101 self.assertEqual(106, self.p.getChildByName('node48').getEffectiveState())
102 self.assertEqual(999, self.p.getChildByName('node13').getEffectiveState())
103 self.assertEqual(888, self.p.getChildByName('node14').getEffectiveState())
104 self.assertEqual(777, self.p.getChildByName('c1').getEffectiveState())
105 self.assertEqual(777, self.p.getChildByName('c1.c1.n2').getEffectiveState())
106 self.assertEqual(106, self.p.getChildByName('c0.c1.n1').getEffectiveState())
107 self.assertEqual(999, self.p.getChildByName('c0.n2').getEffectiveState())
108 self.assertEqual(888, self.p.getChildByName('node62').getEffectiveState())
109 print("================= End of RESUME =========================")
112 def test3_RunWithoutBreakpoints(self):
113 # --- start execution, run without breakpoints
116 print("================= Start of CONTINUE =====================")
117 self.e.setExecMode(0) # YACS::CONTINUE
118 run3 = threading.Thread(None, self.e.RunPy, "continue", (self.p,0))
122 #self.e.displayDot(p)
124 self.assertEqual(106, self.p.getChildByName('node48').getEffectiveState())
125 self.assertEqual(999, self.p.getChildByName('node13').getEffectiveState())
126 self.assertEqual(888, self.p.getChildByName('node14').getEffectiveState())
127 self.assertEqual(777, self.p.getChildByName('c1').getEffectiveState())
128 self.assertEqual(777, self.p.getChildByName('c1.c1.n2').getEffectiveState())
129 self.assertEqual(106, self.p.getChildByName('c0.c1.n1').getEffectiveState())
130 self.assertEqual(999, self.p.getChildByName('c0.n2').getEffectiveState())
131 self.assertEqual(888, self.p.getChildByName('node62').getEffectiveState())
132 print("================= End of CONTINUE =======================")
135 def test4_StopOnError(self):
136 # --- stop execution on first error and save state
139 print("================= Start of STOPONERROR ==================")
140 self.e.setStopOnError()
141 run4 = threading.Thread(None, self.e.RunPy, "continue", (self.p,0))
145 self.e.saveState("dumpErrorASchema.xml")
146 self.e.setStopOnError(False)
147 self.e.resumeCurrentBreakPoint()
151 #self.e.displayDot(self.p)
152 s13 = self.p.getChildByName('node13').getEffectiveState()
153 s43 = self.p.getChildByName('node43').getEffectiveState()
154 self.assertTrue((s13==999) or (s43==999))
155 print("================= End of STOPONERROR =====================")
158 def test5_PartialExec(self):
159 # --- stop execution after breakpoint
162 print("================= Start of PARTIALEXEC ===================")
164 self.e.setListOfBreakPoints(brp)
165 self.e.setExecMode(2) # YACS::STOPBEFORENODES
166 #self.e.displayDot(self.p)
167 run5 = threading.Thread(None, self.e.RunPy, "breakpoint", (self.p,0))
171 #self.e.displayDot(self.p)
172 self.e.saveState('dumpPartialASchema.xml')
173 #self.e.displayDot(self.p)
174 self.e.stopExecution()
176 #self.e.displayDot(self.p)
177 self.assertEqual(106, self.p.getChildByName('node34').getEffectiveState())
178 self.assertEqual(101, self.p.getChildByName('node35').getEffectiveState())
179 print("================= reach BREAKPOINT PARTIAL EXEC ==========")
184 if __name__ == '__main__':
187 U = os.getenv('USER')
188 with open(os.path.join("/tmp", U, "UnitTestsResult"), 'a') as f:
189 f.write(" --- TEST src/yacsloader: testExec.py\n")
190 suite = unittest.makeSuite(TestExec)
191 result = unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
193 sys.exit(not result.wasSuccessful())