-#!/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
-# License as published by the Free Software Foundation; either
-# version 2.1 of the License, or (at your option) any later version.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public
-# License along with this library; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-#
-# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
-#
-
-import os
-import sys
-import multiprocessing
-import unittest
-import logging
-
-def port_reservation(obtained_ports, preferred=None, test=None, expected=None):
- from PortManager import getPort
- if preferred:
- port = getPort(preferred)
- else:
- port = getPort()
- print("obtained port = %s"%port)
-
- obtained_ports.put(port)
-
- if expected:
- test.assertTrue(port == expected, "used = %s, expected = %s"%(port, expected))
-#
-
-class TestMinimalExample(unittest.TestCase):
- def testSequential(self):
- from PortManager import releasePort, getBusyPorts
- print("\nBEGIN testSequential")
- print("Busy ports", getBusyPorts())
- obtained_ports = multiprocessing.Queue()
-
- processes = [
- multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
- for i in range(3)
- ]
-
- for p in processes:
- p.start()
-
- for p in processes:
- p.join()
-
- print("Busy ports", getBusyPorts())
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, self, 2872,))
- p.start()
- p.join()
-
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812, self,))
- p.start()
- p.join()
-
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, self, 2899,))
- p.start()
- p.join()
-
- # Release port
- print ("release port 2899")
- p = multiprocessing.Process(target=releasePort, args=(2899,))
- p.start()
- p.join()
-
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, self, 2899,))
- p.start()
- p.join()
-
- # Release ports
- print("Busy ports", getBusyPorts())
- while not obtained_ports.empty():
- port = obtained_ports.get()
- print("release port", port)
- p = multiprocessing.Process(target=releasePort, args=(port,))
- p.start()
- p.join()
-
- print("END testSequential")
- #
-
- def testConcurrent(self):
- from PortManager import releasePort, getBusyPorts
- print("\nBEGIN testConcurrent")
- print("Busy ports", getBusyPorts())
- obtained_ports = multiprocessing.Queue()
- processes = [
- multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
-
- for i in range(3)
- ]
-
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, self, 2872,))
- processes.append(p)
-
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
- processes.append(p)
-
- # Try to get specific port number
- p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
- processes.append(p)
-
- for p in processes:
- p.start()
-
- for p in processes:
- p.join()
-
- # Release ports
- print("Busy ports", getBusyPorts())
- while not obtained_ports.empty():
- port = obtained_ports.get()
- print("release port", port)
- p = multiprocessing.Process(target=releasePort, args=(port,))
- p.start()
- p.join()
-
- print("END testConcurrent")
- #
-#
-
-if __name__ == "__main__":
- omniorb_user_path = os.getenv("OMNIORB_USER_PATH")
- if not omniorb_user_path:
- msg = "\n"
- msg += "Error: please set OMNIORB_USER_PATH variable.\n"
- msg += " Usually this points to your application USERS directory.\n"
- logging.error(msg)
- sys.exit(1)
-
- try:
- import PortManager
- except ImportError:
- msg = "\n"
- msg += "Error: can't import PortManager; please check PYTHONPATH variable.\n"
- msg += " You need to add <KERNEL_INSTALL_PATH>/bin/salome path.\n"
- logging.error(msg)
- sys.exit(1)
-
- unittest.main()
-#
+#!/usr/bin/env python3\r
+# Copyright (C) 2013-2020 CEA/DEN, EDF R&D, OPEN CASCADE\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+import os\r
+import sys\r
+import multiprocessing\r
+import unittest\r
+import logging\r
+\r
+def port_reservation(obtained_ports, preferred=None, test=None, expected=None):\r
+ from PortManager import getPort\r
+ if preferred:\r
+ port = getPort(preferred)\r
+ else:\r
+ port = getPort()\r
+ print("obtained port = %s"%port)\r
+\r
+ obtained_ports.put(port)\r
+\r
+ if expected:\r
+ test.assertTrue(port == expected, "used = %s, expected = %s"%(port, expected))\r
+#\r
+\r
+class TestMinimalExample(unittest.TestCase):\r
+ def testSequential(self):\r
+ from PortManager import releasePort, getBusyPorts\r
+ print("\nBEGIN testSequential")\r
+ print("Busy ports", getBusyPorts())\r
+ obtained_ports = multiprocessing.Queue()\r
+\r
+ processes = [\r
+ multiprocessing.Process(target=port_reservation, args=(obtained_ports,))\r
+ for i in range(3)\r
+ ]\r
+\r
+ for p in processes:\r
+ p.start()\r
+\r
+ for p in processes:\r
+ p.join()\r
+\r
+ print("Busy ports", getBusyPorts())\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, self, 2872,))\r
+ p.start()\r
+ p.join()\r
+\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812, self,))\r
+ p.start()\r
+ p.join()\r
+\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, self, 2899,))\r
+ p.start()\r
+ p.join()\r
+\r
+ # Release port\r
+ print ("release port 2899")\r
+ p = multiprocessing.Process(target=releasePort, args=(2899,))\r
+ p.start()\r
+ p.join()\r
+\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, self, 2899,))\r
+ p.start()\r
+ p.join()\r
+\r
+ # Release ports\r
+ print("Busy ports", getBusyPorts())\r
+ while not obtained_ports.empty():\r
+ port = obtained_ports.get()\r
+ print("release port", port)\r
+ p = multiprocessing.Process(target=releasePort, args=(port,))\r
+ p.start()\r
+ p.join()\r
+\r
+ print("END testSequential")\r
+ #\r
+\r
+ def testConcurrent(self):\r
+ from PortManager import releasePort, getBusyPorts\r
+ print("\nBEGIN testConcurrent")\r
+ print("Busy ports", getBusyPorts())\r
+ obtained_ports = multiprocessing.Queue()\r
+ processes = [\r
+ multiprocessing.Process(target=port_reservation, args=(obtained_ports,))\r
+\r
+ for i in range(3)\r
+ ]\r
+\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, self, 2872,))\r
+ processes.append(p)\r
+\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))\r
+ processes.append(p)\r
+\r
+ # Try to get specific port number\r
+ p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))\r
+ processes.append(p)\r
+\r
+ for p in processes:\r
+ p.start()\r
+\r
+ for p in processes:\r
+ p.join()\r
+\r
+ # Release ports\r
+ print("Busy ports", getBusyPorts())\r
+ while not obtained_ports.empty():\r
+ port = obtained_ports.get()\r
+ print("release port", port)\r
+ p = multiprocessing.Process(target=releasePort, args=(port,))\r
+ p.start()\r
+ p.join()\r
+\r
+ print("END testConcurrent")\r
+ #\r
+#\r
+\r
+if __name__ == "__main__":\r
+ from salomeContextUtils import setOmniOrbUserPath\r
+ setOmniOrbUserPath()\r
+ omniorb_user_path = os.getenv("OMNIORB_USER_PATH")\r
+ if not omniorb_user_path:\r
+ msg = "\n"\r
+ msg += "Error: please set OMNIORB_USER_PATH variable.\n"\r
+ msg += " Usually this points to your application USERS directory.\n"\r
+ logging.error(msg)\r
+ sys.exit(1)\r
+\r
+ try:\r
+ import PortManager\r
+ except ImportError:\r
+ msg = "\n"\r
+ msg += "Error: can't import PortManager; please check PYTHONPATH variable.\n"\r
+ msg += " You need to add <KERNEL_INSTALL_PATH>/bin/salome path.\n"\r
+ logging.error(msg)\r
+ sys.exit(1)\r
+\r
+ unittest.main()\r
+#\r
-# Copyright (C) 2015-2020 CEA/DEN, EDF R&D
-#
-# This library is free software; you can redistribute it and/or
-# modify it under the terms of the GNU Lesser General Public
-# License as published by the Free Software Foundation; either
-# version 2.1 of the License, or (at your option) any later version.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public
-# License along with this library; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-#
-# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
-#
-
-SET(TEST_NAMES
- TestLauncherSimple
- TestLauncherSessionArgs
- )
-
-IF(WIN32)
- SET(PY_EXEC $ENV{PYTHONBIN})
-ENDIF()
-
-FOREACH(tfile ${TEST_NAMES})
- SET(TEST_NAME ${COMPONENT_NAME}_SALOME_COMMAND_${tfile})
- ADD_TEST(${TEST_NAME} ${PY_EXEC} ${tfile}.py)
- SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT})
-ENDFOREACH()
+# Copyright (C) 2015-2020 CEA/DEN, EDF R&D\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+SET(TEST_NAMES\r
+ TestLauncherSimple\r
+# TestLauncherSessionArgs\r
+ TestLauncherSessionArgs_1\r
+ TestLauncherSessionArgs_2\r
+ TestLauncherSessionArgs_3\r
+ )\r
+\r
+IF(WIN32)\r
+ SET(PY_EXEC $ENV{PYTHONBIN})\r
+ENDIF()\r
+\r
+FOREACH(tfile ${TEST_NAMES})\r
+ SET(TEST_NAME ${COMPONENT_NAME}_SALOME_COMMAND_${tfile})\r
+ ADD_TEST(${TEST_NAME} ${PY_EXEC} ${tfile}.py)\r
+ SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT})\r
+ENDFOREACH()\r
-#!/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
-# License as published by the Free Software Foundation; either
-# version 2.1 of the License, or (at your option) any later version.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public
-# License along with this library; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-#
-# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
-#
-
-import unittest
-
-import os
-import sys
-import imp
-from io import StringIO
-import logging
-
-logger = logging.getLogger("TestLauncherLogger")
-logger.level = logging.DEBUG
-logger.addHandler(logging.StreamHandler())
-
-class TestSessionArgs(unittest.TestCase):
- #
- @classmethod
- def setUpClass(cls):
- # Set some predefined command args and corresponding output messages
- cls.hello0 = ["hello.py", "args:outfile=LOGFILE"]
- cls.hello0Msg = "Hello!"
- cls.hello1 = ["hello.py", "args:you,outfile=LOGFILE"]
- cls.hello1Msg = "Hello to: you"
- cls.helloToAdd = ["hello.py", "args:add.py,1,2,3,outfile=LOGFILE"]
- cls.helloToAddMsg = "Hello to: add.py, 1, 2, 3"
- cls.helloToList = ["hello.py", "args:['file1','file2'],1,2,3,[True,False],'ok',outfile=LOGFILE"]
- cls.helloToListMsg = "Hello to: ['file1','file2'], 1, 2, 3, [True,False], 'ok'"
- cls.add0 = ["add.py", "args:outfile=LOGFILE"]
- cls.add0Msg = "No args!"
- cls.add3 = ["add.py", "args:1,2,3,outfile=LOGFILE"]
- cls.add3Msg = "1+2+3 = 6"
- cls.lines0 = ["lines.py", "args:outfile=LOGFILE"]
- cls.lines0Msg = "No files given"
- cls.lines2 = ["lines.py", "args:hello.py,add.py,outfile=LOGFILE"]
- cls.lines2Msg = "hello.py is 35 lines longadd.py is 37 lines long"
- cls.linesUnreadable = ["lines.py", "args:hello.py,add.py,1,2,outfile=LOGFILE"]
- cls.linesUnreadableMsg = "hello.py is 35 lines longadd.py is 37 lines longFile '1' cannot be readFile '2' cannot be read"
- sys.stdout = StringIO()
- #
- def setUp(self):
- import tempfile
- self.logFile = tempfile.NamedTemporaryFile()
- if sys.platform == "win32": # Close file because of permission denined on Windows
- self.logFile.close()
- #
- def tearDown(self):
- self.logFile.close()
- #
- def session(self, args=None):
- if args is None:
- args = []
- args = [x.replace("LOGFILE",self.logFile.name) for x in args]
- try:
- 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':
- logger.error(e)
- import traceback
- traceback.print_exc()
- pass
- #
- def assertLogFileContentsEqual(self, message):
- with open(self.logFile.name, "r") as f:
- contents = f.read().replace('\n', '')
-
- self.assertTrue(contents==message, "Contents differ!\n\tGenerated contents: %s\n\tExpected contents: %s"%(contents, message))
- #
- def testHello0(self):
- self.session(self.hello0)
- self.assertLogFileContentsEqual(self.hello0Msg)
- #
- def testPythonHello0(self):
- self.session(["python"]+self.hello0)
- self.assertLogFileContentsEqual(self.hello0Msg)
- #
- def testHello1(self):
- self.session(self.hello1)
- self.assertLogFileContentsEqual(self.hello1Msg)
- #
- def testAdd0(self):
- self.session(self.add0)
- self.assertLogFileContentsEqual(self.add0Msg)
- #
- def testAdd3(self):
- self.session(self.add3)
- self.assertLogFileContentsEqual(self.add3Msg)
- #
- def testHello0Add3(self):
- self.session(self.hello0+self.add3)
- self.assertLogFileContentsEqual(self.hello0Msg+self.add3Msg)
- #
- def testHello1Add3(self):
- self.session(self.hello1+self.add3)
- self.assertLogFileContentsEqual(self.hello1Msg+self.add3Msg)
- #
- def testHelloToAdd(self):
- self.session(self.helloToAdd)
- self.assertLogFileContentsEqual(self.helloToAddMsg)
- #
- def testHelloToList(self):
- self.session(self.helloToList)
- self.assertLogFileContentsEqual(self.helloToListMsg)
- #
- def testLines0(self):
- self.session(self.lines0)
- self.assertLogFileContentsEqual(self.lines0Msg)
- #
- def testLines2(self):
- self.session(self.lines2)
- self.assertLogFileContentsEqual(self.lines2Msg)
- #
- def testLines2Add3(self):
- self.session(self.lines2+self.add3)
- self.assertLogFileContentsEqual(self.lines2Msg+self.add3Msg)
- #
- def testLinesUnreadable(self):
- self.session(self.linesUnreadable)
- self.assertLogFileContentsEqual(self.linesUnreadableMsg)
- #
- def testAddAddHello(self):
- self.session(self.add3+self.add3+self.hello1)
- self.assertLogFileContentsEqual(self.add3Msg+self.add3Msg+self.hello1Msg)
- #
- def testHello0Add3Hello0Add3Hello0(self):
- self.session(self.hello1+self.add3+self.hello0+self.add3+self.hello0)
- self.assertLogFileContentsEqual(self.hello1Msg+self.add3Msg+self.hello0Msg+self.add3Msg+self.hello0Msg)
- #
-#
-
-if __name__ == "__main__":
- unittest.main()
-#
+#!/usr/bin/env python3\r
+# Copyright (C) 2013-2020 CEA/DEN, EDF R&D, OPEN CASCADE\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+import unittest\r
+\r
+import os\r
+import sys\r
+import imp\r
+from io import StringIO\r
+import logging\r
+\r
+logger = logging.getLogger("TestLauncherLogger")\r
+logger.level = logging.DEBUG\r
+logger.addHandler(logging.StreamHandler())\r
+\r
+class TestSessionArgs(unittest.TestCase):\r
+ #\r
+ @classmethod\r
+ def setUpClass(cls):\r
+ # Set some predefined command args and corresponding output messages\r
+ cls.hello0 = ["hello.py", "args:outfile=LOGFILE"]\r
+ cls.hello0Msg = "Hello!"\r
+ cls.hello1 = ["hello.py", "args:you,outfile=LOGFILE"]\r
+ cls.hello1Msg = "Hello to: you"\r
+ cls.helloToAdd = ["hello.py", "args:add.py,1,2,3,outfile=LOGFILE"]\r
+ cls.helloToAddMsg = "Hello to: add.py, 1, 2, 3"\r
+ cls.helloToList = ["hello.py", "args:['file1','file2'],1,2,3,[True,False],'ok',outfile=LOGFILE"]\r
+ cls.helloToListMsg = "Hello to: ['file1','file2'], 1, 2, 3, [True,False], 'ok'"\r
+ cls.add0 = ["add.py", "args:outfile=LOGFILE"]\r
+ cls.add0Msg = "No args!"\r
+ cls.add3 = ["add.py", "args:1,2,3,outfile=LOGFILE"]\r
+ cls.add3Msg = "1+2+3 = 6"\r
+ cls.lines0 = ["lines.py", "args:outfile=LOGFILE"]\r
+ cls.lines0Msg = "No files given"\r
+ cls.lines2 = ["lines.py", "args:hello.py,add.py,outfile=LOGFILE"]\r
+ cls.lines2Msg = "hello.py is 35 lines longadd.py is 37 lines long"\r
+ cls.linesUnreadable = ["lines.py", "args:hello.py,add.py,1,2,outfile=LOGFILE"]\r
+ cls.linesUnreadableMsg = "hello.py is 35 lines longadd.py is 37 lines longFile '1' cannot be readFile '2' cannot be read"\r
+ sys.stdout = StringIO()\r
+ #\r
+ def setUp(self):\r
+ import tempfile\r
+ from salomeContextUtils import setOmniOrbUserPath\r
+ setOmniOrbUserPath()\r
+ self.logFile = tempfile.NamedTemporaryFile()\r
+ if sys.platform == "win32": # Close file because of permission denined on Windows\r
+ self.logFile.close()\r
+ #\r
+ def tearDown(self):\r
+ self.logFile.close()\r
+ #\r
+ def session(self, args=None):\r
+ if args is None:\r
+ args = []\r
+ args = [x.replace("LOGFILE",self.logFile.name) for x in args]\r
+ try:\r
+ import setenv\r
+ setenv.main(True)\r
+ import runSession\r
+ params, args = runSession.configureSession(args, exe="salome shell")\r
+ return runSession.runSession(params, args)\r
+ except SystemExit as e:\r
+ if str(e) != '0':\r
+ logger.error(e)\r
+ import traceback\r
+ traceback.print_exc()\r
+ pass\r
+ #\r
+ def assertLogFileContentsEqual(self, message):\r
+ with open(self.logFile.name, "r") as f:\r
+ contents = f.read().replace('\n', '')\r
+\r
+ self.assertTrue(contents==message, "Contents differ!\n\tGenerated contents: %s\n\tExpected contents: %s"%(contents, message))\r
+ #\r
+ def testHello0(self):\r
+ self.session(self.hello0)\r
+ self.assertLogFileContentsEqual(self.hello0Msg)\r
+ #\r
+ def testPythonHello0(self):\r
+ self.session(["python"]+self.hello0)\r
+ self.assertLogFileContentsEqual(self.hello0Msg)\r
+ #\r
+ def testHello1(self):\r
+ self.session(self.hello1)\r
+ self.assertLogFileContentsEqual(self.hello1Msg)\r
+ #\r
+ def testAdd0(self):\r
+ self.session(self.add0)\r
+ self.assertLogFileContentsEqual(self.add0Msg)\r
+ #\r
+ def testAdd3(self):\r
+ self.session(self.add3)\r
+ self.assertLogFileContentsEqual(self.add3Msg)\r
+ #\r
+ def testHello0Add3(self):\r
+ self.session(self.hello0+self.add3)\r
+ self.assertLogFileContentsEqual(self.hello0Msg+self.add3Msg)\r
+ #\r
+ def testHello1Add3(self):\r
+ self.session(self.hello1+self.add3)\r
+ self.assertLogFileContentsEqual(self.hello1Msg+self.add3Msg)\r
+ #\r
+ def testHelloToAdd(self):\r
+ self.session(self.helloToAdd)\r
+ self.assertLogFileContentsEqual(self.helloToAddMsg)\r
+ #\r
+ def testHelloToList(self):\r
+ self.session(self.helloToList)\r
+ self.assertLogFileContentsEqual(self.helloToListMsg)\r
+ #\r
+ def testLines0(self):\r
+ self.session(self.lines0)\r
+ self.assertLogFileContentsEqual(self.lines0Msg)\r
+ #\r
+ def testLines2(self):\r
+ self.session(self.lines2)\r
+ self.assertLogFileContentsEqual(self.lines2Msg)\r
+ #\r
+ def testLines2Add3(self):\r
+ self.session(self.lines2+self.add3)\r
+ self.assertLogFileContentsEqual(self.lines2Msg+self.add3Msg)\r
+ #\r
+ def testLinesUnreadable(self):\r
+ self.session(self.linesUnreadable)\r
+ self.assertLogFileContentsEqual(self.linesUnreadableMsg)\r
+ #\r
+ def testAddAddHello(self):\r
+ self.session(self.add3+self.add3+self.hello1)\r
+ self.assertLogFileContentsEqual(self.add3Msg+self.add3Msg+self.hello1Msg)\r
+ #\r
+ def testHello0Add3Hello0Add3Hello0(self):\r
+ self.session(self.hello1+self.add3+self.hello0+self.add3+self.hello0)\r
+ self.assertLogFileContentsEqual(self.hello1Msg+self.add3Msg+self.hello0Msg+self.add3Msg+self.hello0Msg)\r
+ #\r
+#\r
+\r
+if __name__ == "__main__":\r
+ unittest.main()\r
+#\r
--- /dev/null
+#!/usr/bin/env python3\r
+# Copyright (C) 2013-2020 CEA/DEN, EDF R&D, OPEN CASCADE\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+import unittest\r
+\r
+import os\r
+import sys\r
+import imp\r
+from io import StringIO\r
+import logging\r
+\r
+logger = logging.getLogger("TestLauncherLogger")\r
+logger.level = logging.DEBUG\r
+logger.addHandler(logging.StreamHandler())\r
+\r
+class TestSessionArgs(unittest.TestCase):\r
+ #\r
+ @classmethod\r
+ def setUpClass(cls):\r
+ # Set some predefined command args and corresponding output messages\r
+ cls.hello0 = ["hello.py", "args:outfile=LOGFILE"]\r
+ cls.hello0Msg = "Hello!"\r
+ cls.hello1 = ["hello.py", "args:you,outfile=LOGFILE"]\r
+ cls.hello1Msg = "Hello to: you"\r
+ cls.helloToAdd = ["hello.py", "args:add.py,1,2,3,outfile=LOGFILE"]\r
+ cls.helloToAddMsg = "Hello to: add.py, 1, 2, 3"\r
+ cls.helloToList = ["hello.py", "args:['file1','file2'],1,2,3,[True,False],'ok',outfile=LOGFILE"]\r
+ cls.helloToListMsg = "Hello to: ['file1','file2'], 1, 2, 3, [True,False], 'ok'"\r
+ cls.add0 = ["add.py", "args:outfile=LOGFILE"]\r
+ cls.add0Msg = "No args!"\r
+ cls.add3 = ["add.py", "args:1,2,3,outfile=LOGFILE"]\r
+ cls.add3Msg = "1+2+3 = 6"\r
+ cls.lines0 = ["lines.py", "args:outfile=LOGFILE"]\r
+ cls.lines0Msg = "No files given"\r
+ cls.lines2 = ["lines.py", "args:hello.py,add.py,outfile=LOGFILE"]\r
+ cls.lines2Msg = "hello.py is 35 lines longadd.py is 37 lines long"\r
+ cls.linesUnreadable = ["lines.py", "args:hello.py,add.py,1,2,outfile=LOGFILE"]\r
+ cls.linesUnreadableMsg = "hello.py is 35 lines longadd.py is 37 lines longFile '1' cannot be readFile '2' cannot be read"\r
+ sys.stdout = StringIO()\r
+ #\r
+ def setUp(self):\r
+ import tempfile\r
+ from salomeContextUtils import setOmniOrbUserPath\r
+ setOmniOrbUserPath()\r
+ self.logFile = tempfile.NamedTemporaryFile()\r
+ if sys.platform == "win32": # Close file because of permission denined on Windows\r
+ self.logFile.close()\r
+ #\r
+ def tearDown(self):\r
+ self.logFile.close()\r
+ #\r
+ def session(self, args=None):\r
+ if args is None:\r
+ args = []\r
+ args = [x.replace("LOGFILE",self.logFile.name) for x in args]\r
+ try:\r
+ import setenv\r
+ setenv.main(True)\r
+ import runSession\r
+ params, args = runSession.configureSession(args, exe="salome shell")\r
+ return runSession.runSession(params, args)\r
+ except SystemExit as e:\r
+ if str(e) != '0':\r
+ logger.error(e)\r
+ import traceback\r
+ traceback.print_exc()\r
+ pass\r
+ #\r
+ def assertLogFileContentsEqual(self, message):\r
+ with open(self.logFile.name, "r") as f:\r
+ contents = f.read().replace('\n', '')\r
+\r
+ self.assertTrue(contents==message, "Contents differ!\n\tGenerated contents: %s\n\tExpected contents: %s"%(contents, message))\r
+ #\r
+ def testHello0(self):\r
+ self.session(self.hello0)\r
+ self.assertLogFileContentsEqual(self.hello0Msg)\r
+ #\r
+ def testPythonHello0(self):\r
+ self.session(["python"]+self.hello0)\r
+ self.assertLogFileContentsEqual(self.hello0Msg)\r
+ #\r
+ def testHello1(self):\r
+ self.session(self.hello1)\r
+ self.assertLogFileContentsEqual(self.hello1Msg)\r
+ #\r
+ def testAdd0(self):\r
+ self.session(self.add0)\r
+ self.assertLogFileContentsEqual(self.add0Msg)\r
+ #\r
+ def testAdd3(self):\r
+ self.session(self.add3)\r
+ self.assertLogFileContentsEqual(self.add3Msg)\r
+ #\r
+ def testHello0Add3(self):\r
+ self.session(self.hello0+self.add3)\r
+ self.assertLogFileContentsEqual(self.hello0Msg+self.add3Msg)\r
+ #\r
+ def testHello1Add3(self):\r
+ self.session(self.hello1+self.add3)\r
+ self.assertLogFileContentsEqual(self.hello1Msg+self.add3Msg)\r
+ #\r
+ def testHelloToAdd(self):\r
+ self.session(self.helloToAdd)\r
+ self.assertLogFileContentsEqual(self.helloToAddMsg)\r
+ #\r
+#\r
+\r
+if __name__ == "__main__":\r
+ unittest.main()\r
+#\r
--- /dev/null
+#!/usr/bin/env python3\r
+# Copyright (C) 2013-2020 CEA/DEN, EDF R&D, OPEN CASCADE\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+import unittest\r
+\r
+import os\r
+import sys\r
+import imp\r
+from io import StringIO\r
+import logging\r
+\r
+logger = logging.getLogger("TestLauncherLogger")\r
+logger.level = logging.DEBUG\r
+logger.addHandler(logging.StreamHandler())\r
+\r
+class TestSessionArgs(unittest.TestCase):\r
+ #\r
+ @classmethod\r
+ def setUpClass(cls):\r
+ # Set some predefined command args and corresponding output messages\r
+ cls.hello0 = ["hello.py", "args:outfile=LOGFILE"]\r
+ cls.hello0Msg = "Hello!"\r
+ cls.hello1 = ["hello.py", "args:you,outfile=LOGFILE"]\r
+ cls.hello1Msg = "Hello to: you"\r
+ cls.helloToAdd = ["hello.py", "args:add.py,1,2,3,outfile=LOGFILE"]\r
+ cls.helloToAddMsg = "Hello to: add.py, 1, 2, 3"\r
+ cls.helloToList = ["hello.py", "args:['file1','file2'],1,2,3,[True,False],'ok',outfile=LOGFILE"]\r
+ cls.helloToListMsg = "Hello to: ['file1','file2'], 1, 2, 3, [True,False], 'ok'"\r
+ cls.add0 = ["add.py", "args:outfile=LOGFILE"]\r
+ cls.add0Msg = "No args!"\r
+ cls.add3 = ["add.py", "args:1,2,3,outfile=LOGFILE"]\r
+ cls.add3Msg = "1+2+3 = 6"\r
+ cls.lines0 = ["lines.py", "args:outfile=LOGFILE"]\r
+ cls.lines0Msg = "No files given"\r
+ cls.lines2 = ["lines.py", "args:hello.py,add.py,outfile=LOGFILE"]\r
+ cls.lines2Msg = "hello.py is 35 lines longadd.py is 37 lines long"\r
+ cls.linesUnreadable = ["lines.py", "args:hello.py,add.py,1,2,outfile=LOGFILE"]\r
+ cls.linesUnreadableMsg = "hello.py is 35 lines longadd.py is 37 lines longFile '1' cannot be readFile '2' cannot be read"\r
+ sys.stdout = StringIO()\r
+ #\r
+ def setUp(self):\r
+ import tempfile\r
+ from salomeContextUtils import setOmniOrbUserPath\r
+ setOmniOrbUserPath()\r
+ self.logFile = tempfile.NamedTemporaryFile()\r
+ if sys.platform == "win32": # Close file because of permission denined on Windows\r
+ self.logFile.close()\r
+ #\r
+ def tearDown(self):\r
+ self.logFile.close()\r
+ #\r
+ def session(self, args=None):\r
+ if args is None:\r
+ args = []\r
+ args = [x.replace("LOGFILE",self.logFile.name) for x in args]\r
+ try:\r
+ import setenv\r
+ setenv.main(True)\r
+ import runSession\r
+ params, args = runSession.configureSession(args, exe="salome shell")\r
+ return runSession.runSession(params, args)\r
+ except SystemExit as e:\r
+ if str(e) != '0':\r
+ logger.error(e)\r
+ import traceback\r
+ traceback.print_exc()\r
+ pass\r
+ #\r
+ def assertLogFileContentsEqual(self, message):\r
+ with open(self.logFile.name, "r") as f:\r
+ contents = f.read().replace('\n', '')\r
+\r
+ self.assertTrue(contents==message, "Contents differ!\n\tGenerated contents: %s\n\tExpected contents: %s"%(contents, message))\r
+ #\r
+ def testHelloToList(self):\r
+ self.session(self.helloToList)\r
+ self.assertLogFileContentsEqual(self.helloToListMsg)\r
+ #\r
+ def testLines0(self):\r
+ self.session(self.lines0)\r
+ self.assertLogFileContentsEqual(self.lines0Msg)\r
+ #\r
+ def testLines2(self):\r
+ self.session(self.lines2)\r
+ self.assertLogFileContentsEqual(self.lines2Msg)\r
+ #\r
+ def testLines2Add3(self):\r
+ self.session(self.lines2+self.add3)\r
+ self.assertLogFileContentsEqual(self.lines2Msg+self.add3Msg)\r
+ #\r
+#\r
+\r
+if __name__ == "__main__":\r
+ unittest.main()\r
+#\r
--- /dev/null
+#!/usr/bin/env python3\r
+# Copyright (C) 2013-2020 CEA/DEN, EDF R&D, OPEN CASCADE\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+import unittest\r
+\r
+import os\r
+import sys\r
+import imp\r
+from io import StringIO\r
+import logging\r
+\r
+logger = logging.getLogger("TestLauncherLogger")\r
+logger.level = logging.DEBUG\r
+logger.addHandler(logging.StreamHandler())\r
+\r
+class TestSessionArgs(unittest.TestCase):\r
+ #\r
+ @classmethod\r
+ def setUpClass(cls):\r
+ # Set some predefined command args and corresponding output messages\r
+ cls.hello0 = ["hello.py", "args:outfile=LOGFILE"]\r
+ cls.hello0Msg = "Hello!"\r
+ cls.hello1 = ["hello.py", "args:you,outfile=LOGFILE"]\r
+ cls.hello1Msg = "Hello to: you"\r
+ cls.helloToAdd = ["hello.py", "args:add.py,1,2,3,outfile=LOGFILE"]\r
+ cls.helloToAddMsg = "Hello to: add.py, 1, 2, 3"\r
+ cls.helloToList = ["hello.py", "args:['file1','file2'],1,2,3,[True,False],'ok',outfile=LOGFILE"]\r
+ cls.helloToListMsg = "Hello to: ['file1','file2'], 1, 2, 3, [True,False], 'ok'"\r
+ cls.add0 = ["add.py", "args:outfile=LOGFILE"]\r
+ cls.add0Msg = "No args!"\r
+ cls.add3 = ["add.py", "args:1,2,3,outfile=LOGFILE"]\r
+ cls.add3Msg = "1+2+3 = 6"\r
+ cls.lines0 = ["lines.py", "args:outfile=LOGFILE"]\r
+ cls.lines0Msg = "No files given"\r
+ cls.lines2 = ["lines.py", "args:hello.py,add.py,outfile=LOGFILE"]\r
+ cls.lines2Msg = "hello.py is 35 lines longadd.py is 37 lines long"\r
+ cls.linesUnreadable = ["lines.py", "args:hello.py,add.py,1,2,outfile=LOGFILE"]\r
+ cls.linesUnreadableMsg = "hello.py is 35 lines longadd.py is 37 lines longFile '1' cannot be readFile '2' cannot be read"\r
+ sys.stdout = StringIO()\r
+ #\r
+ def setUp(self):\r
+ import tempfile\r
+ from salomeContextUtils import setOmniOrbUserPath\r
+ setOmniOrbUserPath()\r
+ self.logFile = tempfile.NamedTemporaryFile()\r
+ if sys.platform == "win32": # Close file because of permission denined on Windows\r
+ self.logFile.close()\r
+ #\r
+ def tearDown(self):\r
+ self.logFile.close()\r
+ #\r
+ def session(self, args=None):\r
+ if args is None:\r
+ args = []\r
+ args = [x.replace("LOGFILE",self.logFile.name) for x in args]\r
+ try:\r
+ import setenv\r
+ setenv.main(True)\r
+ import runSession\r
+ params, args = runSession.configureSession(args, exe="salome shell")\r
+ return runSession.runSession(params, args)\r
+ except SystemExit as e:\r
+ if str(e) != '0':\r
+ logger.error(e)\r
+ import traceback\r
+ traceback.print_exc()\r
+ pass\r
+ #\r
+ def assertLogFileContentsEqual(self, message):\r
+ with open(self.logFile.name, "r") as f:\r
+ contents = f.read().replace('\n', '')\r
+\r
+ self.assertTrue(contents==message, "Contents differ!\n\tGenerated contents: %s\n\tExpected contents: %s"%(contents, message))\r
+ #\r
+ def testLinesUnreadable(self):\r
+ self.session(self.linesUnreadable)\r
+ self.assertLogFileContentsEqual(self.linesUnreadableMsg)\r
+ #\r
+ def testAddAddHello(self):\r
+ self.session(self.add3+self.add3+self.hello1)\r
+ self.assertLogFileContentsEqual(self.add3Msg+self.add3Msg+self.hello1Msg)\r
+ #\r
+ def testHello0Add3Hello0Add3Hello0(self):\r
+ self.session(self.hello1+self.add3+self.hello0+self.add3+self.hello0)\r
+ self.assertLogFileContentsEqual(self.hello1Msg+self.add3Msg+self.hello0Msg+self.add3Msg+self.hello0Msg)\r
+ #\r
+#\r
+\r
+if __name__ == "__main__":\r
+ unittest.main()\r
+#\r
-# Copyright (C) 2017-2020 CEA/DEN, EDF R&D
-#
-# This library is free software; you can redistribute it and/or
-# modify it under the terms of the GNU Lesser General Public
-# License as published by the Free Software Foundation; either
-# version 2.1 of the License, or (at your option) any later version.
-#
-# This library is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public
-# License along with this library; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-#
-# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
-#
-
-SET(tname salome_test)
-
-SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_bash)
-ADD_TEST(${TEST_NAME} bash ${tname}.sh)
-SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)
-
-SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_shell)
-ADD_TEST(${TEST_NAME} bash ${tname}_in_shell.sh)
-SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)
-
-SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_py)
-ADD_TEST(${TEST_NAME} python ${tname}.py)
-SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)
-
-SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_driver)
-ADD_TEST(${TEST_NAME} python ${SALOME_TEST_DRIVER} ${TIMEOUT} ${tname}.py)
-SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)
+# Copyright (C) 2017-2020 CEA/DEN, EDF R&D\r
+#\r
+# This library is free software; you can redistribute it and/or\r
+# modify it under the terms of the GNU Lesser General Public\r
+# License as published by the Free Software Foundation; either\r
+# version 2.1 of the License, or (at your option) any later version.\r
+#\r
+# This library is distributed in the hope that it will be useful,\r
+# but WITHOUT ANY WARRANTY; without even the implied warranty of\r
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU\r
+# Lesser General Public License for more details.\r
+#\r
+# You should have received a copy of the GNU Lesser General Public\r
+# License along with this library; if not, write to the Free Software\r
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA\r
+#\r
+# See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com\r
+#\r
+\r
+SET(tname salome_test)\r
+\r
+IF (WIN32)\r
+ SET(PY_EXEC $ENV{PYTHONBIN})\r
+ENDIF()\r
+\r
+IF (NOT WIN32)\r
+ SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_bash)\r
+ ADD_TEST(${TEST_NAME} bash ${tname}.sh)\r
+ SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)\r
+\r
+ SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_shell)\r
+ ADD_TEST(${TEST_NAME} bash ${tname}_in_shell.sh)\r
+ SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)\r
+ENDIF()\r
+\r
+SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_py)\r
+ADD_TEST(${TEST_NAME} ${PY_EXEC} ${tname}.py)\r
+SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)\r
+\r
+SET(TEST_NAME ${COMPONENT_NAME}_SALOME_TEST_${tname}_driver)\r
+ADD_TEST(${TEST_NAME} ${PY_EXEC} ${SALOME_TEST_DRIVER} ${TIMEOUT} ${tname}.py)\r
+SET_TESTS_PROPERTIES(${TEST_NAME} PROPERTIES LABELS "${COMPONENT_NAME}" TIMEOUT ${TIMEOUT} WILL_FAIL ON)\r