Coverage for tests/test_bps_reports.py: 100%

1# This file is part of ctrl_bps. 

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"""Tests for reporting mechanism.""" 

29 

30import dataclasses 

31import io 

32import logging 

33import unittest 

34 

35from astropy.table import Table 

36from wms_test_utils import TEST_REPORT 

37 

38from lsst.ctrl.bps import ( 

39 DISPLAY_SUMMARY_FIELDS, 

40 BaseRunReport, 

41 DetailedRunReport, 

42 ExitCodesReport, 

43 SummaryRunReport, 

44 WmsJobReport, 

45 WmsRunReport, 

46 WmsStates, 

47 compile_code_summary, 

48 compile_job_summary, 

49) 

50 

51 

52class FakeRunReport(BaseRunReport): 

53 """A fake run report.""" 

54 

55 def add(self, run_report, use_global_id=False): 

56 id_ = run_report.global_wms_id if use_global_id else run_report.wms_id 

57 self._table.add_row([id_, run_report.state.name]) 

58 

59 

60class FakeRunReportTestCase(unittest.TestCase): 

61 """Test shared methods.""" 

62 

63 def setUp(self): 

64 self.fields = [("ID", "S"), ("STATE", "S")] 

65 

66 self.report = FakeRunReport(self.fields) 

67 self.report.add(WmsRunReport(wms_id="2.0", state=WmsStates.RUNNING)) 

68 self.report.add(WmsRunReport(wms_id="1.0", state=WmsStates.SUCCEEDED)) 

69 

70 def testEquality(self): 

71 """Test if two reports are identical.""" 

72 other = FakeRunReport(self.fields) 

73 other.add(WmsRunReport(wms_id="2.0", state=WmsStates.RUNNING)) 

74 other.add(WmsRunReport(wms_id="1.0", state=WmsStates.SUCCEEDED)) 

75 self.assertEqual(self.report, other) 

76 

77 def testInequality(self): 

78 """Test if two reports are not identical.""" 

79 other = FakeRunReport(self.fields) 

80 other.add(WmsRunReport(wms_id="1.0", state=WmsStates.FAILED)) 

81 self.assertNotEqual(self.report, other) 

82 

83 def testLength(self): 

84 self.assertEqual(len(self.report), 2) 

85 

86 def testClear(self): 

87 """Test clearing the report.""" 

88 self.report.clear() 

89 self.assertEqual(len(self.report), 0) 

90 

91 def testSortWithKnownKey(self): 

92 """Test sorting the report using known column.""" 

93 expected_output = io.StringIO() 

94 expected = Table(dtype=self.fields) 

95 expected.add_row(["1.0", WmsStates.SUCCEEDED.name]) 

96 expected.add_row(["2.0", WmsStates.RUNNING.name]) 

97 print(expected, file=expected_output) 

98 

99 actual_output = io.StringIO() 

100 self.report.sort("ID") 

101 print(self.report, file=actual_output) 

102 

103 self.assertEqual(actual_output.getvalue(), expected_output.getvalue()) 

104 

105 expected_output.close() 

106 actual_output.close() 

107 

108 def testSortWithUnknownKey(self): 

109 """Test sorting the report using unknown column.""" 

110 with self.assertRaises(AttributeError): 

111 self.report.sort("foo") 

112 

113 

114class SummaryRunReportTestCase(unittest.TestCase): 

115 """Test a summary run report.""" 

116 

117 def setUp(self): 

118 self.fields = DISPLAY_SUMMARY_FIELDS 

119 self.run = WmsRunReport( 

120 wms_id="1.0", 

121 global_wms_id="foo#1.0", 

122 path="/path/to/run", 

123 label="label", 

124 run="run", 

125 site="SITE1", 

126 project="dev", 

127 campaign="testing", 

128 payload="test", 

129 operator="tester", 

130 run_summary="foo:1;bar:1", 

131 state=WmsStates.RUNNING, 

132 jobs=None, 

133 total_number_jobs=2, 

134 job_state_counts={ 

135 state: 1 if state in {WmsStates.SUCCEEDED, WmsStates.RUNNING} else 0 for state in WmsStates 

136 }, 

137 job_summary=None, 

138 ) 

