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 

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 

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

47 

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

49 

50LOCATE_SUCCESS = """[ 

51 CondorPlatform = "$CondorPlatform: X86_64-CentOS_7.9 $"; 

52 MyType = "Scheduler"; 

53 Machine = "testmachine"; 

54 Name = "testmachine"; 

55 CondorVersion = "$CondorVersion: 23.0.3 2024-04-04 $"; 

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

57 ] 

58""" 

59 

60 

61class GetExitCodeSummaryTestCase(unittest.TestCase): 

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

63 

64 def setUp(self): 

65 self.jobs = { 

66 "1.0": { 

67 "JobStatus": JobStatus.IDLE, 

68 "bps_job_label": "foo", 

69 }, 

70 "2.0": { 

71 "JobStatus": JobStatus.RUNNING, 

72 "bps_job_label": "foo", 

73 }, 

74 "3.0": { 

75 "JobStatus": JobStatus.REMOVED, 

76 "bps_job_label": "foo", 

77 }, 

78 "4.0": { 

79 "ExitCode": 0, 

80 "ExitBySignal": False, 

81 "JobStatus": JobStatus.COMPLETED, 

82 "bps_job_label": "bar", 

83 }, 

84 "5.0": { 

85 "ExitCode": 1, 

86 "ExitBySignal": False, 

87 "JobStatus": JobStatus.COMPLETED, 

88 "bps_job_label": "bar", 

89 }, 

90 "6.0": { 

91 "ExitBySignal": True, 

92 "ExitSignal": 11, 

93 "JobStatus": JobStatus.HELD, 

94 "bps_job_label": "baz", 

95 }, 

96 "7.0": { 

97 "ExitBySignal": False, 

98 "ExitCode": 42, 

99 "JobStatus": JobStatus.HELD, 

100 "bps_job_label": "baz", 

101 }, 

102 "8.0": { 

103 "JobStatus": JobStatus.TRANSFERRING_OUTPUT, 

104 "bps_job_label": "qux", 

105 }, 

106 "9.0": { 

107 "JobStatus": JobStatus.SUSPENDED, 

108 "bps_job_label": "qux", 

109 }, 

110 } 

111 

112 def tearDown(self): 

113 pass 

114 

115 def testMainScenario(self): 

116 actual = report_utils._get_exit_code_summary(self.jobs) 

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

118 self.assertEqual(actual, expected) 

119 

120 def testUnknownStatus(self): 

121 jobs = { 

122 "1.0": { 

123 "JobStatus": -1, 

124 "bps_job_label": "foo", 

125 } 

126 } 

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

128 report_utils._get_exit_code_summary(jobs) 

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

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

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

132 

133 def testUnknownKey(self): 

134 jobs = { 

135 "1.0": { 

136 "JobStatus": JobStatus.COMPLETED, 

137 "UnknownKey": None, 

138 "bps_job_label": "foo", 

139 } 

140 } 

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

142 report_utils._get_exit_code_summary(jobs) 

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

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

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

146 

147 

148class GetStateCountsFromDagJobTestCase(unittest.TestCase): 

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

150 

151 def setUp(self): 

152 pass 

153 

154 def tearDown(self): 

155 pass 

156 

157 def testCounts(self): 

158 job = { 

159 "DAG_NodesDone": 1, 

160 "DAG_JobsHeld": 2, 

161 "DAG_NodesFailed": 3, 

162 "DAG_NodesFutile": 4, 

163 "DAG_NodesQueued": 5, 

164 "DAG_NodesReady": 0, 

165 "DAG_NodesUnready": 7, 

166 "DAG_NodesTotal": 22, 

167 } 

168 

169 truth = { 

170 WmsStates.SUCCEEDED: 1, 

171 WmsStates.HELD: 2, 

172 WmsStates.UNREADY: 7, 

173 WmsStates.READY: 0, 

174 WmsStates.FAILED: 3, 

175 WmsStates.PRUNED: 4, 

176 WmsStates.MISFIT: 0, 

177 } 

178 

179 total, result = report_utils._get_state_counts_from_dag_job(job) 

180 self.assertEqual(total, 22) 

181 self.assertEqual(result, truth) 

182 

183 

184class GetInfoFromPathTestCase(unittest.TestCase): 

185 """Test _get_info_from_path function.""" 

186 

187 def test_tmpdir_abort(self): 

188 with temporaryDirectory() as tmp_dir: 

