Coverage for tests/test_htcondor_service.py: 100%
311 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-27 08:46 +0000
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-27 08:46 +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/>.
28"""Unit tests for the HTCondor WMS service class and related functions."""
30import logging
31import os
32import tempfile
33import unittest
34from pathlib import Path
35from unittest.mock import patch
37from htcondor2 import Collector, HTCondorException
39import lsst.ctrl.bps.htcondor.lssthtc as lssthtc
40from lsst.ctrl.bps import BpsConfig, WmsStates
41from lsst.ctrl.bps.htcondor import htcondor_service
42from lsst.ctrl.bps.htcondor.htcondor_config import HTC_DEFAULTS_URI
43from lsst.ctrl.bps.htcondor.htcondor_workflow import HTCondorWorkflow
44from lsst.ctrl.bps.tests.gw_test_utils import make_3_label_workflow
45from lsst.daf.butler import Config
47from .dag_test_utils import LOCATE_SUCCESS, PING_SUCCESS
49logger = logging.getLogger("lsst.ctrl.bps.htcondor")
50TESTDIR = os.path.abspath(os.path.dirname(__file__))
53class HTCondorServiceTestCase(unittest.TestCase):
54 """Test selected methods of the HTCondor WMS service class."""
56 def setUp(self):
57 config = BpsConfig({}, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService")
58 self.service = htcondor_service.HTCondorService(config)
60 def tearDown(self):
61 pass
63 def testDefaults(self):
64 self.assertEqual(self.service.defaults["memoryLimit"], 491520)
66 def testDefaultsPath(self):
67 self.assertEqual(self.service.defaults_uri, HTC_DEFAULTS_URI)
68 self.assertFalse(self.service.defaults_uri.isdir())
70 @patch("lsst.ctrl.bps.htcondor.htcondor_service.ping", return_value=PING_SUCCESS)
71 @patch.object(Collector, "locate", return_value=LOCATE_SUCCESS)
72 def testPingSuccess(self, mock_locate, mock_ping):
73 status, message = self.service.ping(None)
74 self.assertEqual(status, 0)
75 self.assertEqual(message, "")
77 def testPingFailure(self):
78 with patch("lsst.ctrl.bps.htcondor.htcondor_service.Collector.locate") as locate_mock:
79 locate_mock.side_effect = HTCondorException("Unable to locate local daemon.")
80 status, message = self.service.ping(None)
81 self.assertEqual(status, 1)
82 self.assertIn(message, "Could not locate Schedd service.")
84 @patch.object(Collector, "locate", return_value=LOCATE_SUCCESS)
85 def testPingPermission(self, mock_locate):
86 with patch("lsst.ctrl.bps.htcondor.htcondor_service.ping") as ping_mock:
87 ping_mock.side_effect = HTCondorException("Failed to connect to schedd.")
88 status, message = self.service.ping(None)
89 self.assertEqual(status, 1)
90 self.assertEqual(message, "Permission problem with Schedd service.")
92 @patch("lsst.ctrl.bps.htcondor.htcondor_service._get_status_from_id")
93 @patch("lsst.ctrl.bps.htcondor.htcondor_service._locate_schedds")
94 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
95 def testGetStatusLocal(self, mock_type, mock_locate, mock_status):
96 mock_type.return_value = htcondor_service.WmsIdType.LOCAL
97 mock_locate.return_value = {}
98 mock_status.return_value = (WmsStates.RUNNING, "")
100 fake_id = "100"
101 state, message = self.service.get_status(fake_id)
103 mock_type.assert_called_once_with(fake_id)
104 mock_locate.assert_called_once_with(locate_all=False)
105 mock_status.assert_called_once_with(fake_id, 1, schedds={})
107 self.assertEqual(state, WmsStates.RUNNING)
108 self.assertEqual(message, "")
110 @patch("lsst.ctrl.bps.htcondor.htcondor_service._get_status_from_id")
111 @patch("lsst.ctrl.bps.htcondor.htcondor_service._locate_schedds")
112 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
113 def testGetStatusGlobal(self, mock_type, mock_locate, mock_status):
114 mock_type.return_value = htcondor_service.WmsIdType.GLOBAL
115 mock_locate.return_value = {}
116 fake_message = ""
117 mock_status.return_value = (WmsStates.RUNNING, fake_message)
119 fake_id = "100"
120 state, message = self.service.get_status(fake_id, 2)
122 mock_type.assert_called_once_with(fake_id)
123 mock_locate.assert_called_once_with(locate_all=True)
124 mock_status.assert_called_once_with(fake_id, 2, schedds={})
126 self.assertEqual(state, WmsStates.RUNNING)
127 self.assertEqual(message, fake_message)
129 @patch("lsst.ctrl.bps.htcondor.htcondor_service._get_status_from_path")
130 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
131 def testGetStatusPath(self, mock_type, mock_status):
132 fake_message = "fake message"
133 mock_type.return_value = htcondor_service.WmsIdType.PATH
134 mock_status.return_value = (WmsStates.FAILED, fake_message)
136 fake_id = "/fake/path"
137 state, message = self.service.get_status(fake_id)
139 mock_type.assert_called_once_with(fake_id)
140 mock_status.assert_called_once_with(fake_id)
142 self.assertEqual(state, WmsStates.FAILED)
143 self.assertEqual(message, fake_message)
145 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
146 def testGetStatusUnknownType(self, mock_type):
147 mock_type.return_value = htcondor_service.WmsIdType.UNKNOWN
149 fake_id = "100.0"
150 state, message = self.service.get_status(fake_id)
152 mock_type.assert_called_once_with(fake_id)
154 self.assertEqual(state, WmsStates.UNKNOWN)
155 self.assertEqual(message, "Invalid job id")
157 @patch("lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow.write")
158 def testPrepare(self, mock_write):
159 generic_workflow = make_3_label_workflow("test1", True)
160 config = BpsConfig(
161 {
162 "bpsUseShared": True,
163 "overwriteJobFiles": False,
164 "memoryLimit": 491520,
165 "profile": {},
166 "attrs": {},
167 "nodeset": "set1",
168 }
169 )
171 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
172 htc_workflow = self.service.prepare(config, generic_workflow, tmpdir)
173 mock_write.assert_called_once()
174 self.assertEqual(len(htc_workflow.dag), 19) # 3 visit * 2 detectors * 3 labels + init
176 @patch("lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow.write")
177 def testPrepareProvision(self, mock_write):
178 # Leaves testing provisioning code to test_provisioner.py.
179 # Just checking HTCondorService.prepare bits (like nodeset).
180 timestamp = "20260130T211713Z"
181 generic_workflow = make_3_label_workflow("test1", True)
182 config = BpsConfig(
183 {
184 "bpsUseShared": True,
185 "overwriteJobFiles": False,
186 "profile": {"requirements": "dummy_val == 3"},
187 "attrs": {},
188 "nodeset": "set1", # this shouldn't be used with auto-provisioning
189 "provisionResources": True,
190 "provisioning": {"provisioningMaxWallTime": 1200},
191 "bps_defined": {"timestamp": timestamp},
192 },
193 defaults=Config(HTC_DEFAULTS_URI),
194 )
196 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
197 prov_config = Path(f"{tmpdir}/condor-info.py")
198 config[".provisioning.provisioningScriptConfigPath"] = str(prov_config)
199 config[".provisioning.provisioningScriptConfig"] = "foo"
201 htc_workflow = self.service.prepare(config, generic_workflow, tmpdir)
202 mock_write.assert_called_once()
203 self.assertEqual(config[".bps_defined.nodeset"], timestamp)
204 self.assertEqual(len(htc_workflow.dag), 19) # 3 visit * 2 dets * 3 labels + init
205 self.assertIsNotNone(htc_workflow.dag.graph["service_job"])
207 prov_script = Path(tmpdir) / "provisioningJob.bash"
208 self.assertTrue(prov_script.is_file())
209 script_contents = prov_script.read_text()
210 self.assertIn(f"--nodeset '{timestamp}'", script_contents)
212 def testSubmitWithConfigPath(self):
213 """Only testing value for wms_config_path being passed
214 correctly to htc_create_submit_from_dag. Aborting submission
215 after that call to skip rest of submit function.
216 """
218 def _fake_htc_create_submit_from_dag(filename, submit_options, wms_config_path):
219 raise RuntimeError("Fake exception from mock")
221 dag_filename = "should_not_matter.dag"
222 wms_config_path = "dagman.conf"
223 submit_options = {"DAGMAN_MAX_JOBS_SUBMITTED": 30}
224 attribs = {"bps_wms_config_path": wms_config_path}
226 workflow = HTCondorWorkflow("testSuccess")
227 workflow.dag = lssthtc.HTCDag("testSuccess")
228 workflow.dag.graph["dag_filename"] = dag_filename
229 workflow.dag.graph["attr"] = dict(attribs)
230 workflow.dag.graph["submit_options"] = dict(submit_options)
232 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
233 workflow.submit_path = tmpdir
234 with patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_dag") as create_mock:
235 create_mock.side_effect = _fake_htc_create_submit_from_dag
236 with self.assertRaisesRegex(RuntimeError, "Fake exception from mock"):
237 self.service.submit(workflow)
238 create_mock.assert_called_once_with(dag_filename, submit_options, wms_config_path)
240 def testSubmitWithoutConfigPath(self):
241 """Only testing that values are being passed correctly to
242 htc_create_submit_from_dag when there isn't a wms config path.
243 Aborting submission after that call to skip rest of submit function.
244 """
246 def _fake_htc_create_submit_from_dag(filename, submit_options, wms_config_path):
247 raise RuntimeError("Fake exception from mock")
249 dag_filename = "should_not_matter.dag"
250 wms_config_path = None
251 submit_options = {"DAGMAN_MAX_JOBS_SUBMITTED": 30}
252 attribs = {}
254 workflow = HTCondorWorkflow("testSuccess")
255 workflow.dag = lssthtc.HTCDag("testSuccess")
256 workflow.dag.graph["dag_filename"] = dag_filename
257 workflow.dag.graph["attr"] = dict(attribs)
258 workflow.dag.graph["submit_options"] = dict(submit_options)
260 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
261 workflow.submit_path = tmpdir
262 with patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_dag") as create_mock:
263 create_mock.side_effect = _fake_htc_create_submit_from_dag
264 with self.assertRaisesRegex(RuntimeError, "Fake exception from mock"):
265 self.service.submit(workflow)
266 create_mock.assert_called_once_with(dag_filename, submit_options, wms_config_path)
269class RestartTestCase(unittest.TestCase):
270 """Test HTCondorService.restart using mocked lssthtc functions."""
272 def setUp(self):
273 config = BpsConfig({}, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService")
274 self.service = htcondor_service.HTCondorService(config)
276 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
277 def testIdNotFound(self, mock_to_dir):
278 mock_to_dir.return_value = (None, htcondor_service.WmsIdType.UNKNOWN)
279 run_id, run_name, message = self.service.restart("bad_id")
280 self.assertIsNone(run_id)
281 self.assertIsNone(run_name)
282 self.assertIn("not found", message)
283 self.assertIn("submit directory", message)
285 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
286 def testSubmitDirNotFound(self, mock_to_dir):
287 mock_to_dir.return_value = (Path("/does/not/exist"), htcondor_service.WmsIdType.LOCAL)
288 run_id, run_name, message = self.service.restart("100.0")
289 self.assertIsNone(run_id)
290 self.assertIsNone(run_name)
291 self.assertIn("submit directory", message)
292 self.assertIn("not found", message)
294 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
295 def testNoRescueDag(self, mock_to_dir):
296 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
297 mock_to_dir.return_value = (Path(tmpdir), htcondor_service.WmsIdType.PATH)
298 run_id, run_name, message = self.service.restart(tmpdir)
299 self.assertIsNone(run_id)
300 self.assertIsNone(run_name)
301 self.assertIn("rescue DAG", message)
303 @patch("lsst.ctrl.bps.htcondor.htcondor_service.condor_q")
304 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
305 def testAlreadyInQueue(self, mock_to_dir, mock_condor_q):
306 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
307 (Path(tmpdir) / "test.dag.rescue001").touch()
308 mock_to_dir.return_value = (Path(tmpdir), htcondor_service.WmsIdType.PATH)
309 mock_condor_q.return_value = {"schedd": {"1.0": {"GlobalJobId": "schedd#1.0#123"}}}
310 run_id, run_name, message = self.service.restart(tmpdir)
311 self.assertIsNone(run_id)
312 self.assertIsNone(run_name)
313 self.assertIn("already in the job queue", message)
314 self.assertIn("schedd#1.0#123", message)
316 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_status")
317 @patch("lsst.ctrl.bps.htcondor.htcondor_service.condor_q")
318 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
319 def testAllJobsFinished(self, mock_to_dir, mock_condor_q, mock_status):
320 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
321 (Path(tmpdir) / "test.dag.rescue001").touch()
322 mock_to_dir.return_value = (Path(tmpdir), htcondor_service.WmsIdType.PATH)
323 mock_condor_q.return_value = {}
324 mock_status.return_value = {"NodesTotal": 5, "NodesDone": 5}
325 run_id, run_name, message = self.service.restart(tmpdir)
326 self.assertIsNone(run_id)
327 self.assertIsNone(run_name)
328 self.assertIn("finished successfully", message)
330 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_backup_files")
331 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_info")
332 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_status")
333 @patch("lsst.ctrl.bps.htcondor.htcondor_service.condor_q")
334 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
335 def testNoCondorSub(self, mock_to_dir, mock_condor_q, mock_status, mock_read_info, mock_backup):
336 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
337 (Path(tmpdir) / "test.dag.rescue001").touch()
338 mock_to_dir.return_value = (Path(tmpdir), htcondor_service.WmsIdType.PATH)
339 mock_condor_q.return_value = {}
340 mock_status.return_value = {"NodesTotal": 5, "NodesDone": 3}
341 mock_read_info.return_value = (
342 "info.json",
343 {"schedd": {"1.0": {"bps_job_summary": "sum", "bps_run_quanta": "quanta"}}},
344 )
345 mock_backup.return_value = Path(tmpdir) / "test.dag.rescue001"
346 run_id, run_name, message = self.service.restart(tmpdir)
347 self.assertIsNone(run_id)
348 self.assertIsNone(run_name)
349 self.assertIn("submit description file not found", message)
351 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_submit_dag")
352 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_file")
353 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_backup_files")
354 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_info")
355 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_status")
356 @patch("lsst.ctrl.bps.htcondor.htcondor_service.condor_q")
357 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
358 def testSubmitInfoUnavailable(
359 self,
360 mock_to_dir,
361 mock_condor_q,
362 mock_status,
363 mock_read_info,
364 mock_backup,
365 mock_create,
366 mock_submit,
367 ):
368 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
369 (Path(tmpdir) / "test.dag.rescue001").touch()
370 (Path(tmpdir) / "test.dag.condor.sub").touch()
371 mock_to_dir.return_value = (Path(tmpdir), htcondor_service.WmsIdType.PATH)
372 mock_condor_q.return_value = {}
373 mock_status.return_value = {"NodesTotal": 5, "NodesDone": 3}
374 mock_read_info.return_value = (
375 "info.json",
376 {"schedd": {"1.0": {"bps_job_summary": "sum", "bps_run_quanta": "quanta"}}},
377 )
378 mock_backup.return_value = Path(tmpdir) / "test.dag.rescue001"
379 mock_submit.return_value = {}
380 run_id, run_name, message = self.service.restart(tmpdir)
381 self.assertIsNone(run_id)
382 self.assertIsNone(run_name)
383 self.assertEqual(message, "DAGMan job information unavailable")
385 @patch("lsst.ctrl.bps.htcondor.htcondor_service.write_dag_info")
386 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_submit_dag")
387 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_file")
388 @patch("lsst.ctrl.bps.htcondor.htcondor_service.htc_backup_files")
389 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_info")
390 @patch("lsst.ctrl.bps.htcondor.htcondor_service.read_dag_status")
391 @patch("lsst.ctrl.bps.htcondor.htcondor_service.condor_q")
392 @patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_to_dir")
393 def testSuccess(
394 self,
395 mock_to_dir,
396 mock_condor_q,
397 mock_status,
398 mock_read_info,
399 mock_backup,
400 mock_create,
401 mock_submit,
402 mock_write_info,
403 ):
404 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
405 (Path(tmpdir) / "test.dag.rescue001").touch()
406 (Path(tmpdir) / "test.dag.condor.sub").touch()
407 mock_to_dir.return_value = (Path(tmpdir), htcondor_service.WmsIdType.PATH)
408 mock_condor_q.return_value = {}
409 mock_status.return_value = {"NodesTotal": 5, "NodesDone": 3}
410 info_filename = "info.json"
411 mock_read_info.return_value = (
412 info_filename,
413 {"schedd": {"1.0": {"bps_job_summary": "sum", "bps_run_quanta": "quanta"}}},
414 )
415 mock_backup.return_value = Path(tmpdir) / "test.dag.rescue001"
416 schedd_dag_info = {"schedd": {"2.0": {"ClusterId": 2, "ProcId": 0, "bps_run": "myrun"}}}
417 mock_submit.return_value = schedd_dag_info
419 run_id, run_name, message = self.service.restart(tmpdir)
421 self.assertEqual(run_id, "2.0")
422 self.assertEqual(run_name, "myrun")
423 self.assertEqual(message, "")
424 mock_write_info.assert_called_once_with(info_filename, schedd_dag_info)
425 # Summaries from the previous run should be carried forward.
426 dag_ad = schedd_dag_info["schedd"]["2.0"]
427 self.assertEqual(dag_ad["bps_job_summary"], "sum")
428 self.assertEqual(dag_ad["bps_run_quanta"], "quanta")
431class RunSubmissionChecksTestCase(unittest.TestCase):
432 """Test HTCondorService.run_submission_checks."""
434 @staticmethod
435 def _make_service(config_dict):
436 config = BpsConfig(config_dict, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService")
437 return htcondor_service.HTCondorService(config)
439 def testBpsMakeCommandMissing(self):
440 # bpsMakeCommand absent defaults to True, so no checks are performed.
441 service = self._make_service({})
442 self.assertIsNone(service.run_submission_checks())
444 def testBpsMakeCommandTrue(self):
445 service = self._make_service({"bpsMakeCommand": True})
446 self.assertIsNone(service.run_submission_checks())
448 def testMissingPayloadCommand(self):
449 service = self._make_service({"bpsMakeCommand": False})
450 with self.assertRaisesRegex(KeyError, "Missing 'payloadCommand'"):
451 service.run_submission_checks()
453 def testPayloadCommandWithoutSetupEnv(self):
454 # payloadCommand present but does not reference setupEnv, so the
455 # remaining checks are skipped.
456 service = self._make_service({"bpsMakeCommand": False, "payloadCommand": "run_thing --flag"})
457 self.assertIsNone(service.run_submission_checks())
459 def testMissingSetupEnv(self):
460 service = self._make_service({"bpsMakeCommand": False, "payloadCommand": "run_thing {setupEnv}"})
461 with self.assertRaisesRegex(KeyError, "Missing 'setupEnv'"):
462 service.run_submission_checks()
464 def testSetupEnvWithoutLsstVersion(self):
465 service = self._make_service(
466 {
467 "bpsMakeCommand": False,
468 "payloadCommand": "run_thing {setupEnv}",
469 "setupEnv": "source /opt/lsst/setup.sh",
470 }
471 )
472 self.assertIsNone(service.run_submission_checks())
474 def testMissingLsstVersion(self):
475 service = self._make_service(
476 {
477 "bpsMakeCommand": False,
478 "payloadCommand": "run_thing {setupEnv}",
479 "setupEnv": "setup lsst_distrib -t {lsstVersion}",
480 }
481 )
482 with self.assertRaisesRegex(KeyError, "Missing 'lsstVersion'"):
483 service.run_submission_checks()
485 def testAllPresent(self):
486 service = self._make_service(
487 {
488 "bpsMakeCommand": False,
489 "payloadCommand": "run_thing {setupEnv}",
490 "setupEnv": "setup lsst_distrib -t {lsstVersion}",
491 "lsstVersion": "w_2026_01",
492 }
493 )
494 self.assertIsNone(service.run_submission_checks())