2 # -*- coding: utf-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
14 class TestCompo(unittest.TestCase):
17 # Prepare the test directory
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)
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)
32 os.chdir(case_test_dir)
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 -*-
40 # verify import salome
44 f = open('result.txt', 'w')
50 f = open(os.path.join("subdir",'autre.txt'), 'w')
54 f = open(job_script_file, "w")
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
69 job_params.work_directory = "/tmp/job_salome_py" + self.suffix
71 launcher = salome.naming_service.Resolve('/SalomeLauncher')
72 job_id = launcher.createJob(job_params)
74 launcher.launchJob(job_id)
77 jobState = launcher.getJobState(job_id)
78 print "Job %d state: %s" % (job_id,jobState)
79 while jobState != "FINISHED" and jobState != "FAILED" :
81 jobState = launcher.getJobState(job_id)
82 print "Job %d state: %s" % (job_id,jobState)
85 self.assertEqual(jobState, "FINISHED")
87 # getJobResults to default directory (result_directory)
88 launcher.getJobResults(job_id, "")
90 f = open(os.path.join(local_result_dir, "result.txt"), 'r')
93 self.assertEqual(text, "Salut!")
95 self.fail("IO exception:" + str(ex));
98 f = open(os.path.join(local_result_dir, "subdir", "autre.txt"), 'r')
101 self.assertEqual(text, "Hello!")
103 self.fail("IO exception:" + str(ex));
105 # getJobResults to a specific directory
106 mydir = os.path.join(case_test_dir, "custom_result_dir")
107 launcher.getJobResults(job_id, mydir)
109 f = open(os.path.join(mydir, "result.txt"), 'r')
112 self.assertEqual(text, "Salut!")
114 self.fail("IO exception:" + str(ex));
117 f = open(os.path.join(mydir, "subdir", "autre.txt"), 'r')
120 self.assertEqual(text, "Hello!")
122 self.fail("IO exception:" + str(ex));
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)
136 script_file = "myEnvScript.py"
137 script_text = """#! /usr/bin/env python
138 # -*- coding: utf-8 -*-
142 text_result = os.getenv("ENV_TEST_VAR","")
144 f = open('result.txt', 'w')
148 in_f = open("in.txt", "r")
149 in_text = in_f.read()
153 f = open(os.path.join("subdir",'altul.txt'), 'w')
157 abs_script_file = os.path.join(case_test_dir, script_file)
158 f = open(abs_script_file, "w")
161 os.chmod(abs_script_file, 0o755)
164 env_file = "myEnv.sh"
165 env_text = """export ENV_TEST_VAR="expected"
167 f = open(os.path.join(case_test_dir, env_file), "w")
172 f = open(os.path.join(case_test_dir, data_file), "w")
173 f.write("expected data")
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
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)
196 # wait for the end of the job
198 jobState = launcher.getJobState(job_id)
199 print "Job %d state: %s" % (job_id,jobState)
200 while jobState != "FINISHED" and jobState != "FAILED" :
202 jobState = launcher.getJobState(job_id)
203 print "Job %d state: %s" % (job_id,jobState)
207 self.assertEqual(jobState, "FINISHED")
208 launcher.getJobResults(job_id, "")
210 f = open(os.path.join(local_result_dir, "result.txt"), 'r')
213 self.assertEqual(text, "expected")
215 self.fail("IO exception:" + str(ex));
217 ##############################
218 # test of yacs job type
219 ##############################
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.")
225 case_test_dir = os.path.join(TestCompo.test_dir, "yacs")
226 os.mkdir(case_test_dir)
229 env_file = "myEnv.sh"
230 env_text = """export ENV_TEST_VAR="expected"
232 f = open(os.path.join(case_test_dir, env_file), "w")
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"/>
246 <inline name="PyScript0">
247 <script><code><![CDATA[import os
248 text_result = os.getenv("ENV_TEST_VAR","")
249 f = open('result.txt', 'w')
253 <load container="DefaultContainer"/>
257 yacs_file = "mySchema.xml"
258 job_script_file = os.path.join(case_test_dir, yacs_file)
259 f = open(job_script_file, "w")
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
273 # define the interval between two YACS schema dumps (3 seconds)
275 job_params.specific_parameters = [Engines.Parameter("EnableDumpYACS", "3")]
277 job_params.resource_required = salome.ResourceParameters()
278 job_params.resource_required.nb_proc = 1
280 job_params.work_directory = "/tmp/job_yacs" + self.suffix
282 launcher = salome.naming_service.Resolve('/SalomeLauncher')
283 job_id = launcher.createJob(job_params)
285 launcher.launchJob(job_id)
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" :
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)
298 self.assertEqual(jobState, "FINISHED")
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))
305 # Load the schema state from the dump file and verify the state of a node
307 SALOMERuntime.RuntimeSALOME_setRuntime(1)
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())
316 # getJobResults to default directory (result_directory)
317 launcher.getJobResults(job_id, "")
319 f = open(os.path.join(local_result_dir, "result.txt"), 'r')
322 self.assertEqual(text, "expected")
324 self.fail("IO exception:" + str(ex))
326 if __name__ == '__main__':
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)