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     while not running_instances.empty():
  48       instance = running_instances.get()
  49       print("Terminate instance running on port", instance.get_port())
  50       instance.stop()
  51   #
  52
  53   def appli(self, args=None):
  54     if args is None:
  55       args = []
  56     try:
  57       sys.argv = ['runSalome', '-t']
  58       import setenv
  59       setenv.main(True, exeName="salome start")
  60       import runSalome
  61       runSalome.runSalome()
  62     except SystemExit as e:
  63       if str(e) != '0':
  64         logging.error(e)
  65       pass
  66   #
  67   def session(self, args=None):
  68     if args is None:
  69       args = []
  70     try:
  71       import setenv
  72       setenv.main(True)
  73       import runSession
  74       params, args = runSession.configureSession(args, exe="salome shell")
  75       return runSession.runSession(params, args)
  76     except SystemExit as e:
  77       if str(e) != '0':
  78         logging.error(e)
  79       pass
  80   #
  81   def test01_SingleSession(self):
  82     print("** Testing single session **")
  83     self.session(["hello.py"])
  84   #
  85   def test02_MultiSession(self):
  86     print("** Testing multi sessions **")
  87     jobs = []
  88     for i in range(9):
  89       p = multiprocessing.Process(target=self.session, args=(["hello.py"],))
  90       jobs.append(p)
  91       p.start()
  92
  93     for j in jobs:
  94       j.join()
  95   #
  96   def test03_SingleAppli(self):
  97     print("** Testing single appli **")
  98     running_instances, processes = self.__createInstances(1)
  99     for p in processes:
 100       p.start()
 101       pass
 102     for p in processes:
 103       p.join()
 104       pass
 105
 106     self.session(["hello.py"])
 107     self.__terminateInstances(running_instances)
 108   #
 109   def test04_MultiAppli(self):
 110     print("** Testing multi appli **")
 111     running_instances, processes = self.__createInstances(9)
 112     for p in processes:
 113       p.start()
 114       pass
 115     for p in processes:
 116       p.join()
 117       pass
 118
 119     self.session(["hello.py"])
 120     self.__terminateInstances(running_instances)
 121   #
 122 #
 123
 124 if __name__ == "__main__":
 125   unittest.main()
 126 #