Coverage for tests/test_predicted_qg.py: 13%
211 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 01:47 -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 logging
30import os
31import shutil
32import subprocess
33import tempfile
34import unittest
35import uuid
36from io import StringIO
38import lsst.utils.tests
39from lsst.pipe.base import QuantumGraph
40from lsst.pipe.base.dot_tools import graph2dot
41from lsst.pipe.base.mermaid_tools import graph2mermaid
42from lsst.pipe.base.quantum_graph import (
43 AddressReader,
44 PredictedDatasetInfo,
45 PredictedQuantumGraph,
46 PredictedQuantumInfo,
47)
48from lsst.pipe.base.tests.mocks import DynamicConnectionConfig, InMemoryRepo
50logging.getLogger("timer").setLevel(logging.INFO)
51logging.getLogger("lsst").setLevel(logging.INFO)
52logging.getLogger("lsst.pipe.base.quantum_graph").setLevel(logging.DEBUG)
55class PredictedQuantumGraphTestCase(unittest.TestCase):
56 """Unit tests for the predicted quantum graph classes.
58 These tests focus on the interface of the predicted quantum graph
59 especially I/O and conversion to and from the old quantum graph class.
61 Additional test coverage is provided by the tests for quantum graph
62 builders and executors, which all use this class as well.
63 """
65 def setUp(self):
66 self.helper = InMemoryRepo("base.yaml", "spatial.yaml")
67 self.helper.add_task(
68 "calibrate",
69 dimensions=["visit", "detector"],
70 inputs={
71 "input_image": DynamicConnectionConfig(
72 dataset_type_name="raw",
73 dimensions=["visit", "detector"],
74 )
75 },
76 prerequisite_inputs={
77 "refcat": DynamicConnectionConfig(
78 dataset_type_name="references",
79 dimensions=["htm7"],
80 multiple=True,
81 )
82 },
83 init_outputs={
84 "output_schema": DynamicConnectionConfig(
85 dataset_type_name="source_schema",
86 )
87 },
88 outputs={
89 "output_image": DynamicConnectionConfig(
90 dataset_type_name="image",
91 dimensions=["visit", "detector"],
92 ),
93 "output_table": DynamicConnectionConfig(
94 dataset_type_name="source_detector",
95 dimensions=["visit", "detector"],
96 ),
97 },
98 )
99 self.helper.add_task(
100 "consolidate",
101 dimensions=["visit"],
102 init_inputs={
103 "input_schema": DynamicConnectionConfig(
104 dataset_type_name="source_schema",
105 )
106 },
107 inputs={
108 "input_table": DynamicConnectionConfig(
109 dataset_type_name="source_detector",
110 dimensions=["visit", "detector"],
111 multiple=True,
112 )
113 },
114 outputs={
115 "output_table": DynamicConnectionConfig(
116 dataset_type_name="source",
117 dimensions=["visit"],
118 )
119 },
120 )
121 self.helper.add_task(
122 "resample",
123 dimensions=["patch", "visit"],
124 inputs={
125 "input_image": DynamicConnectionConfig(
126 dataset_type_name="image",
127 dimensions=["visit", "detector"],
128 multiple=True,
129 )
130 },
131 outputs={
132 "output_image": DynamicConnectionConfig(
133 dataset_type_name="warp",
134 dimensions=["patch", "visit"],
135 )
136 },
137 )
138 self.helper.add_task(
139 "coadd",
140 dimensions=["patch", "band"],
141 inputs={
142 "input_image": DynamicConnectionConfig(
143 dataset_type_name="warp",
144 dimensions=["patch", "visit"],
145 multiple=True,
146 )
147 },
148 outputs={
149 "output_image": DynamicConnectionConfig(
150 dataset_type_name="coadd",
151 dimensions=["patch", "band"],
152 ),
153 "output_table": DynamicConnectionConfig(
154 dataset_type_name="object",
155 dimensions=["patch", "band"],
156 # Like all other (defaulted) storage classes here,
157 # 'ArrowAstropy' below will be mocked; we pick it so we can
158 # try a component input, and because we know its pytype is
159 # safe to import here.
160 storage_class="ArrowAstropy",
161 ),
162 },
163 )
164 self.helper.add_task(
165 "match",
166 dimensions=["htm6"],
167 inputs={
168 "input_object": DynamicConnectionConfig(
169 dataset_type_name="object",
170 dimensions=["patch", "band"],
171 multiple=True,
172 storage_class="ArrowAstropy",
173 ),
174 "input_source": DynamicConnectionConfig(
175 dataset_type_name="source",
176 dimensions=["visit"],
177 multiple=True,
178 ),
179 "input_object_schema": DynamicConnectionConfig(
180 dataset_type_name="object.schema",
181 dimensions=["visit"],
182 multiple=True,
183 storage_class="ArrowAstropySchema",
184 ),
185 },
186 outputs={
187 "output_table": DynamicConnectionConfig(
188 dataset_type_name="matches",
189 dimensions=["htm6"],
190 )
191 },
192 )
193 self.builder = self.helper.make_quantum_graph_builder()
195 def check_quantum_graph(
196 self,
197 qg: PredictedQuantumGraph,
198 *,
199 all_tasks: bool = True,
200 thin_graph: bool = True,
201 all_quantum_datasets: bool = True,
202 init_quanta: bool = True,
203 dimension_data_loaded: bool = True,
204 dimension_data_deserialized: bool = True,
205 converting_partial: bool = False,
206 ) -> None:
207 """Test the attributes and methods of a quantum graph produced by this
208 test case's builder.
210 Parameters
211 ----------
212 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
213 Graph to test.
214 all_tasks : `bool`, optional
215 Whether all tasks in the pipeline were loaded.
216 thin_graph : `bool`, optional
217 Whether the ``thin_graph`` component was loaded.
218 all_quantum_datasets : `bool`, optional
219 Whether all quantum datasets were loaded.
220 init_quanta : `bool`, optional
221 Whether the ``init_quanta`` component was loaded.
222 dimension_data_loaded : `bool`, optional
223 Whether the ``dimension_data`` component was loaded at all.
224 dimension_data_deserialized : `bool`, optional
225 Whether all dimension records are expected to have been
226 deserialized. Ignored if ``dimension_data_loaded`` is `False`.
227 converting_partial : `bool`, optional
228 Whether this was a partial load that only included some quanta.
229 """
230 self.assertEqual(qg.header.inputs, ["input_chain"])
231 self.assertEqual(qg.header.output, "output_chain")
232 self.assertEqual(qg.header.output_run, "output_run")
233 self.assertEqual(qg.header.metadata["stuff"], "whatever")
234 if not converting_partial:
235 # When partial-reading an old QG, counts reflect what was loaded.
236 # In all other cases they reflect the full original graph.
237 self.assertEqual(qg.header.n_task_quanta["calibrate"], 8)
238 self.assertEqual(qg.header.n_task_quanta["consolidate"], 2)
239 self.assertEqual(qg.header.n_task_quanta["resample"], 10)
240 self.assertEqual(qg.header.n_task_quanta["coadd"], 10)
241 self.assertEqual(qg.header.n_task_quanta["match"], 2)
242 self.assertEqual(qg.header.n_quanta, 32)
243 self.assertEqual(
244 qg.header.n_datasets,
245 63 # non-automatic datasets (see breakdown below)
246 + 32 * 2 # one metadata and one log for each quantam
247 + 5 # one config for each task
248 + 1, # packages
249 )
250 if all_tasks:
251 self.assertFalse(self.helper.pipeline_graph.diff_tasks(qg.pipeline_graph))
252 if thin_graph or all_quantum_datasets:
253 self.assertEqual(len(qg), qg.header.n_quanta)
254 self.assertEqual(len(qg.quantum_only_xgraph), qg.header.n_quanta)
255 self.assertEqual(len(qg.quanta_by_task["calibrate"]), 8)
256 self.assertEqual(len(qg.quanta_by_task["consolidate"]), 2)
257 self.assertEqual(len(qg.quanta_by_task["resample"]), 10)
258 self.assertEqual(len(qg.quanta_by_task["coadd"]), 10)
259 self.assertEqual(len(qg.quanta_by_task["match"]), 2)
260 if init_quanta:
261 if all_tasks:
262 self.assertEqual(len(qg.datasets_by_type["calibrate_config"]), 1)
263 self.assertEqual(len(qg.datasets_by_type["consolidate_config"]), 1)
264 self.assertEqual(len(qg.datasets_by_type["resample_config"]), 1)
265 self.assertEqual(len(qg.datasets_by_type["coadd_config"]), 1)
266 self.assertEqual(len(qg.datasets_by_type["match_config"]), 1)
267 self.assertEqual(len(qg.datasets_by_type["source_schema"]), 1)
268 self.assertEqual(len(qg.datasets_by_type["packages"]), 1)
269 n_init_datasets = 7
270 else:
271 n_init_datasets = 0
272 if all_quantum_datasets:
273 assert all_tasks
274 self.assertEqual(
275 len(qg.bipartite_xgraph),
276 qg.header.n_quanta + qg.header.n_datasets - n_init_datasets,
277 )
278 self.assertEqual(len(qg.datasets_by_type["raw"]), 8)
279 self.assertEqual(len(qg.datasets_by_type["references"]), 4)
280 self.assertEqual(len(qg.datasets_by_type["image"]), 8)
281 self.assertEqual(len(qg.datasets_by_type["source_detector"]), 8)
282 self.assertEqual(len(qg.datasets_by_type["source"]), 2)
283 self.assertEqual(len(qg.datasets_by_type["warp"]), 10)
284 self.assertEqual(len(qg.datasets_by_type["coadd"]), 10)
285 self.assertEqual(len(qg.datasets_by_type["object"]), 10)
286 self.assertEqual(len(qg.datasets_by_type["matches"]), 2)
287 # Spot-check some edges and graph structure.
288 for data_id, dataset_id in qg.datasets_by_type["source_detector"].items():
289 for producer_id, _, edge_data in qg.bipartite_xgraph.in_edges(dataset_id, data=True):
290 self.assertIn(producer_id, qg.quanta_by_task["calibrate"].values())
291 self.assertFalse(edge_data["is_read"])
292 self.assertEqual(
293 edge_data["pipeline_edges"],
294 [qg.pipeline_graph.tasks["calibrate"].get_output_edge("output_table")],
295 )
296 for _, consumer_id, edge_data in qg.bipartite_xgraph.out_edges(dataset_id, data=True):
297 self.assertIn(consumer_id, qg.quanta_by_task["consolidate"].values())
298 self.assertTrue(edge_data["is_read"])
299 self.assertEqual(
300 edge_data["pipeline_edges"],
301 [qg.pipeline_graph.tasks["consolidate"].get_input_edge("input_table")],
302 )
303 # We use 'is' checks instead of just equality because we don't want
304 # duplicates of these objects floating around wasting memory.
305 for task_label, quanta_for_task in qg.quanta_by_task.items():
306 for data_id, quantum_id in quanta_for_task.items():
307 d1: PredictedQuantumInfo = qg.quantum_only_xgraph.nodes[quantum_id]
308 self.assertEqual(d1["task_label"], task_label)
309 self.assertIs(d1["data_id"], data_id)
310 self.assertIs(d1["pipeline_node"], qg.pipeline_graph.tasks[task_label])
311 d2: PredictedQuantumInfo = qg.bipartite_xgraph.nodes[quantum_id]
312 self.assertEqual(d2["task_label"], task_label)
313 self.assertIs(d2["data_id"], data_id)
314 self.assertIs(d2["pipeline_node"], qg.pipeline_graph.tasks[task_label])
315 for dataset_type_name, datasets_for_type in qg.datasets_by_type.items():
316 if (
317 dataset_type_name == "source_schema"
318 or dataset_type_name == "packages"
319 or dataset_type_name.endswith("_config")
320 ):
321 continue
322 for data_id, dataset_id in datasets_for_type.items():
323 d3: PredictedDatasetInfo = qg.bipartite_xgraph.nodes[dataset_id]
324 self.assertIs(d3["data_id"], data_id)
325 self.assertIs(d3["pipeline_node"], qg.pipeline_graph.dataset_types[dataset_type_name])
326 if dimension_data_loaded:
327 self.assertIsNotNone(qg.dimension_data)
328 if dimension_data_deserialized and not converting_partial:
329 self.assertEqual(len(qg.dimension_data.records["instrument"]), 1)
330 self.assertEqual(len(qg.dimension_data.records["visit"]), 2)
331 self.assertEqual(len(qg.dimension_data.records["detector"]), 4)
332 self.assertEqual(len(qg.dimension_data.records["visit_detector_region"]), 8)
333 self.assertEqual(len(qg.dimension_data.records["tract"]), 2)
334 self.assertEqual(len(qg.dimension_data.records["patch"]), 9)
335 self.assertEqual(len(qg.dimension_data.records["band"]), 2)
336 self.assertEqual(len(qg.dimension_data.records["physical_filter"]), 2)
338 def define_partial_read(self, qg: PredictedQuantumGraph) -> list[uuid.UUID]:
339 """Return a list of quantum IDs to test partial loads.
341 Parameters
342 ----------
343 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
344 Full graph to obtain quantum IDs from. Note that quantum IDs are
345 random and change every test run, and hence test behavior should
346 never depend on explicit values or their sort order.
348 Returns
349 -------
350 quanta : `list` [ `uuid.UUID` ]
351 Quanta to use for partial-read tests. These always have the same
352 task labels and data IDs: all ``calibrate`` and ``consolidate``
353 quanta for ``visit==1``.
354 """
355 return [
356 quantum_id
357 for data_id, quantum_id in qg.quanta_by_task["calibrate"].items()
358 if data_id["visit"] == 1
359 ] + [
360 quantum_id
361 for data_id, quantum_id in qg.quanta_by_task["consolidate"].items()
362 if data_id["visit"] == 1
363 ]
365 def check_partial_read(
366 self, qg: PredictedQuantumGraph, quanta: list[uuid.UUID], converting: bool = False
367 ) -> None:
368 """Test the attributes and methods of a partially-loaded quantum graph.
370 Parameters
371 ----------
372 qg : `lsst.pipe.base.quantum_graph.PredictedQuantumGraph`
373 Graph for which only ``quanta`` were loaded.
374 quanta : `list` [ `uuid.UUID` ]
375 IDs of the quanta that were loaded. Must be the result of a call
376 to `define_partial_read`.
377 converting : `bool`, optional
378 Whether this graph was read via the old
379 `lsst.pipe.base.QuantumGraph` class and then converted.
380 """
381 self.check_quantum_graph(
382 qg,
383 all_tasks=False,
384 thin_graph=False,
385 all_quantum_datasets=False,
386 converting_partial=converting,
387 dimension_data_deserialized=converting,
388 )
389 self.assertEqual(len(qg), 5)
390 self.assertEqual(len(qg.quantum_only_xgraph), 5)
391 execution_quanta = qg.build_execution_quanta()
392 self.assertEqual(execution_quanta.keys(), set(quanta))
393 for quantum_id, quantum in execution_quanta.items():
394 self.assertIs(qg.bipartite_xgraph.nodes[quantum_id]["quantum"], quantum)
395 self.assertIs(qg.quantum_only_xgraph.nodes[quantum_id]["quantum"], quantum)
397 def test_build(self) -> None:
398 """Test building a `PredictedQuantumGraph` by
399 inspecting the result.
400 """
401 qg = self.builder.finish(
402 output="output_chain",
403 metadata={"stuff": "whatever"},
404 attach_datastore_records=False,
405 ).assemble()
406 self.check_quantum_graph(qg)
408 def test_from_old_quantum_graph(self) -> None:
409 """Test building a old `QuantumGraph` and then
410 converting it to a `PredictedQuantumGraph`.
411 """
412 old_qg = self.builder.build(
413 metadata={
414 "input": ["input_chain"],
415 "output": "output_chain",
416 "output_run": "output_run",
417 "stuff": "whatever",
418 },
419 attach_datastore_records=False,
420 )
421 new_qg = PredictedQuantumGraph.from_old_quantum_graph(old_qg)
422 self.check_quantum_graph(new_qg)
424 def test_read_execution_quanta_old_file(self) -> None:
425 """Test building a old `QuantumGraph`, saving it, and then reading it
426 via `PredictedQuantumGraph.read_execution_quanta`.
427 """
428 old_qg = self.builder.build(
429 metadata={
430 "input": ["input_chain"],
431 "output": "output_chain",
432 "output_run": "output_run",
433 "stuff": "whatever",
434 },
435 attach_datastore_records=False,
436 )
437 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
438 tmpfile = os.path.join(tmpdir, "old.qgraph")
439 old_qg.saveUri(tmpfile)
440 # Read everything.
441 qg1 = PredictedQuantumGraph.read_execution_quanta(tmpfile)
442 self.check_quantum_graph(qg1, thin_graph=False)
443 # Read just a few quanta.
444 quanta = self.define_partial_read(qg1)
445 qg2 = PredictedQuantumGraph.read_execution_quanta(tmpfile, quantum_ids=quanta)
446 self.check_partial_read(qg2, quanta, converting=True)
448 def test_roundtrip_old_quantum_graph(self) -> None:
449 """Test building a `PredictedQuantumGraph` and round-tripping it
450 through the old `QuantumGraph` class.
451 """
452 qg1 = self.builder.finish(
453 output="output_chain",
454 metadata={"stuff": "whatever"},
455 attach_datastore_records=False,
456 ).assemble()
457 old_qg = qg1.to_old_quantum_graph()
458 qg2 = PredictedQuantumGraph.from_old_quantum_graph(old_qg)
459 self.check_quantum_graph(qg2)
461 def test_write_new_as_old(self) -> None:
462 """Test building a `PredictedQuantumGraphComponents`, saving it with
463 the old extension, and reading it back in both the old class and the
464 new class.
465 """
466 components = self.builder.finish(
467 output="output_chain",
468 metadata={"stuff": "whatever"},
469 attach_datastore_records=False,
470 )
471 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
472 tmpfile = os.path.join(tmpdir, "old.qgraph")
473 components.write(tmpfile)
474 qg1 = PredictedQuantumGraph.read_execution_quanta(tmpfile)
475 self.check_quantum_graph(qg1)
476 old_qg = QuantumGraph.loadUri(tmpfile)
477 qg2 = PredictedQuantumGraph.from_old_quantum_graph(old_qg)
478 self.check_quantum_graph(qg2)
480 def test_read_new_as_old(self) -> None:
481 """Test building a `PredictedQuantumGraphComponents`, saving it with
482 the new extension, and reading it back in via the old class.
483 """
484 components = self.builder.finish(
485 output="output_chain",
486 metadata={"stuff": "whatever"},
487 attach_datastore_records=False,
488 )
489 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
490 tmpfile = os.path.join(tmpdir, "new.qg")
491 components.write(tmpfile)
492 old_qg = QuantumGraph.loadUri(tmpfile)
493 new_qg = PredictedQuantumGraph.from_old_quantum_graph(old_qg)
494 self.check_quantum_graph(new_qg)
496 def test_io(self) -> None:
497 """Test saving a `PredictedQuantumGraphComponents` and reading it back
498 in as a `PredictedQuantumGraph`, both fully and partially, and as a
499 `QuantumGraph`, both fully and partially.
500 """
501 components = self.builder.finish(
502 output="output_chain",
503 metadata={"stuff": "whatever"},
504 attach_datastore_records=False,
505 )
506 # We use a small page size since this is a tiny graph and hence
507 # otherwise all addresses would fit in a single page.
508 four_row_page_size = AddressReader.compute_row_size(int_size=8, n_addresses=1) * 4
509 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
510 tmpfile = os.path.join(tmpdir, "new.qg")
511 components.write(tmpfile, zstd_dict_n_inputs=24) # enable dict compression code path
512 # Test a full read with the new class.
513 with PredictedQuantumGraph.open(tmpfile, page_size=four_row_page_size) as reader:
514 full_qg = reader.read_all().finish()
515 self.check_quantum_graph(full_qg, dimension_data_deserialized=False)
516 # Test a full read with the old class (uses new class and then
517 # converts to old, and we convert back to new for the test).
518 self.check_quantum_graph(
519 PredictedQuantumGraph.from_old_quantum_graph(QuantumGraph.loadUri(tmpfile))
520 )
521 # Test a partial read with the old class.
522 quanta = self.define_partial_read(full_qg)
523 self.check_partial_read(
524 PredictedQuantumGraph.from_old_quantum_graph(QuantumGraph.loadUri(tmpfile, nodes=quanta)),
525 quanta,
526 converting=True,
527 )
528 # Test a thin but shallow read with the new class.
529 with PredictedQuantumGraph.open(tmpfile, page_size=four_row_page_size) as reader:
530 thin_qg = reader.read_thin_graph().finish()
531 self.check_quantum_graph(
532 thin_qg,
533 dimension_data_deserialized=False,
534 all_quantum_datasets=False,
535 dimension_data_loaded=False,
536 init_quanta=False,
537 )
538 # Test a deep read of just a few quanta with the new class.
539 narrow_qg = PredictedQuantumGraph.read_execution_quanta(
540 tmpfile,
541 quantum_ids=quanta,
542 page_size=four_row_page_size,
543 )
544 self.check_partial_read(narrow_qg, quanta)
546 def test_no_compression_dict(self) -> None:
547 """Test saving a `PredictedQuantumGraphComponents` and reading it back
548 in as a `PredictedQuantumGraph` with the compression dictionary
549 disabled.
550 """
551 components = self.builder.finish(
552 output="output_chain",
553 metadata={"stuff": "whatever"},
554 attach_datastore_records=False,
555 )
556 # We use a small page size since this is a tiny graph and hence
557 # otherwise all addresses would fit in a single page.
558 three_row_page_size = AddressReader.compute_row_size(int_size=8, n_addresses=1) * 3
559 with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
560 tmpfile = os.path.join(tmpdir, "new.qg")
561 components.write(tmpfile, zstd_dict_size=0)
562 with PredictedQuantumGraph.open(tmpfile, page_size=three_row_page_size) as reader:
563 full_qg = reader.read_all().finish()
564 self.check_quantum_graph(full_qg, dimension_data_deserialized=False)
566 def test_dot(self) -> None:
567 """Test visualization via GraphViz dot."""
568 qg = self.builder.finish(attach_datastore_records=False).assemble()
569 stream = StringIO()
570 graph2dot(qg, stream)
571 if (dot_cmd := shutil.which("dot")) is None:
572 raise unittest.SkipTest("Aborting test early; 'dot' command is not available.")
573 # Just check that the dot command can parse what we've given it.
574 result = subprocess.run(
575 [dot_cmd, "-Txdot"], input=stream.getvalue().encode(), stdout=subprocess.DEVNULL
576 )
577 result.check_returncode()
579 def test_mermaid(self) -> None:
580 """Test visualization via Mermaid."""
581 qg = self.builder.finish(attach_datastore_records=False).assemble()
582 stream = StringIO()
583 # Just check that it runs without error.
584 graph2mermaid(qg, stream)
586 def test_update_output_run(self) -> None:
587 """Test the PredictedQuantumGraphComponents.output_run method."""
588 components = self.builder.finish(
589 output="output_chain",
590 metadata={"stuff": "whatever"},
591 attach_datastore_records=False,
592 )
593 components.update_output_run("new_output_run")
594 qg = components.assemble()
595 for ref in qg.get_init_outputs("calibrate").values():
596 self.assertEqual(ref.run, "new_output_run")
597 for quantum in qg.build_execution_quanta().values():
598 for ref in itertools.chain.from_iterable(quantum.outputs.values()):
599 self.assertEqual(ref.run, "new_output_run", msg=str(ref))
600 for ref in itertools.chain.from_iterable(quantum.inputs.values()):
601 if ref.datasetType.name not in ("raw", "references"):
602 self.assertEqual(ref.run, "new_output_run", msg=str(ref))
605if __name__ == "__main__":
606 lsst.utils.tests.init()
607 unittest.main()