Coverage for tests/test_lssthtc.py: 20%
540 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-07 02:10 -0800
1# This file is part of ctrl_bps_htcondor.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
27"""Unit tests for classes and functions in lssthtc.py."""
29import io
30import logging
31import os
32import pathlib
33import stat
34import sys
35import tempfile
36import unittest
37from shutil import copy2, copytree, ignore_patterns, rmtree, which
39import htcondor
41from lsst.ctrl.bps.htcondor import lssthtc
42from lsst.utils.tests import temporaryDirectory
44logger = logging.getLogger("lsst.ctrl.bps.htcondor")
45TESTDIR = os.path.abspath(os.path.dirname(__file__))
48class TestLsstHtc(unittest.TestCase):
49 """Test basic usage."""
51 def testHtcEscapeInt(self):
52 self.assertEqual(lssthtc.htc_escape(100), 100)
54 def testHtcEscapeDouble(self):
55 self.assertEqual(lssthtc.htc_escape('"double"'), '""double""')
57 def testHtcEscapeSingle(self):
58 self.assertEqual(lssthtc.htc_escape("'single'"), "''single''")
60 def testHtcEscapeNoSideEffect(self):
61 val = "'val'"
62 self.assertEqual(lssthtc.htc_escape(val), "''val''")
63 self.assertEqual(val, "'val'")
65 def testHtcEscapeQuot(self):
66 self.assertEqual(lssthtc.htc_escape(""val""), '"val"')
68 def testHtcVersion(self):
69 ver = lssthtc.htc_version()
70 self.assertRegex(ver, r"^\d+\.\d+\.\d+$")
73class HtcTweakJobInfoTestCase(unittest.TestCase):
74 """Test the function responsible for massaging job information."""
76 def setUp(self):
77 self.log_dir = tempfile.TemporaryDirectory()
78 self.log_dirname = pathlib.Path(self.log_dir.name)
79 self.job = {
80 "Cluster": 1,
81 "Proc": 0,
82 "Iwd": str(self.log_dirname),
83 "Owner": self.log_dirname.owner(),
84 "MyType": None,
85 "TerminatedNormally": True,
86 }
88 def tearDown(self):
89 self.log_dir.cleanup()
91 def testDirectAssignments(self):
92 lssthtc.htc_tweak_log_info(self.log_dirname, self.job)
93 self.assertEqual(self.job["ClusterId"], self.job["Cluster"])
94 self.assertEqual(self.job["ProcId"], self.job["Proc"])
95 self.assertEqual(self.job["Iwd"], str(self.log_dirname))
96 self.assertEqual(self.job["Owner"], self.log_dirname.owner())
98 def testIncompatibleAdPassThru(self):
99 # Passing a job ad with insufficient information should be a no-op.
100 expected = {"foo": "bar"}
101 result = dict(expected)
102 lssthtc.htc_tweak_log_info(self.log_dirname, result)
103 self.assertEqual(result, expected)
105 def testJobStatusAssignmentJobAbortedEvent(self):
106 job = self.job | {"MyType": "JobAbortedEvent"}
107 lssthtc.htc_tweak_log_info(self.log_dirname, job)
108 self.assertTrue("JobStatus" in job)
109 self.assertEqual(job["JobStatus"], htcondor.JobStatus.REMOVED)
111 def testJobStatusAssignmentExecuteEvent(self):
112 job = self.job | {"MyType": "ExecuteEvent"}
113 lssthtc.htc_tweak_log_info(self.log_dirname, job)
114 self.assertTrue("JobStatus" in job)
115 self.assertEqual(job["JobStatus"], htcondor.JobStatus.RUNNING)
117 def testJobStatusAssignmentSubmitEvent(self):
118 job = self.job | {"MyType": "SubmitEvent"}
119 lssthtc.htc_tweak_log_info(self.log_dirname, job)
120 self.assertTrue("JobStatus" in job)
121 self.assertEqual(job["JobStatus"], htcondor.JobStatus.IDLE)
123 def testJobStatusAssignmentJobHeldEvent(self):
124 job = self.job | {"MyType": "JobHeldEvent"}
125 lssthtc.htc_tweak_log_info(self.log_dirname, job)
126 self.assertTrue("JobStatus" in job)
127 self.assertEqual(job["JobStatus"], htcondor.JobStatus.HELD)
129 def testJobStatusAssignmentJobTerminatedEvent(self):
130 job = self.job | {"MyType": "JobTerminatedEvent"}
131 lssthtc.htc_tweak_log_info(self.log_dirname, job)
132 self.assertTrue("JobStatus" in job)
133 self.assertEqual(job["JobStatus"], htcondor.JobStatus.COMPLETED)
135 def testJobStatusAssignmentPostScriptTerminatedEvent(self):
136 job = self.job | {"MyType": "PostScriptTerminatedEvent"}
137 lssthtc.htc_tweak_log_info(self.log_dirname, job)
138 self.assertTrue("JobStatus" in job)
139 self.assertEqual(job["JobStatus"], htcondor.JobStatus.COMPLETED)
141 def testJobStatusAssignmentReleaseEvent(self):
142 job = self.job | {"MyType": "JobReleaseEvent"}
143 lssthtc.htc_tweak_log_info(self.log_dirname, job)
144 self.assertTrue("JobStatus" in job)
145 self.assertEqual(job["JobStatus"], htcondor.JobStatus.RUNNING)
147 def testAddingExitStatusSuccess(self):
148 job = self.job | {
149 "MyType": "JobTerminatedEvent",
150 "ToE": {"ExitBySignal": False, "ExitCode": 1},
151 }
152 lssthtc.htc_tweak_log_info(self.log_dirname, job)
153 self.assertIn("ExitBySignal", job)
154 self.assertIs(job["ExitBySignal"], False)
155 self.assertIn("ExitCode", job)
156 self.assertEqual(job["ExitCode"], 1)
158 def testAddingExitStatusFailure(self):
159 job = self.job | {
160 "MyType": "JobHeldEvent",
161 }
162 with self.assertLogs(logger=logger, level="ERROR") as cm:
163 lssthtc.htc_tweak_log_info(self.log_dirname, job)
164 self.assertIn("Could not determine exit status", cm.output[0])
166 def testLoggingUnknownLogEvent(self):
167 job = self.job | {"MyType": "Foo"}
168 with self.assertLogs(logger=logger, level="DEBUG") as cm:
169 lssthtc.htc_tweak_log_info(self.log_dirname, job)
170 self.assertIn("Unknown log event", cm.output[1])
172 def testMissingKey(self):
173 job = self.job
174 del job["Cluster"]
175 with self.assertRaises(KeyError) as cm:
176 lssthtc.htc_tweak_log_info(self.log_dirname, job)
177 self.assertEqual(str(cm.exception), "'Cluster'")
180class HtcCheckDagmanOutputTestCase(unittest.TestCase):
181 """Test htc_check_dagman_output function."""
183 def test_missing_output_file(self):
184 with temporaryDirectory() as tmp_dir:
185 with self.assertRaises(FileNotFoundError):
186 _ = lssthtc.htc_check_dagman_output(tmp_dir)
188 def test_permissions_output_file(self):
189 with temporaryDirectory() as tmp_dir:
190 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir)
191 os.chmod(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out", 0o200)
192 print(os.stat(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out"))
193 results = lssthtc.htc_check_dagman_output(tmp_dir)
194 os.chmod(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out", 0o600)
195 self.assertIn("Could not read dagman output file", results)
197 def test_submit_failure(self):
198 with temporaryDirectory() as tmp_dir:
199 copy2(f"{TESTDIR}/data/bad_submit.dag.dagman.out", tmp_dir)
200 results = lssthtc.htc_check_dagman_output(tmp_dir)
201 self.assertIn("Warn: Job submission issues (last: ", results)
203 def test_tmpdir_abort(self):
204 with temporaryDirectory() as tmp_dir:
205 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir)
206 results = lssthtc.htc_check_dagman_output(tmp_dir)
207 self.assertIn("Cannot submit from /tmp", results)
209 def test_no_messages(self):
210 with temporaryDirectory() as tmp_dir:
211 copy2(f"{TESTDIR}/data/test_no_messages.dag.dagman.out", tmp_dir)
212 results = lssthtc.htc_check_dagman_output(tmp_dir)
213 self.assertEqual("", results)
216class SummarizeDagTestCase(unittest.TestCase):
217 """Test summarize_dag function."""
219 def test_no_dag_file(self):
220 with temporaryDirectory() as tmp_dir:
221 summary, job_name_to_pipetask, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
222 self.assertFalse(len(job_name_to_pipetask))
223 self.assertFalse(len(job_name_to_type))
224 self.assertFalse(summary)
226 def test_success(self):
227 with temporaryDirectory() as tmp_dir:
228 copy2(f"{TESTDIR}/data/good.dag", tmp_dir)
229 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
230 self.assertEqual(summary, "pipetaskInit:1;label1:1;label2:1;label3:1;finalJob:1")
231 self.assertEqual(
232 job_name_to_label,
233 {
234 "pipetaskInit": "pipetaskInit",
235 "0682f8f9-12f0-40a5-971e-8b30c7231e5c_label1_val1_val2": "label1",
236 "d0305e2d-f164-4a85-bd24-06afe6c84ed9_label2_val1_val2": "label2",
237 "2806ecc9-1bba-4362-8fff-ab4e6abb9f83_label3_val1_val2": "label3",
238 "finalJob": "finalJob",
239 },
240 )
241 self.assertEqual(
242 job_name_to_type,
243 {
244 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
245 "0682f8f9-12f0-40a5-971e-8b30c7231e5c_label1_val1_val2": lssthtc.WmsNodeType.PAYLOAD,
246 "d0305e2d-f164-4a85-bd24-06afe6c84ed9_label2_val1_val2": lssthtc.WmsNodeType.PAYLOAD,
247 "2806ecc9-1bba-4362-8fff-ab4e6abb9f83_label3_val1_val2": lssthtc.WmsNodeType.PAYLOAD,
248 "finalJob": lssthtc.WmsNodeType.FINAL,
249 },
250 )
252 def test_service(self):
253 with temporaryDirectory() as tmp_dir:
254 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag", tmp_dir)
255 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
256 self.assertEqual(summary, "pipetaskInit:1;label1:2;label2:2;finalJob:1")
257 self.assertEqual(
258 job_name_to_label,
259 {
260 "pipetaskInit": "pipetaskInit",
261 "057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b": "label1",
262 "4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a": "label1",
263 "40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b": "label2",
264 "696ee50d-e711-40d6-9caf-ee29ae4a656d_label2_val1a_val2a": "label2",
265 "finalJob": "finalJob",
266 "provisioningJob": "provisioningJob",
267 },
268 )
269 self.assertEqual(
270 job_name_to_type,
271 {
272 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
273 "057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
274 "4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
275 "40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
276 "696ee50d-e711-40d6-9caf-ee29ae4a656d_label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
277 "finalJob": lssthtc.WmsNodeType.FINAL,
278 "provisioningJob": lssthtc.WmsNodeType.SERVICE,
279 },
280 )
282 def test_noop(self):
283 with temporaryDirectory() as tmp_dir:
284 copy2(f"{TESTDIR}/data/noop_running_1/noop_running_1.dag", tmp_dir)
285 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
286 self.assertEqual(
287 set(summary.split(";")),
288 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"},
289 )
290 self.assertEqual(
291 job_name_to_label,
292 {
293 "label1_val1a_val2a": "label1",
294 "label1_val1a_val2b": "label1",
295 "label1_val1b_val2a": "label1",
296 "label1_val1b_val2b": "label1",
297 "label1_val1c_val2a": "label1",
298 "label1_val1c_val2b": "label1",
299 "label2_val1a_val2a": "label2",
300 "label2_val1a_val2b": "label2",
301 "label2_val1b_val2a": "label2",
302 "label2_val1b_val2b": "label2",
303 "label2_val1c_val2a": "label2",
304 "label2_val1c_val2b": "label2",
305 "label3_val1a_val2a": "label3",
306 "label3_val1a_val2b": "label3",
307 "label3_val1b_val2a": "label3",
308 "label3_val1b_val2b": "label3",
309 "label3_val1c_val2a": "label3",
310 "label3_val1c_val2b": "label3",
311 "label4_val1a_val2a": "label4",
312 "label4_val1a_val2b": "label4",
313 "label4_val1b_val2a": "label4",
314 "label4_val1b_val2b": "label4",
315 "label4_val1c_val2a": "label4",
316 "label4_val1c_val2b": "label4",
317 "label5_val1a_val2a": "label5",
318 "label5_val1a_val2b": "label5",
319 "label5_val1b_val2a": "label5",
320 "label5_val1b_val2b": "label5",
321 "label5_val1c_val2a": "label5",
322 "label5_val1c_val2b": "label5",
323 "finalJob": "finalJob",
324 "pipetaskInit": "pipetaskInit",
325 "wms_noop_order1_val1a": "order1",
326 "wms_noop_order1_val1b": "order1",
327 },
328 )
329 self.assertEqual(
330 job_name_to_type,
331 {
332 "label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
333 "label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
334 "label1_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
335 "label1_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
336 "label1_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
337 "label1_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
338 "label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
339 "label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
340 "label2_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
341 "label2_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
342 "label2_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
343 "label2_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
344 "label3_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
345 "label3_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
346 "label3_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
347 "label3_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
348 "label3_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
349 "label3_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
350 "label4_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
351 "label4_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
352 "label4_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
353 "label4_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
354 "label4_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
355 "label4_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
356 "label5_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
357 "label5_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
358 "label5_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
359 "label5_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
360 "label5_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
361 "label5_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
362 "finalJob": lssthtc.WmsNodeType.FINAL,
363 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
364 "wms_noop_order1_val1a": lssthtc.WmsNodeType.NOOP,
365 "wms_noop_order1_val1b": lssthtc.WmsNodeType.NOOP,
366 },
367 )
369 def test_subdags(self):
370 with temporaryDirectory() as tmp_dir:
371 submit_dir = os.path.join(tmp_dir, "group_running_1")
372 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
373 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(submit_dir)
374 self.assertEqual(
375 set(summary.split(";")),
376 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"},
377 )
379 self.assertEqual(
380 job_name_to_label,
381 {
382 "pipetaskInit": "pipetaskInit",
383 "label1_val1b_val2a": "label1",
384 "label1_val1c_val2a": "label1",
385 "label1_val1a_val2b": "label1",
386 "label1_val1b_val2b": "label1",
387 "label1_val1c_val2b": "label1",
388 "label1_val1a_val2a": "label1",
389 "label2_val1a_val2b": "label2",
390 "label2_val1a_val2a": "label2",
391 "label2_val1b_val2a": "label2",
392 "label2_val1b_val2b": "label2",
393 "label2_val1c_val2a": "label2",
394 "label2_val1c_val2b": "label2",
395 "label3_val1b_val2a": "label3",
396 "label3_val1c_val2a": "label3",
397 "label3_val1a_val2b": "label3",
398 "label3_val1b_val2b": "label3",
399 "label3_val1c_val2b": "label3",
400 "label3_val1a_val2a": "label3",
401 "label4_val1a_val2b": "label4",
402 "label4_val1a_val2a": "label4",
403 "label4_val1b_val2a": "label4",
404 "label4_val1b_val2b": "label4",
405 "label4_val1c_val2a": "label4",
406 "label4_val1c_val2b": "label4",
407 "label5_val1a_val2b": "label5",
408 "label5_val1a_val2a": "label5",
409 "label5_val1b_val2a": "label5",
410 "label5_val1b_val2b": "label5",
411 "label5_val1c_val2a": "label5",
412 "label5_val1c_val2b": "label5",
413 "finalJob": "finalJob",
414 "provisioningJob": "provisioningJob",
415 "wms_group_order1_val1a": "order1",
416 "wms_group_order1_val1b": "order1",
417 "wms_group_order1_val1c": "order1",
418 "wms_check_status_wms_group_order1_val1a": "order1",
419 "wms_check_status_wms_group_order1_val1b": "order1",
420 "wms_check_status_wms_group_order1_val1c": "order1",
421 },
422 )
424 self.assertEqual(
425 job_name_to_type,
426 {
427 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
428 "label1_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
429 "label1_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
430 "label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
431 "label1_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
432 "label1_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
433 "label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
434 "label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
435 "label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
436 "label2_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
437 "label2_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
438 "label2_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
439 "label2_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
440 "label3_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
441 "label3_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
442 "label3_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
443 "label3_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
444 "label3_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
445 "label3_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
446 "label4_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
447 "label4_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
448 "label4_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
449 "label4_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
450 "label4_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
451 "label4_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
452 "label5_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
453 "label5_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
454 "label5_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
455 "label5_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
456 "label5_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
457 "label5_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
458 "finalJob": lssthtc.WmsNodeType.FINAL,
459 "provisioningJob": lssthtc.WmsNodeType.SERVICE,
460 "wms_group_order1_val1a": lssthtc.WmsNodeType.SUBDAG,
461 "wms_group_order1_val1b": lssthtc.WmsNodeType.SUBDAG,
462 "wms_group_order1_val1c": lssthtc.WmsNodeType.SUBDAG,
463 "wms_check_status_wms_group_order1_val1a": lssthtc.WmsNodeType.SUBDAG_CHECK,
464 "wms_check_status_wms_group_order1_val1b": lssthtc.WmsNodeType.SUBDAG_CHECK,
465 "wms_check_status_wms_group_order1_val1c": lssthtc.WmsNodeType.SUBDAG_CHECK,
466 },
467 )
470class ReadDagNodesLogTestCase(unittest.TestCase):
471 """Test read_dag_nodes_log function."""
473 def setUp(self):
474 self.tmpdir = tempfile.mkdtemp()
476 def tearDown(self):
477 rmtree(self.tmpdir, ignore_errors=True)
479 def testFileMissing(self):
480 with self.assertRaisesRegex(FileNotFoundError, "DAGMan node log not found in"):
481 _ = lssthtc.read_dag_nodes_log(self.tmpdir)
483 def testRegular(self):
484 with temporaryDirectory() as tmp_dir:
485 submit_dir = os.path.join(tmp_dir, "tiny_problems")
486 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
487 results = lssthtc.read_dag_nodes_log(submit_dir)
488 self.assertEqual(results["9231.0"]["Cluster"], 9231)
489 self.assertEqual(results["9231.0"]["Proc"], 0)
490 self.assertEqual(results["9231.0"]["ToE"]["ExitCode"], 1)
491 self.assertEqual(len(results), 6)
493 def testSubdags(self):
494 """Making sure it gets data from subdag dirs and doesn't
495 fail if some subdags haven't started running yet.
496 """
497 with temporaryDirectory() as tmp_dir:
498 submit_dir = os.path.join(tmp_dir, "group_running_1")
499 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
500 results = lssthtc.read_dag_nodes_log(submit_dir)
501 # main dag
502 self.assertEqual(results["10094.0"]["Cluster"], 10094)
503 # subdag
504 self.assertEqual(results["10112.0"]["Cluster"], 10112)
505 self.assertEqual(results["10116.0"]["Cluster"], 10116)
508class ReadNodeStatusTestCase(unittest.TestCase):
509 """Test read_node_status function."""
511 def setUp(self):
512 self.tmpdir = tempfile.mkdtemp()
514 def tearDown(self):
515 rmtree(self.tmpdir, ignore_errors=True)
517 def testServiceJobNotSubmitted(self):
518 # tiny_prov_no_submit files have successful workflow
519 # but provisioningJob could not submit.
520 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag.nodes.log", self.tmpdir)
521 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag.dagman.log", self.tmpdir)
522 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.node_status", self.tmpdir)
523 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag", self.tmpdir)
525 jobs = lssthtc.read_node_status(self.tmpdir)
526 found = [
527 id_
528 for id_ in jobs
529 if jobs[id_].get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN) == lssthtc.WmsNodeType.SERVICE
530 ]
531 self.assertEqual(len(found), 1)
532 self.assertEqual(jobs[found[0]]["DAGNodeName"], "provisioningJob")
533 self.assertEqual(jobs[found[0]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY)
535 def testMissingStatusFile(self):
536 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag.nodes.log", self.tmpdir)
537 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag.dagman.log", self.tmpdir)
538 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag", self.tmpdir)
540 jobs = lssthtc.read_node_status(self.tmpdir)
541 self.assertEqual(len(jobs), 7)
542 self.assertEqual(jobs["9230.0"]["DAGNodeName"], "pipetaskInit")
543 self.assertEqual(jobs["9230.0"]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD)
544 found = [
545 id_
546 for id_ in jobs
547 if jobs[id_].get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN) == lssthtc.WmsNodeType.SERVICE
548 ]
549 self.assertEqual(len(found), 1)
550 self.assertEqual(jobs[found[0]]["DAGNodeName"], "provisioningJob")
552 def testSubdagsRunning(self):
553 with temporaryDirectory() as tmp_dir:
554 test_tmp_dir = pathlib.Path(tmp_dir)
555 submit_dir = test_tmp_dir / "submit"
556 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
557 jobs = lssthtc.read_node_status(submit_dir)
558 self.assertEqual(len(jobs), 39) # includes non-payload jobs
559 # not guaranteed ids are same, so use names instead
560 job_name_to_id = {}
561 for id_, info in jobs.items():
562 job_name_to_id[info.get("DAGNodeName", id_)] = id_
563 job_type_to_names = {}
564 for id_, info in jobs.items():
565 job_type_to_names.setdefault(
566 info.get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN), set()
567 ).add(info.get("DAGNodeName", id_))
569 # check counts
570 self.assertNotIn(lssthtc.WmsNodeType.NOOP, job_type_to_names)
571 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.PAYLOAD]), 31)
572 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.FINAL]), 1)
573 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SERVICE]), 1)
574 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG]), 3)
575 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG_CHECK]), 3)
577 # spot check some statuses
578 self.assertEqual(
579 jobs[job_name_to_id["label3_val1a_val2b"]]["NodeStatus"], lssthtc.NodeStatus.DONE
580 )
581 self.assertEqual(
582 jobs[job_name_to_id["wms_group_order1_val1a"]]["NodeStatus"], lssthtc.NodeStatus.SUBMITTED
583 )
584 self.assertEqual(
585 jobs[job_name_to_id["label5_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY
586 )
587 self.assertEqual(
588 jobs[job_name_to_id["label2_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.DONE
589 )
591 def testSubdagsFailed(self):
592 with temporaryDirectory() as tmp_dir:
593 test_tmp_dir = pathlib.Path(tmp_dir)
594 submit_dir = test_tmp_dir / "submit"
595 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
596 jobs = lssthtc.read_node_status(submit_dir)
597 self.assertEqual(len(jobs), 39)
598 # not guaranteed ids are same, so use names instead
599 job_name_to_id = {}
600 for id_, info in jobs.items():
601 job_name_to_id[info.get("DAGNodeName", id_)] = id_
602 job_type_to_names = {}
603 for id_, info in jobs.items():
604 job_type_to_names.setdefault(
605 info.get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN), set()
606 ).add(info.get("DAGNodeName", id_))
608 # check counts
609 self.assertNotIn(lssthtc.WmsNodeType.NOOP, job_type_to_names)
610 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.PAYLOAD]), 31)
611 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.FINAL]), 1)
612 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SERVICE]), 1)
613 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG]), 3)
614 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG_CHECK]), 3)
616 # spot check some statuses
617 self.assertEqual(
618 jobs[job_name_to_id["label3_val1a_val2b"]]["NodeStatus"], lssthtc.NodeStatus.DONE
619 )
620 self.assertEqual(
621 jobs[job_name_to_id["wms_group_order1_val1a"]]["NodeStatus"], lssthtc.NodeStatus.DONE
622 )
623 self.assertEqual(
624 jobs[job_name_to_id["label5_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.DONE
625 )
627 self.assertEqual(
628 jobs[job_name_to_id["label5_val1b_val2a"]]["NodeStatus"], lssthtc.NodeStatus.FUTILE
629 )
630 self.assertEqual(
631 jobs[job_name_to_id["wms_group_order1_val1b"]]["NodeStatus"], lssthtc.NodeStatus.DONE
632 )
633 self.assertEqual(
634 jobs[job_name_to_id["wms_check_status_wms_group_order1_val1b"]]["NodeStatus"],
635 lssthtc.NodeStatus.ERROR,
636 )
639class HTCJobTestCase(unittest.TestCase):
640 """Test HTCJob methods."""
642 def testWriteDagCommandsPayload(self):
643 job = lssthtc.HTCJob(
644 "job1",
645 "label1",
646 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
647 {"dir": "jobs/label1"},
648 )
649 job.subfile = "job1.sub"
651 mockfh = io.StringIO()
652 job.write_dag_commands(mockfh, "../..")
653 self.assertIn('JOB job1 "job1.sub" DIR "../../jobs/label1"', mockfh.getvalue())
655 def testWriteDagCommandsNotJob(self):
656 # Testing giving command_name, no dag_rel_path and no dir
657 job = lssthtc.HTCJob(
658 "finalJob",
659 "finalJob",
660 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
661 )
662 job.subfile = "jobs/finalJob/finalJob.sub"
663 mockfh = io.StringIO()
664 job.write_dag_commands(mockfh, "", "FINAL")
665 self.assertIn('FINAL finalJob "jobs/finalJob/finalJob.sub"', mockfh.getvalue())
667 def testWriteDagCommandsNoop(self):
668 job = lssthtc.HTCJob("wms_noop_job1", "label1", {}, {"noop": True})
669 job.subfile = "notthere.sub"
670 mockfh = io.StringIO()
671 job.write_dag_commands(mockfh, "")
672 self.assertIn("NOOP", mockfh.getvalue())
674 def testWriteSubmitFile(self):
675 job = lssthtc.HTCJob(
676 "job1",
677 "label1",
678 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
679 )
680 with temporaryDirectory() as tmp_dir:
681 filename = pathlib.Path(tmp_dir) / "label1/job1.sub"
682 job.write_submit_file(filename.parent)
683 self.assertTrue(filename.exists())
684 # Try to make Submit object from file to find any syntax issues
685 _ = lssthtc.htc_create_submit_from_file(filename)
687 def testWriteSubmitFileExists(self):
688 job = lssthtc.HTCJob(
689 "job1",
690 "label1",
691 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
692 )
693 with temporaryDirectory() as tmp_dir:
694 filename = pathlib.Path(tmp_dir) / "job1.sub"
695 job.subfile = filename
696 with open(filename, "w"):
697 pass # make empty file
698 job.write_submit_file(filename.parent)
699 # make sure didn't overwrite file
700 self.assertEqual(filename.stat().st_size, 0, "Incorrectly overwrote existing file")
703class HtcWriteJobCommands(unittest.TestCase):
704 """Test _htc_write_job_commands function."""
706 def testAllCommands(self):
707 dag_cmds = {
708 "pre": {
709 "defer": {"status": 1, "time": 120},
710 "debug": {"filename": "debug_pre.txt", "type": "ALL"},
711 "executable": "exec1",
712 "arguments": "arg1 arg2",
713 },
714 "post": {
715 "defer": {"status": 2, "time": 180},
716 "debug": {"filename": "debug_post.txt", "type": "ALL"},
717 "executable": "exec2",
718 "arguments": "arg3 arg4",
719 },
720 "vars": {"num": 8, "spaces": "a space"},
721 "pre_skip": "1",
722 "retry": 3,
723 "retry_unless_exit": 1,
724 "abort_dag_on": {"node_exit": 100, "abort_exit": 4},
725 }
727 truth = """SCRIPT DEFER 1 120 DEBUG debug_pre.txt ALL PRE job1 exec1 arg1 arg2
728SCRIPT DEFER 2 180 DEBUG debug_post.txt ALL POST job1 exec2 arg3 arg4
729VARS job1 num="8"
730VARS job1 spaces="a space"
731PRE_SKIP job1 1
732RETRY job1 3 UNLESS-EXIT 1
733ABORT-DAG-ON job1 100 RETURN 4
734"""
735 mockfh = io.StringIO()
736 lssthtc._htc_write_job_commands(mockfh, "job1", dag_cmds)
737 self.assertEqual(mockfh.getvalue(), truth)
739 def testPartialCommands(self):
740 # Trigger skipping the inner if clauses.
741 dag_cmds = {
742 "pre": {
743 "executable": "exec1",
744 },
745 "post": {
746 "executable": "exec2",
747 },
748 "vars": {"num": 8, "spaces": "a space"},
749 "pre_skip": "1",
750 "retry": 3,
751 }
753 truth = """SCRIPT PRE job1 exec1
754SCRIPT POST job1 exec2
755VARS job1 num="8"
756VARS job1 spaces="a space"
757PRE_SKIP job1 1
758RETRY job1 3
759"""
760 mockfh = io.StringIO()
761 lssthtc._htc_write_job_commands(mockfh, "job1", dag_cmds)
762 self.assertEqual(mockfh.getvalue(), truth)
764 def testNoCommands(self):
765 dag_cmds = {}
766 mockfh = io.StringIO()
767 lssthtc._htc_write_job_commands(mockfh, "job2", dag_cmds)
768 self.assertEqual(mockfh.getvalue(), "")
771class HTCBackupFilesSinglePathTestCase(unittest.TestCase):
772 """Test htc_backup_files_single_path function."""
774 def testSrcDestSame(self):
775 with temporaryDirectory() as tmp_dir:
776 with self.assertRaisesRegex(
777 RuntimeError, "Destination directory is same as the source directory"
778 ):
779 lssthtc.htc_backup_files_single_path(tmp_dir, tmp_dir)
781 def testSuccess(self):
782 with temporaryDirectory() as tmp_dir:
783 test_tmp_dir = pathlib.Path(tmp_dir)
784 submit_dir = test_tmp_dir / "the_src_dir"
785 copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*"))
786 backup_dir = test_tmp_dir / "the_dest_dir"
787 backup_dir.mkdir()
788 lssthtc.htc_backup_files_single_path(submit_dir, backup_dir)
789 result_submit = []
790 for root, _, files in os.walk(submit_dir):
791 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
792 self.assertEqual(
793 set(result_submit),
794 {
795 "./tiny_success.dag.dagman.log",
796 "./tiny_success.dag.dagman.out",
797 "./tiny_success.dag",
798 },
799 )
800 result_backup = []
801 for root, _, files in os.walk(backup_dir):
802 result_backup.extend([str(os.path.join(os.path.relpath(root, backup_dir), f)) for f in files])
803 self.assertEqual(
804 set(result_backup),
805 {
806 "./tiny_success.info.json",
807 "./tiny_success.dag.metrics",
808 "./tiny_success.dag.nodes.log",
809 "./tiny_success.node_status",
810 },
811 )
814class HTCBackupFilesTestCase(unittest.TestCase):
815 """Test htc_backup_files function."""
817 def testDirectoryNotFound(self):
818 with temporaryDirectory() as tmp_dir:
819 test_tmp_dir = pathlib.Path(tmp_dir)
820 submit_dir = test_tmp_dir / "submit"
821 with self.assertRaises(FileNotFoundError):
822 _ = lssthtc.htc_backup_files(submit_dir)
824 def testSuccess(self):
825 with temporaryDirectory() as tmp_dir:
826 test_tmp_dir = pathlib.Path(tmp_dir)
827 submit_dir = test_tmp_dir / "submit"
828 copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*"))
829 result_rescue = lssthtc.htc_backup_files(submit_dir)
830 self.assertIsNone(result_rescue)
831 result_submit = []
832 for root, _, files in os.walk(submit_dir):
833 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
834 self.assertEqual(
835 set(result_submit),
836 {
837 "./tiny_success.dag.dagman.log",
838 "./tiny_success.dag.dagman.out",
839 "./tiny_success.dag",
840 "000/tiny_success.info.json",
841 "000/tiny_success.dag.metrics",
842 "000/tiny_success.dag.nodes.log",
843 "000/tiny_success.node_status",
844 },
845 )
847 def testDestNotInSubmitDir(self):
848 with temporaryDirectory() as tmp_dir:
849 test_tmp_dir = pathlib.Path(tmp_dir)
850 submit_dir = test_tmp_dir / "submit"
851 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
852 with self.assertLogs("lsst.ctrl.bps.htcondor", level="WARNING") as cm:
853 result_rescue = lssthtc.htc_backup_files(submit_dir, test_tmp_dir / "backup")
854 self.assertIn("Invalid backup location:", cm.output[-1])
855 result_rescue = lssthtc.htc_backup_files(submit_dir)
856 self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue))
857 result_submit = []
858 for root, _, files in os.walk(submit_dir):
859 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
860 self.assertEqual(
861 set(result_submit),
862 {
863 "./tiny_problems.dag.dagman.log",
864 "./tiny_problems.dag.dagman.out",
865 "./tiny_problems.dag",
866 "./tiny_problems.dag.rescue001",
867 "001/tiny_problems.info.json",
868 "001/tiny_problems.dag.metrics",
869 "001/tiny_problems.dag.nodes.log",
870 "001/tiny_problems.node_status",
871 },
872 )
874 def testDestInSubmitDir(self):
875 with temporaryDirectory() as tmp_dir:
876 test_tmp_dir = pathlib.Path(tmp_dir)
877 submit_dir = test_tmp_dir / "submit"
878 backup_dir = submit_dir / "subdir"
879 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
880 result_rescue = lssthtc.htc_backup_files(submit_dir, backup_dir)
881 self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue))
882 result_submit = []
883 for root, _, files in os.walk(submit_dir):
884 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
885 self.assertEqual(
886 set(result_submit),
887 {
888 "./tiny_problems.dag.dagman.log",
889 "./tiny_problems.dag.dagman.out",
890 "./tiny_problems.dag",
891 "./tiny_problems.dag.rescue001",
892 "subdir/001/tiny_problems.info.json",
893 "subdir/001/tiny_problems.dag.metrics",
894 "subdir/001/tiny_problems.dag.nodes.log",
895 "subdir/001/tiny_problems.node_status",
896 },
897 )
899 def testRelativeSubdir(self):
900 with temporaryDirectory() as tmp_dir:
901 test_tmp_dir = pathlib.Path(tmp_dir)
902 submit_dir = test_tmp_dir / "submit"
903 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
904 result_rescue = lssthtc.htc_backup_files(submit_dir, "reldir")
905 self.assertTrue((submit_dir / "tiny_problems.dag.rescue001").samefile(result_rescue))
906 result_submit = []
907 for root, _, files in os.walk(submit_dir):
908 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
909 self.assertEqual(
910 set(result_submit),
911 {
912 "./tiny_problems.dag.dagman.log",
913 "./tiny_problems.dag.dagman.out",
914 "./tiny_problems.dag",
915 "./tiny_problems.dag.rescue001",
916 "reldir/001/tiny_problems.info.json",
917 "reldir/001/tiny_problems.dag.metrics",
918 "reldir/001/tiny_problems.dag.nodes.log",
919 "reldir/001/tiny_problems.node_status",
920 },
921 )
923 def testSubdags(self):
924 with temporaryDirectory() as tmp_dir:
925 test_tmp_dir = pathlib.Path(tmp_dir)
926 submit_dir = test_tmp_dir / "submit"
927 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
928 result_rescue = lssthtc.htc_backup_files(submit_dir)
929 self.assertTrue(result_rescue.samefile(submit_dir / "group_failed_1.dag.rescue001"))
930 result_submit = []
931 for root, _, files in os.walk(submit_dir):
932 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
933 self.assertEqual(
934 set(result_submit),
935 {
936 "./group_failed_1.dag",
937 "./group_failed_1.dag.dagman.log",
938 "./group_failed_1.dag.dagman.out",
939 "./group_failed_1.dag.rescue001",
940 "subdags/wms_group_order1_val1a/group_order1_val1a.dag",
941 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.log",
942 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.out",
943 "subdags/wms_group_order1_val1b/group_order1_val1b.dag",
944 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.log",
945 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.out",
946 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.rescue001",
947 "subdags/wms_group_order1_val1c/group_order1_val1c.dag",
948 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.log",
949 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.out",
950 "001/group_failed_1.dag.nodes.log",
951 "001/group_failed_1.info.json",
952 "001/group_failed_1.node_status",
953 "001/subdags/wms_group_order1_val1a/group_order1_val1a.dag.nodes.log",
954 "001/subdags/wms_group_order1_val1a/group_order1_val1a.node_status",
955 "001/subdags/wms_group_order1_val1a/wms_group_order1_val1a.dag.post.out",
956 "001/subdags/wms_group_order1_val1a/wms_group_order1_val1a.status.txt",
957 "001/subdags/wms_group_order1_val1b/group_order1_val1b.dag.nodes.log",
958 "001/subdags/wms_group_order1_val1b/group_order1_val1b.node_status",
959 "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.status.txt",
960 "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.dag.post.out",
961 "001/subdags/wms_group_order1_val1c/group_order1_val1c.dag.nodes.log",
962 "001/subdags/wms_group_order1_val1c/group_order1_val1c.node_status",
963 "001/subdags/wms_group_order1_val1c/wms_group_order1_val1c.dag.post.out",
964 "001/subdags/wms_group_order1_val1c/wms_group_order1_val1c.status.txt",
965 },
966 )
969class UpdateRescueFileTestCase(unittest.TestCase):
970 """Test _update_rescue_file function."""
972 def testSuccess(self):
973 self.maxDiff = None
974 with temporaryDirectory() as tmp_dir:
975 test_tmp_dir = pathlib.Path(tmp_dir)
976 submit_dir = test_tmp_dir / "submit"
977 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
978 rescue_file = submit_dir / "group_failed_1.dag.rescue001"
979 lssthtc._update_rescue_file(rescue_file)
981 with open(rescue_file) as fh:
982 lines = fh.readlines()
983 results = "".join(lines)
985 truth = """# Rescue DAG file, created after running
986# the u_testuser_DM-46294_group_fail_20250310T160455Z.dag DAG file
987# Created 3/10/2025 16:08:56 UTC
988# Rescue DAG version: 2.0.1 (partial)
989#
990# Total number of Nodes: 26
991# Nodes premarked DONE: 21
992# Nodes that failed: 2
993# wms_group_order1_val1b,finalJob,<ENDLIST>
995DONE pipetaskInit
996DONE label1_val1c_val2a
997DONE label1_val1b_val2b
998DONE label1_val1b_val2a
999DONE label1_val1c_val2b
1000DONE label1_val1a_val2a
1001DONE label1_val1a_val2b
1002DONE label3_val1c_val2a
1003DONE label3_val1b_val2b
1004DONE label3_val1b_val2a
1005DONE label3_val1c_val2b
1006DONE label3_val1a_val2a
1007DONE label3_val1a_val2b
1008DONE wms_group_order1_val1a
1009DONE label5_val1a_val2a
1010DONE label5_val1a_val2b
1011DONE wms_group_order1_val1c
1012DONE label5_val1c_val2a
1013DONE label5_val1c_val2b
1014DONE wms_check_status_wms_group_order1_val1a
1015DONE wms_check_status_wms_group_order1_val1c
1016"""
1018 print("results = ", results, file=sys.stderr)
1019 print("truth = ", truth, file=sys.stderr)
1020 self.assertEqual(results, truth)
1023class ReadDagStatusTestCase(unittest.TestCase):
1024 """Test read_dag_status function and read_single_dag_status."""
1026 def testFileMissing(self):
1027 with temporaryDirectory() as tmp_dir:
1028 with self.assertRaisesRegex(FileNotFoundError, "DAGMan node status not found"):
1029 _ = lssthtc.read_dag_status(tmp_dir)
1031 def testRegular(self):
1032 with temporaryDirectory() as tmp_dir:
1033 submit_dir = os.path.join(tmp_dir, "tiny_problems")
1034 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1035 results = lssthtc.read_dag_status(submit_dir)
1036 truth = {
1037 "JobProcsHeld": 0,
1038 "NodesPost": 0,
1039 "JobProcsIdle": 0,
1040 "NodesTotal": 6,
1041 "NodesFailed": 2,
1042 "NodesDone": 3,
1043 "NodesQueued": 0,
1044 "NodesPre": 0,
1045 "NodesFutile": 1,
1046 "NodesUnready": 0,
1047 }
1048 self.assertEqual(results, results | truth)
1050 def testSubdags(self):
1051 """Making sure it gets data from subdag dirs and doesn't
1052 fail if some subdags haven't started running yet.
1053 """
1054 self.maxDiff = None
1055 with temporaryDirectory() as tmp_dir:
1056 submit_dir = os.path.join(tmp_dir, "submit")
1057 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1058 results = lssthtc.read_dag_status(submit_dir)
1059 truth = {
1060 "JobProcsHeld": 0,
1061 "NodesPost": 0,
1062 "JobProcsIdle": 0,
1063 "NodesTotal": 34,
1064 "NodesFailed": 0,
1065 "NodesDone": 17,
1066 "NodesQueued": 3,
1067 "NodesPre": 0,
1068 "NodesFutile": 0,
1069 "NodesUnready": 14,
1070 }
1071 self.assertEqual(results, results | truth)
1074class ReadDagInfoTestCase(unittest.TestCase):
1075 """Test read_dag_info function."""
1077 def testFileMissing(self):
1078 with temporaryDirectory() as tmp_dir:
1079 with self.assertRaisesRegex(FileNotFoundError, "File with DAGMan job information not found in "):
1080 _ = lssthtc.read_dag_info(tmp_dir)
1082 def testSuccess(self):
1083 with temporaryDirectory() as tmp_dir:
1084 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir)
1085 results = lssthtc.read_dag_info(tmp_dir)
1087 truth = {
1088 "test02": {
1089 "9208.0": {
1090 "ClusterId": 9208,
1091 "GlobalJobId": "test02#9208.0#1739465078",
1092 "bps_wms_service": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorService",
1093 "bps_project": "dev",
1094 "bps_payload": "tiny",
1095 "bps_operator": "testuser",
1096 "bps_wms_workflow": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorWorkflow",
1097 "bps_provisioning_job": "provisioningJob",
1098 "bps_run_quanta": "label1:1;label2:1",
1099 "bps_campaign": "quick",
1100 "bps_runsite": "testpool",
1101 "bps_job_summary": "pipetaskInit:1;label1:1;label2:1;finalJob:1",
1102 "bps_run": "u_testuser_tiny_20250213T164427Z",
1103 "bps_isjob": "True",
1104 }
1105 }
1106 }
1108 self.assertEqual(results, truth)
1110 def testPermissionError(self):
1111 with temporaryDirectory() as tmp_dir:
1112 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir)
1113 with unittest.mock.patch("lsst.ctrl.bps.htcondor.lssthtc.open") as mocked_open:
1114 mocked_open.side_effect = PermissionError
1115 with self.assertLogs("lsst.ctrl.bps.htcondor", level="DEBUG") as cm:
1116 results = lssthtc.read_dag_info(tmp_dir)
1117 self.assertIn("Retrieving DAGMan job information failed:", cm.output[-1])
1118 self.assertEqual({}, results)
1121class HtcWriteCondorFileTestCase(unittest.TestCase):
1122 """Test htc_write_condor_file function."""
1124 def testSuccess(self):
1125 with temporaryDirectory() as tmp_dir:
1126 job_name = "job1"
1127 filename = pathlib.Path(tmp_dir) / f"label1/{job_name}.sub"
1128 job = {
1129 "executable": "$(CTRL_MPEXEC_DIR)/bin/pipetask",
1130 "arguments": "-a -b 2 -c",
1131 "request_memory": "2000",
1132 "environment": "one=1 two=\"2\" three='spacey 'quoted' value'",
1133 "log": f"{job_name}.log",
1134 }
1135 job_attrs = {
1136 "bps_job_name": job_name,
1137 "bps_job_label": "label1",
1138 "bps_job_quanta": "task1:8;task2:8",
1139 }
1140 expected = [
1141 "executable=$(CTRL_MPEXEC_DIR)/bin/pipetask\n",
1142 "arguments=-a -b 2 -c\n",
1143 "request_memory=2000\n",
1144 "environment=\"one=1 two=\"2\" three='spacey 'quoted' value'\"\n",
1145 f"output={job_name}.$(Cluster).out\n",
1146 f"error={job_name}.$(Cluster).out\n",
1147 f"log={job_name}.log\n",
1148 f'+bps_job_name = "{job_name}"\n',
1149 '+bps_job_label = "label1"\n',
1150 '+bps_job_quanta = "task1:8;task2:8"\n',
1151 "queue\n",
1152 ]
1154 lssthtc.htc_write_condor_file(filename, job_name, job, job_attrs)
1155 with open(filename, encoding="utf-8") as f:
1156 actual = f.readlines()
1158 self.assertEqual(set(actual), set(expected))
1159 self.assertTrue(filename.exists())
1160 # Try to make Submit object from file to find any syntax issues
1161 _ = lssthtc.htc_create_submit_from_file(filename)
1164class HtcCreateSubmitFromDagTestCase(unittest.TestCase):
1165 """Test htc_create_submit_from_dag function."""
1167 @classmethod
1168 def setUpClass(cls):
1169 cls.bindir = None
1170 # htcondor.Submit.from_dag requires condor_dagman executable in path.
1171 if not which("condor_dagman"): # pragma: no cover
1172 cls.bindir = tempfile.TemporaryDirectory()
1173 fake_dagman_exec = pathlib.Path(cls.bindir.name) / "condor_dagman"
1174 with open(fake_dagman_exec, "w") as fh:
1175 print("#!/bin/bash", file=fh)
1176 print("echo fake_condor_dagman $@", file=fh)
1177 print("exit 0", file=fh)
1178 fake_dagman_exec.chmod(fake_dagman_exec.stat().st_mode | stat.S_IEXEC)
1179 os.environ["PATH"] = f"{os.environ['PATH']}:{cls.bindir.name}"
1181 @classmethod
1182 def tearDownClass(cls):
1183 if cls.bindir:
1184 cls.bindir.cleanup()
1186 @unittest.mock.patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})
1187 def testMaxIdleEnvVar(self):
1188 with temporaryDirectory() as tmp_dir:
1189 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1190 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1191 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})
1192 self.assertIn("-MaxIdle 42", submit["arguments"])
1193 self.assertEqual("true", os.environ["_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV"].lower())
1195 @unittest.mock.patch.dict(os.environ, {})
1196 def testMaxIdleGiven(self):
1197 with temporaryDirectory() as tmp_dir:
1198 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1199 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1200 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"MaxIdle": 37})
1201 self.assertIn("-MaxIdle 37", submit["arguments"])
1202 self.assertEqual("true", os.environ["_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV"].lower())
1204 @unittest.mock.patch.dict(os.environ, {})
1205 def testNoMaxJobsIdle(self):
1206 """Note: Since the produced arguments differ depending on
1207 HTCondor version when no MaxIdle passed to from_dag, not
1208 checking arguments string here. Instead just making sure
1209 lssthtc code doesn't pass MaxIdle value to from_dag.
1210 """
1211 with temporaryDirectory() as tmp_dir:
1212 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1213 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1214 with unittest.mock.patch("htcondor.Submit.from_dag") as submit_mock:
1215 with unittest.mock.patch("htcondor.param") as mock_param:
1216 mock_param.__contains__.return_value = False
1217 _ = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})
1218 self.assertEqual("true", os.environ["_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV"].lower())
1219 submit_mock.assert_called_once_with(str(dag_filename), {})
1221 @unittest.mock.patch.dict(os.environ, {})
1222 def testDoRecurseGivenWithNoEnv(self):
1223 with temporaryDirectory() as tmp_dir:
1224 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1225 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1226 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"do_recurse": True})
1227 self.assertIn("-do_recurse", submit["arguments"])
1228 self.assertEqual("true", os.environ["_CONDOR_DAGMAN_GENERATE_SUBDAG_SUBMITS"].lower())
1229 self.assertEqual("true", os.environ["_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV"].lower())
1231 @unittest.mock.patch.dict(os.environ, {"_CONDOR_DAGMAN_GENERATE_SUBDAG_SUBMITS": "False"})
1232 def testDoRecurseGivenWithEnv(self):
1233 with temporaryDirectory() as tmp_dir:
1234 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1235 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1236 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"do_recurse": True})
1237 self.assertIn("-do_recurse", submit["arguments"])
1238 self.assertEqual("true", os.environ["_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV"].lower())
1239 self.assertEqual("false", os.environ["_CONDOR_DAGMAN_GENERATE_SUBDAG_SUBMITS"].lower())
1241 @unittest.mock.patch.dict(os.environ, {"_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV": "*_DIR"})
1242 def testGetEnvOverridden(self):
1243 with temporaryDirectory() as tmp_dir:
1244 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1245 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1246 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"do_recurse": True})
1247 self.assertIn("-do_recurse", submit["arguments"])
1248 self.assertEqual(os.environ["_CONDOR_DAGMAN_MANAGER_JOB_APPEND_GETENV"], "*_DIR")
1251if __name__ == "__main__":
1252 unittest.main()