bin/appliskel/tests/salomeInstance/instances.py

   1 # Copyright (C) 2015-2016  CEA/DEN, EDF R&D, OPEN CASCADE
   2 #
   3 # This library is free software; you can redistribute it and/or
   4 # modify it under the terms of the GNU Lesser General Public
   5 # License as published by the Free Software Foundation; either
   6 # version 2.1 of the License, or (at your option) any later version.
   7 #
   8 # This library is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11 # Lesser General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU Lesser General Public
  14 # License along with this library; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 #
  17 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  18 #
  19
  20 import unittest
  21 import multiprocessing
  22
  23 def new_instance(running_instances):
  24   from salome_instance import SalomeInstance
  25   instance = SalomeInstance.start()
  26   print("Instance created and now running on port", instance.get_port())
  27   running_instances.put(instance)
  28 #
  29
  30 class TestLauncher(unittest.TestCase):
  31
  32   def __createInstances(self, nb):
  33     running_instances = multiprocessing.Queue()
  34     processes = [
  35       multiprocessing.Process(target=new_instance, args=(running_instances,))
  36       for i in range(nb)
  37       ]
  38     return running_instances, processes
  39   #
  40
  41   def __terminateInstances(self, running_instances):
  42     while not running_instances.empty():
  43       instance = running_instances.get()
  44       print("Terminate instance running on port", instance.get_port())
  45       instance.stop()
  46   #
  47
  48   def __connectToInstance(self, port):
  49     import setenv
  50     setenv.main(True)
  51     import runConsole
  52     return runConsole.connect(["-p", port, "-c", "exit()"]) # auto-logout
  53   #
  54
  55   def testSequential(self):
  56     running_instances, processes = self.__createInstances(3)
  57     for p in processes:
  58       p.start()
  59       p.join()
  60       pass
  61
  62     self.__terminateInstances(running_instances)
  63   #
  64
  65   def testConcurrent(self):
  66     running_instances, processes = self.__createInstances(5)
  67     for p in processes:
  68       p.start()
  69       pass
  70     for p in processes:
  71       p.join()
  72       pass
  73
  74     self.__terminateInstances(running_instances)
  75   #
  76
  77   def testConnectInstance(self):
  78     # Create some instances
  79     running_instances, processes = self.__createInstances(5)
  80     for p in processes:
  81       p.start()
  82       pass
  83     for p in processes:
  84       p.join()
  85       pass
  86
  87     # move queued instances to a list
  88     all_instances = []
  89     while not running_instances.empty():
  90       all_instances.append(running_instances.get())
  91
  92     # Connect to one instance
  93     import runConsole
  94     port = all_instances[len(all_instances)//2].get_port()
  95     print("Connect to instance running on port", port)
  96     self.__connectToInstance(port)
  97
  98     # Connect to another instance
  99     import runConsole
 100     port = all_instances[len(all_instances)//4].get_port()
 101     print("Connect to instance running on port", port)
 102     self.__connectToInstance(port)
 103
 104     # Terminate instances
 105     for instance in all_instances:
 106       print("Terminate instance running on port", instance.get_port())
 107       instance.stop()
 108   #
 109
 110 if __name__ == "__main__":
 111   unittest.main()
 112 #