Coverage for tests/test_init_output_run.py: 14%
253 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-21 01:24 -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/>.
28import itertools
29import tempfile
30import unittest
31from collections.abc import Iterator
32from contextlib import contextmanager
33from typing import ClassVar
35import lsst.utils.tests
36from lsst.daf.butler import (
37 Butler,
38 DatasetRef,
39 DatasetType,
40 MissingDatasetTypeError,
41 QuantumBackedButler,
42 SerializedDatasetType,
43 StorageClassFactory,
44)
45from lsst.daf.butler.registry import ConflictingDefinitionError
46from lsst.pipe.base.all_dimensions_quantum_graph_builder import AllDimensionsQuantumGraphBuilder
47from lsst.pipe.base.pipeline_graph import PipelineGraph
48from lsst.pipe.base.quantum_graph import PredictedQuantumGraph
49from lsst.pipe.base.tests.mocks import (
50 DynamicConnectionConfig,
51 DynamicTestPipelineTask,
52 DynamicTestPipelineTaskConfig,
53 MockDataset,
54)
57def _have_example_storage_classes() -> bool:
58 """Check whether some storage classes work as expected.
60 Given that these have registered converters, it shouldn't actually be
61 necessary to import those types in order to determine that they're
62 convertible, but the storage class machinery is implemented such that types
63 that can't be imported can't be converted, and while that's inconvenient
64 here it's totally fine in non-testing scenarios where you only care about a
65 storage class if you can actually use it.
66 """
67 getter = StorageClassFactory().getStorageClass
68 return (
69 getter("ArrowTable").can_convert(getter("ArrowAstropy"))
70 and getter("ArrowAstropy").can_convert(getter("ArrowTable"))
71 and getter("ArrowTable").can_convert(getter("DataFrame"))
72 and getter("DataFrame").can_convert(getter("ArrowTable"))
73 )
76class InitOutputRunTestCase(unittest.TestCase):
77 """Tests for the init_output_run methods of PipelineGraph and
78 QuantumGraph.
79 """
81 INPUT_COLLECTION: ClassVar[str] = "overall_inputs"
83 @contextmanager
84 def make_butler(self) -> Iterator[Butler]:
85 """Wrap a temporary local butler repository in a context manager."""
86 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as root:
87 Butler.makeRepo(root)
88 butler = Butler.from_config(root, writeable=True)
89 yield butler
91 @contextmanager
92 def prep_butler(self, pipeline_graph: PipelineGraph) -> Iterator[Butler]:
93 """Create a temporary local butler repository with the dataset types
94 and input datasets needed by a pipeline graph.
96 This also resolves the pipeline graph and checks dataset types
97 immediately after they are registered, providing test coverage for the
98 methods that do that.
99 """
100 with self.make_butler() as butler:
101 butler.collections.register(self.INPUT_COLLECTION)
102 pipeline_graph.resolve(butler.registry)
103 with self.assertRaises(MissingDatasetTypeError):
104 pipeline_graph.check_dataset_type_registrations(butler)
105 pipeline_graph.register_dataset_types(butler)
106 pipeline_graph.check_dataset_type_registrations(butler)
107 for _, dataset_type_node in pipeline_graph.iter_overall_inputs():
108 butler.put(
109 MockDataset(
110 dataset_id=None,
111 dataset_type=SerializedDatasetType(
112 name=dataset_type_node.name,
113 dimensions=[],
114 storageClass=dataset_type_node.storage_class_name,
115 ),
116 data_id={},
117 run=self.INPUT_COLLECTION,
118 ),
119 dataset_type_node.name,
120 run=self.INPUT_COLLECTION,
121 )
122 yield butler
124 def find_init_output_refs(
125 self, pipeline_graph: PipelineGraph, butler: Butler
126 ) -> dict[str, list[DatasetRef]]:
127 """Find the init-output datasets of a pipeline graph in a butler
128 repository.
130 Parameters
131 ----------
132 pipeline_graph : `PipelineGraph`
133 Pipeline graph.
134 butler : `Butler`
135 Full butler client.
137 Returns
138 -------
139 init_output_refs : `dict`
140 Dataset references, keyed by task label. Storage classes will
141 match the data repository definitions of the dataset types. The
142 special 'packages' dataset type will be included under a '*' key.
143 """
144 init_output_refs: dict[str, list[DatasetRef]] = {}
145 for task_node in pipeline_graph.tasks.values():
146 init_output_refs_for_task: list[DatasetRef] = []
147 for write_edge in task_node.init.iter_all_outputs():
148 ref = butler.find_dataset(write_edge.dataset_type_name)
149 # Check that the ref we got back uses the dataset type node's
150 # definition of the dataset type (including storage class).
151 self.assertEqual(
152 ref.datasetType, pipeline_graph.dataset_types[write_edge.dataset_type_name].dataset_type
153 )
154 # Remember the version of the ref that has the task's storage
155 # class, in case they differ.
156 init_output_refs_for_task.append(write_edge.adapt_dataset_ref(ref))
157 init_output_refs[task_node.label] = init_output_refs_for_task
158 init_output_refs["*"] = [butler.find_dataset(pipeline_graph.packages_dataset_type)]
159 return init_output_refs
161 def get_quantum_graph_init_output_refs(
162 self, quantum_graph: PredictedQuantumGraph
163 ) -> dict[str, list[DatasetRef]]:
164 """Extract dataset references from a QuantumGraph into the same form
165 as returned by `find_init_output_refs`.
166 """
167 init_output_refs: dict[str, list[DatasetRef]] = {}
168 for task_label in quantum_graph.pipeline_graph.tasks:
169 init_output_refs[task_label] = list(quantum_graph.get_init_outputs(task_label).values())
170 init_output_refs["*"] = list(quantum_graph.get_init_outputs("").values())
171 return init_output_refs
173 def assert_init_output_refs_equal(
174 self, a: dict[str, list[DatasetRef]], b: dict[str, list[DatasetRef]]
175 ) -> None:
176 """Check that two dictionaries of the form returned by
177 `find_init_output_refs` are equal.
178 """
179 self.assertEqual(a.keys(), b.keys())
180 for task_label, init_output_refs_for_task in a.items():
181 self.assertCountEqual(init_output_refs_for_task, b[task_label])
183 def check_qbb_consistency(
184 self, init_output_refs: dict[str, list[DatasetRef]], qbb: QuantumBackedButler
185 ) -> None:
186 """Check that a quantum-backed butler sees all of the given datasets.
188 Parameters
189 ----------
190 init_output_refs : `dict`
191 Dataset references, keyed by task label. Storage classes should
192 match the data repository definitions of the dataset types. The
193 special 'packages' dataset type should be included under a '*' key.
194 qbb : `lsst.daf.butler.QuantumBackedButler`
195 A quantum-backed butler.
196 """
197 for task_label, init_output_refs_for_task in init_output_refs.items():
198 for ref, stored_in in qbb.stored_many(init_output_refs_for_task).items():
199 self.assertTrue(
200 stored_in, msg=f"Init-input {ref} of task {task_label} not stored according to QBB."
201 )
203 def init_with_pipeline_graph_first(
204 self, pipeline_graph: PipelineGraph, butler: Butler, run: str
205 ) -> PredictedQuantumGraph:
206 """Test the init_output_run methods of PipelineGraph and QuantumGraph,
207 using the former to actually write init-outputs (with later attempts
208 correctly failing or doing nothing, depending on parameters).
209 """
210 butler = butler.clone(run=run, collections=[self.INPUT_COLLECTION, run])
211 pipeline_graph.init_output_run(butler)
212 init_output_refs = self.find_init_output_refs(pipeline_graph, butler)
213 # Build a QG with the init outputs already in place.
214 quantum_graph_builder = AllDimensionsQuantumGraphBuilder(
215 pipeline_graph,
216 butler,
217 skip_existing_in=[run],
218 output_run=run,
219 input_collections=[self.INPUT_COLLECTION],
220 )
221 quantum_graph = quantum_graph_builder.finish(metadata={"output_run": run}).assemble()
222 # Check that the QG refs are the same as the ones that were present
223 # already.
224 self.assert_init_output_refs_equal(
225 self.get_quantum_graph_init_output_refs(quantum_graph),
226 init_output_refs,
227 )
228 # Initialize with the pipeline graph, should be a no-op.
229 pipeline_graph.init_output_run(butler)
230 self.assert_init_output_refs_equal(
231 self.find_init_output_refs(pipeline_graph, butler),
232 init_output_refs,
233 )
234 # Initialize again with the QG; should be a no-op.
235 quantum_graph.init_output_run(butler)
236 self.assert_init_output_refs_equal(
237 self.find_init_output_refs(pipeline_graph, butler),
238 init_output_refs,
239 )
240 # Initialize again with the QG but tell it to expect an empty run.
241 with self.assertRaises(ConflictingDefinitionError):
242 quantum_graph.init_output_run(butler, existing=False)
243 with self.assertRaises(ConflictingDefinitionError):
244 quantum_graph.write_configs(butler, compare_existing=False)
245 with self.assertRaises(ConflictingDefinitionError):
246 quantum_graph.write_packages(butler, compare_existing=False)
247 with self.assertRaises(ConflictingDefinitionError):
248 quantum_graph.write_init_outputs(butler, skip_existing=False)
249 # Make a QBB, check that it can see the init outputs.
250 qbb = quantum_graph.make_init_qbb(butler._config)
251 self.check_qbb_consistency(init_output_refs, qbb)
252 # Use QBB to initialize again, should be a no-op.
253 quantum_graph.init_output_run(qbb)
254 self.check_qbb_consistency(init_output_refs, qbb)
255 # Use QBB to initialize again but tell it to expect an empty run.
256 with self.assertRaises(ConflictingDefinitionError):
257 quantum_graph.init_output_run(qbb, existing=False)
258 return quantum_graph
260 def init_with_quantum_graph_first(
261 self, pipeline_graph: PipelineGraph, butler: Butler, run: str
262 ) -> PredictedQuantumGraph:
263 """Test the init_output_run methods of PipelineGraph and QuantumGraph,
264 using the latter to actually write init-outputs (with later attempts
265 correctly failing or doing nothing, depending on parameters).
266 """
267 butler = butler.clone(run=run, collections=[self.INPUT_COLLECTION, run])
268 # Build a QG.
269 quantum_graph_builder = AllDimensionsQuantumGraphBuilder(
270 pipeline_graph,
271 butler,
272 input_collections=[self.INPUT_COLLECTION],
273 )
274 quantum_graph = quantum_graph_builder.finish(metadata={"output_run": run}).assemble()
275 # Initialize with the QG.
276 quantum_graph.init_output_run(butler)
277 # Check that the QG refs are the same as the ones we find in the repo.
278 init_output_refs = self.find_init_output_refs(pipeline_graph, butler)
279 self.assert_init_output_refs_equal(
280 self.get_quantum_graph_init_output_refs(quantum_graph),
281 init_output_refs,
282 )
283 # Initialize again with the QG; should be a no-op.
284 quantum_graph.init_output_run(butler)
285 self.assert_init_output_refs_equal(
286 self.find_init_output_refs(pipeline_graph, butler),
287 init_output_refs,
288 )
289 # Initialize again with the QG but tell it to expect an empty run.
290 with self.assertRaises(ConflictingDefinitionError):
291 quantum_graph.init_output_run(butler, existing=False)
292 with self.assertRaises(ConflictingDefinitionError):
293 quantum_graph.write_configs(butler, compare_existing=False)
294 with self.assertRaises(ConflictingDefinitionError):
295 quantum_graph.write_packages(butler, compare_existing=False)
296 with self.assertRaises(ConflictingDefinitionError):
297 quantum_graph.write_init_outputs(butler, skip_existing=False)
298 # Initialize with the pipeline graph, should be a no-op.
299 pipeline_graph.init_output_run(butler)
300 # Make a QBB, check that it can see the init outputs.
301 qbb = quantum_graph.make_init_qbb(butler._config)
302 self.check_qbb_consistency(init_output_refs, qbb)
303 # Use QBB to initialize again, should be a no-op.
304 quantum_graph.init_output_run(qbb)
305 self.check_qbb_consistency(init_output_refs, qbb)
306 # Use QBB to initialize again but tell it to expect an empty run.
307 with self.assertRaises(ConflictingDefinitionError):
308 quantum_graph.init_output_run(qbb, existing=False)
309 return quantum_graph
311 def init_with_qbb_first(
312 self, pipeline_graph: PipelineGraph, butler: Butler, run: str
313 ) -> PredictedQuantumGraph:
314 """Test the init_output_run methods of PipelineGraph and QuantumGraph,
315 using the latter a quantum-backed butler to actually write init-outputs
316 (with later attempts correctly failing or doing nothing, depending on
317 parameters).
318 """
319 butler = butler.clone(run=run, collections=[self.INPUT_COLLECTION, run])
320 # Build a QG.
321 quantum_graph_builder = AllDimensionsQuantumGraphBuilder(
322 pipeline_graph,
323 butler,
324 input_collections=[self.INPUT_COLLECTION],
325 )
326 quantum_graph = quantum_graph_builder.finish(metadata={"output_run": run}).assemble()
327 # Make a quantum-backed butler and use it to initialize the run.
328 qbb = quantum_graph.make_init_qbb(butler._config)
329 quantum_graph.init_output_run(qbb)
330 init_output_refs = self.get_quantum_graph_init_output_refs(quantum_graph)
331 self.check_qbb_consistency(init_output_refs, qbb)
332 # Use QBB to initialize again, should be a no-op.
333 self.check_qbb_consistency(init_output_refs, qbb)
334 # Use QBB to initialize again but tell it to expect an empty run.
335 with self.assertRaises(ConflictingDefinitionError):
336 quantum_graph.init_output_run(qbb, existing=False)
337 # Transferring datasets back to the main butler (i.e. insert DB entries
338 # for them).
339 butler.transfer_from(qbb, itertools.chain.from_iterable(init_output_refs.values()))
340 # Check that the QG refs are the same as the ones we find in the repo.
341 self.assert_init_output_refs_equal(
342 self.find_init_output_refs(pipeline_graph, butler),
343 init_output_refs,
344 )
345 # Initialize again with the QG; should be a no-op.
346 quantum_graph.init_output_run(butler)
347 self.assert_init_output_refs_equal(
348 self.find_init_output_refs(pipeline_graph, butler),
349 init_output_refs,
350 )
351 # Initialize again with the QG but tell it to expect an empty run.
352 with self.assertRaises(ConflictingDefinitionError):
353 quantum_graph.init_output_run(butler, existing=False)
354 # Initialize with the pipeline graph, should be a no-op.
355 pipeline_graph.init_output_run(butler)
356 return quantum_graph
358 def test_two_tasks_no_conversions(self) -> None:
359 """Test a two-task pipeline with an overall init-input, an overall
360 init-output, and an init-intermediate.
361 """
362 a_config = DynamicTestPipelineTaskConfig()
363 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime")
364 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="intermediate_runtime")
365 a_config.init_inputs["ii"] = DynamicConnectionConfig(dataset_type_name="input_init")
366 a_config.init_outputs["io"] = DynamicConnectionConfig(dataset_type_name="intermediate_init")
367 b_config = DynamicTestPipelineTaskConfig()
368 b_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="intermediate_runtime")
369 b_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
370 b_config.init_inputs["ii"] = DynamicConnectionConfig(dataset_type_name="intermediate_init")
371 b_config.init_outputs["io"] = DynamicConnectionConfig(dataset_type_name="output_init")
372 pipeline_graph = PipelineGraph()
373 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
374 pipeline_graph.add_task("b", DynamicTestPipelineTask, b_config)
375 with self.prep_butler(pipeline_graph) as butler:
376 self.init_with_pipeline_graph_first(pipeline_graph, butler, "run1")
377 self.assertEqual(butler.get("a_config", collections="run1"), a_config)
378 self.assertEqual(butler.get("b_config", collections="run1"), b_config)
379 self.init_with_quantum_graph_first(pipeline_graph, butler, "run2")
380 self.assertEqual(butler.get("a_config", collections="run2"), a_config)
381 self.assertEqual(butler.get("b_config", collections="run2"), b_config)
382 self.init_with_qbb_first(pipeline_graph, butler, "run3")
383 self.assertEqual(butler.get("a_config", collections="run3"), a_config)
384 self.assertEqual(butler.get("b_config", collections="run3"), b_config)
386 def test_optional_input_unregistered(self) -> None:
387 """Test that an optional input dataset type that is not registered is
388 not considered an error.
389 """
390 a_config = DynamicTestPipelineTaskConfig()
391 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime", minimum=0)
392 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
393 pipeline_graph = PipelineGraph()
394 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
395 with self.make_butler() as butler:
396 pipeline_graph.resolve(butler.registry)
397 butler.registry.registerDatasetType(pipeline_graph.dataset_types["a_config"].dataset_type)
398 butler.registry.registerDatasetType(pipeline_graph.dataset_types["a_log"].dataset_type)
399 butler.registry.registerDatasetType(pipeline_graph.dataset_types["a_metadata"].dataset_type)
400 butler.registry.registerDatasetType(pipeline_graph.dataset_types["output_runtime"].dataset_type)
401 pipeline_graph.check_dataset_type_registrations(butler, include_packages=False)
403 def test_registration_changed(self) -> None:
404 """Test that we get an error when dataset type registrations in a data
405 repository change between the time a pipeline graph is resolved (e.g.
406 at QG generation) and when dataset types are checked later (e.g. during
407 execution).
408 """
409 a_config = DynamicTestPipelineTaskConfig()
410 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime")
411 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
412 pipeline_graph = PipelineGraph()
413 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
414 with self.make_butler() as butler:
415 pipeline_graph.resolve(butler.registry)
416 pipeline_graph.register_dataset_types(butler)
417 butler.registry.removeDatasetType("input_runtime")
418 butler.registry.registerDatasetType(
419 DatasetType("input_runtime", {"instrument"}, "StructuredDataList", universe=butler.dimensions)
420 )
421 with self.assertRaises(ConflictingDefinitionError):
422 pipeline_graph.check_dataset_type_registrations(butler)
424 @unittest.skipUnless(
425 _have_example_storage_classes(), "Arrow/Astropy/Pandas storage classes are not available."
426 )
427 def test_init_intermediate_component(self) -> None:
428 """Test init_output_run with an init-intermediate that is written as
429 a composite and read as a component.
430 """
431 a_config = DynamicTestPipelineTaskConfig()
432 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime")
433 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="intermediate_runtime")
434 a_config.init_outputs["io"] = DynamicConnectionConfig(
435 dataset_type_name="intermediate_init", storage_class="ArrowTable"
436 )
437 b_config = DynamicTestPipelineTaskConfig()
438 b_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="intermediate_runtime")
439 b_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
440 b_config.init_inputs["ii"] = DynamicConnectionConfig(
441 dataset_type_name="intermediate_init.schema", storage_class="ArrowSchema"
442 )
443 pipeline_graph = PipelineGraph()
444 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
445 pipeline_graph.add_task("b", DynamicTestPipelineTask, b_config)
446 with self.prep_butler(pipeline_graph) as butler:
447 self.init_with_pipeline_graph_first(pipeline_graph, butler, "run1")
448 self.assertEqual(butler.get("a_config", collections="run1"), a_config)
449 self.assertEqual(butler.get("b_config", collections="run1"), b_config)
450 self.init_with_quantum_graph_first(pipeline_graph, butler, "run2")
451 self.assertEqual(butler.get("a_config", collections="run2"), a_config)
452 self.assertEqual(butler.get("b_config", collections="run2"), b_config)
453 self.init_with_qbb_first(pipeline_graph, butler, "run3")
454 self.assertEqual(butler.get("a_config", collections="run3"), a_config)
455 self.assertEqual(butler.get("b_config", collections="run3"), b_config)
457 def test_no_get_init_input_callback(self) -> None:
458 """Test calling PipelineGraph.instantiate_tasks with no get_init_input
459 callback when one is necessary.
460 """
461 a_config = DynamicTestPipelineTaskConfig()
462 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime")
463 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
464 a_config.init_inputs["ii"] = DynamicConnectionConfig(dataset_type_name="input_init")
465 pipeline_graph = PipelineGraph()
466 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
467 with self.make_butler() as butler:
468 pipeline_graph.resolve(butler.registry)
469 with self.assertRaises(ValueError):
470 pipeline_graph.instantiate_tasks()
472 def test_multiple_init_input_consumers(self) -> None:
473 """Test init_output_run when there are two tasks consuming the same
474 init-input.
475 """
476 a_config = DynamicTestPipelineTaskConfig()
477 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime")
478 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="intermediate_runtime")
479 a_config.init_inputs["ii"] = DynamicConnectionConfig(dataset_type_name="input_init")
480 a_config.init_outputs["io"] = DynamicConnectionConfig(dataset_type_name="output_init")
481 b_config = DynamicTestPipelineTaskConfig()
482 b_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="intermediate_runtime")
483 b_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
484 b_config.init_inputs["ii"] = DynamicConnectionConfig(dataset_type_name="input_init")
485 pipeline_graph = PipelineGraph()
486 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
487 pipeline_graph.add_task("b", DynamicTestPipelineTask, b_config)
488 with self.prep_butler(pipeline_graph) as butler:
489 self.init_with_pipeline_graph_first(pipeline_graph, butler, "run1")
490 self.assertEqual(butler.get("a_config", collections="run1"), a_config)
491 self.assertEqual(butler.get("b_config", collections="run1"), b_config)
492 self.init_with_quantum_graph_first(pipeline_graph, butler, "run2")
493 self.assertEqual(butler.get("a_config", collections="run2"), a_config)
494 self.assertEqual(butler.get("b_config", collections="run2"), b_config)
495 self.init_with_qbb_first(pipeline_graph, butler, "run3")
496 self.assertEqual(butler.get("a_config", collections="run3"), a_config)
497 self.assertEqual(butler.get("b_config", collections="run3"), b_config)
499 def test_config_change(self) -> None:
500 """Test init_output_run when there is an existing config that is
501 inconsistent with the one in the pipeline graph.
502 """
503 a_config = DynamicTestPipelineTaskConfig()
504 a_config.inputs["i"] = DynamicConnectionConfig(dataset_type_name="input_runtime")
505 a_config.outputs["o"] = DynamicConnectionConfig(dataset_type_name="output_runtime")
506 pipeline_graph = PipelineGraph()
507 pipeline_graph.add_task("a", DynamicTestPipelineTask, a_config)
508 with self.prep_butler(pipeline_graph) as butler:
509 butler.collections.register("run1")
510 butler.put(DynamicTestPipelineTaskConfig(), "a_config", run="run1")
511 with self.assertRaises(ConflictingDefinitionError):
512 pipeline_graph.init_output_run(
513 butler.clone(run="run1", collections=[self.INPUT_COLLECTION, "run1"])
514 )
515 quantum_graph_builder = AllDimensionsQuantumGraphBuilder(
516 pipeline_graph,
517 butler,
518 skip_existing_in=["run1"],
519 output_run="run1",
520 input_collections=[self.INPUT_COLLECTION],
521 )
522 quantum_graph = quantum_graph_builder.finish(metadata={"output_run": "run1"}).assemble()
523 with self.assertRaises(ConflictingDefinitionError):
524 quantum_graph.init_output_run(
525 butler.clone(run="run1", collections=[self.INPUT_COLLECTION, "run1"])
526 )
529if __name__ == "__main__":
530 lsst.utils.tests.init()
531 unittest.main()