src/Launcher/Test/test_launcher.py

   1 #! /usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 import unittest
   5 import os
   6 import sys
   7 import time
   8
   9 # Test of SalomeLauncher.
  10 # This test should be run in the salome environment, using "salome shell"
  11 # and salome application should be running.
  12 # If YACS_ROOT_DIR is not set, the test of submitting a YACS schema will be
  13 # skiped.
  14 class TestCompo(unittest.TestCase):
  15   @classmethod
  16   def setUpClass(cls):
  17     # Prepare the test directory
  18     import shutil
  19     cls.test_dir = os.path.join(os.getcwd(), "test_dir")
  20     cls.suffix = time.strftime("-%Y-%m-%d-%H-%M-%S")
  21     shutil.rmtree(cls.test_dir, ignore_errors=True)
  22     os.mkdir(cls.test_dir)
  23
  24   ##############################
  25   # test of python_salome job
  26   ##############################
  27   def test_salome_py_job(self):
  28     case_test_dir = os.path.join(TestCompo.test_dir, "salome_py")
  29     os.mkdir(case_test_dir)
  30
  31     old_dir = os.getcwd()
  32     os.chdir(case_test_dir)
  33
  34     # job script
  35     script_file = "myScript.py"
  36     job_script_file = os.path.join(case_test_dir, script_file)
  37     script_text = """#! /usr/bin/env python
  38 # -*- coding: utf-8 -*-
  39
  40 # verify import salome
  41 import salome
  42 salome.salome_init()
  43
  44 f = open('result.txt', 'w')
  45 f.write("Salut!")
  46 f.close()
  47
  48 import os
  49 os.mkdir("subdir")
  50 f = open(os.path.join("subdir",'autre.txt'), 'w')
  51 f.write("Hello!")
  52 f.close()
  53 """
  54     f = open(job_script_file, "w")
  55     f.write(script_text)
  56     f.close()
  57
  58     local_result_dir = os.path.join(case_test_dir, "result_simple_py_job")
  59     job_params = salome.JobParameters()
  60     job_params.job_name = "MyJob"
  61     job_params.job_type = "python_salome"
  62     job_params.job_file = job_script_file
  63     job_params.in_files = []
  64     job_params.out_files = ["result.txt", "subdir"]
  65     job_params.result_directory = local_result_dir
  66     job_params.resource_required = salome.ResourceParameters()
  67     job_params.resource_required.nb_proc = 1
  68
  69     job_params.work_directory = "/tmp/job_salome_py" + self.suffix
  70
  71     launcher = salome.naming_service.Resolve('/SalomeLauncher')
  72     job_id = launcher.createJob(job_params)
  73
  74     launcher.launchJob(job_id)
  75
  76     import time
  77     jobState = launcher.getJobState(job_id)
  78     print "Job %d state: %s" % (job_id,jobState)
  79     while jobState != "FINISHED" and jobState != "FAILED" :
  80       time.sleep(5)
  81       jobState = launcher.getJobState(job_id)
  82       print "Job %d state: %s" % (job_id,jobState)
  83       pass
  84
  85     self.assertEqual(jobState, "FINISHED")
  86
  87     # getJobResults to default directory (result_directory)
  88     launcher.getJobResults(job_id, "")
  89     try:
  90       f = open(os.path.join(local_result_dir, "result.txt"), 'r')
  91       text = f.read()
  92       f.close()
  93       self.assertEqual(text, "Salut!")
  94     except IOError,ex:
  95       self.fail("IO exception:" + str(ex));
  96
  97     try:
  98       f = open(os.path.join(local_result_dir, "subdir", "autre.txt"), 'r')
  99       text = f.read()
 100       f.close()
 101       self.assertEqual(text, "Hello!")
 102     except IOError,ex:
 103       self.fail("IO exception:" + str(ex));
 104
 105     # getJobResults to a specific directory
 106     mydir = os.path.join(case_test_dir, "custom_result_dir")
 107     launcher.getJobResults(job_id, mydir)
 108     try:
 109       f = open(os.path.join(mydir, "result.txt"), 'r')
 110       text = f.read()
 111       f.close()
 112       self.assertEqual(text, "Salut!")
 113     except IOError,ex:
 114       self.fail("IO exception:" + str(ex));
 115
 116     try:
 117       f = open(os.path.join(mydir, "subdir", "autre.txt"), 'r')
 118       text = f.read()
 119       f.close()
 120       self.assertEqual(text, "Hello!")
 121     except IOError,ex:
 122       self.fail("IO exception:" + str(ex));
 123       pass
 124
 125     os.chdir(old_dir)
 126
 127   ##############################
 128   # test of command job type
 129   ##############################
 130   def test_command(self):
 131     case_test_dir = os.path.join(TestCompo.test_dir, "command")
 132     os.mkdir(case_test_dir)
 133
 134     # job script
 135     data_file = "in.txt"
 136     script_file = "myEnvScript.py"
 137     script_text = """#! /usr/bin/env python
 138 # -*- coding: utf-8 -*-
 139
 140 import os,sys
 141
 142 text_result = os.getenv("ENV_TEST_VAR","")
 143
 144 f = open('result.txt', 'w')
 145 f.write(text_result)
 146 f.close()
 147
 148 in_f = open("in.txt", "r")
 149 in_text = in_f.read()
 150 in_f.close()
 151
 152 os.mkdir("subdir")
 153 f = open(os.path.join("subdir",'altul.txt'), 'w')
 154 f.write(in_text)
 155 f.close()
 156 """
 157     abs_script_file = os.path.join(case_test_dir, script_file)
 158     f = open(abs_script_file, "w")
 159     f.write(script_text)
 160     f.close()
 161     os.chmod(abs_script_file, 0o755)
 162
 163     #environement script
 164     env_file = "myEnv.sh"
 165     env_text = """export ENV_TEST_VAR="expected"
 166 """
 167     f = open(os.path.join(case_test_dir, env_file), "w")
 168     f.write(env_text)
 169     f.close()
 170
 171     # write data file
 172     f = open(os.path.join(case_test_dir, data_file), "w")
 173     f.write("expected data")
 174     f.close()
 175
 176     # job params
 177     local_result_dir = os.path.join(case_test_dir, "resultats_env")
 178     job_params = salome.JobParameters()
 179     job_params.job_name = "CommandJob"
 180     job_params.job_type = "command"
 181     job_params.job_file = script_file
 182     job_params.env_file = env_file
 183     job_params.in_files = [data_file]
 184     job_params.out_files = ["result.txt", "subdir"]
 185     job_params.local_directory = case_test_dir
 186     job_params.result_directory = local_result_dir
 187     job_params.resource_required = salome.ResourceParameters()
 188     job_params.resource_required.nb_proc = 1
 189     job_params.work_directory = "/tmp/command_job" + self.suffix
 190
 191     # create and launch the job
 192     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 193     job_id = launcher.createJob(job_params)
 194     launcher.launchJob(job_id)
 195
 196     # wait for the end of the job
 197     import time
 198     jobState = launcher.getJobState(job_id)
 199     print "Job %d state: %s" % (job_id,jobState)
 200     while jobState != "FINISHED" and jobState != "FAILED" :
 201       time.sleep(3)
 202       jobState = launcher.getJobState(job_id)
 203       print "Job %d state: %s" % (job_id,jobState)
 204       pass
 205
 206     # verify the results
 207     self.assertEqual(jobState, "FINISHED")
 208     launcher.getJobResults(job_id, "")
 209     try:
 210       f = open(os.path.join(local_result_dir, "result.txt"), 'r')
 211       text = f.read()
 212       f.close()
 213       self.assertEqual(text, "expected")
 214     except IOError,ex:
 215       self.fail("IO exception:" + str(ex));
 216
 217   ##############################
 218   # test of yacs job type
 219   ##############################
 220   def test_yacs(self):
 221     yacs_path = os.getenv("YACS_ROOT_DIR", "")
 222     if not os.path.isdir(yacs_path):
 223       self.skipTest("Needs YACS module to run. Please define YACS_ROOT_DIR.")
 224
 225     case_test_dir = os.path.join(TestCompo.test_dir, "yacs")
 226     os.mkdir(case_test_dir)
 227
 228     #environement script
 229     env_file = "myEnv.sh"
 230     env_text = """export ENV_TEST_VAR="expected"
 231 """
 232     f = open(os.path.join(case_test_dir, env_file), "w")
 233     f.write(env_text)
 234     f.close()
 235
 236     # job script
 237     script_text = """<?xml version='1.0' encoding='iso-8859-1' ?>
 238 <proc name="newSchema_1">
 239    <property name="DefaultStudyID" value="1"/>
 240    <container name="DefaultContainer">
 241       <property name="container_kind" value="Salome"/>
 242       <property name="attached_on_cloning" value="0"/>
 243       <property name="container_name" value="FactoryServer"/>
 244       <property name="name" value="localhost"/>
 245    </container>
 246    <inline name="PyScript0">
 247       <script><code><![CDATA[import os
 248 text_result = os.getenv("ENV_TEST_VAR","")
 249 f = open('result.txt', 'w')
 250 f.write(text_result)
 251 f.close()
 252 ]]></code></script>
 253       <load container="DefaultContainer"/>
 254    </inline>
 255 </proc>
 256 """
 257     yacs_file = "mySchema.xml"
 258     job_script_file = os.path.join(case_test_dir, yacs_file)
 259     f = open(job_script_file, "w")
 260     f.write(script_text)
 261     f.close()
 262
 263     local_result_dir = os.path.join(case_test_dir, "result_yacs_job")
 264     job_params = salome.JobParameters()
 265     job_params.job_name = "MyYacsJob"
 266     job_params.job_type = "yacs_file"
 267     job_params.job_file = job_script_file
 268     job_params.env_file = os.path.join(case_test_dir,env_file)
 269     #job_params.in_files = [case_test_dir]
 270     job_params.out_files = ["result.txt"]
 271     job_params.result_directory = local_result_dir
 272
 273     # define the interval between two YACS schema dumps (3 seconds)
 274     import Engines
 275     job_params.specific_parameters = [Engines.Parameter("EnableDumpYACS", "3")]
 276
 277     job_params.resource_required = salome.ResourceParameters()
 278     job_params.resource_required.nb_proc = 1
 279
 280     job_params.work_directory = "/tmp/job_yacs" + self.suffix
 281
 282     launcher = salome.naming_service.Resolve('/SalomeLauncher')
 283     job_id = launcher.createJob(job_params)
 284
 285     launcher.launchJob(job_id)
 286
 287     import time
 288     jobState = launcher.getJobState(job_id)
 289     yacs_dump_success = False
 290     print "Job %d state: %s" % (job_id,jobState)
 291     while jobState != "FINISHED" and jobState != "FAILED" :
 292       time.sleep(5)
 293       jobState = launcher.getJobState(job_id)
 294       yacs_dump_success = launcher.getJobDumpState(job_id, local_result_dir)
 295       print "Job %d state: %s - dump: %s" % (job_id,jobState, yacs_dump_success)
 296       pass
 297
 298     self.assertEqual(jobState, "FINISHED")
 299
 300     # Verify dumpState file is in the results
 301     self.assertTrue(yacs_dump_success)
 302     dump_file_path = os.path.join(local_result_dir, "dumpState_mySchema.xml")
 303     self.assertTrue(os.path.isfile(dump_file_path))
 304
 305     # Load the schema state from the dump file and verify the state of a node
 306     import SALOMERuntime
 307     SALOMERuntime.RuntimeSALOME_setRuntime(1)
 308     import loader
 309     schema = loader.YACSLoader().load(job_script_file)
 310     stateParser = loader.stateParser()
 311     sl = loader.stateLoader(stateParser, schema)
 312     sl.parse(dump_file_path)
 313     # 106 : "DONE" state code
 314     self.assertEqual(106, schema.getChildByName("PyScript0").getEffectiveState())
 315
 316     # getJobResults to default directory (result_directory)
 317     launcher.getJobResults(job_id, "")
 318     try:
 319       f = open(os.path.join(local_result_dir, "result.txt"), 'r')
 320       text = f.read()
 321       f.close()
 322       self.assertEqual(text, "expected")
 323     except IOError,ex:
 324       self.fail("IO exception:" + str(ex))
 325
 326 if __name__ == '__main__':
 327     # creat study
 328     import salome
 329     salome.salome_init()
 330     unittest.main()
 331
 332     # load catalogs
 333 #    mc = salome.naming_service.Resolve('/Kernel/ModulCatalog')
 334 #    ior = salome.orb.object_to_string(mc)
 335 #    import SALOMERuntime
 336 #    SALOMERuntime.RuntimeSALOME_setRuntime()
 337 #    salome_runtime = SALOMERuntime.getSALOMERuntime()
 338 #    session_catalog = salome_runtime.loadCatalog("session", ior)
 339 #    salome_runtime.addCatalog(session_catalog)
 340