Coverage for tests/test_common_utils.py: 100%

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 

28"""Unit tests for the common utility functions.""" 

29 

30import logging 

31import os 

32import unittest 

33from pathlib import Path 

34from unittest.mock import patch 

35 

36from htcondor2 import JobStatus 

37 

38from lsst.ctrl.bps import WmsStates 

39from lsst.ctrl.bps.htcondor import common_utils 

40from lsst.utils.tests import temporaryDirectory 

41 

42logger = logging.getLogger("lsst.ctrl.bps.htcondor") 

43 

44 

45class HtcNodeStatusToWmsStateTestCase(unittest.TestCase): 

46 """Test assigning WMS state base on HTCondor node status.""" 

47 

48 def setUp(self): 

49 pass 

50 

51 def tearDown(self): 

52 pass 

53 

54 def testNotReady(self): 

55 job = {"NodeStatus": common_utils.NodeStatus.NOT_READY} 

56 result = common_utils._htc_node_status_to_wms_state(job) 

57 self.assertEqual(result, WmsStates.UNREADY) 

58 

59 def testReady(self): 

60 job = {"NodeStatus": common_utils.NodeStatus.READY} 

61 result = common_utils._htc_node_status_to_wms_state(job) 

62 self.assertEqual(result, WmsStates.READY) 

63 

64 def testPrerun(self): 

65 job = {"NodeStatus": common_utils.NodeStatus.PRERUN} 

66 result = common_utils._htc_node_status_to_wms_state(job) 

67 self.assertEqual(result, WmsStates.MISFIT) 

68 

69 def testSubmittedHeld(self): 

70 job = { 

71 "NodeStatus": common_utils.NodeStatus.SUBMITTED, 

72 "JobProcsHeld": 1, 

73 "StatusDetails": "", 

74 "JobProcsQueued": 0, 

75 } 

76 result = common_utils._htc_node_status_to_wms_state(job) 

77 self.assertEqual(result, WmsStates.HELD) 

78 

79 def testSubmittedRunning(self): 

80 job = { 

81 "NodeStatus": common_utils.NodeStatus.SUBMITTED, 

82 "JobProcsHeld": 0, 

83 "StatusDetails": "not_idle", 

84 "JobProcsQueued": 0, 

85 } 

86 result = common_utils._htc_node_status_to_wms_state(job) 

87 self.assertEqual(result, WmsStates.RUNNING) 

88 

89 def testSubmittedPending(self): 

90 job = { 

91 "NodeStatus": common_utils.NodeStatus.SUBMITTED, 

92 "JobProcsHeld": 0, 

93 "StatusDetails": "", 

94 "JobProcsQueued": 1, 

95 } 

96 result = common_utils._htc_node_status_to_wms_state(job) 

97 self.assertEqual(result, WmsStates.PENDING) 

98 

99 def testPostrun(self): 

100 job = {"NodeStatus": common_utils.NodeStatus.POSTRUN} 

101 result = common_utils._htc_node_status_to_wms_state(job) 

102 self.assertEqual(result, WmsStates.MISFIT) 

103 

104 def testDone(self): 

105 job = {"NodeStatus": common_utils.NodeStatus.DONE} 

106 result = common_utils._htc_node_status_to_wms_state(job) 

107 self.assertEqual(result, WmsStates.SUCCEEDED) 

108 

109 def testErrorDagmanSuccess(self): 

110 job = { 

111 "NodeStatus": common_utils.NodeStatus.ERROR, 

112 "StatusDetails": "DAGMAN error 0", 

113 } 

114 result = common_utils._htc_node_status_to_wms_state(job) 

115 self.assertEqual(result, WmsStates.SUCCEEDED) 

116 

117 def testErrorDagmanFailure(self): 

118 job = { 

119 "NodeStatus": common_utils.NodeStatus.ERROR, 

120 "StatusDetails": "DAGMAN error 1", 

121 } 

122 result = common_utils._htc_node_status_to_wms_state(job) 

123 self.assertEqual(result, WmsStates.FAILED) 

124 

125 def testFutile(self): 

126 job = {"NodeStatus": common_utils.NodeStatus.FUTILE} 

127 result = common_utils._htc_node_status_to_wms_state(job) 

128 self.assertEqual(result, WmsStates.PRUNED) 

129 

130 def testDeletedJob(self): 

131 job = { 

132 "NodeStatus": common_utils.NodeStatus.ERROR, 

133 "StatusDetails": "HTCondor reported ULOG_JOB_ABORTED event for job proc (1.0.0)", 

134 "JobProcsQueued": 0, 

135 } 

136 result = common_utils._htc_node_status_to_wms_state(job) 

137 self.assertEqual(result, WmsStates.DELETED) 

138 

139 

140class HtcStatusToWmsStateTestCase(unittest.TestCase): 

141 """Test assigning WMS state base on HTCondor status.""" 

142 

143 def testJobStatus(self): 

144 job = { 

145 "ClusterId": 1, 

146 "JobStatus": JobStatus.IDLE, 

147 "bps_job_label": "foo", 

148 } 

149 result = common_utils._htc_status_to_wms_state(job) 

150 self.assertEqual(result, WmsStates.PENDING) 

151 

152 def testNodeStatus(self): 

153 # Hold/Release test case 

154 job = { 

155 "ClusterId": 1, 

156 "JobStatus": None, 

157 "NodeStatus": common_utils.NodeStatus.SUBMITTED, 

158 "JobProcsHeld": 0, 

159 "StatusDetails": "", 

160 "JobProcsQueued": 1, 

161 } 

162 result = common_utils._htc_status_to_wms_state(job) 

163 self.assertEqual(result, WmsStates.PENDING) 