189 copy2(f"{TESTDIR}/data/test_tmpdir_abort.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("Cannot submit from /tmp", message) 

194 

195 def test_no_dagman_messages(self): 

196 with temporaryDirectory() as tmp_dir: 

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

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

199 self.assertEqual(wms_workflow_id, lssthtc.MISSING_ID) 

200 self.assertEqual(jobs, {}) 

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

202 

203 def test_successful_run(self): 

204 with temporaryDirectory() as tmp_dir: 

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

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

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

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

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

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

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

212 self.assertEqual(wms_workflow_id, "1163.0") 

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

214 self.assertEqual(message, "") 

215 

216 def test_relative_path(self): 

217 orig_dir = Path.cwd() 

218 with temporaryDirectory() as tmp_dir: 

219 os.chdir(tmp_dir) 

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

221 abs_path.mkdir() 

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

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

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

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

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

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

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

229 self.assertEqual(wms_workflow_id, "1163.0") 

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

231 self.assertEqual(message, "") 

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

233 os.chdir(orig_dir) 

234 

235 

236class AddServiceJobSpecificInfoTestCase(unittest.TestCase): 

237 """Test _add_service_job_specific_info function. 

238 

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

240 values in the dictionaries come from plugin code as 

241 well as HTCondor. Changes in either of those codes 

242 that produce data for the job_ad can break this 

243 function without breaking these unit tests. 

244 

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

246 various cases with and without job being held just to 

247 ensure get right status in both cases. 

248 """ 

249 

250 def testNotSubmitted(self): 

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

252 # (Typically an plugin bug.) 

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

254 # yet or problem so it never will. 

255 job_ad = { 

256 "ClusterId": -64, 

257 "DAGManJobID": "8997.0", 

258 "DAGNodeName": "provisioningJob", 

259 "NodeStatus": lssthtc.NodeStatus.NOT_READY, 

260 "ProcId": 0, 

261 "bps_job_label": "service_provisioningJob", 

262 } 

263 results = WmsSpecificInfo() 

264 report_utils._add_service_job_specific_info(job_ad, results) 

265 self.assertEqual( 

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

267 ) 

268 

269 def testRunning(self): 

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

271 # Service job is running. 

272 job_ad = { 

273 "ClusterId": 8523, 

274 "ProcId": 0, 

275 "DAGNodeName": "provisioningJob", 

276 "JobStatus": JobStatus.RUNNING, 

277 } 

278 

279 results = WmsSpecificInfo() 

280 report_utils._add_service_job_specific_info(job_ad, results) 

281 self.assertEqual( 

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

283 ) 

284 

285 def testDied(self): 

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

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

288 job_ad = { 

289 "ClusterId": 8761, 

290 "ProcId": 0, 

291 "DAGNodeName": "provisioningJob", 

292 "JobStatus": JobStatus.COMPLETED, 

293 "ExitCode": 4, 

294 } 

295 results = WmsSpecificInfo() 

296 report_utils._add_service_job_specific_info(job_ad, results) 

297 self.assertEqual( 

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

299 ) 

300 

301 def testDeleted(self): 

302 # Deleted by user (never held) 

303 job_ad = { 

304 "ClusterId": 9086, 

305 "DAGNodeName": "provisioningJob", 

306 "JobStatus": JobStatus.REMOVED, 

307 "ProcId": 0, 

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

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

310 } 

311 results = WmsSpecificInfo() 

312 report_utils._add_service_job_specific_info(job_ad, results) 

313 self.assertEqual( 

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

315 ) 

316 

317 def testSucceedEarly(self): 

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

319 # Service job completed with exit code 0 

320 job_ad = { 

321 "ClusterId": 8761, 

322 "ProcId": 0, 

323 "DAGNodeName": "provisioningJob", 

324 "JobStatus": JobStatus.COMPLETED, 

325 "ExitCode": 0, 

326 } 

327 results = WmsSpecificInfo() 

328 report_utils._add_service_job_specific_info(job_ad, results) 

329 self.assertEqual( 

330 results.context, 

331 { 

332 "job_name": "provisioningJob", 

333 "status": "SUCCEEDED", 

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

335 }, 

336 ) 

337 

338 def testSucceedOldRemoveMessage(self): 

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

340 job_ad = { 

341 "ClusterId": 8761, 

342 "ProcId": 0, 

343 "DAGNodeName": "provisioningJob", 

344 "JobStatus": JobStatus.REMOVED, 

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

346 } 

347 results = WmsSpecificInfo() 

348 report_utils._add_service_job_specific_info(job_ad, results) 

349 self.assertEqual( 

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

351 ) 

352 

353 def testSucceed(self): 

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

355 job_ad = { 

356 "ClusterId": 8761, 

357 "ProcId": 0, 

358 "DAGNodeName": "provisioningJob", 

359 "JobStatus": JobStatus.REMOVED, 

360 "Reason": ( 

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

362 " fired when job (8556.0) was removed" 

363 ), 

364 } 

365 results = WmsSpecificInfo() 

366 report_utils._add_service_job_specific_info(job_ad, results) 

367 self.assertEqual( 

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

369 ) 

370 

371 def testUserHeldWhileRunning(self): 

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

373 # user put at least service job on hold 

374 job_ad = { 

375 "ClusterId": 8523, 

376 "ProcId": 0, 

377 "DAGNodeName": "provisioningJob", 

378 "JobStatus": JobStatus.HELD, 

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

380 "HoldReasonCode": 1, 

381 "HoldReasonSubCode": 0, 

382 } 

383 

384 results = WmsSpecificInfo() 

385 report_utils._add_service_job_specific_info(job_ad, results) 

386 self.assertEqual( 

387 results.context, 

388 { 

389 "job_name": "provisioningJob", 

390 "status": "HELD", 

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

392 }, 

393 ) 

394 

395 def testHeldByHTC(self): 

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

397 job_ad = { 

398 "ClusterId": 8693, 

399 "DAGNodeName": "provisioningJob", 

400 "HoldReason": "Failed to execute", 

401 "HoldReasonCode": 6, 

402 "HoldReasonSubCode": 2, 

403 "JobStatus": JobStatus.REMOVED, 

404 "ProcId": 0, 

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

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

407 } 

408 results = WmsSpecificInfo() 

409 report_utils._add_service_job_specific_info(job_ad, results) 

410 self.assertEqual( 

411 results.context, 

412 { 

413 "job_name": "provisioningJob", 

414 "status": "DELETED", 

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

416 }, 

417 ) 

418 

419 def testHeldReleasedRunning(self): 

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

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

422 job_ad = { 

423 "ClusterId": 8625, 

424 "DAGNodeName": "provisioningJob", 

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

426 "HoldReasonCode": 1, 

427 "HoldReasonSubCode": 0, 

428 "JobStatus": JobStatus.RUNNING, 

429 "LogNotes": "DAG Node: provisioningJob", 

430 "ProcId": 0, 

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

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

433 } 

434 results = WmsSpecificInfo() 

435 report_utils._add_service_job_specific_info(job_ad, results) 

436 self.assertEqual( 

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

438 ) 

439 

440 def testHeldReleasedDied(self): 

441 # Since held info will be in job_ad, 

442 # make sure knows status after released. 

443 job_ad = { 

444 "ClusterId": 9120, 

445 "DAGNodeName": "provisioningJob", 

446 "ExitBySignal": False, 

447 "ExitCode": 4, 

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

449 "HoldReasonCode": 1, 

450 "HoldReasonSubCode": 0, 

451 "JobStatus": JobStatus.COMPLETED, 

452 "ProcId": 0, 

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

454 "ReturnValue": 4, 

455 "TerminatedNormally": True, 

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

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

458 } 

459 results = WmsSpecificInfo() 

460 report_utils._add_service_job_specific_info(job_ad, results) 

461 self.assertEqual( 

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

463 ) 

464 

465 def testHeldReleasedSuccessEarly(self): 

466 # Since held info will be in job_ad, 

467 # make sure knows status after released. 

468 job_ad = { 

469 "ClusterId": 9154, 

470 "DAGNodeName": "provisioningJob", 

471 "ExitBySignal": False, 

472 "ExitCode": 0, 

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

474 "HoldReasonCode": 1, 

475 "HoldReasonSubCode": 0, 

476 "JobStatus": JobStatus.COMPLETED, 

477 "ProcId": 0, 

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

479 "TerminatedNormally": True, 

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

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

482 } 

483 results = WmsSpecificInfo() 

484 report_utils._add_service_job_specific_info(job_ad, results) 

485 self.assertEqual( 

486 results.context, 

487 { 

488 "job_name": "provisioningJob", 

489 "status": "SUCCEEDED", 

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

491 }, 

492 ) 

493 

494 def testHeldReleasedSuccess(self): 

495 # DAG has completed. 

496 # Since held info will be in job_ad, 

497 # make sure knows status after released. 

498 job_ad = { 

499 "ClusterId": 8625, 

500 "DAGNodeName": "provisioningJob", 

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

502 "HoldReasonCode": 1, 

503 "HoldReasonSubCode": 0, 

504 "JobStatus": JobStatus.REMOVED, 

505 "ProcId": 0, 

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

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

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

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

510 } 

511 results = WmsSpecificInfo() 

512 report_utils._add_service_job_specific_info(job_ad, results) 

513 self.assertEqual( 

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

515 ) 

516 

517 def testHeldReleasedDeleted(self): 

518 # Since held info will be in job_ad, 

519 # make sure knows status after released. 

520 job_ad = { 

521 "ClusterId": 9086, 

522 "DAGNodeName": "provisioningJob", 

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

524 "HoldReasonCode": 1, 

525 "HoldReasonSubCode": 0, 

526 "JobStatus": JobStatus.REMOVED, 

527 "ProcId": 0, 

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

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

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

531 } 

532 results = WmsSpecificInfo() 

533 report_utils._add_service_job_specific_info(job_ad, results) 

534 self.assertEqual( 

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

536 ) 

537 

538 def testHeldReleasedHeld(self): 

539 # Since release info will be in job_ad, 

540 # make sure knows held after release. 

541 job_ad = { 

542 "ClusterId": 8659, 

543 "DAGNodeName": "provisioningJob", 

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

545 "HoldReasonCode": 1, 

546 "HoldReasonSubCode": 0, 

547 "JobStatus": JobStatus.REMOVED, 

548 "ProcId": 0, 

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

550 "TerminatedNormally": False, 

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

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

553 } 

554 results = WmsSpecificInfo() 

555 report_utils._add_service_job_specific_info(job_ad, results) 

556 self.assertEqual( 

557 results.context, 

558 { 

559 "job_name": "provisioningJob", 

560 "status": "DELETED", 

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

562 }, 

563 ) 

564 

565 

566class GetRunSummaryTestCase(unittest.TestCase): 

567 """Test _get_run_summary function.""" 

568 

569 def testJobSummaryInJobAd(self): 

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

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

572 results = report_utils._get_run_summary(job_ad) 

573 self.assertEqual(results, summary) 

574 

575 def testRunSummaryInJobAd(self): 

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

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

578 results = report_utils._get_run_summary(job_ad) 

579 self.assertEqual(results, summary) 

580 

581 def testSummaryFromDag(self): 

582 with temporaryDirectory() as tmp_dir: 

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

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

585 results = report_utils._get_run_summary(job_ad) 

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

587 

588 def testSummaryNoDag(self): 

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

590 with temporaryDirectory() as tmp_dir: 

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

592 results = report_utils._get_run_summary(job_ad) 

593 self.assertEqual(results, "") 

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

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

596 

597 

598class IsServiceJobTestCase(unittest.TestCase): 

599 """Test is_service_job function.""" 

600 

601 def testNotServiceJob(self): 

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

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

604 

605 def testIsServiceJob(self): 

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

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

608 

609 def testMissingBpsType(self): 

610 job_ad = { 

611 "ClusterId": 8659, 

612 "DAGNodeName": "testJob", 

613 } 

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

615 

616 

617class CreateDetailedReportFromJobsTestCase(unittest.TestCase): 

618 """Test _create_detailed_report_from_jobs function.""" 

619 

620 def testTinySuccess(self): 

621 with temporaryDirectory() as tmp_dir: 

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

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

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

625 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

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

627 report = run_reports[wms_workflow_id] 

628 self.assertEqual(report.wms_id, wms_workflow_id) 

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

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

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

632 self.assertEqual( 

633 report.job_state_counts, 

634 { 

635 WmsStates.UNKNOWN: 0, 

636 WmsStates.MISFIT: 0, 

637 WmsStates.UNREADY: 0, 

638 WmsStates.READY: 0, 

639 WmsStates.PENDING: 0, 

640 WmsStates.RUNNING: 0, 

641 WmsStates.DELETED: 0, 

642 WmsStates.HELD: 0, 

643 WmsStates.SUCCEEDED: 4, 

644 WmsStates.FAILED: 0, 

645 WmsStates.PRUNED: 0, 

646 }, 

647 ) 

648 self.assertEqual( 

649 report.specific_info.context, 

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

651 ) 

652 

653 def testTinyProblems(self): 

654 with temporaryDirectory() as tmp_dir: 

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

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

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

658 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

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

660 report = run_reports[wms_workflow_id] 

661 self.assertEqual(report.wms_id, wms_workflow_id) 

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

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

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

665 self.assertEqual( 

666 report.job_state_counts, 

667 { 

668 WmsStates.UNKNOWN: 0, 

669 WmsStates.MISFIT: 0, 

670 WmsStates.UNREADY: 0, 

671 WmsStates.READY: 0, 

672 WmsStates.PENDING: 0, 

673 WmsStates.RUNNING: 0, 

674 WmsStates.DELETED: 0, 

675 WmsStates.HELD: 0, 

676 WmsStates.SUCCEEDED: 4, 

677 WmsStates.FAILED: 1, 

678 WmsStates.PRUNED: 1, 

679 }, 

680 ) 

681 self.assertEqual( 

682 run_reports[wms_workflow_id].specific_info.context, 

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

684 ) 

685 

686 def testTinyRunning(self): 

687 with temporaryDirectory() as tmp_dir: 

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

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

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

691 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

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

693 report = run_reports[wms_workflow_id] 

694 self.assertEqual(report.wms_id, wms_workflow_id) 

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

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

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

698 self.assertEqual( 

699 report.job_state_counts, 

700 { 

701 WmsStates.UNKNOWN: 0, 

702 WmsStates.MISFIT: 0, 

703 WmsStates.UNREADY: 2, 

704 WmsStates.READY: 0, 

705 WmsStates.PENDING: 0, 

706 WmsStates.RUNNING: 1, 

707 WmsStates.DELETED: 0, 

708 WmsStates.HELD: 0, 

709 WmsStates.SUCCEEDED: 1, 

710 WmsStates.FAILED: 0, 

711 WmsStates.PRUNED: 0, 

712 }, 

713 ) 

714 self.assertEqual( 

715 report.specific_info.context, 

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

717 ) 

718 

719 def testNoopRunning(self): 

720 with temporaryDirectory() as tmp_dir: 

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

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

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

724 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

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

726 report = run_reports[wms_workflow_id] 

727 self.assertEqual(report.wms_id, wms_workflow_id) 

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

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

730 self.assertEqual( 

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

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

733 ) 

734 self.assertEqual( 

735 report.job_state_counts, 

736 { 

737 WmsStates.UNKNOWN: 0, 

738 WmsStates.MISFIT: 0, 

739 WmsStates.UNREADY: 12, 

740 WmsStates.READY: 0, 

741 WmsStates.PENDING: 1, 

742 WmsStates.RUNNING: 10, 

743 WmsStates.DELETED: 0, 

744 WmsStates.HELD: 0, 

745 WmsStates.SUCCEEDED: 9, 

746 WmsStates.FAILED: 0, 

747 WmsStates.PRUNED: 0, 

748 }, 

749 ) 

750 self.assertEqual(report.total_number_jobs, 32) 

751 self.assertIsNone(report.specific_info) 

752 

753 def testNoopFailed(self): 

754 with temporaryDirectory() as tmp_dir: 

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

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

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

758 run_reports = report_utils._create_detailed_report_from_jobs(wms_workflow_id, jobs) 

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

760 report = run_reports[wms_workflow_id] 

761 self.assertEqual(report.wms_id, wms_workflow_id) 

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

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

764 self.assertEqual( 

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

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

767 ) 

768 self.assertEqual( 

769 report.job_state_counts, 

770 { 

771 WmsStates.UNKNOWN: 0, 

772 WmsStates.MISFIT: 0, 

773 WmsStates.UNREADY: 0, 

774 WmsStates.READY: 0, 

775 WmsStates.PENDING: 0, 

776 WmsStates.RUNNING: 0, 

777 WmsStates.DELETED: 0, 

778 WmsStates.HELD: 0, 

779 WmsStates.SUCCEEDED: 27, 

780 WmsStates.FAILED: 1, 

781 WmsStates.PRUNED: 4, 

782 }, 

783 ) 

784 self.assertEqual(report.total_number_jobs, 32) 

785 self.assertIsNone(report.specific_info) 

786 self.assertEqual( 

787 report.exit_code_summary, 

788 { 

789 "pipetaskInit": [], 

790 "label1": [], 

791 "label2": [1], 

792 "label3": [], 

793 "label4": [], 

794 "label5": [], 

795 "finalJob": [], 

796 }, 

797 ) 

798 

799 

800class GetStatusFromIdTestCase(unittest.TestCase): 

801 """Test _get_status_from_id function.""" 

802 

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

804 def testNotFound(self, mock_get): 

805 mock_get.return_value = {} 

806 

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