2 # Copyright (C) 2015-2019 CEA/DEN, EDF R&D, OPEN CASCADE
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
22 import multiprocessing
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)
31 class TestLauncher(unittest.TestCase):
33 def __createInstances(self, nb):
34 running_instances = multiprocessing.Queue()
36 multiprocessing.Process(target=new_instance, args=(running_instances,))
39 return running_instances, processes
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())
49 def __connectToInstance(self, port):
53 return runConsole.connect(["-p", port, "-c", "exit()"]) # auto-logout
56 def testSequential(self):
57 running_instances, processes = self.__createInstances(3)
63 self.__terminateInstances(running_instances)
66 def testConcurrent(self):
67 running_instances, processes = self.__createInstances(5)
75 self.__terminateInstances(running_instances)
78 def testConnectInstance(self):
79 # Create some instances
80 running_instances, processes = self.__createInstances(5)
88 # move queued instances to a list
90 while not running_instances.empty():
91 all_instances.append(running_instances.get())
93 # Connect to one instance
95 port = all_instances[len(all_instances)//2].get_port()
96 print("Connect to instance running on port", port)
97 self.__connectToInstance(port)
99 # Connect to another instance
101 port = all_instances[len(all_instances)//4].get_port()
102 print("Connect to instance running on port", port)
103 self.__connectToInstance(port)
105 # Terminate instances
106 for instance in all_instances:
107 print("Terminate instance running on port", instance.get_port())
111 if __name__ == "__main__":