Coverage for tests/test_lssthtc.py: 99%
850 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 08:59 +0000
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 tempfile
35import unittest
36from shutil import copy2, copytree, ignore_patterns, rmtree, which
37from unittest.mock import patch
39from htcondor2 import JobStatus
41from lsst.ctrl.bps import BpsConfig
42from lsst.ctrl.bps.bps_utils import chdir
43from lsst.ctrl.bps.htcondor import dagman_configurator, htcondor_config, lssthtc
44from lsst.daf.butler import Config
45from lsst.utils.tests import temporaryDirectory
47from .dag_test_utils import make_lazy_dag
49logger = logging.getLogger("lsst.ctrl.bps.htcondor")
50TESTDIR = os.path.abspath(os.path.dirname(__file__))
53class TestLsstHtc(unittest.TestCase):
54 """Test basic usage."""
56 def testHtcEscapeInt(self):
57 self.assertEqual(lssthtc.htc_escape(100), 100)
59 def testHtcEscapeDouble(self):
60 self.assertEqual(lssthtc.htc_escape('"double"'), '""double""')
62 def testHtcEscapeSingle(self):
63 self.assertEqual(lssthtc.htc_escape("'single'"), "''single''")
65 def testHtcEscapeNoSideEffect(self):
66 val = "'val'"
67 self.assertEqual(lssthtc.htc_escape(val), "''val''")
68 self.assertEqual(val, "'val'")
70 def testHtcEscapeQuot(self):
71 self.assertEqual(lssthtc.htc_escape(""val""), '"val"')
74class HtcTweakJobInfoTestCase(unittest.TestCase):
75 """Test the function responsible for massaging job information."""
77 def setUp(self):
78 self.log_dir = tempfile.TemporaryDirectory()
79 self.log_dirname = pathlib.Path(self.log_dir.name)
80 self.job = {
81 "Cluster": 1,
82 "Proc": 0,
83 "Iwd": str(self.log_dirname),
84 "Owner": self.log_dirname.owner(),
85 "MyType": None,
86 "TerminatedNormally": True,
87 }
89 def tearDown(self):
90 self.log_dir.cleanup()
92 def testDirectAssignments(self):
93 lssthtc.htc_tweak_log_info(self.log_dirname, self.job)
94 self.assertEqual(self.job["ClusterId"], self.job["Cluster"])
95 self.assertEqual(self.job["ProcId"], self.job["Proc"])
96 self.assertEqual(self.job["Iwd"], str(self.log_dirname))
97 self.assertEqual(self.job["Owner"], self.log_dirname.owner())
99 def testIncompatibleAdPassThru(self):
100 # Passing a job ad with insufficient information should be a no-op.
101 expected = {"foo": "bar"}
102 result = dict(expected)
103 lssthtc.htc_tweak_log_info(self.log_dirname, result)
104 self.assertEqual(result, expected)
106 def testJobStatusAssignmentJobAbortedEvent(self):
107 job = self.job | {"MyType": "JobAbortedEvent"}
108 lssthtc.htc_tweak_log_info(self.log_dirname, job)
109 self.assertTrue("JobStatus" in job)
110 self.assertEqual(job["JobStatus"], JobStatus.REMOVED)
112 def testJobStatusAssignmentExecuteEvent(self):
113 job = self.job | {"MyType": "ExecuteEvent"}
114 lssthtc.htc_tweak_log_info(self.log_dirname, job)
115 self.assertTrue("JobStatus" in job)
116 self.assertEqual(job["JobStatus"], JobStatus.RUNNING)
118 def testJobStatusAssignmentSubmitEvent(self):
119 job = self.job | {"MyType": "SubmitEvent"}
120 lssthtc.htc_tweak_log_info(self.log_dirname, job)
121 self.assertTrue("JobStatus" in job)
122 self.assertEqual(job["JobStatus"], JobStatus.IDLE)
124 def testJobStatusAssignmentJobHeldEvent(self):
125 job = self.job | {"MyType": "JobHeldEvent"}
126 lssthtc.htc_tweak_log_info(self.log_dirname, job)
127 self.assertTrue("JobStatus" in job)
128 self.assertEqual(job["JobStatus"], JobStatus.HELD)
130 def testJobStatusAssignmentJobTerminatedEvent(self):
131 job = self.job | {"MyType": "JobTerminatedEvent"}
132 lssthtc.htc_tweak_log_info(self.log_dirname, job)
133 self.assertTrue("JobStatus" in job)
134 self.assertEqual(job["JobStatus"], JobStatus.COMPLETED)
136 def testJobStatusAssignmentPostScriptTerminatedEvent(self):
137 job = self.job | {"MyType": "PostScriptTerminatedEvent"}
138 lssthtc.htc_tweak_log_info(self.log_dirname, job)
139 self.assertTrue("JobStatus" in job)
140 self.assertEqual(job["JobStatus"], JobStatus.COMPLETED)
142 def testJobStatusAssignmentReleaseEventMainDagJob(self):
143 job = self.job | {"MyType": "JobReleaseEvent"}
144 lssthtc.htc_tweak_log_info(self.log_dirname, job)
145 self.assertTrue("JobStatus" in job)
146 self.assertEqual(job["JobStatus"], JobStatus.RUNNING)
148 def testJobStatusAssignmentReleaseEventForNodeJob(self):
149 job = self.job | {"MyType": "JobReleaseEvent", "DAGNodeName": "test_payload_job"}
150 lssthtc.htc_tweak_log_info(self.log_dirname, job)
151 self.assertTrue("JobStatus" in job)
152 self.assertEqual(job["JobStatus"], None)
154 def testAddingExitStatusSuccess(self):
155 job = self.job | {
156 "MyType": "JobTerminatedEvent",
157 "ExitBySignal": False,
158 "ReturnValue": 1,
159 }
160 lssthtc.htc_tweak_log_info(self.log_dirname, job)
161 self.assertIn("ExitBySignal", job)
162 self.assertIs(job["ExitBySignal"], False)
163 self.assertIn("ExitCode", job)
164 self.assertEqual(job["ExitCode"], 1)
166 def testAddingExitStatusFailure(self):
167 job = self.job | {
168 "MyType": "JobHeldEvent",
169 }
170 with self.assertLogs(logger=logger, level="ERROR") as cm:
171 lssthtc.htc_tweak_log_info(self.log_dirname, job)
172 self.assertIn("Could not determine exit status", cm.output[0])
174 def testLoggingUnknownLogEvent(self):
175 job = self.job | {"MyType": "Foo"}
176 with self.assertLogs(logger=logger, level="DEBUG") as cm:
177 lssthtc.htc_tweak_log_info(self.log_dirname, job)
178 self.assertIn("Unknown log event", cm.output[1])
180 def testMissingKey(self):
181 job = self.job
182 del job["Cluster"]
183 with self.assertRaises(KeyError) as cm:
184 lssthtc.htc_tweak_log_info(self.log_dirname, job)
185 self.assertEqual(str(cm.exception), "'Cluster'")
188class HtcCheckDagmanOutputTestCase(unittest.TestCase):
189 """Test htc_check_dagman_output function."""
191 def test_missing_output_file(self):
192 with temporaryDirectory() as tmp_dir:
193 with self.assertRaises(FileNotFoundError):
194 _ = lssthtc.htc_check_dagman_output(tmp_dir)
196 def test_permissions_output_file(self):
197 with temporaryDirectory() as tmp_dir:
198 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir)
199 os.chmod(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out", 0o200)
200 print(os.stat(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out"))
201 results = lssthtc.htc_check_dagman_output(tmp_dir)
202 os.chmod(f"{tmp_dir}/test_tmpdir_abort.dag.dagman.out", 0o600)
203 self.assertIn("Could not read dagman output file", results)
205 def test_submit_failure(self):
206 with temporaryDirectory() as tmp_dir:
207 copy2(f"{TESTDIR}/data/bad_submit.dag.dagman.out", tmp_dir)
208 results = lssthtc.htc_check_dagman_output(tmp_dir)
209 self.assertIn("Warn: Job submission issues (last: ", results)
211 def test_tmpdir_abort(self):
212 with temporaryDirectory() as tmp_dir:
213 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir)
214 results = lssthtc.htc_check_dagman_output(tmp_dir)
215 self.assertIn("Cannot submit from /tmp", results)
217 def test_no_messages(self):
218 with temporaryDirectory() as tmp_dir:
219 copy2(f"{TESTDIR}/data/test_no_messages.dag.dagman.out", tmp_dir)
220 results = lssthtc.htc_check_dagman_output(tmp_dir)
221 self.assertEqual("", results)
223 def test_missing_warning(self):
224 with temporaryDirectory() as tmp_dir:
225 with open(os.path.join(tmp_dir, "test_missing_warning.dag.dagman.out"), "w") as fh:
226 print(
227 "01/01/26 01:01:01 ERROR: Warning is fatal error because of DAGMAN_USE_STRICT setting",
228 file=fh,
229 )
230 results = lssthtc.htc_check_dagman_output(tmp_dir)
231 self.assertIn("Missing warning", results)
234class SummarizeDagTestCase(unittest.TestCase):
235 """Test summarize_dag function."""
237 def test_no_dag_file(self):
238 with temporaryDirectory() as tmp_dir:
239 summary, job_name_to_pipetask, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
240 self.assertFalse(len(job_name_to_pipetask))
241 self.assertFalse(len(job_name_to_type))
242 self.assertFalse(summary)
244 def test_success(self):
245 with temporaryDirectory() as tmp_dir:
246 copy2(f"{TESTDIR}/data/good.dag", tmp_dir)
247 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
248 self.assertEqual(summary, "pipetaskInit:1;label1:1;label2:1;label3:1;finalJob:1")
249 self.assertEqual(
250 job_name_to_label,
251 {
252 "pipetaskInit": "pipetaskInit",
253 "0682f8f9-12f0-40a5-971e-8b30c7231e5c_label1_val1_val2": "label1",
254 "d0305e2d-f164-4a85-bd24-06afe6c84ed9_label2_val1_val2": "label2",
255 "2806ecc9-1bba-4362-8fff-ab4e6abb9f83_label3_val1_val2": "label3",
256 "finalJob": "finalJob",
257 },
258 )
259 self.assertEqual(
260 job_name_to_type,
261 {
262 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
263 "0682f8f9-12f0-40a5-971e-8b30c7231e5c_label1_val1_val2": lssthtc.WmsNodeType.PAYLOAD,
264 "d0305e2d-f164-4a85-bd24-06afe6c84ed9_label2_val1_val2": lssthtc.WmsNodeType.PAYLOAD,
265 "2806ecc9-1bba-4362-8fff-ab4e6abb9f83_label3_val1_val2": lssthtc.WmsNodeType.PAYLOAD,
266 "finalJob": lssthtc.WmsNodeType.FINAL,
267 },
268 )
270 def test_service(self):
271 with temporaryDirectory() as tmp_dir:
272 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag", tmp_dir)
273 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
274 self.assertEqual(summary, "pipetaskInit:1;label1:2;label2:2;finalJob:1")
275 self.assertEqual(
276 job_name_to_label,
277 {
278 "pipetaskInit": "pipetaskInit",
279 "057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b": "label1",
280 "4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a": "label1",
281 "40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b": "label2",
282 "696ee50d-e711-40d6-9caf-ee29ae4a656d_label2_val1a_val2a": "label2",
283 "finalJob": "finalJob",
284 "provisioningJob": "provisioningJob",
285 },
286 )
287 self.assertEqual(
288 job_name_to_type,
289 {
290 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
291 "057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
292 "4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
293 "40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
294 "696ee50d-e711-40d6-9caf-ee29ae4a656d_label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
295 "finalJob": lssthtc.WmsNodeType.FINAL,
296 "provisioningJob": lssthtc.WmsNodeType.SERVICE,
297 },
298 )
300 def test_noop(self):
301 with temporaryDirectory() as tmp_dir:
302 copy2(f"{TESTDIR}/data/noop_running_1/noop_running_1.dag", tmp_dir)
303 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(tmp_dir)
304 self.assertEqual(
305 set(summary.split(";")),
306 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"},
307 )
308 self.assertEqual(
309 job_name_to_label,
310 {
311 "label1_val1a_val2a": "label1",
312 "label1_val1a_val2b": "label1",
313 "label1_val1b_val2a": "label1",
314 "label1_val1b_val2b": "label1",
315 "label1_val1c_val2a": "label1",
316 "label1_val1c_val2b": "label1",
317 "label2_val1a_val2a": "label2",
318 "label2_val1a_val2b": "label2",
319 "label2_val1b_val2a": "label2",
320 "label2_val1b_val2b": "label2",
321 "label2_val1c_val2a": "label2",
322 "label2_val1c_val2b": "label2",
323 "label3_val1a_val2a": "label3",
324 "label3_val1a_val2b": "label3",
325 "label3_val1b_val2a": "label3",
326 "label3_val1b_val2b": "label3",
327 "label3_val1c_val2a": "label3",
328 "label3_val1c_val2b": "label3",
329 "label4_val1a_val2a": "label4",
330 "label4_val1a_val2b": "label4",
331 "label4_val1b_val2a": "label4",
332 "label4_val1b_val2b": "label4",
333 "label4_val1c_val2a": "label4",
334 "label4_val1c_val2b": "label4",
335 "label5_val1a_val2a": "label5",
336 "label5_val1a_val2b": "label5",
337 "label5_val1b_val2a": "label5",
338 "label5_val1b_val2b": "label5",
339 "label5_val1c_val2a": "label5",
340 "label5_val1c_val2b": "label5",
341 "finalJob": "finalJob",
342 "pipetaskInit": "pipetaskInit",
343 "wms_noop_order1_val1a": "order1",
344 "wms_noop_order1_val1b": "order1",
345 },
346 )
347 self.assertEqual(
348 job_name_to_type,
349 {
350 "label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
351 "label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
352 "label1_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
353 "label1_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
354 "label1_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
355 "label1_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
356 "label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
357 "label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
358 "label2_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
359 "label2_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
360 "label2_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
361 "label2_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
362 "label3_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
363 "label3_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
364 "label3_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
365 "label3_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
366 "label3_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
367 "label3_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
368 "label4_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
369 "label4_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
370 "label4_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
371 "label4_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
372 "label4_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
373 "label4_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
374 "label5_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
375 "label5_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
376 "label5_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
377 "label5_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
378 "label5_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
379 "label5_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
380 "finalJob": lssthtc.WmsNodeType.FINAL,
381 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
382 "wms_noop_order1_val1a": lssthtc.WmsNodeType.NOOP,
383 "wms_noop_order1_val1b": lssthtc.WmsNodeType.NOOP,
384 },
385 )
387 def test_subdags(self):
388 self.maxDiff = None
389 with temporaryDirectory() as tmp_dir:
390 submit_dir = os.path.join(tmp_dir, "group_running_1")
391 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
392 summary, job_name_to_label, job_name_to_type = lssthtc.summarize_dag(submit_dir)
393 self.assertEqual(
394 set(summary.split(";")),
395 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"},
396 )
398 self.assertEqual(
399 job_name_to_label,
400 {
401 "pipetaskInit": "pipetaskInit",
402 "label1_val1b_val2a": "label1",
403 "label1_val1c_val2a": "label1",
404 "label1_val1a_val2b": "label1",
405 "label1_val1b_val2b": "label1",
406 "label1_val1c_val2b": "label1",
407 "label1_val1a_val2a": "label1",
408 "label2_val1a_val2b": "label2",
409 "label2_val1a_val2a": "label2",
410 "label2_val1b_val2a": "label2",
411 "label2_val1b_val2b": "label2",
412 "label2_val1c_val2a": "label2",
413 "label2_val1c_val2b": "label2",
414 "label3_val1b_val2a": "label3",
415 "label3_val1c_val2a": "label3",
416 "label3_val1a_val2b": "label3",
417 "label3_val1b_val2b": "label3",
418 "label3_val1c_val2b": "label3",
419 "label3_val1a_val2a": "label3",
420 "label4_val1a_val2b": "label4",
421 "label4_val1a_val2a": "label4",
422 "label4_val1b_val2a": "label4",
423 "label4_val1b_val2b": "label4",
424 "label4_val1c_val2a": "label4",
425 "label4_val1c_val2b": "label4",
426 "label5_val1a_val2b": "label5",
427 "label5_val1a_val2a": "label5",
428 "label5_val1b_val2a": "label5",
429 "label5_val1b_val2b": "label5",
430 "label5_val1c_val2a": "label5",
431 "label5_val1c_val2b": "label5",
432 "finalJob": "finalJob",
433 "provisioningJob": "provisioningJob",
434 "wms_group_order1_val1a": "order1",
435 "wms_group_order1_val1b": "order1",
436 "wms_group_order1_val1c": "order1",
437 "wms_check_status_wms_group_order1_val1a": "order1",
438 "wms_check_status_wms_group_order1_val1b": "order1",
439 "wms_check_status_wms_group_order1_val1c": "order1",
440 },
441 )
443 self.assertEqual(
444 job_name_to_type,
445 {
446 "pipetaskInit": lssthtc.WmsNodeType.PAYLOAD,
447 "label1_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
448 "label1_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
449 "label1_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
450 "label1_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
451 "label1_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
452 "label1_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
453 "label2_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
454 "label2_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
455 "label2_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
456 "label2_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
457 "label2_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
458 "label2_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
459 "label3_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
460 "label3_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
461 "label3_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
462 "label3_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
463 "label3_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
464 "label3_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
465 "label4_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
466 "label4_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
467 "label4_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
468 "label4_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
469 "label4_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
470 "label4_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
471 "label5_val1a_val2b": lssthtc.WmsNodeType.PAYLOAD,
472 "label5_val1a_val2a": lssthtc.WmsNodeType.PAYLOAD,
473 "label5_val1b_val2a": lssthtc.WmsNodeType.PAYLOAD,
474 "label5_val1b_val2b": lssthtc.WmsNodeType.PAYLOAD,
475 "label5_val1c_val2a": lssthtc.WmsNodeType.PAYLOAD,
476 "label5_val1c_val2b": lssthtc.WmsNodeType.PAYLOAD,
477 "finalJob": lssthtc.WmsNodeType.FINAL,
478 "provisioningJob": lssthtc.WmsNodeType.SERVICE,
479 "wms_group_order1_val1a": lssthtc.WmsNodeType.SUBDAG,
480 "wms_group_order1_val1b": lssthtc.WmsNodeType.SUBDAG,
481 "wms_group_order1_val1c": lssthtc.WmsNodeType.SUBDAG,
482 "wms_check_status_wms_group_order1_val1a": lssthtc.WmsNodeType.SUBDAG_CHECK,
483 "wms_check_status_wms_group_order1_val1b": lssthtc.WmsNodeType.SUBDAG_CHECK,
484 "wms_check_status_wms_group_order1_val1c": lssthtc.WmsNodeType.SUBDAG_CHECK,
485 },
486 )
489class ReadDagNodesLogTestCase(unittest.TestCase):
490 """Test read_dag_nodes_log function."""
492 def setUp(self):
493 self.tmpdir = tempfile.mkdtemp()
495 def tearDown(self):
496 rmtree(self.tmpdir, ignore_errors=True)
498 def testFileMissing(self):
499 with self.assertRaisesRegex(FileNotFoundError, "DAGMan node log not found in"):
500 _ = lssthtc.read_dag_nodes_log(self.tmpdir)
502 def testRegular(self):
503 with temporaryDirectory() as tmp_dir:
504 submit_dir = os.path.join(tmp_dir, "tiny_problems")
505 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
506 results = lssthtc.read_dag_nodes_log(submit_dir)
507 self.assertEqual(results["9231.0"]["Cluster"], 9231)
508 self.assertEqual(results["9231.0"]["Proc"], 0)
509 self.assertEqual(results["9231.0"]["ReturnValue"], 1)
510 self.assertEqual(len(results), 6)
512 def testSubdags(self):
513 """Making sure it gets data from subdag dirs and doesn't
514 fail if some subdags haven't started running yet.
515 """
516 with temporaryDirectory() as tmp_dir:
517 submit_dir = os.path.join(tmp_dir, "group_running_1")
518 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
519 results = lssthtc.read_dag_nodes_log(submit_dir)
520 # main dag
521 self.assertEqual(results["10094.0"]["Cluster"], 10094)
522 # subdag
523 self.assertEqual(results["10112.0"]["Cluster"], 10112)
524 self.assertEqual(results["10116.0"]["Cluster"], 10116)
527class ReadNodeStatusTestCase(unittest.TestCase):
528 """Test read_node_status function."""
530 def setUp(self):
531 self.tmpdir = tempfile.mkdtemp()
533 def tearDown(self):
534 rmtree(self.tmpdir, ignore_errors=True)
536 def testServiceJobNotSubmitted(self):
537 # tiny_prov_no_submit files have successful workflow
538 # but provisioningJob could not submit.
539 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag.nodes.log", self.tmpdir)
540 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag.dagman.log", self.tmpdir)
541 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.node_status", self.tmpdir)
542 copy2(f"{TESTDIR}/data/tiny_prov_no_submit/tiny_prov_no_submit.dag", self.tmpdir)
544 jobs = lssthtc.read_node_status(self.tmpdir)
545 found = [
546 id_
547 for id_ in jobs
548 if jobs[id_].get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN) == lssthtc.WmsNodeType.SERVICE
549 ]
550 self.assertEqual(len(found), 1)
551 self.assertEqual(jobs[found[0]]["DAGNodeName"], "provisioningJob")
552 self.assertEqual(jobs[found[0]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY)
554 def testMissingStatusFile(self):
555 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag.nodes.log", self.tmpdir)
556 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag.dagman.log", self.tmpdir)
557 copy2(f"{TESTDIR}/data/tiny_problems/tiny_problems.dag", self.tmpdir)
559 jobs = lssthtc.read_node_status(self.tmpdir)
560 self.assertEqual(len(jobs), 7)
561 self.assertEqual(jobs["9230.0"]["DAGNodeName"], "pipetaskInit")
562 self.assertEqual(jobs["9230.0"]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD)
563 found = [
564 id_
565 for id_ in jobs
566 if jobs[id_].get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN) == lssthtc.WmsNodeType.SERVICE
567 ]
568 self.assertEqual(len(found), 1)
569 self.assertEqual(jobs[found[0]]["DAGNodeName"], "provisioningJob")
571 def testSubdagsRunning(self):
572 with temporaryDirectory() as tmp_dir:
573 test_tmp_dir = pathlib.Path(tmp_dir)
574 submit_dir = test_tmp_dir / "submit"
575 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
576 jobs = lssthtc.read_node_status(submit_dir)
577 self.assertEqual(len(jobs), 39) # includes non-payload jobs
578 # not guaranteed ids are same, so use names instead
579 job_name_to_id = {}
580 for id_, info in jobs.items():
581 job_name_to_id[info.get("DAGNodeName", id_)] = id_
582 job_type_to_names = {}
583 for id_, info in jobs.items():
584 job_type_to_names.setdefault(
585 info.get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN), set()
586 ).add(info.get("DAGNodeName", id_))
588 # check counts
589 self.assertNotIn(lssthtc.WmsNodeType.NOOP, job_type_to_names)
590 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.PAYLOAD]), 31)
591 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.FINAL]), 1)
592 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SERVICE]), 1)
593 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG]), 3)
594 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG_CHECK]), 3)
596 # spot check some statuses
597 self.assertEqual(
598 jobs[job_name_to_id["label3_val1a_val2b"]]["NodeStatus"], lssthtc.NodeStatus.DONE
599 )
600 self.assertEqual(
601 jobs[job_name_to_id["wms_group_order1_val1a"]]["NodeStatus"], lssthtc.NodeStatus.SUBMITTED
602 )
603 self.assertEqual(
604 jobs[job_name_to_id["label5_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY
605 )
606 self.assertEqual(
607 jobs[job_name_to_id["label2_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.DONE
608 )
610 def testSubdagsFailed(self):
611 with temporaryDirectory() as tmp_dir:
612 test_tmp_dir = pathlib.Path(tmp_dir)
613 submit_dir = test_tmp_dir / "submit"
614 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
615 jobs = lssthtc.read_node_status(submit_dir)
616 self.assertEqual(len(jobs), 39)
617 # not guaranteed ids are same, so use names instead
618 job_name_to_id = {}
619 for id_, info in jobs.items():
620 job_name_to_id[info.get("DAGNodeName", id_)] = id_
621 job_type_to_names = {}
622 for id_, info in jobs.items():
623 job_type_to_names.setdefault(
624 info.get("wms_node_type", lssthtc.WmsNodeType.UNKNOWN), set()
625 ).add(info.get("DAGNodeName", id_))
627 # check counts
628 self.assertNotIn(lssthtc.WmsNodeType.NOOP, job_type_to_names)
629 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.PAYLOAD]), 31)
630 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.FINAL]), 1)
631 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SERVICE]), 1)
632 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG]), 3)
633 self.assertEqual(len(job_type_to_names[lssthtc.WmsNodeType.SUBDAG_CHECK]), 3)
635 # spot check some statuses
636 self.assertEqual(
637 jobs[job_name_to_id["label3_val1a_val2b"]]["NodeStatus"], lssthtc.NodeStatus.DONE
638 )
639 self.assertEqual(
640 jobs[job_name_to_id["wms_group_order1_val1a"]]["NodeStatus"], lssthtc.NodeStatus.DONE
641 )
642 self.assertEqual(
643 jobs[job_name_to_id["label5_val1a_val2a"]]["NodeStatus"], lssthtc.NodeStatus.DONE
644 )
646 self.assertEqual(
647 jobs[job_name_to_id["label5_val1b_val2a"]]["NodeStatus"], lssthtc.NodeStatus.FUTILE
648 )
649 self.assertEqual(
650 jobs[job_name_to_id["wms_group_order1_val1b"]]["NodeStatus"], lssthtc.NodeStatus.DONE
651 )
652 self.assertEqual(
653 jobs[job_name_to_id["wms_check_status_wms_group_order1_val1b"]]["NodeStatus"],
654 lssthtc.NodeStatus.ERROR,
655 )
658class ReadSingleNodeStatusTestCase(unittest.TestCase):
659 """Test read_single_node_status function."""
661 def setUp(self):
662 self.tmpdir = tempfile.mkdtemp()
664 def tearDown(self):
665 rmtree(self.tmpdir, ignore_errors=True)
667 def _copyFiles(self, data_subdir, suffixes):
668 """Copy files with given suffixes from tests/data/<data_subdir>/."""
669 for suffix in suffixes:
670 copy2(f"{TESTDIR}/data/{data_subdir}/{data_subdir}{suffix}", self.tmpdir)
672 def _jobNameToId(self, jobs):
673 return {info["DAGNodeName"]: id_ for id_, info in jobs.items()}
675 def testAllDone(self):
676 self._copyFiles(
677 "tiny_success",
678 [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"],
679 )
680 filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status"
681 jobs = lssthtc.read_single_node_status(filename, -1)
683 self.assertEqual(len(jobs), 5)
684 name_to_id = self._jobNameToId(jobs)
686 # All four submitted nodes are marked DONE.
687 for name in [
688 "pipetaskInit",
689 "5bba27bd-8df7-4668-a9c5-e911192c5cdb_label1_val1_val2",
690 "0b225f1f-6edf-4380-b546-76c97947a88f_label2_val1_val2",
691 "finalJob",
692 ]:
693 self.assertIn(name, name_to_id, msg=f"Missing job {name}")
694 self.assertEqual(
695 jobs[name_to_id[name]]["NodeStatus"],
696 lssthtc.NodeStatus.DONE,
697 msg=f"Expected DONE for {name}",
698 )
700 # Service job not tracked by node_status; it came from the event log so
701 # it has a real positive ClusterId but no NodeStatus field.
702 self.assertIn("provisioningJob", name_to_id)
703 self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0)
705 # Spot-check labels and types.
706 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["bps_job_label"], "pipetaskInit")
707 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD)
708 self.assertEqual(jobs[name_to_id["finalJob"]]["wms_node_type"], lssthtc.WmsNodeType.FINAL)
709 self.assertEqual(jobs[name_to_id["provisioningJob"]]["wms_node_type"], lssthtc.WmsNodeType.SERVICE)
711 # DAGManJobID is populated from the dagman log for every job.
712 for job in jobs.values():
713 self.assertIn("DAGManJobID", job)
715 def testMixedStatuses(self):
716 self._copyFiles(
717 "tiny_problems",
718 [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"],
719 )
720 filename = pathlib.Path(self.tmpdir) / "tiny_problems.node_status"
721 jobs = lssthtc.read_single_node_status(filename, -1)
723 self.assertEqual(len(jobs), 7)
724 name_to_id = self._jobNameToId(jobs)
726 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE)
727 self.assertEqual(
728 jobs[name_to_id["057c8caf-66f6-4612-abf7-cdea5b666b1b_label1_val1a_val2b"]]["NodeStatus"],
729 lssthtc.NodeStatus.ERROR,
730 )
731 self.assertEqual(
732 jobs[name_to_id["4a7f478b-2e9b-435c-a730-afac3f621658_label1_val1a_val2a"]]["NodeStatus"],
733 lssthtc.NodeStatus.DONE,
734 )
735 self.assertEqual(
736 jobs[name_to_id["40040b97-606d-4997-98d3-e0493055fe7e_label2_val1a_val2b"]]["NodeStatus"],
737 lssthtc.NodeStatus.FUTILE,
738 )
739 self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.ERROR)
740 # Service job not tracked by node_status; came from event log so
741 # it has a real positive ClusterId but no NodeStatus field.
742 self.assertIn("provisioningJob", name_to_id)
743 self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0)
745 def testRunningWorkflow(self):
746 self._copyFiles(
747 "tiny_running",
748 [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"],
749 )
750 filename = pathlib.Path(self.tmpdir) / "tiny_running.node_status"
751 jobs = lssthtc.read_single_node_status(filename, -1)
753 self.assertEqual(len(jobs), 5)
754 name_to_id = self._jobNameToId(jobs)
756 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE)
757 self.assertEqual(
758 jobs[name_to_id["ca27ea57-c014-44c1-838a-78c06bc3ec1b_label1_val1_val2"]]["NodeStatus"],
759 lssthtc.NodeStatus.SUBMITTED,
760 )
761 self.assertEqual(
762 jobs[name_to_id["dbf919fa-5453-4b05-8806-ad6390fda0a3_label2_val1_val2"]]["NodeStatus"],
763 lssthtc.NodeStatus.NOT_READY,
764 )
765 self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY)
766 # Service job appeared in the event log; has a real positive ClusterId.
767 self.assertIn("provisioningJob", name_to_id)
768 self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0)
770 def testMissingNodeStatusFile(self):
771 # Omit the .node_status file; jobs must be built from the event log
772 # and dag.
773 self._copyFiles(
774 "tiny_problems",
775 [".dag", ".dag.dagman.log", ".dag.nodes.log"],
776 )
777 filename = pathlib.Path(self.tmpdir) / "tiny_problems.node_status"
778 jobs = lssthtc.read_single_node_status(filename, -1)
780 self.assertEqual(len(jobs), 7)
781 name_to_id = self._jobNameToId(jobs)
783 # Jobs that appeared in the event log have real (positive) cluster IDs.
784 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["DAGNodeName"], "pipetaskInit")
785 self.assertGreater(jobs[name_to_id["pipetaskInit"]]["ClusterId"], 0)
787 # The service job appeared in the event log and has a real positive ID.
788 self.assertGreater(jobs[name_to_id["provisioningJob"]]["ClusterId"], 0)
790 # All jobs carry the correct label and type from the dag file.
791 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["wms_node_type"], lssthtc.WmsNodeType.PAYLOAD)
792 self.assertEqual(jobs[name_to_id["provisioningJob"]]["wms_node_type"], lssthtc.WmsNodeType.SERVICE)
794 def testMissingLogFiles(self):
795 # Omit both log files; every job should get a fake negative ID.
796 self._copyFiles("tiny_success", [".dag", ".node_status"])
797 filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status"
798 jobs = lssthtc.read_single_node_status(filename, -1)
800 self.assertEqual(len(jobs), 5)
801 for job in jobs.values():
802 self.assertLess(job["ClusterId"], 0)
803 self.assertEqual(job["DAGManJobID"], lssthtc.MISSING_ID)
805 # NodeStatus values from the node_status file must still be preserved.
806 name_to_id = self._jobNameToId(jobs)
807 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE)
808 self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.DONE)
809 self.assertEqual(jobs[name_to_id["provisioningJob"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY)
811 def testPermissionLogFiles(self):
812 # Check what happens if can't read dag log file.
813 self._copyFiles("tiny_success", [".dag", ".node_status", ".dag.dagman.log"])
814 filename = os.path.join(self.tmpdir, "tiny_success.dag.dagman.log")
815 current_mode = os.stat(filename).st_mode
816 no_read_mode = current_mode & ~stat.S_IRUSR & ~stat.S_IRGRP & ~stat.S_IROTH
817 os.chmod(filename, no_read_mode)
819 filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status"
820 jobs = lssthtc.read_single_node_status(filename, -1)
822 self.assertEqual(len(jobs), 5)
823 for job in jobs.values():
824 self.assertLess(job["ClusterId"], 0)
825 self.assertEqual(job["DAGManJobID"], lssthtc.MISSING_ID)
827 # NodeStatus values from the node_status file must still be preserved.
828 name_to_id = self._jobNameToId(jobs)
829 self.assertEqual(jobs[name_to_id["pipetaskInit"]]["NodeStatus"], lssthtc.NodeStatus.DONE)
830 self.assertEqual(jobs[name_to_id["finalJob"]]["NodeStatus"], lssthtc.NodeStatus.DONE)
831 self.assertEqual(jobs[name_to_id["provisioningJob"]]["NodeStatus"], lssthtc.NodeStatus.NOT_READY)
833 def testInitFakeId(self):
834 # Verify fake IDs count down from the given starting value.
835 self._copyFiles("tiny_success", [".dag", ".node_status"])
836 filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status"
837 init_fake_id = -10
838 jobs = lssthtc.read_single_node_status(filename, init_fake_id)
840 self.assertEqual(len(jobs), 5)
841 cluster_ids = [job["ClusterId"] for job in jobs.values()]
842 # All IDs must be at most init_fake_id (i.e., -10 or lower).
843 for cid in cluster_ids:
844 self.assertLessEqual(cid, init_fake_id)
845 # All IDs must be unique.
846 self.assertEqual(len(set(cluster_ids)), len(cluster_ids))
848 def testFromDagJobAttribute(self):
849 self._copyFiles(
850 "tiny_success",
851 [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"],
852 )
853 filename = pathlib.Path(self.tmpdir) / "tiny_success.node_status"
854 jobs = lssthtc.read_single_node_status(filename, -1)
856 for id_, job in jobs.items():
857 self.assertEqual(
858 job["from_dag_job"],
859 "wms_tiny_success",
860 msg=f"Job {id_} has wrong from_dag_job",
861 )
863 def testServiceJobPlaceholder(self):
864 self._copyFiles(
865 "tiny_prov_no_submit",
866 [".dag", ".dag.dagman.log", ".dag.nodes.log", ".node_status"],
867 )
868 filename = pathlib.Path(self.tmpdir) / "tiny_prov_no_submit.node_status"
869 jobs = lssthtc.read_single_node_status(filename, -1)
871 service_jobs = [
872 (id_, info)
873 for id_, info in jobs.items()
874 if info.get("wms_node_type") == lssthtc.WmsNodeType.SERVICE
875 ]
876 self.assertEqual(len(service_jobs), 1)
877 service_id, service_job = service_jobs[0]
878 self.assertEqual(service_job["DAGNodeName"], "provisioningJob")
879 self.assertEqual(service_job["NodeStatus"], lssthtc.NodeStatus.NOT_READY)
880 self.assertLess(service_job["ClusterId"], 0)
883class HTCJobTestCase(unittest.TestCase):
884 """Test HTCJob methods."""
886 def testWriteDagCommandsPayload(self):
887 job = lssthtc.HTCJob(
888 "job1",
889 "label1",
890 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
891 {"dir": "jobs/label1"},
892 )
893 job.subfile = "job1.sub"
895 mockfh = io.StringIO()
896 job.write_dag_commands(mockfh, "../..")
897 self.assertIn('JOB job1 "job1.sub" DIR "../../jobs/label1"', mockfh.getvalue())
899 def testWriteDagCommandsNotJob(self):
900 # Testing giving command_name, no dag_rel_path and no dir
901 job = lssthtc.HTCJob(
902 "finalJob",
903 "finalJob",
904 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
905 )
906 job.subfile = "jobs/finalJob/finalJob.sub"
907 mockfh = io.StringIO()
908 job.write_dag_commands(mockfh, "", "FINAL")
909 self.assertIn('FINAL finalJob "jobs/finalJob/finalJob.sub"', mockfh.getvalue())
911 def testWriteDagCommandsNoop(self):
912 job = lssthtc.HTCJob("wms_noop_job1", "label1", {}, {"noop": True})
913 job.subfile = "notthere.sub"
914 mockfh = io.StringIO()
915 job.write_dag_commands(mockfh, "")
916 self.assertIn("NOOP", mockfh.getvalue())
918 def testWriteSubmitFile(self):
919 job = lssthtc.HTCJob(
920 "job1",
921 "label1",
922 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
923 )
924 with temporaryDirectory() as tmp_dir:
925 filename = pathlib.Path(tmp_dir) / "label1/job1.sub"
926 job.write_submit_file(filename.parent)
927 self.assertTrue(filename.exists())
928 # Try to make Submit object from file to find any syntax issues
929 _ = lssthtc.htc_create_submit_from_file(filename)
931 def testWriteSubmitFileExists(self):
932 job = lssthtc.HTCJob(
933 "job1",
934 "label1",
935 {"executable": "/bin/sleep", "arguments": "60", "log": "job1.log"},
936 )
937 with temporaryDirectory() as tmp_dir:
938 filename = pathlib.Path(tmp_dir) / "job1.sub"
939 job.subfile = filename
940 with open(filename, "w"):
941 pass # make empty file
942 job.write_submit_file(filename.parent)
943 # make sure didn't overwrite file
944 self.assertEqual(filename.stat().st_size, 0, "Incorrectly overwrote existing file")
947class HtcWriteJobCommands(unittest.TestCase):
948 """Test _htc_write_job_commands function."""
950 def testAllCommands(self):
951 dag_cmds = {
952 "pre": {
953 "defer": {"status": 1, "time": 120},
954 "debug": {"filename": "debug_pre.txt", "type": "ALL"},
955 "executable": "exec1",
956 "arguments": "arg1 arg2",
957 },
958 "post": {
959 "defer": {"status": 2, "time": 180},
960 "debug": {"filename": "debug_post.txt", "type": "ALL"},
961 "executable": "exec2",
962 "arguments": "arg3 arg4",
963 },
964 "vars": {"num": 8, "spaces": "a space"},
965 "pre_skip": "1",
966 "retry": 3,
967 "retry_unless_exit": 1,
968 "abort_dag_on": {"node_exit": 100, "abort_exit": 4},
969 "priority": 123,
970 }
972 truth = """SCRIPT DEFER 1 120 DEBUG debug_pre.txt ALL PRE job1 exec1 arg1 arg2
973SCRIPT DEFER 2 180 DEBUG debug_post.txt ALL POST job1 exec2 arg3 arg4
974VARS job1 num="8"
975VARS job1 spaces="a space"
976PRE_SKIP job1 1
977RETRY job1 3 UNLESS-EXIT 1
978ABORT-DAG-ON job1 100 RETURN 4
979PRIORITY job1 123
980"""
981 mockfh = io.StringIO()
982 lssthtc._htc_write_job_commands(mockfh, "job1", dag_cmds)
983 self.assertEqual(mockfh.getvalue(), truth)
985 def testPartialCommands(self):
986 # Trigger skipping the inner if clauses.
987 dag_cmds = {
988 "pre": {
989 "executable": "exec1",
990 },
991 "post": {
992 "executable": "exec2",
993 },
994 "vars": {"num": 8, "spaces": "a space"},
995 "pre_skip": "1",
996 "retry": 3,
997 }
999 truth = """SCRIPT PRE job1 exec1
1000SCRIPT POST job1 exec2
1001VARS job1 num="8"
1002VARS job1 spaces="a space"
1003PRE_SKIP job1 1
1004RETRY job1 3
1005"""
1006 mockfh = io.StringIO()
1007 lssthtc._htc_write_job_commands(mockfh, "job1", dag_cmds)
1008 self.assertEqual(mockfh.getvalue(), truth)
1010 def testNoCommands(self):
1011 dag_cmds = {}
1012 mockfh = io.StringIO()
1013 lssthtc._htc_write_job_commands(mockfh, "job2", dag_cmds)
1014 self.assertEqual(mockfh.getvalue(), "")
1016 def testFinal(self):
1017 self.maxDiff = None
1018 dag_cmds = {
1019 "pre": {
1020 "defer": {"status": 1, "time": 120},
1021 "debug": {"filename": "debug_pre.txt", "type": "ALL"},
1022 "executable": "exec1",
1023 "arguments": "arg1 arg2",
1024 },
1025 "post": {
1026 "defer": {"status": 2, "time": 180},
1027 "debug": {"filename": "debug_post.txt", "type": "ALL"},
1028 "executable": "exec2",
1029 "arguments": "arg3 arg4",
1030 },
1031 "vars": {"num": 8, "spaces": "a space"},
1032 "pre_skip": "1",
1033 "retry": 3,
1034 "retry_unless_exit": 1,
1035 "abort_dag_on": {"node_exit": 100, "abort_exit": 4},
1036 "priority": 123,
1037 }
1039 truth = """SCRIPT DEFER 1 120 DEBUG debug_pre.txt ALL PRE finalJob exec1 arg1 arg2
1040SCRIPT DEFER 2 180 DEBUG debug_post.txt ALL POST finalJob exec2 arg3 arg4
1041VARS finalJob num="8"
1042VARS finalJob spaces="a space"
1043PRE_SKIP finalJob 1
1044"""
1045 mockfh = io.StringIO()
1046 lssthtc._htc_write_job_commands(mockfh, "finalJob", dag_cmds, "FINAL")
1047 self.assertEqual(mockfh.getvalue(), truth)
1050class HTCBackupFilesSinglePathTestCase(unittest.TestCase):
1051 """Test htc_backup_files_single_path function."""
1053 def testSrcDestSame(self):
1054 with temporaryDirectory() as tmp_dir:
1055 with self.assertRaisesRegex(
1056 RuntimeError, "Destination directory is same as the source directory"
1057 ):
1058 lssthtc.htc_backup_files_single_path(tmp_dir, tmp_dir)
1060 def testSuccess(self):
1061 with temporaryDirectory() as tmp_dir:
1062 test_tmp_dir = pathlib.Path(tmp_dir)
1063 submit_dir = test_tmp_dir / "the_src_dir"
1064 copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1065 backup_dir = test_tmp_dir / "the_dest_dir"
1066 backup_dir.mkdir()
1067 lssthtc.htc_backup_files_single_path(submit_dir, backup_dir)
1068 result_submit = []
1069 for root, _, files in os.walk(submit_dir):
1070 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
1071 self.assertEqual(
1072 set(result_submit),
1073 {
1074 "./tiny_success.dag.dagman.log",
1075 "./tiny_success.dag.dagman.out",
1076 "./tiny_success.dag",
1077 },
1078 )
1079 result_backup = []
1080 for root, _, files in os.walk(backup_dir):
1081 result_backup.extend([str(os.path.join(os.path.relpath(root, backup_dir), f)) for f in files])
1082 self.assertEqual(
1083 set(result_backup),
1084 {
1085 "./tiny_success.info.json",
1086 "./tiny_success.dag.metrics",
1087 "./tiny_success.dag.nodes.log",
1088 "./tiny_success.node_status",
1089 },
1090 )
1093class HTCBackupFilesTestCase(unittest.TestCase):
1094 """Test htc_backup_files function."""
1096 def testDirectoryNotFound(self):
1097 with temporaryDirectory() as tmp_dir:
1098 test_tmp_dir = pathlib.Path(tmp_dir)
1099 submit_dir = test_tmp_dir / "submit"
1100 with self.assertRaises(FileNotFoundError):
1101 lssthtc.htc_backup_files(submit_dir)
1103 def testSuccess(self):
1104 with temporaryDirectory() as tmp_dir:
1105 test_tmp_dir = pathlib.Path(tmp_dir)
1106 submit_dir = test_tmp_dir / "submit"
1107 copytree(f"{TESTDIR}/data/tiny_success", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1108 lssthtc.htc_backup_files(submit_dir)
1109 result_submit = []
1110 for root, _, files in os.walk(submit_dir):
1111 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
1112 self.assertEqual(
1113 set(result_submit),
1114 {
1115 "./tiny_success.dag.dagman.log",
1116 "./tiny_success.dag.dagman.out",
1117 "./tiny_success.dag",
1118 "000/tiny_success.info.json",
1119 "000/tiny_success.dag.metrics",
1120 "000/tiny_success.dag.nodes.log",
1121 "000/tiny_success.node_status",
1122 },
1123 )
1125 def testDestNotInSubmitDir(self):
1126 with temporaryDirectory() as tmp_dir:
1127 test_tmp_dir = pathlib.Path(tmp_dir)
1128 submit_dir = test_tmp_dir / "submit"
1129 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1130 with self.assertLogs("lsst.ctrl.bps.htcondor", level="WARNING") as cm:
1131 lssthtc.htc_backup_files(submit_dir, test_tmp_dir / "backup")
1132 self.assertIn("Invalid backup location:", cm.output[-1])
1133 lssthtc.htc_backup_files(submit_dir)
1134 result_submit = []
1135 for root, _, files in os.walk(submit_dir):
1136 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
1137 self.assertEqual(
1138 set(result_submit),
1139 {
1140 "./tiny_problems.dag.dagman.log",
1141 "./tiny_problems.dag.dagman.out",
1142 "./tiny_problems.dag",
1143 "./tiny_problems.dag.rescue001",
1144 "001/tiny_problems.info.json",
1145 "001/tiny_problems.dag.metrics",
1146 "001/tiny_problems.dag.nodes.log",
1147 "001/tiny_problems.node_status",
1148 },
1149 )
1151 def testDestInSubmitDir(self):
1152 with temporaryDirectory() as tmp_dir:
1153 test_tmp_dir = pathlib.Path(tmp_dir)
1154 submit_dir = test_tmp_dir / "submit"
1155 backup_dir = submit_dir / "subdir"
1156 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1157 lssthtc.htc_backup_files(submit_dir, backup_dir)
1158 result_submit = []
1159 for root, _, files in os.walk(submit_dir):
1160 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
1161 self.assertEqual(
1162 set(result_submit),
1163 {
1164 "./tiny_problems.dag.dagman.log",
1165 "./tiny_problems.dag.dagman.out",
1166 "./tiny_problems.dag",
1167 "./tiny_problems.dag.rescue001",
1168 "subdir/001/tiny_problems.info.json",
1169 "subdir/001/tiny_problems.dag.metrics",
1170 "subdir/001/tiny_problems.dag.nodes.log",
1171 "subdir/001/tiny_problems.node_status",
1172 },
1173 )
1175 def testRelativeSubdir(self):
1176 with temporaryDirectory() as tmp_dir:
1177 test_tmp_dir = pathlib.Path(tmp_dir)
1178 submit_dir = test_tmp_dir / "submit"
1179 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1180 lssthtc.htc_backup_files(submit_dir, "reldir")
1181 result_submit = []
1182 for root, _, files in os.walk(submit_dir):
1183 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
1184 self.assertEqual(
1185 set(result_submit),
1186 {
1187 "./tiny_problems.dag.dagman.log",
1188 "./tiny_problems.dag.dagman.out",
1189 "./tiny_problems.dag",
1190 "./tiny_problems.dag.rescue001",
1191 "reldir/001/tiny_problems.info.json",
1192 "reldir/001/tiny_problems.dag.metrics",
1193 "reldir/001/tiny_problems.dag.nodes.log",
1194 "reldir/001/tiny_problems.node_status",
1195 },
1196 )
1198 def testSubdags(self):
1199 with temporaryDirectory() as tmp_dir:
1200 test_tmp_dir = pathlib.Path(tmp_dir)
1201 submit_dir = test_tmp_dir / "submit"
1202 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1203 lssthtc.htc_backup_files(submit_dir)
1204 result_submit = []
1205 for root, _, files in os.walk(submit_dir):
1206 result_submit.extend([str(os.path.join(os.path.relpath(root, submit_dir), f)) for f in files])
1207 self.assertEqual(
1208 set(result_submit),
1209 {
1210 "./group_failed_1.dag",
1211 "./group_failed_1.dag.dagman.log",
1212 "./group_failed_1.dag.dagman.out",
1213 "./group_failed_1.dag.rescue001",
1214 "subdags/wms_group_order1_val1a/group_order1_val1a.dag",
1215 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.log",
1216 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.dagman.out",
1217 "subdags/wms_group_order1_val1a/group_order1_val1a.dag.nodes.log",
1218 "subdags/wms_group_order1_val1a/group_order1_val1a.node_status",
1219 "subdags/wms_group_order1_val1a/wms_group_order1_val1a.dag.post.out",
1220 "subdags/wms_group_order1_val1a/wms_group_order1_val1a.status.txt",
1221 "subdags/wms_group_order1_val1b/group_order1_val1b.dag",
1222 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.log",
1223 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.dagman.out",
1224 "subdags/wms_group_order1_val1b/group_order1_val1b.dag.rescue001",
1225 "subdags/wms_group_order1_val1c/group_order1_val1c.dag",
1226 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.log",
1227 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.dagman.out",
1228 "subdags/wms_group_order1_val1c/group_order1_val1c.dag.nodes.log",
1229 "subdags/wms_group_order1_val1c/group_order1_val1c.node_status",
1230 "subdags/wms_group_order1_val1c/wms_group_order1_val1c.dag.post.out",
1231 "subdags/wms_group_order1_val1c/wms_group_order1_val1c.status.txt",
1232 "001/group_failed_1.dag.nodes.log",
1233 "001/group_failed_1.info.json",
1234 "001/group_failed_1.node_status",
1235 "001/subdags/wms_group_order1_val1b/group_order1_val1b.dag.nodes.log",
1236 "001/subdags/wms_group_order1_val1b/group_order1_val1b.node_status",
1237 "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.status.txt",
1238 "001/subdags/wms_group_order1_val1b/wms_group_order1_val1b.dag.post.out",
1239 },
1240 )
1243class UpdateRescueFileTestCase(unittest.TestCase):
1244 """Test _update_rescue_file function."""
1246 def testSuccess(self):
1247 self.maxDiff = None
1248 with temporaryDirectory() as tmp_dir:
1249 test_tmp_dir = pathlib.Path(tmp_dir)
1250 submit_dir = test_tmp_dir / "submit"
1251 copytree(f"{TESTDIR}/data/group_failed_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1252 rescue_file = submit_dir / "group_failed_1.dag.rescue001"
1253 failed_subdags = lssthtc._update_rescue_file(rescue_file)
1254 self.assertEqual(set(failed_subdags), {"wms_group_order1_val1b"})
1255 with open(rescue_file) as fh:
1256 lines = fh.readlines()
1257 results = "".join(lines)
1259 truth = """# Rescue DAG file, created after running
1260# the u_testuser_DM-46294_group_fail_20250310T160455Z.dag DAG file
1261# Created 3/10/2025 16:08:56 UTC
1262# Rescue DAG version: 2.0.1 (partial)
1263#
1264# Total number of Nodes: 26
1265# Nodes premarked DONE: 21
1266# Nodes that failed: 2
1267# wms_group_order1_val1b,finalJob,<ENDLIST>
1269DONE pipetaskInit
1270DONE label1_val1c_val2a
1271DONE label1_val1b_val2b
1272DONE label1_val1b_val2a
1273DONE label1_val1c_val2b
1274DONE label1_val1a_val2a
1275DONE label1_val1a_val2b
1276DONE label3_val1c_val2a
1277DONE label3_val1b_val2b
1278DONE label3_val1b_val2a
1279DONE label3_val1c_val2b
1280DONE label3_val1a_val2a
1281DONE label3_val1a_val2b
1282DONE wms_group_order1_val1a
1283DONE label5_val1a_val2a
1284DONE label5_val1a_val2b
1285DONE wms_group_order1_val1c
1286DONE label5_val1c_val2a
1287DONE label5_val1c_val2b
1288DONE wms_check_status_wms_group_order1_val1a
1289DONE wms_check_status_wms_group_order1_val1c
1290"""
1292 self.assertEqual(results, truth)
1295class ReadRescueHeadersTestCase(unittest.TestCase):
1296 """Test _read_rescue_headers function."""
1298 def testTypical(self):
1299 content = "# Header line 1\n# Header line 2\n\nDONE somenode\n"
1300 result = lssthtc._read_rescue_headers(io.StringIO(content))
1301 self.assertEqual(result, ["# Header line 1", "# Header line 2"])
1303 def testEmptyFile(self):
1304 result = lssthtc._read_rescue_headers(io.StringIO(""))
1305 self.assertEqual(result, [])
1307 def testOnlyHeaderLines(self):
1308 content = "# Line 1\n# Line 2\n# Line 3\n"
1309 result = lssthtc._read_rescue_headers(io.StringIO(content))
1310 self.assertEqual(result, ["# Line 1", "# Line 2", "# Line 3"])
1312 def testFirstLineNotComment(self):
1313 content = "DONE somenode\n# Header\n"
1314 result = lssthtc._read_rescue_headers(io.StringIO(content))
1315 self.assertEqual(result, [])
1317 def testWhitespaceStripped(self):
1318 content = " # Header line 1 \n # Header line 2 \n\n"
1319 result = lssthtc._read_rescue_headers(io.StringIO(content))
1320 self.assertEqual(result, ["# Header line 1", "# Header line 2"])
1323class WriteRescueHeadersTestCase(unittest.TestCase):
1324 """Test _write_rescue_headers function."""
1326 def testTypical(self):
1327 header_lines = ["# Header line 1", "# Header line 2", "# Header line 3"]
1328 outfh = io.StringIO()
1329 lssthtc._write_rescue_headers(header_lines, outfh)
1330 self.assertEqual(outfh.getvalue(), "# Header line 1\n# Header line 2\n# Header line 3\n\n")
1332 def testEmptyList(self):
1333 outfh = io.StringIO()
1334 lssthtc._write_rescue_headers([], outfh)
1335 self.assertEqual(outfh.getvalue(), "\n")
1337 def testSingleLine(self):
1338 outfh = io.StringIO()
1339 lssthtc._write_rescue_headers(["# Only line"], outfh)
1340 self.assertEqual(outfh.getvalue(), "# Only line\n\n")
1343class UpdateRescueHeadersTestCase(unittest.TestCase):
1344 """Test _update_rescue_headers function."""
1346 def testWithFailedSubdag(self):
1347 header_lines = [
1348 "# Total number of Nodes: 26",
1349 "# Nodes premarked DONE: 22",
1350 "# Nodes that failed: 2",
1351 "# wms_check_status_wms_group_order1_val1b,finalJob,<ENDLIST>",
1352 ]
1353 result = lssthtc._update_rescue_headers(header_lines)
1354 self.assertEqual(result, ["wms_group_order1_val1b"])
1355 self.assertEqual(header_lines[1], "# Nodes premarked DONE: 21")
1356 self.assertEqual(header_lines[3], "# wms_group_order1_val1b,finalJob,<ENDLIST>")
1358 def testNoSubdagFailures(self):
1359 header_lines = [
1360 "# Nodes premarked DONE: 5",
1361 "# Nodes that failed: 1",
1362 "# finalJob,<ENDLIST>",
1363 ]
1364 result = lssthtc._update_rescue_headers(header_lines)
1365 self.assertEqual(result, [])
1366 self.assertEqual(header_lines[0], "# Nodes premarked DONE: 5")
1367 self.assertEqual(header_lines[2], "# finalJob,<ENDLIST>")
1369 def testMultipleFailedSubdags(self):
1370 header_lines = [
1371 "# Nodes premarked DONE: 10",
1372 "# Nodes that failed: 3",
1373 "# wms_check_status_subdag_a,wms_check_status_subdag_b,finalJob,<ENDLIST>",
1374 ]
1375 result = lssthtc._update_rescue_headers(header_lines)
1376 self.assertEqual(result, ["subdag_a", "subdag_b"])
1377 self.assertEqual(header_lines[0], "# Nodes premarked DONE: 8")
1378 self.assertEqual(header_lines[2], "# subdag_a,subdag_b,finalJob,<ENDLIST>")
1380 def testNoFailedNodesLine(self):
1381 header_lines = [
1382 "# Total number of Nodes: 5",
1383 "# Nodes premarked DONE: 5",
1384 ]
1385 original = list(header_lines)
1386 result = lssthtc._update_rescue_headers(header_lines)
1387 self.assertEqual(result, [])
1388 self.assertEqual(header_lines, original)
1390 def testEmptyHeader(self):
1391 result = lssthtc._update_rescue_headers([])
1392 self.assertEqual(result, [])
1395class ReadDagStatusTestCase(unittest.TestCase):
1396 """Test read_dag_status function and read_single_dag_status."""
1398 def testFileMissing(self):
1399 with temporaryDirectory() as tmp_dir:
1400 with self.assertRaisesRegex(FileNotFoundError, "DAGMan node status not found"):
1401 _ = lssthtc.read_dag_status(tmp_dir)
1403 def testRegular(self):
1404 with temporaryDirectory() as tmp_dir:
1405 submit_dir = os.path.join(tmp_dir, "tiny_problems")
1406 copytree(f"{TESTDIR}/data/tiny_problems", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1407 results = lssthtc.read_dag_status(submit_dir)
1408 truth = {
1409 "JobProcsHeld": 0,
1410 "NodesPost": 0,
1411 "JobProcsIdle": 0,
1412 "NodesTotal": 6,
1413 "NodesFailed": 2,
1414 "NodesDone": 3,
1415 "NodesQueued": 0,
1416 "NodesPre": 0,
1417 "NodesFutile": 1,
1418 "NodesUnready": 0,
1419 }
1420 self.assertEqual(results, results | truth)
1422 def testSubdags(self):
1423 """Making sure it gets data from subdag dirs and doesn't
1424 fail if some subdags haven't started running yet.
1425 """
1426 self.maxDiff = None
1427 with temporaryDirectory() as tmp_dir:
1428 submit_dir = os.path.join(tmp_dir, "submit")
1429 copytree(f"{TESTDIR}/data/group_running_1", submit_dir, ignore=ignore_patterns("*~", ".???*"))
1430 results = lssthtc.read_dag_status(submit_dir)
1431 truth = {
1432 "JobProcsHeld": 0,
1433 "NodesPost": 0,
1434 "JobProcsIdle": 0,
1435 "NodesTotal": 34,
1436 "NodesFailed": 0,
1437 "NodesDone": 17,
1438 "NodesQueued": 3,
1439 "NodesPre": 0,
1440 "NodesFutile": 0,
1441 "NodesUnready": 14,
1442 }
1443 self.assertEqual(results, results | truth)
1446class ReadDagInfoTestCase(unittest.TestCase):
1447 """Test read_dag_info function."""
1449 def testFileMissing(self):
1450 with temporaryDirectory() as tmp_dir:
1451 with self.assertRaisesRegex(FileNotFoundError, "File with DAGMan job information not found in "):
1452 _ = lssthtc.read_dag_info(tmp_dir)
1454 def testSuccess(self):
1455 with temporaryDirectory() as tmp_dir:
1456 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir)
1457 filename, results = lssthtc.read_dag_info(tmp_dir)
1458 self.assertIn("info.json", str(filename))
1459 self.assertTrue(pathlib.Path(filename).is_file())
1461 truth = {
1462 "test02": {
1463 "9208.0": {
1464 "ClusterId": 9208,
1465 "GlobalJobId": "test02#9208.0#1739465078",
1466 "bps_wms_service": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorService",
1467 "bps_project": "dev",
1468 "bps_payload": "tiny",
1469 "bps_operator": "testuser",
1470 "bps_wms_workflow": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorWorkflow",
1471 "bps_provisioning_job": "provisioningJob",
1472 "bps_run_quanta": "label1:1;label2:1",
1473 "bps_campaign": "quick",
1474 "bps_runsite": "testpool",
1475 "bps_job_summary": "pipetaskInit:1;label1:1;label2:1;finalJob:1",
1476 "bps_run": "u_testuser_tiny_20250213T164427Z",
1477 "bps_isjob": "True",
1478 }
1479 }
1480 }
1482 self.assertEqual(results, truth)
1484 def testPermissionError(self):
1485 with temporaryDirectory() as tmp_dir:
1486 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.info.json", tmp_dir)
1487 with patch("lsst.ctrl.bps.htcondor.lssthtc.open") as mocked_open:
1488 mocked_open.side_effect = PermissionError
1489 with self.assertLogs("lsst.ctrl.bps.htcondor", level="DEBUG") as cm:
1490 _, results = lssthtc.read_dag_info(tmp_dir)
1491 self.assertIn("Retrieving DAGMan job information failed:", cm.output[-1])
1492 self.assertEqual({}, results)
1495class HtcWriteCondorFileTestCase(unittest.TestCase):
1496 """Test htc_write_condor_file function."""
1498 def testSuccess(self):
1499 with temporaryDirectory() as tmp_dir:
1500 job_name = "job1"
1501 filename = pathlib.Path(tmp_dir) / f"label1/{job_name}.sub"
1502 job = {
1503 "executable": "$(CTRL_MPEXEC_DIR)/bin/pipetask",
1504 "arguments": "-a -b 2 -c",
1505 "request_memory": "2000",
1506 "environment": "one=1 two=\"2\" three='spacey 'quoted' value'",
1507 "log": f"{job_name}.log",
1508 }
1509 job_attrs = {
1510 "bps_job_name": job_name,
1511 "bps_job_label": "label1",
1512 "bps_job_quanta": "task1:8;task2:8",
1513 }
1514 expected = [
1515 "executable=$(CTRL_MPEXEC_DIR)/bin/pipetask\n",
1516 'arguments="-a -b 2 -c"\n',
1517 "request_memory=2000\n",
1518 "environment=\"one=1 two=\"2\" three='spacey 'quoted' value'\"\n",
1519 f"output={job_name}.$(Cluster).out\n",
1520 f"error={job_name}.$(Cluster).out\n",
1521 f"log={job_name}.log\n",
1522 f'+bps_job_name = "{job_name}"\n',
1523 '+bps_job_label = "label1"\n',
1524 '+bps_job_quanta = "task1:8;task2:8"\n',
1525 "queue\n",
1526 ]
1528 lssthtc.htc_write_condor_file(filename, job_name, job, job_attrs)
1529 with open(filename, encoding="utf-8") as f:
1530 actual = f.readlines()
1532 self.assertEqual(set(actual), set(expected))
1533 self.assertTrue(filename.exists())
1534 # Try to make Submit object from file to find any syntax issues
1535 _ = lssthtc.htc_create_submit_from_file(filename)
1538class HtcCreateSubmitFromDagTestCase(unittest.TestCase):
1539 """Test htc_create_submit_from_dag function."""
1541 @classmethod
1542 def setUpClass(cls):
1543 cls.bindir = None
1544 # htcondor.Submit.from_dag requires condor_dagman executable in path.
1545 if not which("condor_dagman"): # pragma: no cover
1546 cls.bindir = tempfile.TemporaryDirectory()
1547 fake_dagman_exec = pathlib.Path(cls.bindir.name) / "condor_dagman"
1548 with open(fake_dagman_exec, "w") as fh:
1549 print("#!/bin/bash", file=fh)
1550 print("echo fake_condor_dagman $@", file=fh)
1551 print("exit 0", file=fh)
1552 fake_dagman_exec.chmod(fake_dagman_exec.stat().st_mode | stat.S_IEXEC)
1553 os.environ["PATH"] = f"{os.environ['PATH']}:{cls.bindir.name}"
1555 @classmethod
1556 def tearDownClass(cls):
1557 if cls.bindir: 1557 ↛ 1558line 1557 didn't jump to line 1558 because the condition on line 1557 was never true
1558 cls.bindir.cleanup()
1560 @patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})
1561 def testMaxIdleEnvVar(self):
1562 with temporaryDirectory() as tmp_dir:
1563 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1564 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1565 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})
1566 self.assertIn("-MaxIdle 42", submit["arguments"])
1568 @patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})
1569 def testMaxIdleInDAGManConfig(self):
1570 with temporaryDirectory() as tmp_dir:
1571 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1572 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1573 config_filename = pathlib.Path(tmp_dir) / "dagman.conf"
1574 with open(config_filename, "w") as fh:
1575 print("DAGMAN_MAX_JOBS_IDLE = 300", file=fh)
1576 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {}, config_filename)
1577 self.assertIn("-MaxIdle 300", submit["arguments"])
1579 @patch.dict(os.environ, {"_CONDOR_DAGMAN_MAX_JOBS_IDLE": "42"})
1580 def testMaxIdleNotInDAGManConfig(self):
1581 with temporaryDirectory() as tmp_dir:
1582 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1583 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1584 config_filename = pathlib.Path(tmp_dir) / "dagman.conf"
1585 with open(config_filename, "w") as fh:
1586 print("DAGMAN_MAX_JOBS_SUBMITTED = 300", file=fh)
1587 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {}, config_filename)
1588 self.assertIn("-MaxIdle 42", submit["arguments"])
1590 @patch.dict(os.environ, {})
1591 def testMaxIdleGiven(self):
1592 with temporaryDirectory() as tmp_dir:
1593 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1594 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1595 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {"MaxIdle": 37})
1596 self.assertIn("-MaxIdle 37", submit["arguments"])
1598 @patch.dict(os.environ, {})
1599 def testMaxJobsIdleParam(self):
1600 def _fake_params_contains(key):
1601 if key == "DAGMAN_MAX_JOBS_IDLE":
1602 return True
1603 return False # pragma: no cover
1605 def _fake_params_get(key):
1606 if key == "DAGMAN_MAX_JOBS_IDLE":
1607 return 16
1608 return "FAKE_VAL" # pragma: no cover
1610 with temporaryDirectory() as tmp_dir:
1611 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1612 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1613 with patch("lsst.ctrl.bps.htcondor.lssthtc.param") as mock_param:
1614 mock_param.__contains__.side_effect = _fake_params_contains
1615 mock_param.__getitem__.side_effect = _fake_params_get
1616 submit = lssthtc.htc_create_submit_from_dag(str(dag_filename), {}, None)
1617 self.assertIn("-MaxIdle 16", submit["arguments"])
1619 @patch.dict(os.environ, {})
1620 def testNoMaxJobsIdle(self):
1621 """Note: Since the produced arguments differ depending on
1622 HTCondor version when no MaxIdle passed to from_dag, not
1623 checking arguments string here. Instead just making sure
1624 lssthtc code doesn't pass MaxIdle value to from_dag.
1625 """
1626 with temporaryDirectory() as tmp_dir:
1627 copy2(f"{TESTDIR}/data/tiny_success/tiny_success.dag", tmp_dir)
1628 dag_filename = pathlib.Path(tmp_dir) / "tiny_success.dag"
1629 with patch("lsst.ctrl.bps.htcondor.lssthtc.Submit.from_dag") as submit_mock:
1630 with patch("lsst.ctrl.bps.htcondor.lssthtc.param") as mock_param:
1631 mock_param.__contains__.return_value = False
1632 _ = lssthtc.htc_create_submit_from_dag(str(dag_filename), {})
1633 submit_mock.assert_called_once_with(str(dag_filename), {})
1636class HtcDagTestCase(unittest.TestCase):
1637 """Test for HTCDag class."""
1639 def setUp(self):
1640 job = lssthtc.HTCJob(name="test_job")
1641 job.add_job_cmds(
1642 {
1643 "executable": "/usr/bin/echo",
1644 "arguments": "foo",
1645 "output": "test_job.$(Cluster).out",
1646 "error": "test_job.$(Cluster).out",
1647 "log": "test_job.$(Cluster).log",
1648 }
1649 )
1650 job.subfile = f"{job.name}.sub"
1652 self.dag = lssthtc.HTCDag(name="test_workflow")
1653 self.dag.add_job(job)
1655 self.subfile_expected = [
1656 "executable=/usr/bin/echo\n",
1657 'arguments="foo"\n',
1658 "output=test_job.$(Cluster).out\n",
1659 "error=test_job.$(Cluster).out\n",
1660 "log=test_job.$(Cluster).log\n",
1661 "queue\n",
1662 ]
1664 def tearDown(self):
1665 pass
1667 def testWriteWithDagConfig(self):
1668 with temporaryDirectory() as tmp_dir:
1669 config = BpsConfig(Config(htcondor_config.HTC_DEFAULTS_URI))
1670 job = self.dag.nodes["test_job"]["data"]
1671 wms_config_filename = "dagman.conf"
1672 wms_configurator = dagman_configurator.DagmanConfigurator(config)
1673 wms_configurator.prepare(wms_config_filename, prefix=tmp_dir)
1674 wms_configurator.configure(self.dag)
1675 dagfile_expected = [
1676 f"CONFIG {wms_config_filename}\n",
1677 f'JOB {job.name} "{job.subfile}"\n',
1678 f"NODE_STATUS_FILE {self.dag.name}.node_status\n",
1679 f'SET_JOB_ATTR bps_wms_config_path= "{wms_config_filename}"\n',
1680 ]
1682 self.dag.write(tmp_dir, "", "")
1684 self.assertIn("submit_path", self.dag.graph)
1685 self.assertEqual(self.dag.graph["submit_path"], tmp_dir)
1686 self.assertIn("dag_filename", self.dag.graph)
1687 self.assertEqual(self.dag.graph["dag_filename"], f"{self.dag.graph['name']}.dag")
1688 with open(os.path.join(tmp_dir, self.dag.graph["dag_filename"]), encoding="utf-8") as f:
1689 dagfile_actual = f.readlines()
1690 self.assertEqual(dagfile_actual, dagfile_expected)
1691 with open(os.path.join(tmp_dir, job.subfile), encoding="utf-8") as f:
1692 subfile_actual = f.readlines()
1693 self.assertEqual(subfile_actual, self.subfile_expected)
1695 def testWriteWithoutDagConfig(self):
1696 with temporaryDirectory() as tmp_dir:
1697 job = self.dag.nodes["test_job"]["data"]
1698 dagfile_expected = [
1699 f'JOB {job.name} "{job.subfile}"\n',
1700 f"NODE_STATUS_FILE {self.dag.name}.node_status\n",
1701 ]
1703 self.dag.write(tmp_dir, "", "")
1705 self.assertIn("submit_path", self.dag.graph)
1706 self.assertEqual(self.dag.graph["submit_path"], tmp_dir)
1707 self.assertIn("dag_filename", self.dag.graph)
1708 self.assertEqual(self.dag.graph["dag_filename"], f"{self.dag.graph['name']}.dag")
1709 with open(os.path.join(tmp_dir, self.dag.graph["dag_filename"]), encoding="utf-8") as f:
1710 dagfile_actual = f.readlines()
1711 self.assertEqual(dagfile_actual, dagfile_expected)
1712 with open(os.path.join(tmp_dir, job.subfile), encoding="utf-8") as f:
1713 subfile_actual = f.readlines()
1714 self.assertEqual(subfile_actual, self.subfile_expected)
1716 def testWriteLazySubdag(self):
1717 self.maxDiff = None
1718 dag, truth_files = make_lazy_dag("test1", True)
1719 dag.graph["write_dot"] = True
1720 with temporaryDirectory() as tmp_dir:
1721 dag.write(tmp_dir, "", "")
1722 with open(os.path.join(tmp_dir, dag.graph["dag_filename"]), encoding="utf-8") as f:
1723 dagfile_actual = f.readlines()
1724 self.assertIn("DOT test1.dot\n", dagfile_actual)
1726 all_files = []
1727 for root, _, files in pathlib.Path(tmp_dir).walk():
1728 relroot = pathlib.Path(root).relative_to(tmp_dir)
1729 all_files.extend([str(relroot / f) for f in files])
1730 self.assertEqual(sorted(all_files), sorted(truth_files))
1732 @staticmethod
1733 def _make_simple_job(name):
1734 job = lssthtc.HTCJob(name=name)
1735 job.add_job_cmds({"executable": "/usr/bin/echo", "arguments": name})
1736 job.subfile = f"{name}.sub"
1737 return job
1739 def testWriteEdgesAndSpecialJobs(self):
1740 self.maxDiff = None
1741 dag = lssthtc.HTCDag(name="test_special")
1742 dag.add_attribs({"bps_run": "myrun"})
1743 dag.add_job(self._make_simple_job("jobA"))
1744 dag.add_job(self._make_simple_job("jobB"))
1745 dag.add_job_relationships(["jobA"], ["jobB"])
1746 dag.add_final_job(self._make_simple_job("finalJob"))
1747 dag.add_service_job(self._make_simple_job("svc"))
1749 dagfile_expected = [
1750 'JOB jobA "jobA.sub"\n',
1751 'JOB jobB "jobB.sub"\n',
1752 "PARENT jobA CHILD jobB\n",
1753 f"NODE_STATUS_FILE {dag.name}.node_status\n",
1754 'SET_JOB_ATTR bps_run= "myrun"\n',
1755 'FINAL finalJob "finalJob.sub"\n',
1756 'SERVICE svc "svc.sub"\n',
1757 ]
1759 with temporaryDirectory() as tmp_dir:
1760 dag.write(tmp_dir, "", "")
1761 with open(os.path.join(tmp_dir, dag.graph["dag_filename"]), encoding="utf-8") as f:
1762 self.assertEqual(f.readlines(), dagfile_expected)
1763 # Submit files are written for regular and special jobs.
1764 for subfile in ("jobA.sub", "jobB.sub", "finalJob.sub", "svc.sub"):
1765 self.assertTrue(os.path.exists(os.path.join(tmp_dir, subfile)))
1767 def testWriteDagSubdir(self):
1768 # A non-empty dag_subdir places the .dag file (and DIR clause) under
1769 # that subdirectory, creating it as needed.
1770 dag = lssthtc.HTCDag(name="test_subdir")
1771 dag.add_job(self._make_simple_job("jobA"))
1773 with temporaryDirectory() as tmp_dir:
1774 dag.write(tmp_dir, "", "daglevel")
1775 self.assertEqual(dag.graph["dag_filename"], "daglevel/test_subdir.dag")
1776 self.assertTrue(os.path.exists(os.path.join(tmp_dir, "daglevel", "test_subdir.dag")))
1778 def testWriteMissingJobData(self):
1779 # A node without the "data" key should raise KeyError.
1780 dag = lssthtc.HTCDag(name="test_baddata")
1781 dag.add_node("orphan") # no data=... provided
1782 with temporaryDirectory() as tmp_dir:
1783 with self.assertRaises(KeyError):
1784 dag.write(tmp_dir, "", "")
1787class WriteDagInfoTestCase(unittest.TestCase):
1788 """Test for write_dag_info function."""
1790 def setUp(self):
1791 self.run = "u_testuser_DM-53494_20260220T001651Z"
1792 self.data = {
1793 "mycomputer": {
1794 "24390.0": {
1795 "ClusterId": 24390,
1796 "GlobalJobId": "mycomputer#24390.0#1771546612",
1797 "bps_run": self.run,
1798 "bps_isjob": "True",
1799 "bps_payload": "DM-53494",
1800 "bps_project": "dev",
1801 "bps_runsite": "site1",
1802 "bps_campaign": "ci_rc2",
1803 "bps_operator": "testuser",
1804 "bps_run_quanta": "",
1805 "bps_job_summary": "buildQuantumGraph:1;preparePayloadWorkflow:1;dummyJob:1",
1806 "bps_wms_service": "lsst.ctrl.bps.htcondor.htcondor_service.HTCondorService",
1807 "bps_wms_workflow": "lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow",
1808 "bps_wms_config_path": "dagman.conf",
1809 }
1810 }
1811 }
1813 def testWrite(self):
1814 with temporaryDirectory() as tmp_dir:
1815 with chdir(tmp_dir):
1816 path = pathlib.Path(tmp_dir) / "test.info.json"
1817 filename = lssthtc.write_dag_info(path, self.data)
1818 self.assertTrue(path.is_file(), f"File not found at {path}")
1819 self.assertEqual(filename, path)
1821 read_filename, read_data = lssthtc.read_dag_info(tmp_dir)
1822 self.assertEqual(read_filename, path)
1823 self.assertEqual(read_data, self.data)
1826if __name__ == "__main__":
1827 unittest.main()