Coverage for tests/test_report_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 report utilities.""" 

29 

30import logging 

31import os 

32import unittest 

33from pathlib import Path 

34from shutil import copy2, copytree 

35from unittest.mock import patch 

36 

37from htcondor2 import JobStatus, Schedd 

38 

39from lsst.ctrl.bps import ( 

40 WmsSpecificInfo, 

41 WmsStates, 

42) 

43from lsst.ctrl.bps.htcondor import lssthtc, report_utils 

44from lsst.utils.tests import temporaryDirectory 

45 

46from .dag_test_utils import LOCATE_SUCCESS 

47 

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

49 

50TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

51 

52 

53class GetExitCodeSummaryTestCase(unittest.TestCase): 

54 """Test the function responsible for creating exit code summary.""" 

55 

56 def setUp(self): 

57 self.jobs = { 

58 "1.0": { 

59 "JobStatus": JobStatus.IDLE, 

60 "bps_job_label": "foo", 

61 }, 

62 "2.0": { 

63 "JobStatus": JobStatus.RUNNING, 

64 "bps_job_label": "foo", 

65 }, 

66 "3.0": { 

67 "JobStatus": JobStatus.REMOVED, 

68 "bps_job_label": "foo", 

69 }, 

70 "4.0": { 

71 "ExitCode": 0, 

72 "ExitBySignal": False, 

73 "JobStatus": JobStatus.COMPLETED, 

74 "bps_job_label": "bar", 

75 }, 

76 "5.0": { 

77 "ExitCode": 1, 

78 "ExitBySignal": False, 

79 "JobStatus": JobStatus.COMPLETED, 

80 "bps_job_label": "bar", 

81 }, 

82 "6.0": { 

83 "ExitBySignal": True, 

84 "ExitSignal": 11, 

85 "JobStatus": JobStatus.HELD, 

86 "bps_job_label": "baz", 

87 }, 

88 "7.0": { 

89 "ExitBySignal": False, 

90 "ExitCode": 42, 

91 "JobStatus": JobStatus.HELD, 

92 "bps_job_label": "baz", 

93 }, 

94 "8.0": { 

95 "JobStatus": JobStatus.TRANSFERRING_OUTPUT, 

96 "bps_job_label": "qux", 

97 }, 

98 "9.0": { 

99 "JobStatus": JobStatus.SUSPENDED, 

100 "bps_job_label": "qux", 

101 }, 

102 } 

103 

104 def tearDown(self): 

105 pass 

106 

107 def testMainScenario(self): 

108 actual = report_utils._get_exit_code_summary(self.jobs) 

109 expected = {"foo": [], "bar": [1], "baz": [11, 42], "qux": []} 

110 self.assertEqual(actual, expected) 

111 

112 def testUnknownStatus(self): 

113 jobs = { 

114 "1.0": { 

115 "JobStatus": -1, 

116 "bps_job_label": "foo", 

117 } 

118 } 

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

120 report_utils._get_exit_code_summary(jobs) 

121 self.assertIn("lsst.ctrl.bps.htcondor", cm.records[0].name) 

122 self.assertIn("Unknown", cm.output[0]) 

123 self.assertIn("JobStatus", cm.output[0]) 

124 

125 def testUnknownKey(self): 

126 jobs = { 

127 "1.0": { 

128 "JobStatus": JobStatus.COMPLETED, 

129 "UnknownKey": None, 

130 "bps_job_label": "foo", 

131 } 

132 } 

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

134 report_utils._get_exit_code_summary(jobs) 

135 self.assertIn("lsst.ctrl.bps.htcondor", cm.records[0].name) 

136 self.assertIn("Attribute", cm.output[0]) 

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

138 

139 

140class GetStateCountsFromDagJobTestCase(unittest.TestCase): 

141 """Test counting number of jobs per WMS state.""" 

142 

143 def setUp(self): 

144 pass 

145 

146 def tearDown(self): 

147 pass 

148 

149 def testCounts(self): 

150 job = { 

151 "DAG_NodesDone": 1, 

152 "DAG_JobsHeld": 2, 

153 "DAG_NodesFailed": 3, 

154 "DAG_NodesFutile": 4, 

155 "DAG_NodesQueued": 5, 

156 "DAG_NodesReady": 0, 

157 "DAG_NodesUnready": 7, 

158 "DAG_NodesTotal": 22, 

159 } 

160 

161 truth = { 

162 WmsStates.SUCCEEDED: 1, 

163 WmsStates.HELD: 2, 

164 WmsStates.UNREADY: 7, 

165 WmsStates.READY: 0, 

166 WmsStates.FAILED: 3, 

167 WmsStates.PRUNED: 4, 

168 WmsStates.MISFIT: 0, 

169 } 

170 

171 total, result = report_utils._get_state_counts_from_dag_job(job) 

172 self.assertEqual(total, 22) 

173 self.assertEqual(result, truth) 

174 

175 

176class GetInfoFromPathTestCase(unittest.TestCase): 

177 """Test _get_info_from_path function.""" 

178 

179 def test_tmpdir_abort(self): 

180 with temporaryDirectory() as tmp_dir: 

181 copy2(f"{TESTDIR}/data/test_tmpdir_abort.dag.dagman.out", tmp_dir) 

182 wms_workflow_id, jobs, message = report_utils._get_info_from_path(tmp_dir) 

183 self.assertEqual(wms_workflow_id, lssthtc.MISSING_ID) 

184 self.assertEqual(jobs, {}) 

185 self.assertIn("Cannot submit from /tmp", message) 

186 

187 def test_no_dagman_messages(self): 

188 with temporaryDirectory() as tmp_dir: 

189 copy2(f"{TESTDIR}/data/test_no_messages.dag.dagman.out", tmp_dir) 

190 wms_workflow_id, jobs, message = report_utils._get_info_from_path(tmp_dir) 

191 self.assertEqual(wms_workflow_id, lssthtc.MISSING_ID) 

192 self.assertEqual(jobs, {}) 

193 self.assertIn("Could not find HTCondor files", message) 

194 

195 def test_successful_run(self): 

196 with temporaryDirectory() as tmp_dir: 

197 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag", tmp_dir) 

198 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag.dagman.log", tmp_dir) 

199 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag.dagman.out", tmp_dir) 

200 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag.nodes.log", tmp_dir) 

201 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.node_status", tmp_dir) 

202 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.info.json", tmp_dir) 

203 wms_workflow_id, jobs, message = report_utils._get_info_from_path(tmp_dir) 

204 self.assertEqual(wms_workflow_id, "1163.0") 

205 self.assertEqual(len(jobs), 6) # dag, pipetaskInit, 3 science, finalJob 

206 self.assertEqual(message, "") 

207 

208 def test_relative_path(self): 

209 orig_dir = Path.cwd() 

210 with temporaryDirectory() as tmp_dir: 

211 os.chdir(tmp_dir) 

212 abs_path = Path(tmp_dir).resolve() / "subdir" 

213 abs_path.mkdir() 

214 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag", abs_path) 

215 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag.dagman.log", abs_path) 

216 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag.dagman.out", abs_path) 

217 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.dag.nodes.log", abs_path) 

218 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.node_status", abs_path) 

219 copy2(f"{TESTDIR}/data/test_pipelines_check_20240727T003507Z.info.json", abs_path) 

220 wms_workflow_id, jobs, message = report_utils._get_info_from_path("subdir") 

221 self.assertEqual(wms_workflow_id, "1163.0") 

222 self.assertEqual(len(jobs), 6) # dag, pipetaskInit, 3 science, finalJob 

223 self.assertEqual(message, "") 

224 self.assertEqual(jobs["1163.0"]["Iwd"], str(abs_path)) 

225 os.chdir(orig_dir) 

226 

227 

228class AddServiceJobSpecificInfoTestCase(unittest.TestCase): 

229 """Test _add_service_job_specific_info function. 

