Coverage for tests/test_cliCmdReport.py: 9%
164 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-14 01:21 -0800
1# This file is part of ctrl_mpexec.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://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 <http://www.gnu.org/licenses/>.
28"""Unit tests for ctrl_mpexec CLI update-graph-run subcommand."""
30import os
31import unittest
33import yaml
34from yaml.loader import SafeLoader
36from lsst.ctrl.mpexec.cli.pipetask import cli as pipetask_cli
37from lsst.ctrl.mpexec.cli.script.report import print_summary
38from lsst.daf.butler.cli.utils import LogCliRunner, clickResultMsg
39from lsst.daf.butler.tests.utils import makeTestTempDir, removeTestTempDir
40from lsst.pipe.base.quantum_provenance_graph import Summary
41from lsst.pipe.base.tests.simpleQGraph import makeSimpleQGraph
42from lsst.pipe.base.tests.util import check_output_run
44TESTDIR = os.path.abspath(os.path.dirname(__file__))
46expected_mock_datasets = [
47 "add_dataset1",
48 "add2_dataset1",
49 "task0_metadata",
50 "task0_log",
51 "add_dataset2",
52 "add2_dataset2",
53 "task1_metadata",
54 "task1_log",
55 "add_dataset3",
56 "add2_dataset3",
57 "task2_metadata",
58 "task2_log",
59 "add_dataset4",
60 "add2_dataset4",
61 "task3_metadata",
62 "task3_log",
63 "add_dataset5",
64 "add2_dataset5",
65 "task4_metadata",
66 "task4_log",
67]
70class ReportTest(unittest.TestCase):
71 """Test executing "pipetask report" command."""
73 def setUp(self) -> None:
74 self.runner = LogCliRunner()
75 self.root = makeTestTempDir(TESTDIR)
77 def tearDown(self) -> None:
78 removeTestTempDir(self.root)
80 def test_report(self):
81 """Test for making a report on the produced, missing and expected
82 datasets in a quantum graph.
83 """
84 metadata = {"output_run": "run"}
85 butler, qgraph = makeSimpleQGraph(
86 run="run",
87 root=self.root,
88 metadata=metadata,
89 )
90 # Check that we can get the proper run collection from the qgraph
91 self.assertEqual(check_output_run(qgraph, "run"), [])
93 graph_uri = os.path.join(self.root, "graph.qgraph")
94 qgraph.saveUri(graph_uri)
96 test_filename = os.path.join(self.root, "report_test.yaml")
98 result = self.runner.invoke(
99 pipetask_cli,
100 ["report", self.root, graph_uri, "--full-output-filename", test_filename, "--no-logs"],
101 input="no",
102 )
104 # Check that we can read from the command line
105 self.assertEqual(result.exit_code, 0, clickResultMsg(result))
107 # Check that we can open and read the file produced by make_reports
108 with open(test_filename) as f:
109 report_output_dict = yaml.load(f, Loader=SafeLoader)
111 self.assertIsNotNone(report_output_dict["task0"])
112 self.assertIsNotNone(report_output_dict["task0"]["failed_quanta"])
113 self.assertIsInstance(report_output_dict["task0"]["n_expected"], int)
115 result_hr = self.runner.invoke(
116 pipetask_cli,
117 ["report", self.root, graph_uri, "--no-logs"],
118 input="no",
119 )
121 # Check that we can read from the command line
122 self.assertEqual(result_hr.exit_code, 0, clickResultMsg(result_hr))
124 # Check that we get string output
125 self.assertIsInstance(result_hr.stdout, str)
127 # Check that task0 and the failed quanta for task0 exist in the string
128 self.assertIn("task0", result_hr.stdout)
129 self.assertIn("Failed", result_hr.stdout)
130 self.assertIn("Expected", result_hr.stdout)
131 self.assertIn("Succeeded", result_hr.stdout)
133 # Check brief option for pipetask report
134 result_brief = self.runner.invoke(
135 pipetask_cli,
136 ["report", self.root, graph_uri, "--no-logs", "--brief"],
137 input="no",
138 )
139 self.assertIsInstance(result_brief.stdout, str)
141 # Check that task0 and the failed quanta for task0 exist in the string
142 self.assertIn("task0", result_brief.stdout)
143 self.assertIn("Failed", result_brief.stdout)
144 self.assertIn("Expected", result_brief.stdout)
145 self.assertIn("Succeeded", result_brief.stdout)
147 # Test cli for the QPG
148 result_v2_terminal_out = self.runner.invoke(
149 pipetask_cli,
150 ["report", self.root, graph_uri, "--no-logs", "--force-v2"],
151 input="no",
152 )
154 # Check that we can read from the command line
155 self.assertEqual(result_v2_terminal_out.exit_code, 0, clickResultMsg(result_v2_terminal_out))
157 # Check that we get string output
158 self.assertIsInstance(result_v2_terminal_out.stdout, str)
160 # Check that task0 and the quanta for task0 exist in the string
161 self.assertIn("task0", result_v2_terminal_out.stdout)
162 self.assertIn("Unknown", result_v2_terminal_out.stdout)
163 self.assertIn("Successful", result_v2_terminal_out.stdout)
164 self.assertIn("Blocked", result_v2_terminal_out.stdout)
165 self.assertIn("Failed", result_v2_terminal_out.stdout)
166 self.assertIn("Wonky", result_v2_terminal_out.stdout)
167 self.assertIn("TOTAL", result_v2_terminal_out.stdout)
168 self.assertIn("EXPECTED", result_v2_terminal_out.stdout)
170 # Test cli for the QPG brief option
171 result_v2_brief = self.runner.invoke(
172 pipetask_cli,
173 ["report", self.root, graph_uri, "--no-logs", "--force-v2", "--brief"],
174 input="no",
175 )
177 # Check that we can read from the command line
178 self.assertEqual(result_v2_brief.exit_code, 0, clickResultMsg(result_v2_brief))
180 # Check that we get string output
181 self.assertIsInstance(result_v2_brief.stdout, str)
183 # Check that task0 and the quanta for task0 exist in the string
184 self.assertIn("task0", result_v2_brief.stdout)
185 self.assertIn("Unknown", result_v2_brief.stdout)
186 self.assertIn("Successful", result_v2_brief.stdout)
187 self.assertIn("Blocked", result_v2_brief.stdout)
188 self.assertIn("Failed", result_v2_brief.stdout)
189 self.assertIn("Wonky", result_v2_brief.stdout)
190 self.assertIn("TOTAL", result_v2_brief.stdout)
191 self.assertIn("EXPECTED", result_v2_brief.stdout)
193 # Check that the full output option works
194 test_filename_v2 = os.path.join(self.root, "report_test.json")
195 result_v2_full = self.runner.invoke(
196 pipetask_cli,
197 [
198 "report",
199 self.root,
200 graph_uri,
201 "--no-logs",
202 "--full-output-filename",
203 test_filename_v2,
204 "--force-v2",
205 ],
206 input="no",
207 )
209 self.assertEqual(result_v2_full.exit_code, 0, clickResultMsg(result_v2_full))
210 # Check the "brief" output that prints to the terminal first:
211 # Check that we get string output
212 self.assertIsInstance(result_v2_full.stdout, str)
214 # Check that task0 and the quanta for task0 exist in the string
215 self.assertIn("task0", result_v2_full.stdout)
216 self.assertIn("Unknown", result_v2_full.stdout)
217 self.assertIn("Successful", result_v2_full.stdout)
218 self.assertIn("Blocked", result_v2_full.stdout)
219 self.assertIn("Failed", result_v2_full.stdout)
220 self.assertIn("Wonky", result_v2_full.stdout)
221 self.assertIn("TOTAL", result_v2_full.stdout)
222 self.assertIn("EXPECTED", result_v2_full.stdout)
224 # Then validate the full output json file:
225 with open(test_filename_v2) as f:
226 output = f.read()
227 model = Summary.model_validate_json(output)
228 # Below is the same set of tests as in `pipe_base`:
229 for task_summary in model.tasks.values():
230 self.assertEqual(task_summary.n_successful, 0)
231 self.assertEqual(task_summary.n_blocked, 0)
232 self.assertEqual(task_summary.n_unknown, 1)
233 self.assertEqual(task_summary.n_expected, 1)
234 self.assertListEqual(task_summary.failed_quanta, [])
235 self.assertListEqual(task_summary.recovered_quanta, [])
236 self.assertListEqual(task_summary.wonky_quanta, [])
237 self.assertEqual(task_summary.n_wonky, 0)
238 self.assertEqual(task_summary.n_failed, 0)
239 for dataset_type_name, dataset_type_summary in model.datasets.items():
240 self.assertListEqual(
241 dataset_type_summary.unsuccessful_datasets,
242 [{"instrument": "INSTR", "detector": 0}],
243 )
244 self.assertEqual(dataset_type_summary.n_visible, 0)
245 self.assertEqual(dataset_type_summary.n_shadowed, 0)
246 self.assertEqual(dataset_type_summary.n_predicted_only, 0)
247 self.assertEqual(dataset_type_summary.n_expected, 1)
248 self.assertEqual(dataset_type_summary.n_cursed, 0)
249 self.assertEqual(dataset_type_summary.n_unsuccessful, 1)
250 self.assertListEqual(dataset_type_summary.cursed_datasets, [])
251 self.assertIn(dataset_type_name, expected_mock_datasets)
252 match dataset_type_name:
253 case name if name in ["add_dataset1", "add2_dataset1", "task0_metadata", "task0_log"]:
254 self.assertEqual(dataset_type_summary.producer, "task0")
255 case name if name in ["add_dataset2", "add2_dataset2", "task1_metadata", "task1_log"]:
256 self.assertEqual(dataset_type_summary.producer, "task1")
257 case name if name in ["add_dataset3", "add2_dataset3", "task2_metadata", "task2_log"]:
258 self.assertEqual(dataset_type_summary.producer, "task2")
259 case name if name in ["add_dataset4", "add2_dataset4", "task3_metadata", "task3_log"]:
260 self.assertEqual(dataset_type_summary.producer, "task3")
261 case name if name in ["add_dataset5", "add2_dataset5", "task4_metadata", "task4_log"]:
262 self.assertEqual(dataset_type_summary.producer, "task4")
264 def test_aggregate_reports(self):
265 """Test `pipetask aggregate-reports` command. We make one
266 `SimpleQgraph` and then fake a copy in a couple of different ways,
267 making sure we can aggregate the similar graphs.
268 """
269 metadata = {"output_run": "run1"}
270 butler, qgraph1 = makeSimpleQGraph(
271 run="run",
272 root=self.root,
273 metadata=metadata,
274 )
276 # Check that we can get the proper run collection from the qgraph
277 self.assertEqual(check_output_run(qgraph1, "run"), [])
279 # Save the graph
280 graph_uri_1 = os.path.join(self.root, "graph1.qgraph")
281 qgraph1.saveUri(graph_uri_1)
283 file1 = os.path.join(self.root, "report_test_1.json")
284 file2 = os.path.join(self.root, "report_test_2.json")
285 aggregate_file = os.path.join(self.root, "aggregate_report.json")
287 report1 = self.runner.invoke(
288 pipetask_cli,
289 [
290 "report",
291 self.root,
292 graph_uri_1,
293 "--no-logs",
294 "--full-output-filename",
295 file1,
296 "--force-v2",
297 ],
298 input="no",
299 )
301 self.assertEqual(report1.exit_code, 0, clickResultMsg(report1))
302 # Now, copy the json output into a duplicate file and aggregate
303 with open(file1) as f:
304 sum1 = Summary.model_validate_json(f.read())
305 sum2 = sum1.model_copy(deep=True)
306 print_summary(sum2, file2, brief=False)
308 # Then use these file outputs as the inputs to aggregate reports:
309 aggregate_report = self.runner.invoke(
310 pipetask_cli,
311 [
312 "aggregate-reports",
313 file1,
314 file2,
315 "--full-output-filename",
316 aggregate_file,
317 ],
318 )
319 # Check that aggregate command had a zero exit code:
320 self.assertEqual(aggregate_report.exit_code, 0, clickResultMsg(aggregate_report))
322 # Check that it aggregates as expected:
323 with open(aggregate_file) as f:
324 agg_sum = Summary.model_validate_json(f.read())
325 for task_label, task_summary in agg_sum.tasks.items():
326 self.assertEqual(task_summary.n_successful, 0)
327 self.assertEqual(task_summary.n_blocked, 0)
328 self.assertEqual(task_summary.n_unknown, 2)
329 self.assertEqual(task_summary.n_expected, 2)
330 self.assertListEqual(task_summary.failed_quanta, [])
331 self.assertListEqual(task_summary.recovered_quanta, [])
332 self.assertListEqual(task_summary.wonky_quanta, [])
333 self.assertEqual(task_summary.n_wonky, 0)
334 self.assertEqual(task_summary.n_failed, 0)
335 for dataset_type_name, dataset_type_summary in agg_sum.datasets.items():
336 self.assertListEqual(
337 dataset_type_summary.unsuccessful_datasets,
338 [{"instrument": "INSTR", "detector": 0}, {"instrument": "INSTR", "detector": 0}],
339 )
340 self.assertEqual(dataset_type_summary.n_visible, 0)
341 self.assertEqual(dataset_type_summary.n_shadowed, 0)
342 self.assertEqual(dataset_type_summary.n_predicted_only, 0)
343 self.assertEqual(dataset_type_summary.n_expected, 2)
344 self.assertEqual(dataset_type_summary.n_cursed, 0)
345 self.assertEqual(dataset_type_summary.n_unsuccessful, 2)
346 self.assertListEqual(dataset_type_summary.cursed_datasets, [])
347 self.assertIn(dataset_type_name, expected_mock_datasets)
348 match dataset_type_name:
349 case name if name in ["add_dataset1", "add2_dataset1", "task0_metadata", "task0_log"]:
350 self.assertEqual(dataset_type_summary.producer, "task0")
351 case name if name in ["add_dataset2", "add2_dataset2", "task1_metadata", "task1_log"]:
352 self.assertEqual(dataset_type_summary.producer, "task1")
353 case name if name in ["add_dataset3", "add2_dataset3", "task2_metadata", "task2_log"]:
354 self.assertEqual(dataset_type_summary.producer, "task2")
355 case name if name in ["add_dataset4", "add2_dataset4", "task3_metadata", "task3_log"]:
356 self.assertEqual(dataset_type_summary.producer, "task3")
357 case name if name in ["add_dataset5", "add2_dataset5", "task4_metadata", "task4_log"]:
358 self.assertEqual(dataset_type_summary.producer, "task4")
361if __name__ == "__main__":
362 unittest.main()