2 # -*- coding: utf-8 -*-
14 except OSError as exc: # Python >2.5
15 if exc.errno == errno.EEXIST and os.path.isdir(path):
20 # Test of SalomeLauncher.
21 # This test should be run in the salome environment, using "salome shell"
22 # and a salome application should be running.
23 # The test will try to launch batch jobs on every available resources which
24 # have the can_launch_batch_jobs parameter set to True.
25 # You can use the environment variable USER_CATALOG_RESOURCES_FILE in order to
26 # define a customised resource catalog.
27 # If YACS_ROOT_DIR is not set, the test of submitting a YACS schema will be
29 class TestCompo(unittest.TestCase):
32 # Prepare the test directory
33 temp = tempfile.NamedTemporaryFile()
34 cls.test_dir = os.path.join(temp.name, "test_dir")
35 name = os.path.basename(temp.name)
37 cls.suffix = time.strftime("-%Y-%m-%d-%H-%M-%S")+"-%s"%(os.getpid())
41 # mc = salome.naming_service.Resolve('/Kernel/ModulCatalog')
42 # ior = salome.orb.object_to_string(mc)
43 # import SALOMERuntime
44 # SALOMERuntime.RuntimeSALOME_setRuntime()
45 # salome_runtime = SALOMERuntime.getSALOMERuntime()
46 # session_catalog = salome_runtime.loadCatalog("session", ior)
47 # salome_runtime.addCatalog(session_catalog)
49 # Get the list of possible ressources
50 ressource_param = salome.ResourceParameters()
51 ressource_param.can_launch_batch_jobs = True
52 rm = salome.lcc.getResourcesManager()
53 cls.ressources = rm.GetFittingResources(ressource_param)
55 def verifyFile(self, path, content):
60 self.assertEqual(text, content)
62 self.fail("IO exception:" + str(ex));
64 ##############################
65 # test of python_salome job
66 ##############################
67 def test_salome_py_job(self):
68 case_test_dir = os.path.join(TestCompo.test_dir, "salome_py")
69 mkdir_p(case_test_dir)
72 os.chdir(case_test_dir)
75 script_file = "myScript.py"
76 job_script_file = os.path.join(case_test_dir, script_file)
77 script_text = """#! /usr/bin/env python
78 # -*- coding: utf-8 -*-
80 # verify import salome
84 f = open('result.txt', 'w')
90 f = open(os.path.join("subdir",'autre.txt'), 'w')
94 f = open(job_script_file, "w")
98 local_result_dir = os.path.join(case_test_dir, "result_py_job-")
99 job_params = salome.JobParameters()
100 job_params.job_type = "python_salome"
101 job_params.job_file = job_script_file
102 job_params.in_files = []
103 job_params.out_files = ["result.txt", "subdir"]
104 job_params.resource_required = salome.ResourceParameters()
105 job_params.resource_required.nb_proc = 1
107 launcher = salome.naming_service.Resolve('/SalomeLauncher')
109 for resource in self.ressources:
110 print("Testing python_salome job on ", resource)
111 job_params.result_directory = local_result_dir + resource
112 job_params.job_name = "PyJob" + resource
113 job_params.resource_required.name = resource
114 # use default working directory for this test
116 job_id = launcher.createJob(job_params)
117 launcher.launchJob(job_id)
119 jobState = launcher.getJobState(job_id)
120 print("Job %d state: %s" % (job_id,jobState))
121 while jobState != "FINISHED" and jobState != "FAILED" :
123 jobState = launcher.getJobState(job_id)
124 print("Job %d state: %s" % (job_id,jobState))
127 self.assertEqual(jobState, "FINISHED")
129 # getJobResults to default directory (result_directory)
130 launcher.getJobResults(job_id, "")
131 self.verifyFile(os.path.join(job_params.result_directory, "result.txt"),
133 self.verifyFile(os.path.join(job_params.result_directory,
134 "subdir", "autre.txt"),
137 # getJobResults to a specific directory
138 mydir = os.path.join(case_test_dir, "custom_result_dir" + resource)
139 launcher.getJobResults(job_id, mydir)
140 self.verifyFile(os.path.join(mydir, "result.txt"), "Salut!")
141 self.verifyFile(os.path.join(mydir, "subdir", "autre.txt"), "Hello!")
146 ##############################
147 # test of command job type
148 ##############################
149 def test_command(self):
150 case_test_dir = os.path.join(TestCompo.test_dir, "command")
151 mkdir_p(case_test_dir)
155 script_file = "myEnvScript.py"
156 script_text = """#! /usr/bin/env python
157 # -*- coding: utf-8 -*-
161 text_result = os.getenv("ENV_TEST_VAR","")
163 f = open('result.txt', 'w')
167 in_f = open("in.txt", "r")
168 in_text = in_f.read()
172 f = open(os.path.join("copie",'copie.txt'), 'w')
176 abs_script_file = os.path.join(case_test_dir, script_file)
177 f = open(abs_script_file, "w")
180 os.chmod(abs_script_file, 0o755)
183 env_file = "myEnv.sh"
184 env_text = """export ENV_TEST_VAR="expected"
186 f = open(os.path.join(case_test_dir, env_file), "w")
191 f = open(os.path.join(case_test_dir, data_file), "w")
192 f.write("to be copied")
196 local_result_dir = os.path.join(case_test_dir, "result_com_job-")
197 job_params = salome.JobParameters()
198 job_params.job_type = "command"
199 job_params.job_file = script_file
200 job_params.env_file = env_file
201 job_params.in_files = [data_file]
202 job_params.out_files = ["result.txt", "copie"]
203 job_params.local_directory = case_test_dir
204 job_params.resource_required = salome.ResourceParameters()
205 job_params.resource_required.nb_proc = 1
207 # create and launch the job
208 launcher = salome.naming_service.Resolve('/SalomeLauncher')
209 resManager= salome.lcc.getResourcesManager()
211 for resource in self.ressources:
212 print("Testing command job on ", resource)
213 job_params.result_directory = local_result_dir + resource
214 job_params.job_name = "CommandJob_" + resource
215 job_params.resource_required.name = resource
217 # use the working directory of the resource
218 resParams = resManager.GetResourceDefinition(resource)
219 wd = os.path.join(resParams.working_directory,
220 "CommandJob" + self.suffix)
221 job_params.work_directory = wd
223 job_id = launcher.createJob(job_params)
224 launcher.launchJob(job_id)
225 # wait for the end of the job
226 jobState = launcher.getJobState(job_id)
227 print("Job %d state: %s" % (job_id,jobState))
228 while jobState != "FINISHED" and jobState != "FAILED" :
230 jobState = launcher.getJobState(job_id)
231 print("Job %d state: %s" % (job_id,jobState))
235 self.assertEqual(jobState, "FINISHED")
236 launcher.getJobResults(job_id, "")
237 self.verifyFile(os.path.join(job_params.result_directory, "result.txt"),
239 self.verifyFile(os.path.join(job_params.result_directory,
240 "copie",'copie.txt'),
243 # verify getJobWorkFile
244 mydir = os.path.join(case_test_dir, "work_dir" + resource)
245 success = launcher.getJobWorkFile(job_id, "result.txt", mydir)
246 self.assertEqual(success, True)
247 self.verifyFile(os.path.join(mydir, "result.txt"), "expected")
249 success = launcher.getJobWorkFile(job_id, "copie", mydir)
250 self.assertEqual(success, True)
251 self.verifyFile(os.path.join(mydir, "copie", "copie.txt"),
255 ##############################
256 # test of yacs job type
257 ##############################
259 yacs_path = os.getenv("YACS_ROOT_DIR", "")
260 if not os.path.isdir(yacs_path):
261 self.skipTest("Needs YACS module to run. Please define YACS_ROOT_DIR.")
263 case_test_dir = os.path.join(TestCompo.test_dir, "yacs")
264 mkdir_p(case_test_dir)
267 env_file = "myEnv.sh"
268 env_text = """export ENV_TEST_VAR="expected"
270 f = open(os.path.join(case_test_dir, env_file), "w")
275 script_text = """<?xml version='1.0' encoding='iso-8859-1' ?>
276 <proc name="newSchema_1">
277 <container name="DefaultContainer">
278 <property name="container_kind" value="Salome"/>
279 <property name="attached_on_cloning" value="0"/>
280 <property name="container_name" value="FactoryServer"/>
281 <property name="name" value="localhost"/>
283 <inline name="PyScript0">
284 <script><code><![CDATA[import os
285 text_result = os.getenv("ENV_TEST_VAR","")
286 f = open('result.txt', 'w')
290 <load container="DefaultContainer"/>
294 yacs_file = "mySchema.xml"
295 job_script_file = os.path.join(case_test_dir, yacs_file)
296 f = open(job_script_file, "w")
300 local_result_dir = os.path.join(case_test_dir, "result_yacs_job-")
301 job_params = salome.JobParameters()
302 job_params.job_type = "yacs_file"
303 job_params.job_file = job_script_file
304 job_params.env_file = os.path.join(case_test_dir,env_file)
305 job_params.out_files = ["result.txt"]
307 # define the interval between two YACS schema dumps (3 seconds)
309 job_params.specific_parameters = [Engines.Parameter("EnableDumpYACS", "3")]
310 job_params.resource_required = salome.ResourceParameters()
311 job_params.resource_required.nb_proc = 1
313 launcher = salome.naming_service.Resolve('/SalomeLauncher')
314 resManager= salome.lcc.getResourcesManager()
316 for resource in self.ressources:
317 print("Testing yacs job on ", resource)
318 job_params.result_directory = local_result_dir + resource
319 job_params.job_name = "YacsJob_" + resource
320 job_params.resource_required.name = resource
322 # use the working directory of the resource
323 resParams = resManager.GetResourceDefinition(resource)
324 wd = os.path.join(resParams.working_directory,
325 "YacsJob" + self.suffix)
326 job_params.work_directory = wd
328 job_id = launcher.createJob(job_params)
329 launcher.launchJob(job_id)
330 jobState = launcher.getJobState(job_id)
332 yacs_dump_success = False
333 print("Job %d state: %s" % (job_id,jobState))
334 while jobState != "FINISHED" and jobState != "FAILED" :
336 jobState = launcher.getJobState(job_id)
337 # yacs_dump_success = launcher.getJobWorkFile(job_id, "dumpState_mySchema.xml",
338 yacs_dump_success = launcher.getJobDumpState(job_id,
339 job_params.result_directory)
340 print("Job %d state: %s - dump: %s" % (job_id,jobState, yacs_dump_success))
343 self.assertEqual(jobState, "FINISHED")
345 # Verify dumpState file is in the results
346 self.assertTrue(yacs_dump_success)
347 dump_file_path = os.path.join(job_params.result_directory,
348 "dumpState_mySchema.xml")
349 self.assertTrue(os.path.isfile(dump_file_path))
351 # Load the schema state from the dump file and verify the state of a node
353 SALOMERuntime.RuntimeSALOME_setRuntime(1)
355 schema = loader.YACSLoader().load(job_script_file)
356 stateParser = loader.stateParser()
357 sl = loader.stateLoader(stateParser, schema)
358 sl.parse(dump_file_path)
359 # 106 : "DONE" state code
360 self.assertEqual(106, schema.getChildByName("PyScript0").getEffectiveState())
362 # getJobResults to default directory (result_directory)
363 launcher.getJobResults(job_id, "")
364 self.verifyFile(os.path.join(job_params.result_directory, "result.txt"),
367 ##############################
368 # test of yacs job type using "--init_port" driver option
369 ##############################
370 def test_yacsopt(self):
371 yacs_path = os.getenv("YACS_ROOT_DIR", "")
372 if not os.path.isdir(yacs_path):
373 self.skipTest("Needs YACS module to run. Please define YACS_ROOT_DIR.")
375 case_test_dir = os.path.join(TestCompo.test_dir, "yacs_opt")
376 mkdir_p(case_test_dir)
379 script_text = """<?xml version='1.0' encoding='iso-8859-1' ?>
380 <proc name="myschema">
381 <type name="string" kind="string"/>
382 <type name="bool" kind="bool"/>
383 <type name="double" kind="double"/>
384 <type name="int" kind="int"/>
385 <container name="DefaultContainer">
386 <property name="container_kind" value="Salome"/>
387 <property name="attached_on_cloning" value="0"/>
388 <property name="container_name" value="FactoryServer"/>
389 <property name="name" value="localhost"/>
391 <inline name="mynode">
392 <script><code><![CDATA[
393 text_result = "i=%s,d=%s,b=%s,s=%s" % (i,d,b,s)
394 f = open('result.txt', 'w')
398 <load container="DefaultContainer"/>
399 <inport name="i" type="int"/>
400 <inport name="d" type="double"/>
401 <inport name="b" type="bool"/>
402 <inport name="s" type="string"/>
406 yacs_file = "simpleSchema.xml"
407 job_script_file = os.path.join(case_test_dir, yacs_file)
408 f = open(job_script_file, "w")
412 local_result_dir = os.path.join(case_test_dir, "result_yacsopt_job-")
413 job_params = salome.JobParameters()
414 job_params.job_type = "yacs_file"
415 job_params.job_file = job_script_file
416 #job_params.env_file = os.path.join(case_test_dir,env_file)
417 job_params.out_files = ["result.txt"]
419 # define the interval between two YACS schema dumps (3 seconds)
421 job_params.specific_parameters = [Engines.Parameter("YACSDriverOptions",
422 "-imynode.i=5 -imynode.d=3.7 -imynode.b=False -imynode.s=lili")]
423 expected_result="i=5,d=3.7,b=False,s=lili"
424 job_params.resource_required = salome.ResourceParameters()
425 job_params.resource_required.nb_proc = 1
427 launcher = salome.naming_service.Resolve('/SalomeLauncher')
428 resManager= salome.lcc.getResourcesManager()
430 for resource in self.ressources:
431 print("Testing yacs job with options on ", resource)
432 job_params.result_directory = local_result_dir + resource
433 job_params.job_name = "YacsJobOpt_" + resource
434 job_params.resource_required.name = resource
436 # use the working directory of the resource
437 resParams = resManager.GetResourceDefinition(resource)
438 wd = os.path.join(resParams.working_directory,
439 "YacsJobOpt" + self.suffix)
440 job_params.work_directory = wd
442 job_id = launcher.createJob(job_params)
443 launcher.launchJob(job_id)
444 jobState = launcher.getJobState(job_id)
446 yacs_dump_success = False
447 print("Job %d state: %s" % (job_id,jobState))
448 while jobState != "FINISHED" and jobState != "FAILED" :
450 jobState = launcher.getJobState(job_id)
451 print("Job %d state: %s " % (job_id,jobState))
454 self.assertEqual(jobState, "FINISHED")
456 # getJobResults to default directory (result_directory)
457 launcher.getJobResults(job_id, "")
458 self.verifyFile(os.path.join(job_params.result_directory, "result.txt"),
461 if __name__ == '__main__':