bin/appliskel/tests/salomeInstance/instances.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2015-2019  CEA/DEN, EDF R&D, OPEN CASCADE
   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 unittest
  22 import multiprocessing
  23
  24 def new_instance(running_instances):
  25   from salome_instance import SalomeInstance
  26   instance = SalomeInstance.start()
  27   print("Instance created and now running on port", instance.get_port())
  28   running_instances.put(instance)
  29 #
  30
  31 class TestLauncher(unittest.TestCase):
  32
  33   def __createInstances(self, nb):
  34     running_instances = multiprocessing.Queue()
  35     processes = [
  36       multiprocessing.Process(target=new_instance, args=(running_instances,))
  37       for i in range(nb)
  38       ]
  39     return running_instances, processes
  40   #
  41
  42   def __terminateInstances(self, running_instances):
  43     while not running_instances.empty():
  44       instance = running_instances.get()
  45       print("Terminate instance running on port", instance.get_port())
  46       instance.stop()
  47   #
  48
  49   def __connectToInstance(self, port):
  50     import setenv
  51     setenv.main(True)
  52     import runConsole
  53     return runConsole.connect(["-p", port, "-c", "exit()"]) # auto-logout
  54   #
  55
  56   def testSequential(self):
  57     running_instances, processes = self.__createInstances(3)
  58     for p in processes:
  59       p.start()
  60       p.join()
  61       pass
  62
  63     self.__terminateInstances(running_instances)
  64   #
  65
  66   def testConcurrent(self):
  67     running_instances, processes = self.__createInstances(5)
  68     for p in processes:
  69       p.start()
  70       pass
  71     for p in processes:
  72       p.join()
  73       pass
  74
  75     self.__terminateInstances(running_instances)
  76   #
  77
  78   def testConnectInstance(self):
  79     # Create some instances
  80     running_instances, processes = self.__createInstances(5)
  81     for p in processes:
  82       p.start()
  83       pass
  84     for p in processes:
  85       p.join()
  86       pass
  87
  88     # move queued instances to a list
  89     all_instances = []
  90     while not running_instances.empty():
  91       all_instances.append(running_instances.get())
  92
  93     # Connect to one instance
  94     import runConsole
  95     port = all_instances[len(all_instances)//2].get_port()
  96     print("Connect to instance running on port", port)
  97     self.__connectToInstance(port)
  98
  99     # Connect to another instance
 100     import runConsole
 101     port = all_instances[len(all_instances)//4].get_port()
 102     print("Connect to instance running on port", port)
 103     self.__connectToInstance(port)
 104
 105     # Terminate instances
 106     for instance in all_instances:
 107       print("Terminate instance running on port", instance.get_port())
 108       instance.stop()
 109   #
 110
 111 if __name__ == "__main__":
 112   unittest.main()
 113 #