src/yacsloader_swig/Test/testResume.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2006-2022  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 import tempfile
  25 import os
  26
  27 import SALOMERuntime
  28 import loader
  29 import pilot
  30 import salome
  31
  32 class TestResume(unittest.TestCase):
  33
  34     def setUp(self):
  35         SALOMERuntime.RuntimeSALOME_setRuntime(1)
  36         self.l = loader.YACSLoader()
  37         self.e = pilot.ExecutorSwig()
  38         self.p = self.l.load("samples/bloc2.xml")
  39         workdir = tempfile.mkdtemp(suffix=".yacstest")
  40         self.statefile = os.path.join(workdir, 'dumpPartialBloc2.xml')
  41         pass
  42
  43     def tearDown(self):
  44       salome.salome_init()
  45       cm = salome.lcc.getContainerManager()
  46       cm.ShutdownContainers()
  47       pass
  48
  49     def test1_PartialExec(self):
  50         # --- stop execution after breakpoint
  51         time.sleep(1)
  52
  53         print("================= Start of PARTIALEXEC ===================")
  54         brp=['b1.b2.node1']
  55         self.e.setListOfBreakPoints(brp)
  56         self.e.setExecMode(2) # YACS::STOPBEFORENODES
  57         #self.e.displayDot(self.p)
  58         run1 = threading.Thread(None, self.e.RunPy, "breakpoint", (self.p,0))
  59         run1.start()
  60         time.sleep(0.1)
  61         self.e.waitPause()
  62         #self.e.displayDot(self.p)
  63         self.e.saveState(self.statefile)
  64         #self.e.displayDot(self.p)
  65         self.e.stopExecution()
  66         #self.e.displayDot(self.p)
  67         self.assertEqual(101, self.p.getChildByName('b1.b2.node1').getEffectiveState())
  68         self.assertEqual(106, self.p.getChildByName('b1.node1').getEffectiveState())
  69         print("================= reach BREAKPOINT PARTIAL EXEC ==========")
  70
  71         # --- reload state from previous partial execution then exec
  72         time.sleep(1)
  73
  74         print("================= Start of EXECLOADEDSTATE ===============")
  75         sp = loader.stateParser()
  76         sl = loader.stateLoader(sp,self.p)
  77         sl.parse(self.statefile)
  78         #self.e.displayDot(self.p)
  79         self.e.setExecMode(0) # YACS::CONTINUE
  80         run2 = threading.Thread(None, self.e.RunPy, "loadState", (self.p,0,True,True))
  81         run2.start()
  82         time.sleep(0.1)
  83         self.e.waitPause()
  84         #self.e.displayDot(self.p)
  85         run2.join()
  86         self.assertEqual(106, self.p.getChildByName('node1').getEffectiveState())
  87         self.assertEqual(106, self.p.getChildByName('node2').getEffectiveState())
  88         self.assertEqual(106, self.p.getChildByName('b1.node1').getEffectiveState())
  89         self.assertEqual(106, self.p.getChildByName('b1.node2').getEffectiveState())
  90         self.assertEqual(106, self.p.getChildByName('b1.b2.node1').getEffectiveState())
  91         self.assertEqual(106, self.p.getChildByName('b1.b2.node2').getEffectiveState())
  92         self.assertEqual(106, self.p.getChildByName('b1.b2.loop1.node1').getEffectiveState())
  93         print("================= End of EXECLOADEDSTATE =================")
  94
  95     pass
  96
  97 if __name__ == '__main__':
  98   dir_test = tempfile.mkdtemp(suffix=".yacstest")
  99   file_test = os.path.join(dir_test,"UnitTestsResult")
 100   with open(file_test, 'a') as f:
 101       f.write("  --- TEST src/yacsloader: testResume.py\n")
 102       suite = unittest.makeSuite(TestResume)
 103       result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
 104   sys.exit(not result.wasSuccessful())