139 self.report = SummaryRunReport(self.fields) 

140 

141 self.expected = Table(dtype=self.fields) 

142 self.expected.add_row( 

143 ["", "RUNNING", "50", "1.0", "tester", "dev", "testing", "SITE1", "test", "run"] 

144 ) 

145 

146 self.expected_output = io.StringIO() 

147 self.actual_output = io.StringIO() 

148 

149 def tearDown(self): 

150 self.expected_output.close() 

151 self.actual_output.close() 

152 

153 def testAddWithNoFlag(self): 

154 """Test adding a report for a run with no issues.""" 

155 self.maxDiff = None 

156 

157 print("\n".join(self.expected.pformat(max_lines=-1, max_width=-1)), file=self.expected_output) 

158 

159 self.report.add(self.run) 

160 print(self.report, file=self.actual_output) 

161 

162 self.assertEqual(self.actual_output.getvalue(), self.expected_output.getvalue()) 

163 

164 def testAddWithFailedFlag(self): 

165 """Test adding a run with a failed job.""" 

166 self.maxDiff = None 

167 

168 self.expected["X"][0] = "F" 

169 print("\n".join(self.expected.pformat(max_lines=-1, max_width=-1)), file=self.expected_output) 

170 

171 # Alter the run report to include a failed job. 

172 self.run.job_state_counts = { 

173 state: 1 if state in {WmsStates.FAILED, WmsStates.SUCCEEDED} else 0 for state in WmsStates 

174 } 

175 self.report.add(self.run) 

176 print(self.report, file=self.actual_output) 

177 

178 self.assertEqual(self.actual_output.getvalue(), self.expected_output.getvalue()) 

179 

180 def testAddWithHeldFlag(self): 

181 """Test adding a run with a held job.""" 

182 self.maxDiff = None 

183 

184 self.expected["X"][0] = "H" 

185 print("\n".join(self.expected.pformat(max_lines=-1, max_width=-1)), file=self.expected_output) 

186 

187 # Alter the run report to include a held job. 

188 self.run.job_state_counts = { 

189 state: 1 if state in {WmsStates.SUCCEEDED, WmsStates.HELD} else 0 for state in WmsStates 

190 } 

191 self.report.add(self.run) 

192 print(self.report, file=self.actual_output) 

193 

194 self.assertEqual(self.actual_output.getvalue(), self.expected_output.getvalue()) 

195 

196 def testAddWithDeletedFlag(self): 

197 """Test adding a run with a deleted job.""" 

198 self.maxDiff = None 

199 

200 self.expected["X"][0] = "D" 

201 print("\n".join(self.expected.pformat(max_lines=-1, max_width=-1)), file=self.expected_output) 

202 

203 # Alter the run report to include a deleted job. 

204 self.run.job_state_counts = { 

205 state: 1 if state in {WmsStates.SUCCEEDED, WmsStates.DELETED} else 0 for state in WmsStates 

206 } 

207 self.report.add(self.run) 

208 print(self.report, file=self.actual_output) 

209 

210 self.assertEqual(self.actual_output.getvalue(), self.expected_output.getvalue()) 

211 

212 def testNoneValues(self): 

213 """Test no exception if plugin lets values default to None.""" 

214 self.maxDiff = None 

215 

216 self.run.operator = None 

217 self.run.project = None 

218 self.run.campaign = None 

219 self.run.operator = None 

220 self.run.payload = None 

221 self.run.site = None 

222 self.report.add(self.run) 

223 print(self.report, file=self.actual_output) 

224 

225 self.expected = Table(dtype=self.fields) 

226 self.expected.add_row(["", "RUNNING", "50", "1.0", "", "", "", "", "", "run"]) 

227 print("\n".join(self.expected.pformat(max_lines=-1, max_width=-1)), file=self.expected_output) 

228 

229 self.assertEqual(self.actual_output.getvalue(), self.expected_output.getvalue()) 

230 

231 def testValueError(self): 

232 """Test no exception if plugin lets values default to None.""" 

233 self.maxDiff = None 

234 

235 logger = logging.getLogger("test_bps_report") 

236 logger.setLevel(logging.INFO) 

237 

238 self.run.run = None 