230 

231 Note: The job_ad's are hardcoded in these tests. The 

232 values in the dictionaries come from plugin code as 

233 well as HTCondor. Changes in either of those codes 

234 that produce data for the job_ad can break this 

235 function without breaking these unit tests. 

236 

237 Also, since hold status/messages stick around, testing 

238 various cases with and without job being held just to 

239 ensure get right status in both cases. 

240 """ 

241 

242 def testNotSubmitted(self): 

243 # Service job not submitted yet or can't be submitted. 

244 # (Typically an plugin bug.) 

245 # At this function level, can't tell if not submitted 

246 # yet or problem so it never will. 

247 job_ad = { 

248 "ClusterId": -64, 

249 "DAGManJobID": "8997.0", 

250 "DAGNodeName": "provisioningJob", 

251 "NodeStatus": lssthtc.NodeStatus.NOT_READY, 

252 "ProcId": 0, 

253 "bps_job_label": "service_provisioningJob", 

254 } 

255 results = WmsSpecificInfo() 

256 report_utils._add_service_job_specific_info(job_ad, results) 

257 self.assertEqual( 

258 results.context, {"job_name": "provisioningJob", "status": "UNREADY", "status_details": ""} 

259 ) 

260 

261 def testRunning(self): 

262 # DAG hasn't completed (Running or held), 

263 # Service job is running. 

264 job_ad = { 

265 "ClusterId": 8523, 

266 "ProcId": 0, 

267 "DAGNodeName": "provisioningJob", 

268 "JobStatus": JobStatus.RUNNING, 

269 } 

270 

271 results = WmsSpecificInfo() 

272 report_utils._add_service_job_specific_info(job_ad, results) 

273 self.assertEqual( 

274 results.context, {"job_name": "provisioningJob", "status": "RUNNING", "status_details": ""} 

275 ) 

276 

277 def testDied(self): 

278 # DAG hasn't completed (Running or held), 

279 # Service job failed (completed non-zero exit code) 

280 job_ad = { 

281 "ClusterId": 8761, 

282 "ProcId": 0, 

283 "DAGNodeName": "provisioningJob", 

284 "JobStatus": JobStatus.COMPLETED, 

285 "ExitCode": 4, 

286 } 

287 results = WmsSpecificInfo() 

288 report_utils._add_service_job_specific_info(job_ad, results) 

289 self.assertEqual( 

290 results.context, {"job_name": "provisioningJob", "status": "FAILED", "status_details": ""} 

291 ) 

292 

293 def testDeleted(self): 

294 # Deleted by user (never held) 

295 job_ad = { 

296 "ClusterId": 9086, 

297 "DAGNodeName": "provisioningJob", 

298 "JobStatus": JobStatus.REMOVED, 

299 "ProcId": 0, 

300 "Reason": "via condor_rm (by user mgower)", 

301 "job_evicted_time": "2025-02-11T11:35:04", 

302 } 

303 results = WmsSpecificInfo() 

304 report_utils._add_service_job_specific_info(job_ad, results) 

305 self.assertEqual( 

306 results.context, {"job_name": "provisioningJob", "status": "DELETED", "status_details": ""} 

307 ) 

308 

309 def testSucceedEarly(self): 

310 # DAG hasn't completed (Running or held), 

311 # Service job completed with exit code 0 

312 job_ad = { 

313 "ClusterId": 8761, 

314 "ProcId": 0, 

315 "DAGNodeName": "provisioningJob", 

316 "JobStatus": JobStatus.COMPLETED, 

317 "ExitCode": 0, 

318 } 

319 results = WmsSpecificInfo() 

320 report_utils._add_service_job_specific_info(job_ad, results) 

321 self.assertEqual( 

322 results.context, 

323 { 

324 "job_name": "provisioningJob", 

325 "status": "SUCCEEDED", 

326 "status_details": "(Note: Finished before workflow.)", 

327 }, 

328 ) 

329 

330 def testSucceedOldRemoveMessage(self): 

331 # DAG completed, job was in running state when removed. 

332 job_ad = { 

333 "ClusterId": 8761, 

334 "ProcId": 0, 

335 "DAGNodeName": "provisioningJob", 

336 "JobStatus": JobStatus.REMOVED, 

337 "Reason": "Removed by DAGMan (by user mgower)", 

338 } 

339 results = WmsSpecificInfo() 

340 report_utils._add_service_job_specific_info(job_ad, results) 

341 self.assertEqual( 

342 results.context, {"job_name": "provisioningJob", "status": "SUCCEEDED", "status_details": ""} 

343 ) 

344 

345 def testSucceed(self): 

346 # DAG completed, job was in running state when removed. 

347 job_ad = { 

348 "ClusterId": 8761, 

349 "ProcId": 0, 

350 "DAGNodeName": "provisioningJob", 

351 "JobStatus": JobStatus.REMOVED, 

352 "Reason": ( 

353 "removed because <OtherJobRemoveRequirements = DAGManJobId =?= 8556>" 

354 " fired when job (8556.0) was removed" 

355 ), 

356 } 

357 results = WmsSpecificInfo() 

358 report_utils._add_service_job_specific_info(job_ad, results) 

359 self.assertEqual( 

360 results.context, {"job_name": "provisioningJob", "status": "SUCCEEDED", "status_details": ""} 

361 ) 

362 

363 def testUserHeldWhileRunning(self): 

364 # DAG hasn't completed (Running or held), 

365 # user put at least service job on hold 

366 job_ad = { 

367 "ClusterId": 8523, 

368 "ProcId": 0, 

369 "DAGNodeName": "provisioningJob", 

370 "JobStatus": JobStatus.HELD, 

371 "HoldReason": "via condor_hold (by user mgower)", 

372 "HoldReasonCode": 1, 

373 "HoldReasonSubCode": 0, 

374 } 

375 

376 results = WmsSpecificInfo() 

377 report_utils._add_service_job_specific_info(job_ad, results) 

378 self.assertEqual( 

379 results.context, 

380 { 

381 "job_name": "provisioningJob", 

382 "status": "HELD", 

383 "status_details": "(via condor_hold (by user mgower))", 

384 }, 

385 ) 

386 

387 def testHeldByHTC(self): 

388 # Job put on hold by HTCondor, removed when DAG ends 

389 job_ad = { 

390 "ClusterId": 8693, 

391 "DAGNodeName": "provisioningJob", 

392 "HoldReason": "Failed to execute", 

393 "HoldReasonCode": 6, 

394 "HoldReasonSubCode": 2, 

395 "JobStatus": JobStatus.REMOVED, 

396 "ProcId": 0, 

397 "Reason": "Removed by DAGMan (by user mgower)", 

398 "job_held_time": "2025-02-07T12:50:07", 

399 } 

400 results = WmsSpecificInfo() 

401 report_utils._add_service_job_specific_info(job_ad, results) 

402 self.assertEqual( 

403 results.context, 

404 { 

405 "job_name": "provisioningJob", 

406 "status": "DELETED", 

407 "status_details": "(Job was held for the following reason: Failed to execute)", 

408 }, 

409 ) 

410 

411 def testHeldReleasedRunning(self): 

412 # DAG hasn't completed (Running or held), 

413 # Since held info will be in job_ad, make sure knows released. 

414 job_ad = { 

415 "ClusterId": 8625, 

416 "DAGNodeName": "provisioningJob", 

417 "HoldReason": "via condor_hold (by user mgower)", 

418 "HoldReasonCode": 1, 

419 "HoldReasonSubCode": 0, 

420 "JobStatus": JobStatus.RUNNING, 

421 "LogNotes": "DAG Node: provisioningJob", 

422 "ProcId": 0, 

423 "job_held_time": "2025-02-07T12:33:34", 

424 "job_released_time": "2025-02-07T12:33:47", 

425 } 

426 results = WmsSpecificInfo() 

427 report_utils._add_service_job_specific_info(job_ad, results) 

428 self.assertEqual( 

429 results.context, {"job_name": "provisioningJob", "status": "RUNNING", "status_details": ""} 

430 ) 

431 

432 def testHeldReleasedDied(self): 

433 # Since held info will be in job_ad, 

434 # make sure knows status after released. 

435 job_ad = { 

436 "ClusterId": 9120, 

437 "DAGNodeName": "provisioningJob", 

438 "ExitBySignal": False, 

439 "ExitCode": 4, 

440 "HoldReason": "via condor_hold (by user mgower)", 

441 "HoldReasonCode": 1, 

442 "HoldReasonSubCode": 0, 

443 "JobStatus": JobStatus.COMPLETED, 

444 "ProcId": 0, 

445 "Reason": "via condor_release (by user mgower)", 

446 "ReturnValue": 4, 

447 "TerminatedNormally": True, 

448 "job_held_time": "2025-02-11T11:46:40", 

449 "job_released_time": "2025-02-11T11:46:47", 

450 } 

451 results = WmsSpecificInfo() 

452 report_utils._add_service_job_specific_info(job_ad, results) 

453 self.assertEqual( 

454 results.context, {"job_name": "provisioningJob", "status": "FAILED", "status_details": ""} 

455 ) 

456 

457 def testHeldReleasedSuccessEarly(self): 

458 # Since held info will be in job_ad, 

459 # make sure knows status after released. 

460 job_ad = { 

461 "ClusterId": 9154, 

462 "DAGNodeName": "provisioningJob", 

463 "ExitBySignal": False, 

464 "ExitCode": 0, 

465 "HoldReason": "via condor_hold (by user mgower)", 

466 "HoldReasonCode": 1, 

467 "HoldReasonSubCode": 0, 

468 "JobStatus": JobStatus.COMPLETED, 

469 "ProcId": 0, 

470 "Reason": "via condor_release (by user mgower)", 

471 "TerminatedNormally": True, 

472 "job_held_time": "2025-02-11T11:55:20", 

473 "job_released_time": "2025-02-11T11:55:25", 

474 } 

475 results = WmsSpecificInfo() 

476 report_utils._add_service_job_specific_info(job_ad, results) 

477 self.assertEqual( 

478 results.context, 

479 { 

480 "job_name": "provisioningJob", 

481 "status": "SUCCEEDED", 

482 "status_details": "(Note: Finished before workflow.)", 

483 }, 

484 ) 

485 

486 def testHeldReleasedSuccess(self): 

487 # DAG has completed. 

488 # Since held info will be in job_ad, 

489 # make sure knows status after released. 

490 job_ad = { 

491 "ClusterId": 8625, 

492 "DAGNodeName": "provisioningJob", 

493 "HoldReason": "via condor_hold (by user mgower)", 

494 "HoldReasonCode": 1, 

495 "HoldReasonSubCode": 0, 

496 "JobStatus": JobStatus.REMOVED, 

497 "ProcId": 0, 

498 "Reason": "removed because <OtherJobRemoveRequirements = DAGManJobId =?= " 

499 "8624> fired when job (8624.0) was removed", 

500 "job_held_time": "2025-02-07T12:33:34", 

501 "job_released_time": "2025-02-07T12:33:47", 

502 } 

503 results = WmsSpecificInfo() 

504 report_utils._add_service_job_specific_info(job_ad, results) 

505 self.assertEqual( 

506 results.context, {"job_name": "provisioningJob", "status": "SUCCEEDED", "status_details": ""} 

507 ) 

508 

509 def testHeldReleasedDeleted(self): 

510 # Since held info will be in job_ad, 

511 # make sure knows status after released. 

512 job_ad = { 

513 "ClusterId": 9086, 

514 "DAGNodeName": "provisioningJob", 

515 "HoldReason": "via condor_hold (by user mgower)", 

516 "HoldReasonCode": 1, 

517 "HoldReasonSubCode": 0, 

518 "JobStatus": JobStatus.REMOVED, 

519 "ProcId": 0, 

520 "Reason": "via condor_rm (by user mgower)", 

521 "job_evicted_time": "2025-02-11T11:35:04", 

522 "job_held_time": "2025-02-11T11:35:04", 

523 } 

524 results = WmsSpecificInfo() 

525 report_utils._add_service_job_specific_info(job_ad, results) 

526 self.assertEqual( 

527 results.context, {"job_name": "provisioningJob", "status": "DELETED", "status_details": ""} 

528 ) 

529 

530 def testHeldReleasedHeld(self): 

531 # Since release info will be in job_ad, 

532 # make sure knows held after release. 

533 job_ad = { 

534 "ClusterId": 8659, 

535 "DAGNodeName": "provisioningJob", 

536 "HoldReason": "via condor_hold (by user mgower)", 

537 "HoldReasonCode": 1, 

538 "HoldReasonSubCode": 0, 

539 "JobStatus": JobStatus.REMOVED, 

540 "ProcId": 0, 

541 "Reason": "Removed by DAGMan (by user mgower)", 

542 "TerminatedNormally": False, 

543 "job_held_time": "2025-02-07T12:36:15", 

544 "job_released_time": "2025-02-07T12:36:07", 

545 } 

546 results = WmsSpecificInfo() 

547 report_utils._add_service_job_specific_info(job_ad, results) 

548 self.assertEqual( 

549 results.context, 

550 { 

551 "job_name": "provisioningJob", 

552 "status": "DELETED", 

553 "status_details": "(Job was held for the following reason: via condor_hold (by user mgower))", 

554 }, 

555 ) 

556 

557 

558class GetRunSummaryTestCase(unittest.TestCase): 

559 """Test _get_run_summary function.""" 

560 

561 def testJobSummaryInJobAd(self): 

562 summary = "pipetaskInit:1;label1:2;label2:2;finalJob:1" 

563 job_ad = {"ClusterId": 8659, "DAGNodeName": "testJob", "bps_job_summary": summary} 

564 results = report_utils._get_run_summary(job_ad) 

565 self.assertEqual(results, summary) 

566 

567 def testRunSummaryInJobAd(self): 

568 summary = "pipetaskInit:1;label1:2;label2:2;finalJob:1" 

569 job_ad = {"ClusterId": 8659, "DAGNodeName": "testJob", "bps_run_summary": summary} 

570 results = report_utils._get_run_summary(job_ad) 

571 self.assertEqual(results, summary) 

572 

573 def testSummaryFromDag(self): 

574 with temporaryDirectory() as tmp_dir: 

575 copy2(f"{TESTDIR}/data/good.dag", tmp_dir) 

576 job_ad = {"ClusterId": 8659, "DAGNodeName": "testJob", "Iwd": tmp_dir} 

577 results = report_utils._get_run_summary(job_ad) 

578 self.assertEqual(results, "pipetaskInit:1;label1:1;label2:1;label3:1;finalJob:1") 

579 

580 def testSummaryNoDag(self): 

581 with self.assertLogs(logger=logger, level="WARNING") as cm: 

582 with temporaryDirectory() as tmp_dir: 

583 job_ad = {"ClusterId": 8659, "DAGNodeName": "testJob", "Iwd": tmp_dir} 

584 results = report_utils._get_run_summary(job_ad) 

585 self.assertEqual(results, "") 

586 self.assertIn("lsst.ctrl.bps.htcondor", cm.records[0].name) 

587 self.assertIn("Could not get run summary for htcondor job", cm.output[0]) 

588 

589 

590class IsServiceJobTestCase(unittest.TestCase): 

591 """Test is_service_job function.""" 

592 

593 def testNotServiceJob(self): 

594 job_ad = {"ClusterId": 8659, "DAGNodeName": "testJob", "wms_node_type": lssthtc.WmsNodeType.PAYLOAD} 

595 self.assertFalse(report_utils.is_service_job(job_ad)) 

596 

597 def testIsServiceJob(self): 

598 job_ad = {"ClusterId": 8659, "DAGNodeName": "testJob", "wms_node_type": lssthtc.WmsNodeType.SERVICE} 

599 self.assertTrue(report_utils.is_service_job(job_ad)) 

600 

601 def testMissingBpsType(self): 

602 job_ad = { 

603 "ClusterId": 8659, 

604 "DAGNodeName": "testJob", 

605 } 

606 self.assertFalse(report_utils.is_service_job(job_ad)) 

607 

608 

609class CreateDetailedReportFromJobsTestCase(unittest.TestCase): 

610 """Test _create_detailed_report_from_jobs function.""" 

611 

612 def testTinySuccess(self): 

613 with temporaryDirectory() as tmp_dir: 

614 test_submit_dir = os.path.join(tmp_dir, "tiny_success") 

615 copytree(f"{TESTDIR}/data/tiny_success", test_submit_dir) 

616 wms_workflow_id, jobs, message = report_utils._get_info_from_path(test_submit_dir) 

617 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

618 self.assertEqual(len(run_reports), 1) 

619 report = run_reports[wms_workflow_id] 

620 self.assertEqual(report.wms_id, wms_workflow_id) 

621 self.assertEqual(report.state, WmsStates.SUCCEEDED) 

622 self.assertTrue(os.path.samefile(report.path, test_submit_dir)) 

623 self.assertEqual(report.run_summary, "pipetaskInit:1;label1:1;label2:1;finalJob:1") 

624 self.assertEqual( 

625 report.job_state_counts, 

626 { 

627 WmsStates.UNKNOWN: 0, 

628 WmsStates.MISFIT: 0, 

629 WmsStates.UNREADY: 0, 

630 WmsStates.READY: 0, 

631 WmsStates.PENDING: 0, 

632 WmsStates.RUNNING: 0, 

633 WmsStates.DELETED: 0, 

634 WmsStates.HELD: 0, 

635 WmsStates.SUCCEEDED: 4, 

636 WmsStates.FAILED: 0, 

637 WmsStates.PRUNED: 0, 

638 }, 

639 ) 

640 self.assertEqual( 

641 report.specific_info.context, 

642 {"job_name": "provisioningJob", "status": "SUCCEEDED", "status_details": ""}, 

643 ) 

644 

645 def testTinyProblems(self): 

646 with temporaryDirectory() as tmp_dir: 

647 test_submit_dir = os.path.join(tmp_dir, "tiny_problems") 

648 copytree(f"{TESTDIR}/data/tiny_problems", test_submit_dir) 

649 wms_workflow_id, jobs, message = report_utils._get_info_from_path(test_submit_dir) 

650 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

651 self.assertEqual(len(run_reports), 1) 

652 report = run_reports[wms_workflow_id] 

653 self.assertEqual(report.wms_id, wms_workflow_id) 

654 self.assertEqual(report.state, WmsStates.FAILED) 

655 self.assertTrue(os.path.samefile(report.path, test_submit_dir)) 

656 self.assertEqual(report.run_summary, "pipetaskInit:1;label1:2;label2:2;finalJob:1") 

657 self.assertEqual( 

658 report.job_state_counts, 

659 { 

660 WmsStates.UNKNOWN: 0, 

661 WmsStates.MISFIT: 0, 

662 WmsStates.UNREADY: 0, 

663 WmsStates.READY: 0, 

664 WmsStates.PENDING: 0, 

665 WmsStates.RUNNING: 0, 

666 WmsStates.DELETED: 0, 

667 WmsStates.HELD: 0, 

668 WmsStates.SUCCEEDED: 4, 

669 WmsStates.FAILED: 1, 

670 WmsStates.PRUNED: 1, 

671 }, 

672 ) 

673 self.assertEqual( 

674 run_reports[wms_workflow_id].specific_info.context, 

675 {"job_name": "provisioningJob", "status": "SUCCEEDED", "status_details": ""}, 

676 ) 

677 

678 def testTinyRunning(self): 

679 with temporaryDirectory() as tmp_dir: 

680 test_submit_dir = os.path.join(tmp_dir, "tiny_running") 

681 copytree(f"{TESTDIR}/data/tiny_running", test_submit_dir) 

682 wms_workflow_id, jobs, message = report_utils._get_info_from_path(test_submit_dir) 

683 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

684 self.assertEqual(len(run_reports), 1) 

685 report = run_reports[wms_workflow_id] 

686 self.assertEqual(report.wms_id, wms_workflow_id) 

687 self.assertEqual(report.state, WmsStates.RUNNING) 

688 self.assertTrue(os.path.samefile(report.path, test_submit_dir)) 

689 self.assertEqual(report.run_summary, "pipetaskInit:1;label1:1;label2:1;finalJob:1") 

690 self.assertEqual( 

691 report.job_state_counts, 

692 { 

693 WmsStates.UNKNOWN: 0, 

694 WmsStates.MISFIT: 0, 

695 WmsStates.UNREADY: 2, 

696 WmsStates.READY: 0, 

697 WmsStates.PENDING: 0, 

698 WmsStates.RUNNING: 1, 

699 WmsStates.DELETED: 0, 

700 WmsStates.HELD: 0, 

701 WmsStates.SUCCEEDED: 1, 

702 WmsStates.FAILED: 0, 

703 WmsStates.PRUNED: 0, 

704 }, 

705 ) 

706 self.assertEqual( 

707 report.specific_info.context, 

708 {"job_name": "provisioningJob", "status": "RUNNING", "status_details": ""}, 

709 ) 

710 

711 def testNoopRunning(self): 

712 with temporaryDirectory() as tmp_dir: 

713 test_submit_dir = os.path.join(tmp_dir, "noop_running_1") 

714 copytree(f"{TESTDIR}/data/noop_running_1", test_submit_dir) 

715 wms_workflow_id, jobs, message = report_utils._get_info_from_path(test_submit_dir) 

716 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

717 self.assertEqual(len(run_reports), 1) 

718 report = run_reports[wms_workflow_id] 

719 self.assertEqual(report.wms_id, wms_workflow_id) 

720 self.assertEqual(report.state, WmsStates.RUNNING) 

721 self.assertTrue(os.path.samefile(report.path, test_submit_dir)) 

722 self.assertEqual( 

723 set(report.run_summary.split(";")), 

724 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"}, 

725 ) 

726 self.assertEqual( 

727 report.job_state_counts, 

728 { 

729 WmsStates.UNKNOWN: 0, 

730 WmsStates.MISFIT: 0, 

731 WmsStates.UNREADY: 12, 

732 WmsStates.READY: 0, 

733 WmsStates.PENDING: 1, 

734 WmsStates.RUNNING: 10, 

735 WmsStates.DELETED: 0, 

736 WmsStates.HELD: 0, 

737 WmsStates.SUCCEEDED: 9, 

738 WmsStates.FAILED: 0, 

739 WmsStates.PRUNED: 0, 

740 }, 

741 ) 

742 self.assertEqual(report.total_number_jobs, 32) 

743 self.assertIsNone(report.specific_info) 

744 

745 def testNoopFailed(self): 

746 with temporaryDirectory() as tmp_dir: 

747 test_submit_dir = os.path.join(tmp_dir, "noop_failed_1") 

748 copytree(f"{TESTDIR}/data/noop_failed_1", test_submit_dir) 

749 wms_workflow_id, jobs, message = report_utils._get_info_from_path(test_submit_dir) 

750 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

751 self.assertEqual(len(run_reports), 1) 

752 report = run_reports[wms_workflow_id] 

753 self.assertEqual(report.wms_id, wms_workflow_id) 

754 self.assertEqual(report.state, WmsStates.FAILED) 

755 self.assertTrue(os.path.samefile(report.path, test_submit_dir)) 

756 self.assertEqual( 

757 set(report.run_summary.split(";")), 

758 {"pipetaskInit:1", "label1:6", "label2:6", "label3:6", "label4:6", "label5:6", "finalJob:1"}, 

759 ) 

760 self.assertEqual( 

761 report.job_state_counts, 

762 { 

763 WmsStates.UNKNOWN: 0, 

764 WmsStates.MISFIT: 0, 

765 WmsStates.UNREADY: 0, 

766 WmsStates.READY: 0, 

767 WmsStates.PENDING: 0, 

768 WmsStates.RUNNING: 0, 

769 WmsStates.DELETED: 0, 

770 WmsStates.HELD: 0, 

771 WmsStates.SUCCEEDED: 27, 

772 WmsStates.FAILED: 1, 

773 WmsStates.PRUNED: 4, 

774 }, 

775 ) 

776 self.assertEqual(report.total_number_jobs, 32) 

777 self.assertIsNone(report.specific_info) 

778 self.assertEqual( 

779 report.exit_code_summary, 

780 { 

781 "pipetaskInit": [], 

782 "label1": [], 

783 "label2": [1], 

784 "label3": [], 

785 "label4": [], 

786 "label5": [], 

787 "finalJob": [], 

788 }, 

789 ) 

790 

791 

792class GetStatusFromIdTestCase(unittest.TestCase): 

793 """Test _get_status_from_id function.""" 

794 

795 @patch("lsst.ctrl.bps.htcondor.report_utils._get_info_from_schedd") 

796 def testNotFound(self, mock_get): 

797 mock_get.return_value = {} 

798 

799 state, message = report_utils._get_status_from_id("100", 0, {}) 

800 

801 mock_get.assert_called_once_with("100", 0, {}, skip_history=True) 

802 

803 self.assertEqual(state, WmsStates.UNKNOWN)