bin/appliskel/tests/concurrentSession/TestConcurrentSession.py

   1 # Copyright (C) 2013-2017  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 tempfile
  22
  23 import os
  24 import sys
  25 import imp
  26 from io import StringIO
  27 import multiprocessing
  28 import logging
  29
  30 def new_instance(running_instances):
  31   from salome_instance import SalomeInstance
  32   instance = SalomeInstance.start()
  33   print("Instance created and now running on port", instance.get_port())
  34   running_instances.put(instance)
  35 #
  36
  37 class TestConcurrentLaunch(unittest.TestCase):
  38   def __createInstances(self, nb):
  39     running_instances = multiprocessing.Queue()
  40     processes = [
  41       multiprocessing.Process(target=new_instance, args=(running_instances,))
  42       for i in range(nb)
  43       ]
  44     return running_instances, processes
  45   #
  46   def __terminateInstances(self, running_instances):
  47     import time
  48     timeout = time.time() + 60*10 # the test duration is about 50 s, we reasonably assume a max duration of 10mn
  49
  50     while not running_instances.empty() and time.time() < timeout:
  51       instance = running_instances.get()
  52       print("Terminate instance running on port", instance.get_port())
  53       instance.stop()
  54   #
  55
  56   def appli(self, args=None):
  57     if args is None:
  58       args = []
  59     try:
  60       sys.argv = ['runSalome', '-t']
  61       import setenv
  62       setenv.main(True, exeName="salome start")
  63       import runSalome
  64       runSalome.runSalome()
  65     except SystemExit as e:
  66       if str(e) != '0':
  67         logging.error(e)
  68       pass
  69   #
  70   def session(self, args=None):
  71     if args is None:
  72       args = []
  73     try:
  74       import setenv
  75       setenv.main(True)
  76       import runSession
  77       params, args = runSession.configureSession(args, exe="salome shell")
  78       return runSession.runSession(params, args)
  79     except SystemExit as e:
  80       if str(e) != '0':
  81         logging.error(e)
  82       pass
  83   #
  84   def test01_SingleSession(self):
  85     print("** Testing single session **")
  86     self.session(["hello.py"])
  87   #
  88   def test02_MultiSession(self):
  89     print("** Testing multi sessions **")
  90     jobs = []
  91     for i in range(9):
  92       p = multiprocessing.Process(target=self.session, args=(["hello.py"],))
  93       jobs.append(p)
  94       p.start()
  95
  96     for j in jobs:
  97       j.join()
  98   #
  99   def test03_SingleAppli(self):
 100     print("** Testing single appli **")
 101     running_instances, processes = self.__createInstances(1)
 102     for p in processes:
 103       p.start()
 104       pass
 105     for p in processes:
 106       p.join()
 107       pass
 108
 109     self.session(["hello.py"])
 110     self.__terminateInstances(running_instances)
 111   #
 112   def test04_MultiAppli(self):
 113     print("** Testing multi appli **")
 114     running_instances, processes = self.__createInstances(9)
 115     for p in processes:
 116       p.start()
 117       pass
 118     for p in processes:
 119       p.join()
 120       pass
 121
 122     self.session(["hello.py"])
 123     self.__terminateInstances(running_instances)
 124   #
 125 #
 126
 127 if __name__ == "__main__":
 128   unittest.main()
 129 #