239 with self.assertLogs(level=logging.ERROR) as cm: 

240 self.report.add(self.run) 

241 print(self.report, file=self.actual_output) 

242 self.assertRegex(cm.records[0].getMessage(), "Error when adding summary report row") 

243 

244 self.expected = Table(dtype=self.fields) 

245 print("\n".join(self.expected.pformat(max_lines=-1, max_width=-1)), file=self.expected_output) 

246 

247 self.assertEqual(self.actual_output.getvalue(), self.expected_output.getvalue()) 

248 

249 

250class DetailedRunReportTestCase(unittest.TestCase): 

251 """Test a detailed run report.""" 

252 

253 def setUp(self): 

254 self.fields = [("", "S")] + [(state.name, "I") for state in WmsStates] + [("EXPECTED", "i")] 

255 

256 table = Table(dtype=self.fields) 

257 table.add_row( 

258 ["TOTAL"] 

259 + [1 if state in {WmsStates.RUNNING, WmsStates.SUCCEEDED} else 0 for state in WmsStates] 

260 + [2] 

261 ) 

262 table.add_row(["foo"] + [1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates] + [1]) 

263 table.add_row(["bar"] + [1 if state == WmsStates.RUNNING else 0 for state in WmsStates] + [1]) 

264 self.expected = DetailedRunReport.from_table(table) 

265 

266 self.run = WmsRunReport( 

267 wms_id="1.0", 

268 global_wms_id="foo#1.0", 

269 path="/path/to/run", 

270 label="label", 

271 run="run", 

272 site="TEST", 

273 project="dev", 

274 campaign="testing", 

275 payload="test", 

276 operator="tester", 

277 run_summary="foo:1;bar:1", 

278 state=WmsStates.RUNNING, 

279 jobs=[ 

280 WmsJobReport(wms_id="1.0", name="", label="foo", state=WmsStates.SUCCEEDED), 

281 WmsJobReport(wms_id="2.0", name="", label="bar", state=WmsStates.RUNNING), 

282 ], 

283 total_number_jobs=2, 

284 job_state_counts={ 

285 state: 1 if state in {WmsStates.SUCCEEDED, WmsStates.RUNNING} else 0 for state in WmsStates 

286 }, 

287 job_summary={ 

288 "foo": {state: 1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates}, 

289 "bar": {state: 1 if state == WmsStates.RUNNING else 0 for state in WmsStates}, 

290 }, 

291 ) 

292 

293 self.actual = DetailedRunReport(self.fields) 

294 

295 def testAddWithJobSummary(self): 

296 """Test adding a run with a job summary.""" 

297 self.run.jobs = None 

298 self.actual.add(self.run) 

299 

300 self.assertEqual(self.actual, self.expected) 

301 

302 def testAddWithoutJobSummary(self): 

303 """Test adding a run without either a job summary or job info.""" 

304 self.run.jobs = None 

305 self.run.job_summary = None 

306 self.actual.add(self.run) 

307 

308 self.assertEqual(len(self.actual), 1) 

309 self.assertRegex(self.actual.message, r"^WARNING.*incomplete") 

310 

311 def testAddWithoutRunSummary(self): 

312 """Test adding a run without a run summary.""" 

313 table = Table(dtype=self.fields) 

314 table.add_row( 

315 ["TOTAL"] 

316 + [1 if state in {WmsStates.RUNNING, WmsStates.SUCCEEDED} else 0 for state in WmsStates] 

317 + [2] 

318 ) 

319 table.add_row(["bar"] + [1 if state == WmsStates.RUNNING else 0 for state in WmsStates] + [-1]) 

320 table.add_row(["foo"] + [1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates] + [-1]) 

321 expected = DetailedRunReport.from_table(table) 

322 

323 self.run.run_summary = None 

324 self.actual.add(self.run) 

325 

326 self.assertRegex(self.actual.message, r"^WARNING.*sorted alphabetically") 

327 self.assertEqual(self.actual, expected) 

328 

329 

330class ExitCodesReportTestCase(unittest.TestCase): 

331 """Test an exit code report.""" 

332 

333 def setUp(self): 

