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