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