Coverage for tests/test_simple_pipeline_executor.py: 10%
292 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-20 01:53 -0800
1# This file is part of pipe_base.
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/>.
28from __future__ import annotations
30import os
31import shutil
32import tempfile
33import unittest
35import lsst.daf.butler
36import lsst.pipe.base.quantum_provenance_graph as qpg
37import lsst.utils.tests
38from lsst.pipe.base import PipelineGraph, QuantumSuccessCaveats, RepeatableQuantumError
39from lsst.pipe.base.simple_pipeline_executor import SimplePipelineExecutor
40from lsst.pipe.base.tests.mocks import (
41 DirectButlerRepo,
42 DynamicConnectionConfig,
43 DynamicTestPipelineTask,
44 DynamicTestPipelineTaskConfig,
45 MockStorageClass,
46 get_mock_name,
47)
49TESTDIR = os.path.abspath(os.path.dirname(__file__))
52class SimplePipelineExecutorTests(lsst.utils.tests.TestCase):
53 """Test the SimplePipelineExecutor API.
55 Because SimplePipelineExecutor is the easiest way to run simple pipelines
56 in tests, this has also become a home for tests of execution edge cases
57 that don't have a clear home in other test files.
58 """
60 def setUp(self):
61 self.path = tempfile.mkdtemp()
62 # standalone parameter forces the returned config to also include
63 # the information from the search paths.
64 config = lsst.daf.butler.Butler.makeRepo(
65 self.path, standalone=True, searchPaths=[os.path.join(TESTDIR, "config")]
66 )
67 self.butler = SimplePipelineExecutor.prep_butler(config, [], "fake")
68 self.butler.registry.registerDatasetType(
69 lsst.daf.butler.DatasetType(
70 "input",
71 dimensions=self.butler.dimensions.empty,
72 storageClass="StructuredDataDict",
73 )
74 )
75 self.butler.put({"zero": 0}, "input")
76 MockStorageClass.get_or_register_mock("StructuredDataDict")
77 MockStorageClass.get_or_register_mock("TaskMetadataLike")
79 def tearDown(self):
80 shutil.rmtree(self.path, ignore_errors=True)
82 def test_from_task_class(self):
83 """Test executing a single quantum with an executor created by the
84 `from_task_class` factory method, and the
85 `SimplePipelineExecutor.as_generator` method.
86 """
87 config = DynamicTestPipelineTaskConfig()
88 config.inputs["i"] = DynamicConnectionConfig(
89 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
90 )
91 config.outputs["o"] = DynamicConnectionConfig(
92 dataset_type_name="output", storage_class="StructuredDataDict"
93 )
94 executor = SimplePipelineExecutor.from_task_class(
95 DynamicTestPipelineTask,
96 config=config,
97 butler=self.butler,
98 label="a",
99 )
100 (quantum,) = executor.as_generator(register_dataset_types=True)
101 self.assertEqual(self.butler.get("output").storage_class, get_mock_name("StructuredDataDict"))
103 def test_metadata_input(self):
104 """Test two tasks where the output uses metadata from input."""
105 config_a = DynamicTestPipelineTaskConfig()
106 config_a.inputs["i"] = DynamicConnectionConfig(
107 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
108 )
109 config_b = DynamicTestPipelineTaskConfig()
110 config_b.outputs["o"] = DynamicConnectionConfig(
111 dataset_type_name="output", storage_class="StructuredDataDict"
112 )
113 config_b.inputs["in_metadata"] = DynamicConnectionConfig(
114 dataset_type_name="a_metadata", storage_class="TaskMetadata"
115 )
116 pipeline_graph = PipelineGraph()
117 pipeline_graph.add_task("a", DynamicTestPipelineTask, config_a)
118 pipeline_graph.add_task("b", DynamicTestPipelineTask, config_b)
119 executor = SimplePipelineExecutor.from_pipeline_graph(pipeline_graph, butler=self.butler)
120 quanta = executor.run(register_dataset_types=True, save_versions=False)
121 self.assertEqual(len(quanta), 2)
122 output = self.butler.get("output")
123 self.assertEqual(output.quantum.inputs["in_metadata"][0].original_type, "lsst.pipe.base.TaskMetadata")
125 def test_optional_intermediate(self):
126 """Test a pipeline task with an optional regular input that is produced
127 by another task.
128 """
129 config_a = DynamicTestPipelineTaskConfig()
130 config_a.inputs["i"] = DynamicConnectionConfig(
131 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
132 )
133 config_a.fail_exception = "lsst.pipe.base.NoWorkFound"
134 config_a.fail_condition = "1=1" # butler query expression that is true
135 config_a.outputs["o"] = DynamicConnectionConfig(
136 dataset_type_name="intermediate", storage_class="StructuredDataDict"
137 )
138 config_b = DynamicTestPipelineTaskConfig()
139 config_b.inputs["i"] = DynamicConnectionConfig(
140 dataset_type_name="intermediate", storage_class="StructuredDataDict", minimum=0
141 )
142 config_b.outputs["o"] = DynamicConnectionConfig(
143 dataset_type_name="output", storage_class="StructuredDataDict"
144 )
145 pipeline_graph = PipelineGraph()
146 pipeline_graph.add_task("a", DynamicTestPipelineTask, config_a)
147 pipeline_graph.add_task("b", DynamicTestPipelineTask, config_b)
148 executor = SimplePipelineExecutor.from_pipeline_graph(pipeline_graph, butler=self.butler)
149 quanta = executor.run(register_dataset_types=True, save_versions=False)
150 self.assertEqual(len(quanta), 2)
151 # Both quanta ran successfully (NoWorkFound is a success).
152 self.assertTrue(self.butler.exists("a_metadata"))
153 self.assertTrue(self.butler.exists("b_metadata"))
154 # The intermediate dataset was not written, but the final output was.
155 self.assertFalse(self.butler.exists("intermediate"))
156 self.assertTrue(self.butler.exists("output"))
158 def test_optional_input(self):
159 """Test a pipeline task with an optional regular input that is an
160 overall input to the pipeline.
161 """
162 config_a = DynamicTestPipelineTaskConfig()
163 config_a.inputs["i1"] = DynamicConnectionConfig(
164 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
165 )
166 config_a.outputs["i2"] = DynamicConnectionConfig(
167 dataset_type_name="input_2",
168 storage_class="StructuredDataDict", # will never exist
169 )
170 config_a.outputs["o"] = DynamicConnectionConfig(
171 dataset_type_name="output", storage_class="StructuredDataDict"
172 )
173 pipeline_graph = PipelineGraph()
174 pipeline_graph.add_task("a", DynamicTestPipelineTask, config_a)
175 executor = SimplePipelineExecutor.from_pipeline_graph(pipeline_graph, butler=self.butler)
176 quanta = executor.run(register_dataset_types=True, save_versions=False)
177 self.assertEqual(len(quanta), 1)
178 # The quanta ran successfully.
179 self.assertTrue(self.butler.exists("a_metadata"))
180 # The final output was written.
181 self.assertTrue(self.butler.exists("output"))
183 def test_from_pipeline_file(self) -> None:
184 """Test executing a two quanta from different configurations of the
185 same task, with an executor created by the `from_pipeline_filename`
186 factory method, and the `SimplePipelineExecutor.run` method.
187 """
188 filename = os.path.join(TESTDIR, "pipelines", "pipeline_simple.yaml")
189 executor = SimplePipelineExecutor.from_pipeline_filename(filename, butler=self.butler)
190 self._test_pipeline_file(executor)
192 def test_use_local_butler(self) -> None:
193 """Test generating a local butler repository from a pipeline, then
194 running that pipeline using the local butler.
195 """
196 # Test a trivial pipeline that has only dataset types with empty
197 # dimensions.
198 filename = os.path.join(TESTDIR, "pipelines", "pipeline_simple.yaml")
199 executor = SimplePipelineExecutor.from_pipeline_filename(
200 filename, butler=self.butler, output="u/someone/pipeline"
201 )
202 with tempfile.TemporaryDirectory() as tempdir:
203 root = os.path.join(tempdir, "butler_root")
204 executor.use_local_butler(root)
205 self._test_pipeline_file(executor)
207 # Test a more complicated pipeline involving dataset types with
208 # non-empty dimensions. This will require dimension records to be
209 # copied into the destination repository.
210 with (
211 tempfile.TemporaryDirectory() as tempdir,
212 DirectButlerRepo.make_temporary("base.yaml", "spatial.yaml") as (helper, root),
213 ):
214 helper.add_task(dimensions=["visit", "detector"])
215 qg = helper.make_quantum_graph(output="out_chain")
216 executor = SimplePipelineExecutor(qg, helper.butler)
217 output_butler_root = os.path.join(tempdir, "root")
218 executor.use_local_butler(output_butler_root)
219 executor.run(register_dataset_types=True)
220 output_butler = lsst.daf.butler.Butler(output_butler_root)
221 ref = output_butler.find_dataset(
222 "dataset_auto1",
223 collections="out_chain",
224 dimension_records=True,
225 instrument="Cam1",
226 detector=1,
227 visit=1,
228 )
229 self.assertIsNotNone(ref)
230 self.assertIsNotNone(output_butler.get(ref))
231 # Check that dimension records are present in the output Butler.
232 self.assertEqual(ref.dataId.records["visit"].science_program, "test_survey")
233 self.assertTrue(ref.dataId.records["visit_detector_region"].region.contains(0.004, 0.020))
235 def _test_pipeline_file(self, executor: SimplePipelineExecutor) -> None:
236 quanta = executor.run(register_dataset_types=True, save_versions=False)
237 self.assertEqual(len(quanta), 2)
238 self.assertEqual(
239 executor.butler.get("intermediate").storage_class, get_mock_name("StructuredDataDict")
240 )
241 self.assertEqual(executor.butler.get("output").storage_class, get_mock_name("StructuredDataDict"))
243 def test_partial_outputs_success(self):
244 """Test executing two quanta where the first raises
245 `lsst.pipe.base.AnnotatedPartialOutputsError` and its output is an
246 optional input to the second, while configuring the executor to
247 consider this a success.
248 """
249 config_a = DynamicTestPipelineTaskConfig()
250 config_a.inputs["i"] = DynamicConnectionConfig(
251 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
252 )
253 config_a.fail_exception = "lsst.pipe.base.AnnotatedPartialOutputsError"
254 config_a.fail_condition = "1=1" # butler query expression that is true
255 config_a.outputs["o"] = DynamicConnectionConfig(
256 dataset_type_name="intermediate", storage_class="StructuredDataDict"
257 )
258 config_b = DynamicTestPipelineTaskConfig()
259 config_b.inputs["i"] = DynamicConnectionConfig(
260 dataset_type_name="intermediate", storage_class="StructuredDataDict", minimum=0
261 )
262 config_b.outputs["o"] = DynamicConnectionConfig(
263 dataset_type_name="output", storage_class="StructuredDataDict"
264 )
265 pipeline_graph = PipelineGraph()
266 pipeline_graph.add_task("a", DynamicTestPipelineTask, config_a)
267 pipeline_graph.add_task("b", DynamicTestPipelineTask, config_b)
268 # Consider the partial a success and proceed.
269 executor = SimplePipelineExecutor.from_pipeline_graph(
270 pipeline_graph, butler=self.butler, raise_on_partial_outputs=False
271 )
272 (_, _) = executor.as_generator(register_dataset_types=True)
273 self.assertFalse(self.butler.exists("intermediate"))
274 self.assertEqual(self.butler.get("output").storage_class, get_mock_name("StructuredDataDict"))
275 prov = qpg.QuantumProvenanceGraph(self.butler, [executor.quantum_graph], read_caveats="exhaustive")
276 (quantum_key_a,) = prov.quanta["a"]
277 quantum_info_a = prov.get_quantum_info(quantum_key_a)
278 _, quantum_run_a = qpg.QuantumRun.find_final(quantum_info_a)
279 self.assertEqual(
280 quantum_run_a.caveats,
281 QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
282 | QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
283 | QuantumSuccessCaveats.PARTIAL_OUTPUTS_ERROR,
284 )
285 self.assertEqual(
286 quantum_run_a.exception.type_name,
287 "lsst.pipe.base.tests.mocks.MockAlgorithmError",
288 )
289 self.assertEqual(
290 quantum_run_a.exception.metadata,
291 {"badness": 12},
292 )
293 (quantum_key_b,) = prov.quanta["b"]
294 quantum_info_b = prov.get_quantum_info(quantum_key_b)
295 _, quantum_run_b = qpg.QuantumRun.find_final(quantum_info_b)
296 self.assertEqual(quantum_run_b.caveats, QuantumSuccessCaveats.NO_CAVEATS)
297 prov_summary = prov.to_summary(self.butler)
298 # One partial-outputs case, with an empty data ID:
299 self.assertEqual(prov_summary.tasks["a"].caveats, {"*P": [{}]})
300 self.assertEqual(
301 prov_summary.tasks["a"].exceptions.keys(), {"lsst.pipe.base.tests.mocks.MockAlgorithmError"}
302 )
303 self.assertEqual(
304 prov_summary.tasks["a"]
305 .exceptions["lsst.pipe.base.tests.mocks.MockAlgorithmError"][0]
306 .exception.metadata,
307 {"badness": 12},
308 )
309 # No caveats for the second task, since it didn't need the first task's
310 # output anyway.
311 self.assertEqual(prov_summary.tasks["b"].caveats, {})
312 self.assertEqual(prov_summary.tasks["b"].exceptions, {})
313 # Check table forms for summaries of the same information.
314 quantum_table = prov_summary.make_quantum_table()
315 self.assertEqual(list(quantum_table["Task"]), ["a", "b"])
316 self.assertEqual(list(quantum_table["Unknown"]), [0, 0])
317 self.assertEqual(list(quantum_table["Successful"]), [1, 1])
318 self.assertEqual(list(quantum_table["Caveats"]), ["*P(1)", ""])
319 self.assertEqual(list(quantum_table["Blocked"]), [0, 0])
320 self.assertEqual(list(quantum_table["Failed"]), [0, 0])
321 self.assertEqual(list(quantum_table["Wonky"]), [0, 0])
322 self.assertEqual(list(quantum_table["TOTAL"]), [1, 1])
323 self.assertEqual(list(quantum_table["EXPECTED"]), [1, 1])
324 dataset_table = prov_summary.make_dataset_table()
325 self.assertEqual(
326 list(dataset_table["Dataset"]),
327 ["intermediate", "a_metadata", "a_log", "output", "b_metadata", "b_log"],
328 )
329 self.assertEqual(list(dataset_table["Visible"]), [0, 1, 1, 1, 1, 1])
330 self.assertEqual(list(dataset_table["Shadowed"]), [0, 0, 0, 0, 0, 0])
331 self.assertEqual(list(dataset_table["Predicted Only"]), [1, 0, 0, 0, 0, 0])
332 self.assertEqual(list(dataset_table["Unsuccessful"]), [0, 0, 0, 0, 0, 0])
333 self.assertEqual(list(dataset_table["Cursed"]), [0, 0, 0, 0, 0, 0])
334 self.assertEqual(list(dataset_table["TOTAL"]), [1, 1, 1, 1, 1, 1])
335 self.assertEqual(list(dataset_table["EXPECTED"]), [1, 1, 1, 1, 1, 1])
336 exception_table = prov_summary.make_exception_table()
337 self.assertEqual(list(exception_table["Task"]), ["a"])
338 self.assertEqual(
339 list(exception_table["Exception"]), ["lsst.pipe.base.tests.mocks.MockAlgorithmError"]
340 )
341 self.assertEqual(list(exception_table["Count"]), [1])
342 bad_quantum_tables = prov_summary.make_bad_quantum_tables()
343 self.assertEqual(bad_quantum_tables.keys(), {"a"})
344 self.assertEqual(list(bad_quantum_tables["a"]["Status(Caveats)"]), ["SUCCESSFUL(P)"])
345 self.assertEqual(list(bad_quantum_tables["a"]["Exception"]), ["MockAlgorithmError"])
346 self.assertFalse(prov_summary.make_bad_dataset_tables())
348 def test_no_work_found(self):
349 """Test executing two quanta where the first raises
350 `NoWorkFound` in `runQuantum`, leading the next to raise `NoWorkFound`
351 in `adjustQuantum`.
352 """
353 config_a = DynamicTestPipelineTaskConfig()
354 config_a.inputs["i"] = DynamicConnectionConfig(
355 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
356 )
357 config_a.fail_exception = "lsst.pipe.base.NoWorkFound"
358 config_a.fail_condition = "1=1" # butler query expression that is true
359 config_a.outputs["o"] = DynamicConnectionConfig(
360 dataset_type_name="intermediate", storage_class="StructuredDataDict"
361 )
362 config_b = DynamicTestPipelineTaskConfig()
363 config_b.inputs["i"] = DynamicConnectionConfig(
364 dataset_type_name="intermediate", storage_class="StructuredDataDict"
365 )
366 config_b.outputs["o"] = DynamicConnectionConfig(
367 dataset_type_name="output", storage_class="StructuredDataDict"
368 )
369 pipeline_graph = PipelineGraph()
370 pipeline_graph.add_task("a", DynamicTestPipelineTask, config_a)
371 pipeline_graph.add_task("b", DynamicTestPipelineTask, config_b)
372 # Consider the partial a success and proceed.
373 executor = SimplePipelineExecutor.from_pipeline_graph(
374 pipeline_graph, butler=self.butler, raise_on_partial_outputs=False
375 )
376 (_, _) = executor.as_generator(register_dataset_types=True)
377 prov = qpg.QuantumProvenanceGraph()
378 prov.assemble_quantum_provenance_graph(self.butler, [executor.quantum_graph])
379 (quantum_key_a,) = prov.quanta["a"]
380 quantum_info_a = prov.get_quantum_info(quantum_key_a)
381 _, quantum_run_a = qpg.QuantumRun.find_final(quantum_info_a)
382 self.assertEqual(
383 quantum_run_a.caveats,
384 QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
385 | QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
386 | QuantumSuccessCaveats.NO_WORK,
387 )
388 (quantum_key_b,) = prov.quanta["b"]
389 quantum_info_b = prov.get_quantum_info(quantum_key_b)
390 _, quantum_run_b = qpg.QuantumRun.find_final(quantum_info_b)
391 self.assertEqual(
392 quantum_run_b.caveats,
393 QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
394 | QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
395 | QuantumSuccessCaveats.ADJUST_QUANTUM_RAISED
396 | QuantumSuccessCaveats.NO_WORK,
397 )
398 prov_summary = prov.to_summary(self.butler)
399 # One NoWorkFound, raised by runQuantum, with an empty data ID:
400 self.assertEqual(prov_summary.tasks["a"].caveats, {"*N": [{}]})
401 self.assertEqual(prov_summary.tasks["a"].exceptions, {})
402 # One NoWorkFound, raised by adjustQuantum, with an empty data ID.
403 self.assertEqual(prov_summary.tasks["b"].caveats, {"*A": [{}]})
404 self.assertEqual(prov_summary.tasks["b"].exceptions, {})
405 # Check table forms for summaries of the same information.
406 quantum_table = prov_summary.make_quantum_table()
407 self.assertEqual(list(quantum_table["Task"]), ["a", "b"])
408 self.assertEqual(list(quantum_table["Unknown"]), [0, 0])
409 self.assertEqual(list(quantum_table["Successful"]), [1, 1])
410 self.assertEqual(list(quantum_table["Caveats"]), ["*N(1)", "*A(1)"])
411 self.assertEqual(list(quantum_table["Blocked"]), [0, 0])
412 self.assertEqual(list(quantum_table["Failed"]), [0, 0])
413 self.assertEqual(list(quantum_table["Wonky"]), [0, 0])
414 self.assertEqual(list(quantum_table["TOTAL"]), [1, 1])
415 self.assertEqual(list(quantum_table["EXPECTED"]), [1, 1])
416 dataset_table = prov_summary.make_dataset_table()
417 self.assertEqual(
418 list(dataset_table["Dataset"]),
419 ["intermediate", "a_metadata", "a_log", "output", "b_metadata", "b_log"],
420 )
421 self.assertEqual(list(dataset_table["Visible"]), [0, 1, 1, 0, 1, 1])
422 self.assertEqual(list(dataset_table["Shadowed"]), [0, 0, 0, 0, 0, 0])
423 self.assertEqual(list(dataset_table["Predicted Only"]), [1, 0, 0, 1, 0, 0])
424 self.assertEqual(list(dataset_table["Unsuccessful"]), [0, 0, 0, 0, 0, 0])
425 self.assertEqual(list(dataset_table["Cursed"]), [0, 0, 0, 0, 0, 0])
426 self.assertEqual(list(dataset_table["TOTAL"]), [1, 1, 1, 1, 1, 1])
427 self.assertEqual(list(dataset_table["EXPECTED"]), [1, 1, 1, 1, 1, 1])
428 self.assertFalse(prov_summary.make_exception_table())
429 self.assertFalse(prov_summary.make_bad_quantum_tables())
430 self.assertFalse(prov_summary.make_bad_dataset_tables())
432 def test_partial_outputs_failure(self):
433 """Test executing two quanta where the first raises
434 `lsst.pipe.base.AnnotatedPartialOutputsError` and its output is an
435 optional input to the second, while configuring the executor to
436 consider this a failure.
437 """
438 config_a = DynamicTestPipelineTaskConfig()
439 config_a.inputs["i"] = DynamicConnectionConfig(
440 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
441 )
442 config_a.fail_exception = "lsst.pipe.base.AnnotatedPartialOutputsError"
443 config_a.fail_condition = "1=1" # butler query expression that is true
444 config_a.outputs["o"] = DynamicConnectionConfig(
445 dataset_type_name="intermediate", storage_class="StructuredDataDict"
446 )
447 config_b = DynamicTestPipelineTaskConfig()
448 config_b.inputs["i"] = DynamicConnectionConfig(
449 dataset_type_name="intermediate", storage_class="StructuredDataDict", minimum=0
450 )
451 config_b.outputs["o"] = DynamicConnectionConfig(
452 dataset_type_name="output", storage_class="StructuredDataDict"
453 )
454 pipeline_graph = PipelineGraph()
455 pipeline_graph.add_task("a", DynamicTestPipelineTask, config_a)
456 pipeline_graph.add_task("b", DynamicTestPipelineTask, config_b)
457 executor = SimplePipelineExecutor.from_pipeline_graph(
458 pipeline_graph,
459 butler=self.butler,
460 raise_on_partial_outputs=True,
461 )
462 # The executor should raise the chained exception
463 # (RepeatableQuantumError, since that's what the mocking system in
464 # pipe_base uses here), not AnnotatedPartialOutputsError.
465 with self.assertRaises(RepeatableQuantumError):
466 executor.run(register_dataset_types=True)
467 self.assertFalse(self.butler.exists("intermediate"))
468 self.assertFalse(self.butler.exists("output"))
469 prov = qpg.QuantumProvenanceGraph()
470 prov.assemble_quantum_provenance_graph(self.butler, [executor.quantum_graph])
471 (quantum_key_a,) = prov.quanta["a"]
472 quantum_info_a = prov.get_quantum_info(quantum_key_a)
473 _, quantum_run_a = qpg.QuantumRun.find_final(quantum_info_a)
474 self.assertEqual(quantum_run_a.status, qpg.QuantumRunStatus.FAILED)
475 self.assertIsNone(quantum_run_a.caveats)
476 self.assertIsNone(quantum_run_a.exception)
477 (quantum_key_b,) = prov.quanta["b"]
478 quantum_info_b = prov.get_quantum_info(quantum_key_b)
479 self.assertEqual(quantum_info_b["status"], qpg.QuantumInfoStatus.BLOCKED)
480 prov_summary = prov.to_summary(self.butler)
481 # One partial-outputs failure case for the first task.
482 self.assertEqual(prov_summary.tasks["a"].n_failed, 1)
483 # No direct failures, but one blocked for the second
484 self.assertEqual(prov_summary.tasks["b"].n_failed, 0)
485 self.assertEqual(prov_summary.tasks["b"].n_blocked, 1)
486 # Check table forms for summaries of the same information.
487 quantum_table = prov_summary.make_quantum_table()
488 self.assertEqual(list(quantum_table["Task"]), ["a", "b"])
489 self.assertEqual(list(quantum_table["Unknown"]), [0, 0])
490 self.assertEqual(list(quantum_table["Successful"]), [0, 0])
491 self.assertEqual(list(quantum_table["Caveats"]), ["", ""])
492 self.assertEqual(list(quantum_table["Blocked"]), [0, 1])
493 self.assertEqual(list(quantum_table["Failed"]), [1, 0])
494 self.assertEqual(list(quantum_table["Wonky"]), [0, 0])
495 self.assertEqual(list(quantum_table["TOTAL"]), [1, 1])
496 self.assertEqual(list(quantum_table["EXPECTED"]), [1, 1])
497 dataset_table = prov_summary.make_dataset_table()
498 self.assertEqual(
499 list(dataset_table["Dataset"]),
500 ["intermediate", "a_metadata", "a_log", "output", "b_metadata", "b_log"],
501 )
502 # Note that a_log is UNSUCCESSFUL, not VISIBLE, despite being present
503 # in butler because those categories are mutually exclusive and we
504 # don't want to consider any outputs of failed quanta to be successful.
505 self.assertEqual(list(dataset_table["Visible"]), [0, 0, 0, 0, 0, 0])
506 self.assertEqual(list(dataset_table["Shadowed"]), [0, 0, 0, 0, 0, 0])
507 self.assertEqual(list(dataset_table["Predicted Only"]), [0, 0, 0, 0, 0, 0])
508 self.assertEqual(list(dataset_table["Unsuccessful"]), [1, 1, 1, 1, 1, 1])
509 self.assertEqual(list(dataset_table["Cursed"]), [0, 0, 0, 0, 0, 0])
510 self.assertEqual(list(dataset_table["TOTAL"]), [1, 1, 1, 1, 1, 1])
511 self.assertEqual(list(dataset_table["EXPECTED"]), [1, 1, 1, 1, 1, 1])
512 self.assertFalse(prov_summary.make_exception_table())
513 bad_quantum_tables = prov_summary.make_bad_quantum_tables()
514 self.assertEqual(bad_quantum_tables.keys(), {"a"})
515 self.assertEqual(list(bad_quantum_tables["a"]["Status(Caveats)"]), ["FAILED"])
516 self.assertEqual(list(bad_quantum_tables["a"]["Exception"]), [""])
517 self.assertFalse(prov_summary.make_bad_dataset_tables())
519 def test_existence_check_skips(self):
520 """Test that pre-execution existence checks are not performed for
521 overall-input datasets, as this those checks could otherwise mask
522 repository configuration problems or downtime as NoWorkFound cases.
523 """
524 # First we configure and execute task A, which is just a way to get a
525 # MockDataset in the repo for us to play with; the important test can't
526 # use the non-mock 'input' dataset because the mock runQuantum only
527 # actually reads MockDatasets.
528 config_a = DynamicTestPipelineTaskConfig()
529 config_a.inputs["i"] = DynamicConnectionConfig(
530 dataset_type_name="input", storage_class="StructuredDataDict", mock_storage_class=False
531 )
532 config_a.outputs["o"] = DynamicConnectionConfig(
533 dataset_type_name="intermediate", storage_class="StructuredDataDict"
534 )
535 executor_a = SimplePipelineExecutor.from_task_class(
536 DynamicTestPipelineTask,
537 config=config_a,
538 butler=self.butler,
539 label="a",
540 )
541 executor_a.run(register_dataset_types=True)
542 # Now we can do the real test.
543 config_b = DynamicTestPipelineTaskConfig()
544 config_b.inputs["i"] = DynamicConnectionConfig(
545 dataset_type_name="intermediate", storage_class="StructuredDataDict", minimum=0
546 )
547 config_b.outputs["o"] = DynamicConnectionConfig(
548 dataset_type_name="output", storage_class="StructuredDataDict"
549 )
550 butler = self.butler.clone(run="new_run")
551 executor_b = SimplePipelineExecutor.from_task_class(
552 DynamicTestPipelineTask,
553 config=config_b,
554 butler=butler,
555 label="b",
556 attach_datastore_records=True,
557 )
558 # Delete the input dataset after the QG has already been built.
559 intermediate_refs = butler.query_datasets("intermediate")
560 self.assertEqual(len(intermediate_refs), 1)
561 butler.pruneDatasets(intermediate_refs, purge=True, unstore=True)
562 with self.assertRaises(FileNotFoundError):
563 # We should get an exception rather than NoWorkFound, because for
564 # this single-task pipeline, the missing dataset is an
565 # overall-input (name notwithstanding).
566 executor_b.run(register_dataset_types=True)
569class MemoryTester(lsst.utils.tests.MemoryTestCase):
570 """Generic tests for file leaks."""
573def setup_module(module):
574 """Set up the module for pytest.
576 Parameters
577 ----------
578 module : `~types.ModuleType`
579 Module to set up.
580 """
581 lsst.utils.tests.init()
584if __name__ == "__main__":
585 lsst.utils.tests.init()
586 unittest.main()