334 self.fields = [ 

335 (" ", "S"), 

336 ("PAYLOAD ERROR COUNT", "i"), 

337 ("PAYLOAD ERROR CODES", "S"), 

338 ("INFRASTRUCTURE ERROR COUNT", "i"), 

339 ("INFRASTRUCTURE ERROR CODES", "S"), 

340 ] 

341 

342 table = Table(dtype=self.fields) 

343 table.add_row(["foo", 0, "None", 0, "None"]) 

344 table.add_row(["bar", 2, "1, 2", 2, "3, 4"]) 

345 self.expected = ExitCodesReport.from_table(table) 

346 

347 self.run = WmsRunReport( 

348 wms_id="1.0", 

349 global_wms_id="foo#1.0", 

350 path="/path/to/run", 

351 label="label", 

352 run="run", 

353 project="dev", 

354 campaign="testing", 

355 payload="test", 

356 operator="tester", 

357 run_summary="foo:1;bar:1", 

358 state=WmsStates.RUNNING, 

359 jobs=[ 

360 WmsJobReport(wms_id="1.0", name="", label="foo", state=WmsStates.SUCCEEDED), 

361 WmsJobReport(wms_id="2.0", name="", label="bar", state=WmsStates.RUNNING), 

362 ], 

363 total_number_jobs=2, 

364 job_state_counts={ 

365 state: 1 if state in {WmsStates.SUCCEEDED, WmsStates.RUNNING} else 0 for state in WmsStates 

366 }, 

367 job_summary={ 

368 "foo": {state: 1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates}, 

369 "bar": {state: 1 if state == WmsStates.RUNNING else 0 for state in WmsStates}, 

370 }, 

371 exit_code_summary={ 

372 "foo": [], 

373 "bar": [1, 2, 3, 4], 

374 }, 

375 ) 

376 

377 self.actual = ExitCodesReport(self.fields) 

378 

379 def testAddSuccess(self): 

380 """Test adding a run successfully.""" 

381 self.actual.add(self.run) 

382 

383 self.assertEqual(len(self.actual), 2) 

384 self.assertEqual(self.actual, self.expected) 

385 

386 def testAddFailure(self): 

387 """Test adding a run unsuccessfully.""" 

388 self.run.job_summary = {} 

389 self.run.exit_code_summary = {} 

390 

391 self.actual.add(self.run) 

392 

393 self.assertEqual(len(self.actual), 0) 

394 self.assertRegex(self.actual.message, r"^WARNING.*report.*incomplete") 

395 

396 def testAddWithoutRunSummary(self): 

397 """Test adding a run without a run summary.""" 

398 self.run.run_summary = None 

399 

400 self.actual.add(self.run) 

401 

402 self.assertRegex(self.actual.message, r"^WARNING.*sorted alphabetically") 

403 

404 

405class CompileJobSummaryTestCase(unittest.TestCase): 

406 """Test compiling a job summary.""" 

407 

408 def setUp(self): 

409 self.report = dataclasses.replace(TEST_REPORT) 

410 

411 def tearDown(self): 

412 pass 

413 

414 def testSummaryExists(self): 

415 """Test if the existing report is not altered.""" 

416 # Create a report with a "fake" job summary, i.e., a summary which 

417 # differs from the one which would be compiled from the information 

418 # about individual jobs. 

419 expected = dataclasses.replace( 

420 self.report, 

421 job_summary={"foo": {state: 1 if state == WmsStates.FAILED else 0 for state in WmsStates}}, 

422 ) 

423 result = dataclasses.replace(expected) 

424 

425 messages = compile_job_summary(result) 

426 

427 self.assertEqual(result, expected) 

428 self.assertFalse(messages) 

429 

430 def testSummaryMissing(self): 

431 """Test if the summary is compiled if necessary.""" 

432 result = dataclasses.replace(self.report, job_summary=None) 

433 

434 messages = compile_job_summary(result) 

435 

436 self.assertEqual(result, self.report) 

437 self.assertFalse(messages) 

438 

439 def testCompilationError(self): 

440 """Test if a warning is issued if the summary cannot be compiled.""" 

441 result = dataclasses.replace(self.report, jobs=None, job_summary=None) 

442 

443 messages = compile_job_summary(result) 

444 

445 self.assertEqual(len(messages), 1) 

