2 # Copyright (C) 2006-2015 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.stopExecution()
148 #self.e.displayDot(self.p)
149 s13 = self.p.getChildByName('node13').getEffectiveState()
150 s43 = self.p.getChildByName('node43').getEffectiveState()
151 self.assert_((s13==999) or (s43==999))
152 print "================= End of STOPONERROR ====================="
155 def test5_PartialExec(self):
156 # --- stop execution after breakpoint
159 print "================= Start of PARTIALEXEC ==================="
161 self.e.setListOfBreakPoints(brp)
162 self.e.setExecMode(2) # YACS::STOPBEFORENODES
163 #self.e.displayDot(self.p)
164 run5 = threading.Thread(None, self.e.RunPy, "breakpoint", (self.p,0))
168 #self.e.displayDot(self.p)
169 self.e.saveState('dumpPartialASchema.xml')
170 #self.e.displayDot(self.p)
171 self.e.stopExecution()
173 #self.e.displayDot(self.p)
174 self.assertEqual(106, self.p.getChildByName('node34').getEffectiveState())
175 self.assertEqual(101, self.p.getChildByName('node35').getEffectiveState())
176 print "================= reach BREAKPOINT PARTIAL EXEC =========="
181 if __name__ == '__main__':
183 U = os.getenv('USER')
184 f=open("/tmp/" + U + "/UnitTestsResult", 'a')
185 f.write(" --- TEST src/yacsloader: testExec.py\n")
186 suite = unittest.makeSuite(TestExec)
187 result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
189 sys.exit(not result.wasSuccessful())