Coverage for tests/test_graphBuilder.py: 21%
87 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-19 01:44 -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/>.
28"""Tests of things related to the GraphBuilder class."""
30import io
31import logging
32import unittest
34import lsst.utils.tests
35from lsst.daf.butler import Butler, DatasetType
36from lsst.daf.butler.registry import UserExpressionError
37from lsst.pipe.base import PipelineGraph, QuantumGraph
38from lsst.pipe.base.all_dimensions_quantum_graph_builder import (
39 AllDimensionsQuantumGraphBuilder,
40 DatasetQueryConstraintVariant,
41)
42from lsst.pipe.base.tests import simpleQGraph
43from lsst.pipe.base.tests.mocks import (
44 DynamicConnectionConfig,
45 DynamicTestPipelineTask,
46 DynamicTestPipelineTaskConfig,
47 MockDataset,
48 MockStorageClass,
49)
50from lsst.utils.tests import temporaryDirectory
52_LOG = logging.getLogger(__name__)
55class GraphBuilderTestCase(unittest.TestCase):
56 """Test graph building."""
58 def _assertGraph(self, graph: QuantumGraph) -> None:
59 """Check basic structure of the graph."""
60 for taskDef in graph.iterTaskGraph():
61 refs = graph.initOutputRefs(taskDef)
62 # task has one initOutput, second ref is for config dataset
63 self.assertEqual(len(refs), 2)
65 self.assertEqual(len(list(graph.inputQuanta)), 1)
67 # This includes only "packages" dataset for now.
68 refs = graph.globalInitOutputRefs()
69 self.assertEqual(len(refs), 1)
71 def testDefault(self):
72 """Simple test to verify makeSimpleQGraph can be used to make a Quantum
73 Graph.
74 """
75 with temporaryDirectory() as root:
76 # makeSimpleQGraph calls GraphBuilder.
77 butler, qgraph = simpleQGraph.makeSimpleQGraph(root=root)
78 # by default makeSimpleQGraph makes a graph with 5 nodes
79 self.assertEqual(len(qgraph), 5)
80 self._assertGraph(qgraph)
81 constraint = DatasetQueryConstraintVariant.OFF
82 _, qgraph2 = simpleQGraph.makeSimpleQGraph(
83 butler=butler, datasetQueryConstraint=constraint, callPopulateButler=False
84 )
85 # When all outputs are random resolved refs, direct comparison
86 # of graphs does not work because IDs are different. Can only
87 # verify the number of quanta in the graph without doing something
88 # terribly complicated.
89 self.assertEqual(len(qgraph2), 5)
90 constraint = DatasetQueryConstraintVariant.fromExpression("add_dataset0")
91 _, qgraph3 = simpleQGraph.makeSimpleQGraph(
92 butler=butler, datasetQueryConstraint=constraint, callPopulateButler=False
93 )
94 self.assertEqual(len(qgraph3), 5)
96 def test_empty_qg(self):
97 """Test that making an empty QG doesn't raise exceptions."""
98 config = DynamicTestPipelineTaskConfig()
99 config.inputs["i"] = DynamicConnectionConfig(
100 dataset_type_name="input",
101 storage_class="StructuredDataDict",
102 dimensions=["detector"],
103 )
104 config.init_inputs["ii"] = DynamicConnectionConfig(
105 dataset_type_name="init_input",
106 storage_class="StructuredDataDict",
107 )
108 config.outputs["o"] = DynamicConnectionConfig(
109 dataset_type_name="output",
110 storage_class="StructuredDataDict",
111 dimensions=["detector"],
112 )
113 config.init_outputs["io"] = DynamicConnectionConfig(
114 dataset_type_name="init_output",
115 storage_class="StructuredDataDict",
116 )
117 pipeline_graph = PipelineGraph()
118 pipeline_graph.add_task("a", DynamicTestPipelineTask, config)
119 with temporaryDirectory() as repo_path:
120 Butler.makeRepo(repo_path)
121 butler = Butler.from_config(repo_path, writeable=True, run="test_empty_qg")
122 MockStorageClass.get_or_register_mock("StructuredDataDict")
123 butler.registry.registerDatasetType(
124 DatasetType(
125 "input",
126 dimensions=butler.dimensions.conform(["detector"]),
127 storageClass="_mock_StructuredDataDict",
128 )
129 )
130 init_input_dataset_type = DatasetType(
131 "init_input",
132 dimensions=butler.dimensions.empty,
133 storageClass="_mock_StructuredDataDict",
134 )
135 butler.registry.registerDatasetType(init_input_dataset_type)
136 # Init-input initially exists, but input does not (hence empty QG).
137 butler.put(
138 MockDataset(
139 dataset_id=None,
140 dataset_type=init_input_dataset_type.to_simple(),
141 data_id={},
142 run=butler.run,
143 ),
144 "init_input",
145 )
146 # Attempt to make QG; should just be empty, with no exceptions.
147 self.assertFalse(AllDimensionsQuantumGraphBuilder(pipeline_graph, butler).build())
148 # Initialize the output run, try again, with same expected result.
149 pipeline_graph.register_dataset_types(butler)
150 pipeline_graph.init_output_run(butler)
151 self.assertFalse(AllDimensionsQuantumGraphBuilder(pipeline_graph, butler).build())
153 # Inconsistent governor dimensions are no longer an error, so this test
154 # fails with the new query system. We should probably check instead that
155 # logging includes an explanation for the empty QG, but it might not
156 # because explain_no_results isn't good enough yet.
157 @unittest.expectedFailure
158 def testAddInstrumentMismatch(self):
159 """Verify that a RuntimeError is raised if the instrument in the user
160 query does not match the instrument in the pipeline.
161 """
162 with temporaryDirectory() as root:
163 pipeline = simpleQGraph.makeSimplePipeline(
164 nQuanta=5, instrument="lsst.pipe.base.tests.simpleQGraph.SimpleInstrument"
165 )
166 with self.assertRaises(UserExpressionError):
167 simpleQGraph.makeSimpleQGraph(root=root, pipeline=pipeline, userQuery="instrument = 'foo'")
169 def testUserQueryBind(self):
170 """Verify that bind values work for user query."""
171 pipeline = simpleQGraph.makeSimplePipeline(
172 nQuanta=5, instrument="lsst.pipe.base.tests.simpleQGraph.SimpleInstrument"
173 )
174 instr = simpleQGraph.SimpleInstrument.getName()
175 # With a literal in the user query
176 with temporaryDirectory() as root:
177 simpleQGraph.makeSimpleQGraph(root=root, pipeline=pipeline, userQuery=f"instrument = '{instr}'")
178 # With a bind for the user query
179 with temporaryDirectory() as root:
180 simpleQGraph.makeSimpleQGraph(
181 root=root, pipeline=pipeline, userQuery="instrument = :instr", bind={"instr": instr}
182 )
184 def test_datastore_records(self):
185 """Test for generating datastore records."""
186 with temporaryDirectory() as root:
187 # need FileDatastore for this tests
188 butler, qgraph1 = simpleQGraph.makeSimpleQGraph(
189 root=root, inMemory=False, makeDatastoreRecords=True
190 )
192 # save and reload
193 buffer = io.BytesIO()
194 qgraph1.save(buffer)
195 buffer.seek(0)
196 qgraph2 = QuantumGraph.load(buffer, universe=butler.dimensions)
197 del buffer
199 for qgraph in (qgraph1, qgraph2):
200 self.assertEqual(len(qgraph), 5)
201 for i, qnode in enumerate(qgraph):
202 quantum = qnode.quantum
203 self.assertIsNotNone(quantum.datastore_records)
204 # only the first quantum has a pre-existing input
205 if i == 0:
206 datastore_name = "FileDatastore@<butlerRoot>"
207 self.assertEqual(set(quantum.datastore_records.keys()), {datastore_name})
208 records_data = quantum.datastore_records[datastore_name]
209 records = dict(records_data.records)
210 self.assertEqual(len(records), 1)
211 _, records = records.popitem()
212 records = records["file_datastore_records"]
213 self.assertEqual(
214 [record.path for record in records],
215 ["test/add_dataset0/add_dataset0_INSTR_det0_test.pickle"],
216 )
217 else:
218 self.assertEqual(quantum.datastore_records, {})
221if __name__ == "__main__":
222 lsst.utils.tests.init()
223 unittest.main()