164 

165 def testNeitherStatus(self): 

166 job = {"ClusterId": 1} 

167 result = common_utils._htc_status_to_wms_state(job) 

168 self.assertEqual(result, WmsStates.MISFIT) 

169 

170 def testRetrySuccess(self): 

171 job = { 

172 "NodeStatus": 5, 

173 "Node": "8e62c569-ae2e-44e8-be36-d1aee333a129_isr_903342_10", 

174 "RetryCount": 0, 

175 "ClusterId": 851, 

176 "ProcId": 0, 

177 "MyType": "JobTerminatedEvent", 

178 "EventTypeNumber": 5, 

179 "HoldReasonCode": 3, 

180 "HoldReason": "Job raised a signal 9. Handling signal as if job has gone over memory limit.", 

181 "HoldReasonSubCode": 34, 

182 "JobStatus": JobStatus.COMPLETED, 

183 "ExitBySignal": False, 

184 "ReturnValue": 0, 

185 } 

186 result = common_utils._htc_status_to_wms_state(job) 

187 self.assertEqual(result, WmsStates.SUCCEEDED) 

188 

189 

190class WmsIdToDirTestCase(unittest.TestCase): 

191 """Test _wms_id_to_dir function.""" 

192 

193 @patch("lsst.ctrl.bps.htcondor.common_utils._wms_id_type") 

194 def testInvalidIdType(self, _wms_id_type_mock): 

195 _wms_id_type_mock.return_value = common_utils.WmsIdType.UNKNOWN 

196 with self.assertRaises(TypeError) as cm: 

197 _, _ = common_utils._wms_id_to_dir("not_used") 

198 self.assertIn("Invalid job id type", str(cm.exception)) 

199 

200 @patch("lsst.ctrl.bps.htcondor.common_utils._wms_id_type") 

201 def testAbsPathId(self, mock_wms_id_type): 

202 mock_wms_id_type.return_value = common_utils.WmsIdType.PATH 

203 with temporaryDirectory() as tmp_dir: 

204 wms_path, id_type = common_utils._wms_id_to_dir(tmp_dir) 

205 self.assertEqual(id_type, common_utils.WmsIdType.PATH) 

206 self.assertEqual(Path(tmp_dir).resolve(), wms_path) 

207 

208 @patch("lsst.ctrl.bps.htcondor.common_utils._wms_id_type") 

209 def testRelPathId(self, _wms_id_type_mock): 

210 _wms_id_type_mock.return_value = common_utils.WmsIdType.PATH 

211 orig_dir = Path.cwd() 

212 with temporaryDirectory() as tmp_dir: 

213 os.chdir(tmp_dir) 

214 abs_path = Path(tmp_dir) / "newdir" 

215 abs_path.mkdir() 

216 wms_path, id_type = common_utils._wms_id_to_dir("newdir") 

217 self.assertEqual(id_type, common_utils.WmsIdType.PATH) 

218 self.assertEqual(abs_path.resolve(), wms_path) 

219 os.chdir(orig_dir) 

220 

221 

222class WmsIdTypeTestCase(unittest.TestCase): 

223 """Test _wms_id_type function.""" 

224 

225 def testIntId(self): 

226 id_type = common_utils._wms_id_type("4") 

227 self.assertEqual(id_type, common_utils.WmsIdType.LOCAL) 

228 

229 def testPathId(self): 

230 with temporaryDirectory() as tmp_dir: 

231 id_type = common_utils._wms_id_type(str(tmp_dir)) 

232 self.assertEqual(id_type, common_utils.WmsIdType.PATH) 

233 

234 def testGlobalId(self): 

235 id_type = common_utils._wms_id_type("testmachine#5044.0#1757720957") 

236 self.assertEqual(id_type, common_utils.WmsIdType.GLOBAL) 

237 

238 def testUnknownType(self): 

239 id_type = common_utils._wms_id_type(["bad param"]) 

240 self.assertEqual(id_type, common_utils.WmsIdType.UNKNOWN) 

241 

242 

243class WmsIdToClusterTestCase(unittest.TestCase): 

244 """Test _wms_id_to_cluster function.""" 

245 

246 class _MockCollector: 

247 def locate(self, daemon_type, schedd_name): 

248 return """[ 

249 CondorPlatform = "$CondorPlatform: X86_64-AlmaLinux_9.6 $"; 

250 MyType = "Scheduler"; 

251 Machine = "testmachine"; 

252 Name = "testmachine"; 

253 CondorVersion = "$CondorVersion: 24.0.10 2025-08-05 $"; 

254 MyAddress = "<127.0.0.1:9618?addrs=127.0.0.1-9618+snip>" 

255 ]""" 

256 

257 @patch("lsst.ctrl.bps.htcondor.common_utils.Collector", new=_MockCollector) 

258 @patch("lsst.ctrl.bps.htcondor.common_utils.read_dag_info") 

259 def testPath(self, mock_read): 

260 # path must exist or _wms_id_type will assume GLOBAL string 

261 with temporaryDirectory() as tmp_dir: 

262 mock_read.return_value = [ 

263 "dummy_str", 

264 { 

265 "testmachine": { 

266 "1163.0": { 

267 "testmachine": { 

268 "ClusterId": 1163, 

269 "GlobalJobId": "testmachine#1163.0#1722040518", 

270 } 

271 } 

272 } 

273 }, 

274 ] 

275 schedd_ad, cluster_id, id_type = common_utils._wms_id_to_cluster(tmp_dir) 

276 self.assertEqual(id_type, common_utils.WmsIdType.PATH) 

277 self.assertEqual(cluster_id, 1163) 

278 

279 

280if __name__ == "__main__": 

281 unittest.main()