446 self.assertRegex(messages[0], r"information.*not available") 

447 

448 

449class CompileCodeSummaryTestCase(unittest.TestCase): 

450 """Test compiling a code summary.""" 

451 

452 def setUp(self): 

453 self.report = WmsRunReport( 

454 wms_id="1.0", 

455 global_wms_id="foo#1.0", 

456 path="/path/to/run", 

457 label="label", 

458 run="run", 

459 project="dev", 

460 campaign="testing", 

461 payload="test", 

462 operator="tester", 

463 run_summary="foo:1;bar:1;baz:1", 

464 state=WmsStates.RUNNING, 

465 jobs=[ 

466 WmsJobReport(wms_id="1.0", name="", label="foo", state=WmsStates.SUCCEEDED), 

467 WmsJobReport(wms_id="2.0", name="", label="bar", state=WmsStates.FAILED), 

468 WmsJobReport(wms_id="3.0", name="", label="baz", state=WmsStates.RUNNING), 

469 ], 

470 total_number_jobs=3, 

471 job_state_counts={ 

472 state: 1 if state in {WmsStates.SUCCEEDED, WmsStates.FAILED, WmsStates.RUNNING} else 0 

473 for state in WmsStates 

474 }, 

475 job_summary={ 

476 "foo": {state: 1 if state == WmsStates.SUCCEEDED else 0 for state in WmsStates}, 

477 "bar": {state: 1 if state == WmsStates.FAILED else 0 for state in WmsStates}, 

478 "baz": {state: 1 if state == WmsStates.RUNNING else 0 for state in WmsStates}, 

479 }, 

480 exit_code_summary={"foo": [], "bar": [1], "baz": []}, 

481 ) 

482 

483 def tearDown(self): 

484 pass 

485 

486 def testAddingMissingEntries(self): 

487 """Test if the missing entries are added to the summary.""" 

488 result = dataclasses.replace(self.report, exit_code_summary={"bar": [1]}) 

489 

490 messages = compile_code_summary(result) 

491 

492 self.assertEqual(result, self.report) 

493 self.assertFalse(messages) 

494 

495 def testDetectingMismatches(self): 

496 """Test if a mismatch between exit codes and failures is reported.""" 

497 expected = dataclasses.replace(self.report, exit_code_summary={"foo": [1], "bar": [1], "baz": []}) 

498 result = dataclasses.replace(expected) 

499 

500 messages = compile_code_summary(result) 

501 

502 self.assertEqual(result, expected) 

503 self.assertEqual(len(messages), 1) 

504 self.assertRegex(messages[0], r"exit codes.*differs.*failures.*labels: foo") 

505 

506 def testDetectingOmissions(self): 

507 """Test if a failure not reflected in exit codes is reported.""" 

508 expected = dataclasses.replace(self.report, exit_code_summary={"foo": [], "baz": []}) 

509 result = dataclasses.replace(expected) 

510 

511 messages = compile_code_summary(result) 

512 

513 self.assertEqual(result, expected) 

514 self.assertEqual(len(messages), 1) 

515 self.assertRegex(messages[0], r"exit codes.*not available.*labels: bar") 

516 

517 def testDetectingDiscrepancies(self): 

518 """Test if multiple discrepancies are reported.""" 

519 expected = dataclasses.replace(self.report, exit_code_summary={"foo": [], "baz": [1]}) 

520 result = dataclasses.replace(expected) 

521 

522 messages = compile_code_summary(result) 

523 

524 self.assertEqual(result, expected) 

525 self.assertEqual(len(messages), 2) 

526 self.assertRegex(messages[0], r"exit codes.*differs.*failures.*labels: baz") 

527 self.assertRegex(messages[1], r"exit codes.*not available.*labels: bar") 

528 

529 def testHandlingNoJobSummary(self): 

530 """Test if the existing report is not altered if no job summary.""" 

531 expected = dataclasses.replace(self.report, job_summary=None) 

532 result = dataclasses.replace(expected) 

533 

534 messages = compile_code_summary(result) 

535 

536 self.assertEqual(result, expected) 

537 self.assertFalse(messages) 

538 

539 

540if __name__ == "__main__": 

541 unittest.main()