src/yacsloader_swig/Test/testExec.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2006-2019  CEA/DEN, EDF R&D
   3 #
   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.
   8 #
   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.
  13 #
  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
  17 #
  18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  19 #
  20
  21 import time
  22 import unittest
  23 import threading
  24
  25 import SALOMERuntime
  26 import loader
  27 import pilot
  28
  29 class TestExec(unittest.TestCase):
  30
  31     def setUp(self):
  32         SALOMERuntime.RuntimeSALOME_setRuntime(1)
  33         self.l = loader.YACSLoader()
  34         self.e = pilot.ExecutorSwig()
  35         self.p = self.l.load("samples/aschema.xml")
  36         pass
  37
  38     def test1_StepByStep(self):
  39         # --- execution step by step
  40
  41         print("================= Start of STEPBYSTEP ===================")
  42         self.e.setExecMode(1) # YACS::STEPBYSTEP
  43
  44         run1 = threading.Thread(None, self.e.RunPy, "stepbystep", (self.p,0))
  45         run1.start()
  46         time.sleep(0.1)       # let the thread be initialised
  47         #e.displayDot(self.p)
  48
  49         tocont = True
  50         while tocont:
  51             self.e.waitPause()
  52             #e.displayDot(p)
  53             bp = self.e.getTasksToLoad()
  54             print("nexts possible steps = ", bp)
  55             if len(bp) > 0:
  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()
  60             else:
  61                 tocont = False
  62                 pass
  63             print("toContinue = ", tocont)
  64             pass
  65
  66         self.e.resumeCurrentBreakPoint()
  67         run1.join()
  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 =====================")
  77         pass
  78
  79     def test2_StopToBreakpoint(self):
  80         # --- start execution, set a breakpoint before node48, then continue
  81         time.sleep(1)
  82         print("================= Start of BREAKPOINT ===================")
  83         brp=['node48']
  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))
  87         self.run2.start()
  88         time.sleep(0.1)
  89         self.e.waitPause()
  90         #self.e.displayDot(p)
  91         print("================= reach BREAKPOINT ======================")
  92         # --- resume from breakpoint
  93         print("=========== BREAKPOINT, start RESUME ====================")
  94         time.sleep(1)
  95         self.e.setExecMode(0) # YACS::CONTINUE
  96         self.e.resumeCurrentBreakPoint()
  97         time.sleep(0.1)
  98         self.e.waitPause()
  99         #self.e.displayDot(p)
 100         self.run2.join()
 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 =========================")
 110         pass
 111
 112     def test3_RunWithoutBreakpoints(self):
 113         # --- start execution, run without breakpoints
 114         time.sleep(1)
 115
 116         print("================= Start of CONTINUE =====================")
 117         self.e.setExecMode(0) # YACS::CONTINUE
 118         run3 = threading.Thread(None, self.e.RunPy, "continue", (self.p,0))
 119         run3.start()
 120         time.sleep(0.1)
 121         self.e.waitPause()
 122         #self.e.displayDot(p)
 123         run3.join()
 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 =======================")
 133         pass
 134
 135     def test4_StopOnError(self):
 136         # --- stop execution on first error and save state
 137         time.sleep(1)
 138
 139         print("================= Start of STOPONERROR ==================")
 140         self.e.setStopOnError()
 141         run4 = threading.Thread(None, self.e.RunPy, "continue", (self.p,0))
 142         run4.start()
 143         time.sleep(0.1)
 144         self.e.waitPause()
 145         self.e.saveState("dumpErrorASchema.xml")
 146         self.e.setStopOnError(False)
 147         self.e.resumeCurrentBreakPoint()
 148         time.sleep(0.1)
 149         self.e.waitPause()
 150         run4.join()
 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 =====================")
 156         pass
 157
 158     def test5_PartialExec(self):
 159         # --- stop execution after breakpoint
 160         time.sleep(1)
 161
 162         print("================= Start of PARTIALEXEC ===================")
 163         brp=['node35']
 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))
 168         run5.start()
 169         time.sleep(0.1)
 170         self.e.waitPause()
 171         #self.e.displayDot(self.p)
 172         self.e.saveState('dumpPartialASchema.xml')
 173         #self.e.displayDot(self.p)
 174         self.e.stopExecution()
 175         run5.join()
 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 ==========")
 180         pass
 181
 182     pass
 183
 184 if __name__ == '__main__':
 185     import os
 186     import sys
 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)
 192
 193     sys.exit(not result.wasSuccessful())