-# Copyright (C) 2013-2014 CEA/DEN, EDF R&D, OPEN CASCADE
+#!/usr/bin/env python3
+# Copyright (C) 2013-2020 CEA/DEN, EDF R&D, OPEN CASCADE
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
import os
import sys
import imp
-from cStringIO import StringIO
+from io import StringIO
import multiprocessing
import logging
-class TestConcurrentLaunch(unittest.TestCase):
- def setUp(self):
- # Initialize path to SALOME application
- path_to_launcher = os.getenv("SALOME_LAUNCHER")
- appli_dir = os.path.dirname(path_to_launcher)
- sys.path[:0] = [os.path.join(appli_dir, "bin", "salome", "appliskel")]
+def new_instance(running_instances):
+ from salome_instance import SalomeInstance
+ instance = SalomeInstance.start()
+ print("Instance created and now running on port", instance.get_port())
+ running_instances.put(instance)
+#
- # Configure session startup
- self.SALOME = imp.load_source("SALOME", os.path.join(appli_dir,"salome"))
- self.SALOME_appli_args = ["start", "-t"]
- self.SALOME_shell_args = ["shell"]
+class TestConcurrentLaunch(unittest.TestCase):
+ def __createInstances(self, nb):
+ running_instances = multiprocessing.Queue()
+ processes = [
+ multiprocessing.Process(target=new_instance, args=(running_instances,))
+ for i in range(nb)
+ ]
+ return running_instances, processes
#
- def tearDown(self):
- pass
+ def __terminateInstances(self, running_instances):
+ import time
+ timeout = time.time() + 60*10 # the test duration is about 50 s, we reasonably assume a max duration of 10mn
+
+ while not running_instances.empty() and time.time() < timeout:
+ instance = running_instances.get()
+ print("Terminate instance running on port", instance.get_port())
+ instance.stop()
#
+
def appli(self, args=None):
if args is None:
args = []
try:
- self.SALOME.main(self.SALOME_appli_args + args)
- except SystemExit, e:
+ sys.argv = ['runSalome', '-t']
+ import setenv
+ setenv.main(True, exeName="salome start")
+ import runSalome
+ runSalome.runSalome()
+ except SystemExit as e:
if str(e) != '0':
logging.error(e)
pass
if args is None:
args = []
try:
- self.SALOME.main(self.SALOME_shell_args + args)
- except SystemExit, e:
+ import setenv
+ setenv.main(True)
+ import runSession
+ params, args = runSession.configureSession(args, exe="salome shell")
+ return runSession.runSession(params, args)
+ except SystemExit as e:
if str(e) != '0':
logging.error(e)
pass
#
def test01_SingleSession(self):
- print "** Testing single session **"
+ print("** Testing single session **")
self.session(["hello.py"])
#
def test02_MultiSession(self):
- print "** Testing multi sessions **"
+ print("** Testing multi sessions **")
jobs = []
for i in range(9):
p = multiprocessing.Process(target=self.session, args=(["hello.py"],))
j.join()
#
def test03_SingleAppli(self):
- print "** Testing single appli **"
- current_directory = os.path.dirname(os.path.abspath(__file__))
- session_log = tempfile.NamedTemporaryFile(prefix='session_', suffix='.log')
- self.appli(["--ns-port-log=%s"%session_log.name])
- port_number = None
- with open(session_log.name, "r") as f:
- port_number = f.readline()
+ print("** Testing single appli **")
+ running_instances, processes = self.__createInstances(1)
+ for p in processes:
+ p.start()
+ pass
+ for p in processes:
+ p.join()
+ pass
+
self.session(["hello.py"])
- self.session(["-p", port_number, "killSalomeWithPort.py", "args:%s"%port_number])
- session_log.close()
+ self.__terminateInstances(running_instances)
#
def test04_MultiAppli(self):
- print "** Testing multi appli **"
- jobs = []
- for i in range(9):
- p = multiprocessing.Process(target=self.test03_SingleAppli)
- jobs.append(p)
+ print("** Testing multi appli **")
+ running_instances, processes = self.__createInstances(9)
+ for p in processes:
p.start()
+ pass
+ for p in processes:
+ p.join()
+ pass
- for j in jobs:
- j.join()
+ self.session(["hello.py"])
+ self.__terminateInstances(running_instances)
#
#
if __name__ == "__main__":
- path_to_launcher = os.getenv("SALOME_LAUNCHER")
- if not path_to_launcher:
- msg = "\n"
- msg += "Error: please set SALOME_LAUNCHER variable to the salome command in your application folder.\n"
- logging.error(msg)
- sys.exit(1)
-
- if not os.path.isfile("hello.py"):
- with open("hello.py", "w") as f:
- f.write("print 'Hello!'")
-
unittest.main()
#
