Coverage for tests/test_handlers.py: 99%

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 job ClassAd handlers.""" 

29 

30import logging 

31import unittest 

32from typing import Any 

33 

34from lsst.ctrl.bps.htcondor.handlers import ( 

35 Chain, 

36 Handler, 

37 JobAbortedByPeriodicRemoveHandler, 

38 JobAbortedByUserHandler, 

39 JobCompletedHandler, 

40 JobHeldByOtherHandler, 

41 JobHeldBySignalHandler, 

42 JobHeldByUserHandler, 

43) 

44 

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

46 

47 

48class DummyHandler(Handler): 

49 """A concrete handler that does nothing.""" 

50 

51 def handle(self, ad: dict[str, Any]) -> dict[str, Any] | None: 

52 pass 

53 

54 

55class RaisingHandler(Handler): 

56 """A concrete handler that raises KeyError exception.""" 

57 

58 def handle(self, ad: dict[str, Any]) -> dict[str, Any]: 

59 raise KeyError("foo") 

60 

61 

62class ChainTestCase(unittest.TestCase): 

63 """Test the Chain class.""" 

64 

65 def setUp(self): 

66 pass 

67 

68 def tearDown(self): 

69 pass 

70 

71 def testDefaultInitialization(self): 

72 chain = Chain() 

73 self.assertEqual(len(chain), 0) 

74 

75 def testCustomInitialization(self): 

76 handler = DummyHandler() 

77 chain = Chain(handlers=[handler]) 

78 self.assertEqual(len(chain), 1) 

79 self.assertIs(chain[0], handler) 

80 

81 def testAppendingHandler(self): 

82 first = DummyHandler() 

83 second = DummyHandler() 

84 chain = Chain(handlers=[first]) 

85 chain.append(second) 

86 self.assertEqual(len(chain), 2) 

87 self.assertIs(chain[0], first) 

88 self.assertIs(chain[1], second) 

89 

90 def testAppendingNonHandler(self): 

91 handler = "foo" 

92 chain = Chain() 

93 with self.assertRaises(TypeError): 

94 chain.append(handler) # type: ignore 

95 

96 

97class JobCompletedHandlerTestCase(unittest.TestCase): 

98 """Test the handler for a completed job.""" 

99 

100 def setUp(self): 

101 self.ad = {"ClusterId": 1, "ProcId": 0, "MyType": "JobTerminatedEvent"} 

102 self.handler = JobCompletedHandler() 

103 

104 def tearDown(self): 

105 pass 

106 

107 def testNormalTermination(self): 

108 ad = self.ad | {"TerminatedNormally": True, "ReturnValue": 0} 

109 result = self.handler.handle(ad) 

110 assert result is not None 

111 self.assertIn("ExitBySignal", result) 

112 self.assertFalse(result["ExitBySignal"]) 

113 self.assertIn("ExitCode", result) 

114 self.assertEqual(result["ExitCode"], 0) 

115 

116 def testAbnormalTermination(self): 

117 ad = self.ad | {"TerminatedNormally": False, "TerminatedBySignal": 9} 

118 result = self.handler.handle(ad) 

119 assert result is not None 

120 self.assertIn("ExitBySignal", result) 

121 self.assertTrue(result["ExitBySignal"]) 

122 self.assertIn("ExitSignal", result) 

123 self.assertEqual(result["ExitSignal"], 9) 

124 

125 def testNotHandlingJobNotCompleted(self): 

126 ad = self.ad | {"MyType": "foo"} 

127 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

128 result = self.handler.handle(ad) 

129 self.assertIsNone(result) 

130 self.assertIn("job not completed", cm.output[0]) 

131 

132 

133class JobHeldOtherTestCase(unittest.TestCase): 

134 """Test the handler for a held job.""" 

135 

136 def setUp(self): 

137 self.ad = {"ClusterId": 1, "ProcId": 0, "MyType": "JobHeldEvent"} 

138 self.handler = JobHeldByOtherHandler() 

139 

140 def tearDown(self): 

141 pass 

142 

143 def testHeld(self): 

144 ad = self.ad | {"HoldReasonCode": 42} 

145 result = self.handler.handle(ad) 

146 assert result is not None 

147 self.assertIn("ExitBySignal", result) 

148 self.assertFalse(result["ExitBySignal"]) 

149 self.assertIn("ExitCode", result) 

150 self.assertEqual(result["ExitCode"], 42) 

151 

152 def testHeldBySignal(self): 

153 ad = self.ad | {"HoldReasonCode": 3} 

154 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

155 result = self.handler.handle(ad) 

156 self.assertIsNone(result) 

157 self.assertIn("invalid hold reason code", cm.output[0]) 

158 self.assertIn("HoldReasonCode = 3", cm.output[0]) 

159 

160 def testHeldByUser(self): 

161 ad = self.ad | {"HoldReasonCode": 1} 

162 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

163 result = self.handler.handle(ad) 

164 self.assertIsNone(result) 

165 self.assertIn("invalid hold reason code", cm.output[0]) 

166 self.assertIn("HoldReasonCode = 1", cm.output[0]) 

167 

168 def testNotHandlingJobNotHeld(self): 

169 ad = self.ad | {"MyType": "foo"} 

170 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

171 result = self.handler.handle(ad) 

172 self.assertIsNone(result) 

173 self.assertIn("job not held", cm.output[0]) 

174 

175 

176class JobHeldBySignalHandlerTestCase(unittest.TestCase): 

177 """Test the handler for a job held by a signal.""" 

178 

179 def setUp(self): 

180 self.ad = {"ClusterId": 1, "ProcId": 0, "MyType": "JobHeldEvent"} 

181 self.handler = JobHeldBySignalHandler() 

182 

183 def tearDown(self): 

184 pass 

185 

186 def testSignalAvailable(self): 

187 ad = self.ad | {"HoldReasonCode": 3, "HoldReason": "Job raised a signal 9."} 

188 result = self.handler.handle(ad) 

189 assert result is not None 

190 self.assertIn("ExitBySignal", result) 

191 self.assertTrue(result["ExitBySignal"]) 

192 self.assertIn("ExitSignal", result) 

193 self.assertEqual(int(result["ExitSignal"]), 9) 

194 

195 def testSignalNotAvailable(self): 

196 ad = self.ad | {"HoldReasonCode": 3, "HoldReason": "foo"} 

197 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

198 result = self.handler.handle(ad) 

199 self.assertIsNone(result) 

200 self.assertIn("signal not found", cm.output[0]) 

201 

202 def testNotHandlingInvalidHoldReasonCode(self): 

203 ad = self.ad | {"HoldReasonCode": 1, "HoldReason": "via condor_hold (by user foo)"} 

204 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

205 result = self.handler.handle(ad) 

206 self.assertIsNone(result) 

207 self.assertIn("not held by a signal", cm.output[0]) 

208 

209 def testNotHandlingJobNotHeld(self): 

210 ad = self.ad | {"MyType": "foo"} 

211 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

212 result = self.handler.handle(ad) 

213 self.assertIsNone(result) 

214 self.assertIn("job not held", cm.output[0]) 

215 

216 

217class JobHeldByUserHandlerTestCase(unittest.TestCase): 

218 """Test the handler for a job held by the user.""" 

219 

220 def setUp(self): 

221 self.ad = {"ClusterId": 1, "ProcId": 0, "MyType": "JobHeldEvent"} 

222 self.handler = JobHeldByUserHandler() 

223 

224 def tearDown(self): 

225 pass 

226 

227 def testHandling(self): 

228 ad = self.ad | {"HoldReasonCode": 1} 

229 result = self.handler.handle(ad) 

230 assert result is not None 

231 self.assertIn("ExitBySignal", result) 

232 self.assertFalse(result["ExitBySignal"]) 

233 self.assertIn("ExitCode", result) 

234 self.assertEqual(result["ExitCode"], 0) 

235 

236 def testNotHandlingInvalidHoldReaconCode(self): 

237 ad = self.ad | {"HoldReasonCode": 3, "HoldReason": "Job raised a signal 9."} 

238 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

239 result = self.handler.handle(ad) 

240 self.assertIsNone(result) 

241 self.assertIn("not held by the user", cm.output[0]) 

242 

243 def testNotHandlingJobNotHeld(self): 

244 ad = self.ad | {"MyType": "foo"} 

245 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

246 result = self.handler.handle(ad) 

247 self.assertIsNone(result) 

248 self.assertIn("job not held", cm.output[0]) 

249 

250 

251class JobAbortedByPeriodicRemoveHandlerTestCase(unittest.TestCase): 

252 """Test the handler for jobs deleted by periodic removal policy.""" 

253 

254 def setUp(self): 

255 self.ad = { 

256 "ClusterId": 1, 

257 "ProcId": 0, 

258 "MyType": "JobAbortedEvent", 

259 "Reason": "The job attribute PeriodicRemove expression 'foo' evaluated to TRUE", 

260 } 

261 self.handler = JobAbortedByPeriodicRemoveHandler() 

262 

263 def tearDown(self): 

264 pass 

265 

266 def testHandling(self): 

267 self.ad |= {"HoldReason": "Job raised a signal 9."} 

268 result = self.handler.handle(self.ad) 

269 assert result is not None 

270 self.assertIn("ExitBySignal", result) 

271 self.assertTrue(result["ExitBySignal"]) 

272 self.assertIn("ExitSignal", result) 

273 self.assertEqual(result["ExitSignal"], 9) 

274 

275 def testHandlingWithHoldReasonNoExitSignal(self): 

276 self.ad |= {"HoldReason": "Job raised a signal."} 

277 result = self.handler.handle(self.ad) 

278 assert result is not None 

279 self.assertIn("ExitBySignal", result) 

280 self.assertTrue(result["ExitBySignal"]) 

281 self.assertIn("ExitSignal", result) 

282 self.assertEqual(result["ExitSignal"], -1) 

283 

284 def testHandlingWithoutHoldReason(self): 

285 result = self.handler.handle(self.ad) 

286 assert result is not None 

287 self.assertIn("ExitBySignal", result) 

288 self.assertTrue(result["ExitBySignal"]) 

289 self.assertIn("ExitSignal", result) 

290 self.assertEqual(result["ExitSignal"], -1) 

291 

292 def testNotHandlingJobNotRemoved(self): 

293 self.ad["MyType"] = "foo" 

294 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

295 result = self.handler.handle(self.ad) 

296 self.assertIsNone(result) 

297 self.assertIn("job not removed", cm.output[0]) 

298 

299 def testNotHandlingJobNotRemovedByPeriodicRemoval(self): 

300 self.ad["Reason"] = "DAG Abort" 

301 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

302 result = self.handler.handle(self.ad) 

303 self.assertIsNone(result) 

304 self.assertIn("not removed by the periodic removal policy", cm.output[0]) 

305 

306 def testNotHandlingNoReason(self): 

307 del self.ad["Reason"] 

308 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

309 result = self.handler.handle(self.ad) 

310 self.assertIsNone(result) 

311 self.assertIn("unable to determine the reason", cm.output[0]) 

312 

313 

314class JobAbortedByUserHandlerTestCase(unittest.TestCase): 

315 """Test the handler for jobs deleted by the user.""" 

316 

317 def setUp(self): 

318 self.ad = { 

319 "ClusterId": 1, 

320 "ProcId": 0, 

321 "MyType": "JobAbortedEvent", 

322 } 

323 self.handler = JobAbortedByUserHandler() 

324 

325 def tearDown(self): 

326 pass 

327 

328 def testHandlingAbortedDagmanJob(self): 

329 self.ad |= {"Reason": "Python-initiated action"} 

330 result = self.handler.handle(self.ad) 

331 assert result is not None 

332 self.assertIn("ExitBySignal", result) 

333 self.assertFalse(result["ExitBySignal"]) 

334 self.assertIn("ExitCode", result) 

335 self.assertEqual(result["ExitCode"], 0) 

336 

337 def testHandlingAbortedPayloadJob(self): 

338 self.ad |= {"Reason": "DAG Removed"} 

339 result = self.handler.handle(self.ad) 

340 assert result is not None 

341 self.assertIn("ExitBySignal", result) 

342 self.assertFalse(result["ExitBySignal"]) 

343 self.assertIn("ExitCode", result) 

344 self.assertEqual(result["ExitCode"], 0) 

345 

346 def testHandlingAbortedSubdagJob(self): 

347 self.ad |= {"Reason": "OtherJobRemoveRequirements = DAGManJobId =?= 78"} 

348 result = self.handler.handle(self.ad) 

349 assert result is not None 

350 self.assertIn("ExitBySignal", result) 

351 self.assertFalse(result["ExitBySignal"]) 

352 self.assertIn("ExitCode", result) 

353 self.assertEqual(result["ExitCode"], 0) 

354 

355 def testNotHandlingJobNotRemoved(self): 

356 self.ad["MyType"] = "foo" 

357 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

358 result = self.handler.handle(self.ad) 

359 self.assertIsNone(result) 

360 self.assertIn("job not removed", cm.output[0]) 

361 

362 def testNotHandlingJobNotRemovedByUser(self): 

363 self.ad |= {"Reason": "The job attribute PeriodicRemove expression 'foo' evaluated to TRUE"} 

364 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

365 result = self.handler.handle(self.ad) 

366 self.assertIsNone(result) 

367 self.assertIn("job not removed", cm.output[0]) 

368 

369 def testNotHandlingNoReason(self): 

370 with self.assertLogs(logger=logger, level="DEBUG") as cm: 

371 result = self.handler.handle(self.ad) 

372 self.assertIsNone(result) 

373 self.assertIn("unable to determine the reason", cm.output[0])