2 # -*- coding: utf-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
13 class TestCompo(unittest.TestCase):
16 # Prepare the test directory
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)
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)
31 os.chdir(case_test_dir)
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 -*-
39 # verify import salome
43 f = open('result.txt', 'w')
49 f = open(os.path.join("subdir",'autre.txt'), 'w')
53 f = open(job_script_file, "w")
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
68 job_params.work_directory = "/tmp/job_salome_py" + self.suffix
70 launcher = salome.naming_service.Resolve('/SalomeLauncher')
71 job_id = launcher.createJob(job_params)
73 launcher.launchJob(job_id)
76 jobState = launcher.getJobState(job_id)
77 print "Job %d state: %s" % (job_id,jobState)
78 while jobState != "FINISHED" and jobState != "FAILED" :
80 jobState = launcher.getJobState(job_id)
81 print "Job %d state: %s" % (job_id,jobState)
84 self.assertEqual(jobState, "FINISHED")
86 # getJobResults to default directory (result_directory)
87 launcher.getJobResults(job_id, "")
89 f = open(os.path.join(local_result_dir, "result.txt"), 'r')
92 self.assertEqual(text, "Salut!")
94 self.fail("IO exception:" + str(ex));
97 f = open(os.path.join(local_result_dir, "subdir", "autre.txt"), 'r')
100 self.assertEqual(text, "Hello!")
102 self.fail("IO exception:" + str(ex));
104 # getJobResults to a specific directory
105 mydir = os.path.join(case_test_dir, "custom_result_dir")
106 launcher.getJobResults(job_id, mydir)
108 f = open(os.path.join(mydir, "result.txt"), 'r')
111 self.assertEqual(text, "Salut!")
113 self.fail("IO exception:" + str(ex));
116 f = open(os.path.join(mydir, "subdir", "autre.txt"), 'r')
119 self.assertEqual(text, "Hello!")
121 self.fail("IO exception:" + str(ex));
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)
135 script_file = "myEnvScript.py"
136 script_text = """#! /usr/bin/env python
137 # -*- coding: utf-8 -*-
141 text_result = os.getenv("ENV_TEST_VAR","")
143 f = open('result.txt', 'w')
147 in_f = open("in.txt", "r")
148 in_text = in_f.read()
152 f = open(os.path.join("subdir",'altul.txt'), 'w')
156 abs_script_file = os.path.join(case_test_dir, script_file)
157 f = open(abs_script_file, "w")
160 os.chmod(abs_script_file, 0o755)
163 env_file = "myEnv.sh"
164 env_text = """export ENV_TEST_VAR="expected"
166 f = open(os.path.join(case_test_dir, env_file), "w")
171 f = open(os.path.join(case_test_dir, data_file), "w")
172 f.write("expected data")
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
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)
195 # wait for the end of the job
197 jobState = launcher.getJobState(job_id)
198 print "Job %d state: %s" % (job_id,jobState)
199 while jobState != "FINISHED" and jobState != "FAILED" :
201 jobState = launcher.getJobState(job_id)
202 print "Job %d state: %s" % (job_id,jobState)
206 self.assertEqual(jobState, "FINISHED")
207 launcher.getJobResults(job_id, "")
209 f = open(os.path.join(local_result_dir, "result.txt"), 'r')
212 self.assertEqual(text, "expected")
214 self.fail("IO exception:" + str(ex));
216 ##############################
217 # test of yacs job type
218 ##############################
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.")
224 case_test_dir = os.path.join(TestCompo.test_dir, "yacs")
225 os.mkdir(case_test_dir)
228 env_file = "myEnv.sh"
229 env_text = """export ENV_TEST_VAR="expected"
231 f = open(os.path.join(case_test_dir, env_file), "w")
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"/>
245 <inline name="PyScript0">
246 <script><code><![CDATA[import os
247 text_result = os.getenv("ENV_TEST_VAR","")
248 f = open('result.txt', 'w')
252 <load container="DefaultContainer"/>
256 yacs_file = "mySchema.xml"
257 job_script_file = os.path.join(case_test_dir, yacs_file)
258 f = open(job_script_file, "w")
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
274 job_params.work_directory = "/tmp/job_yacs" + self.suffix
276 launcher = salome.naming_service.Resolve('/SalomeLauncher')
277 job_id = launcher.createJob(job_params)
279 launcher.launchJob(job_id)
282 jobState = launcher.getJobState(job_id)
283 print "Job %d state: %s" % (job_id,jobState)
284 while jobState != "FINISHED" and jobState != "FAILED" :
286 jobState = launcher.getJobState(job_id)
287 print "Job %d state: %s" % (job_id,jobState)
290 self.assertEqual(jobState, "FINISHED")
292 # getJobResults to default directory (result_directory)
293 launcher.getJobResults(job_id, "")
295 f = open(os.path.join(local_result_dir, "result.txt"), 'r')
298 self.assertEqual(text, "expected")
300 self.fail("IO exception:" + str(ex));
302 if __name__ == '__main__':
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)