Salome HOME
Merge 'master' branch into 'V9_dev' branch
[modules/kernel.git] / 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 #