Coverage for tests/test_htcondor_service.py: 100%
160 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 02:12 -0700
1# This file is part of ctrl_bps_htcondor.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <https://www.gnu.org/licenses/>.
28"""Unit tests for the HTCondor WMS service class and related functions."""
30import logging
31import os
32import tempfile
33import unittest
34from pathlib import Path
36import htcondor
38import lsst.ctrl.bps.htcondor.lssthtc as lssthtc
39from lsst.ctrl.bps import BpsConfig, WmsStates
40from lsst.ctrl.bps.htcondor import htcondor_service
41from lsst.ctrl.bps.htcondor.htcondor_config import HTC_DEFAULTS_URI
42from lsst.ctrl.bps.htcondor.htcondor_workflow import HTCondorWorkflow
43from lsst.ctrl.bps.tests.gw_test_utils import make_3_label_workflow
44from lsst.daf.butler import Config
46logger = logging.getLogger("lsst.ctrl.bps.htcondor")
47TESTDIR = os.path.abspath(os.path.dirname(__file__))
49LOCATE_SUCCESS = """[
50 CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $";
51 MyType = "Scheduler";
52 Machine = "testmachine";
53 Name = "testmachine";
54 CondorVersion = "$CondorVersion: 23.0.3 2024-04-04 $";
55 MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>"
56 ]
57"""
59PING_SUCCESS = """[
60 AuthCommand = 60011;
61 AuthMethods = "FS_REMOTE";
62 Command = 60040;
63 AuthorizationSucceeded = true;
64 ValidCommands = "60002,60003,60011,60014,60045,60046,60047,60048,60049,60050,60052,523";
65 TriedAuthentication = true;
66 RemoteVersion = "$CondorVersion: 10.9.0 2023-09-28 BuildID: 678228 PackageID: 10.9.0-1 $";
67 MyRemoteUserName = "testuser@testmachine";
68 Authentication = "YES";
69 ]
70"""
73class HTCondorServiceTestCase(unittest.TestCase):
74 """Test selected methods of the HTCondor WMS service class."""
76 def setUp(self):
77 config = BpsConfig({}, wms_service_class_fqn="lsst.ctrl.bps.htcondor.HTCondorService")
78 self.service = htcondor_service.HTCondorService(config)
80 def tearDown(self):
81 pass
83 def testDefaults(self):
84 self.assertEqual(self.service.defaults["memoryLimit"], 491520)
86 def testDefaultsPath(self):
87 self.assertEqual(self.service.defaults_uri, HTC_DEFAULTS_URI)
88 self.assertFalse(self.service.defaults_uri.isdir())
90 @unittest.mock.patch.object(htcondor.SecMan, "ping", return_value=PING_SUCCESS)
91 @unittest.mock.patch.object(htcondor.Collector, "locate", return_value=LOCATE_SUCCESS)
92 def testPingSuccess(self, mock_locate, mock_ping):
93 status, message = self.service.ping(None)
94 self.assertEqual(status, 0)
95 self.assertEqual(message, "")
97 def testPingFailure(self):
98 with unittest.mock.patch("htcondor.Collector.locate") as locate_mock:
99 locate_mock.side_effect = htcondor.HTCondorLocateError()
100 status, message = self.service.ping(None)
101 self.assertEqual(status, 1)
102 self.assertEqual(message, "Could not locate Schedd service.")
104 @unittest.mock.patch.object(htcondor.Collector, "locate", return_value=LOCATE_SUCCESS)
105 def testPingPermission(self, mock_locate):
106 with unittest.mock.patch("htcondor.SecMan.ping") as ping_mock:
107 ping_mock.side_effect = htcondor.HTCondorIOError()
108 status, message = self.service.ping(None)
109 self.assertEqual(status, 1)
110 self.assertEqual(message, "Permission problem with Schedd service.")
112 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._get_status_from_id")
113 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._locate_schedds")
114 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
115 def testGetStatusLocal(self, mock_type, mock_locate, mock_status):
116 mock_type.return_value = htcondor_service.WmsIdType.LOCAL
117 mock_locate.return_value = {}
118 mock_status.return_value = (WmsStates.RUNNING, "")
120 fake_id = "100"
121 state, message = self.service.get_status(fake_id)
123 mock_type.assert_called_once_with(fake_id)
124 mock_locate.assert_called_once_with(locate_all=False)
125 mock_status.assert_called_once_with(fake_id, 1, schedds={})
127 self.assertEqual(state, WmsStates.RUNNING)
128 self.assertEqual(message, "")
130 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._get_status_from_id")
131 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._locate_schedds")
132 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
133 def testGetStatusGlobal(self, mock_type, mock_locate, mock_status):
134 mock_type.return_value = htcondor_service.WmsIdType.GLOBAL
135 mock_locate.return_value = {}
136 fake_message = ""
137 mock_status.return_value = (WmsStates.RUNNING, fake_message)
139 fake_id = "100"
140 state, message = self.service.get_status(fake_id, 2)
142 mock_type.assert_called_once_with(fake_id)
143 mock_locate.assert_called_once_with(locate_all=True)
144 mock_status.assert_called_once_with(fake_id, 2, schedds={})
146 self.assertEqual(state, WmsStates.RUNNING)
147 self.assertEqual(message, fake_message)
149 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._get_status_from_path")
150 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
151 def testGetStatusPath(self, mock_type, mock_status):
152 fake_message = "fake message"
153 mock_type.return_value = htcondor_service.WmsIdType.PATH
154 mock_status.return_value = (WmsStates.FAILED, fake_message)
156 fake_id = "/fake/path"
157 state, message = self.service.get_status(fake_id)
159 mock_type.assert_called_once_with(fake_id)
160 mock_status.assert_called_once_with(fake_id)
162 self.assertEqual(state, WmsStates.FAILED)
163 self.assertEqual(message, fake_message)
165 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_service._wms_id_type")
166 def testGetStatusUnknownType(self, mock_type):
167 mock_type.return_value = htcondor_service.WmsIdType.UNKNOWN
169 fake_id = "100.0"
170 state, message = self.service.get_status(fake_id)
172 mock_type.assert_called_once_with(fake_id)
174 self.assertEqual(state, WmsStates.UNKNOWN)
175 self.assertEqual(message, "Invalid job id")
177 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow.write")
178 def testPrepare(self, mock_write):
179 generic_workflow = make_3_label_workflow("test1", True)
180 config = BpsConfig(
181 {
182 "bpsUseShared": True,
183 "overwriteJobFiles": False,
184 "memoryLimit": 491520,
185 "profile": {},
186 "attrs": {},
187 "nodeset": "set1",
188 }
189 )
191 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
192 htc_workflow = self.service.prepare(config, generic_workflow, tmpdir)
193 mock_write.assert_called_once()
194 self.assertEqual(len(htc_workflow.dag), 19) # 3 visit * 2 detectors * 3 labels + init
196 @unittest.mock.patch("lsst.ctrl.bps.htcondor.htcondor_workflow.HTCondorWorkflow.write")
197 def testPrepareProvision(self, mock_write):
198 # Leaves testing provisioning code to test_provisioner.py.
199 # Just checking HTCondorService.prepare bits (like nodeset).
200 timestamp = "20260130T211713Z"
201 generic_workflow = make_3_label_workflow("test1", True)
202 config = BpsConfig(
203 {
204 "bpsUseShared": True,
205 "overwriteJobFiles": False,
206 "profile": {"requirements": "dummy_val == 3"},
207 "attrs": {},
208 "nodeset": "set1", # this shouldn't be used with auto-provisioning
209 "provisionResources": True,
210 "provisioning": {"provisioningMaxWallTime": 1200},
211 "bps_defined": {"timestamp": timestamp},
212 },
213 defaults=Config(HTC_DEFAULTS_URI),
214 )
216 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
217 prov_config = Path(f"{tmpdir}/condor-info.py")
218 config[".provisioning.provisioningScriptConfigPath"] = str(prov_config)
219 config[".provisioning.provisioningScriptConfig"] = "foo"
221 htc_workflow = self.service.prepare(config, generic_workflow, tmpdir)
222 mock_write.assert_called_once()
223 self.assertEqual(config[".bps_defined.nodeset"], timestamp)
224 self.assertEqual(len(htc_workflow.dag), 19) # 3 visit * 2 dets * 3 labels + init
225 self.assertIsNotNone(htc_workflow.dag.graph["service_job"])
227 prov_script = Path(tmpdir) / "provisioningJob.bash"
228 self.assertTrue(prov_script.is_file())
229 script_contents = prov_script.read_text()
230 self.assertIn(f"--nodeset {timestamp}", script_contents)
232 def testSubmitWithConfigPath(self):
233 """Only testing value for wms_config_path being passed
234 correctly to htc_create_submit_from_dag. Aborting submission
235 after that call to skip rest of submit function.
236 """
238 def _fake_htc_create_submit_from_dag(filename, submit_options, wms_config_path):
239 raise RuntimeError("Fake exception from mock")
241 dag_filename = "should_not_matter.dag"
242 wms_config_path = "dagman.conf"
243 submit_options = {"DAGMAN_MAX_JOBS_SUBMITTED": 30}
244 attribs = {"bps_wms_config_path": wms_config_path}
246 workflow = HTCondorWorkflow("testSuccess")
247 workflow.dag = lssthtc.HTCDag("testSuccess")
248 workflow.dag.graph["dag_filename"] = dag_filename
249 workflow.dag.graph["attr"] = dict(attribs)
250 workflow.dag.graph["submit_options"] = dict(submit_options)
252 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
253 workflow.submit_path = tmpdir
254 with unittest.mock.patch(
255 "lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_dag"
256 ) as create_mock:
257 create_mock.side_effect = _fake_htc_create_submit_from_dag
258 with self.assertRaisesRegex(RuntimeError, "Fake exception from mock"):
259 self.service.submit(workflow)
260 create_mock.assert_called_once_with(dag_filename, submit_options, wms_config_path)
262 def testSubmitWithoutConfigPath(self):
263 """Only testing that values are being passed correctly to
264 htc_create_submit_from_dag when there isn't a wms config path.
265 Aborting submission after that call to skip rest of submit function.
266 """
268 def _fake_htc_create_submit_from_dag(filename, submit_options, wms_config_path):
269 raise RuntimeError("Fake exception from mock")
271 dag_filename = "should_not_matter.dag"
272 wms_config_path = None
273 submit_options = {"DAGMAN_MAX_JOBS_SUBMITTED": 30}
274 attribs = {}
276 workflow = HTCondorWorkflow("testSuccess")
277 workflow.dag = lssthtc.HTCDag("testSuccess")
278 workflow.dag.graph["dag_filename"] = dag_filename
279 workflow.dag.graph["attr"] = dict(attribs)
280 workflow.dag.graph["submit_options"] = dict(submit_options)
282 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
283 workflow.submit_path = tmpdir
284 with unittest.mock.patch(
285 "lsst.ctrl.bps.htcondor.htcondor_service.htc_create_submit_from_dag"
286 ) as create_mock:
287 create_mock.side_effect = _fake_htc_create_submit_from_dag
288 with self.assertRaisesRegex(RuntimeError, "Fake exception from mock"):
289 self.service.submit(workflow)
290 create_mock.assert_called_once_with(dag_filename, submit_options, wms_config_path)