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