Coverage for tests/test_pipelines.py: 96%
123 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-29 02:52 -0700
1# This file is part of ap_pipe.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://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 program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
22import itertools
23import tempfile
24import unittest
26import lsst.daf.butler.tests as butlerTests
27import lsst.pipe.base
28from lsst.pipe.base.tests.pipelineStepTester import PipelineStepTester # Can't use fully-qualified name
29import lsst.utils
30import lsst.utils.tests
32from lsst.resources import ResourcePath
35class PipelineDefintionsTestSuite(lsst.utils.tests.TestCase):
36 """Tests of the self-consistency of our pipeline definitions.
37 """
38 def setUp(self):
39 self.path = ResourcePath("eups://ap_pipe/pipelines/", forceDirectory=True)
40 # Each pipeline file should have a subset that represents it in
41 # higher-level pipelines.
42 self.synonyms = {"ApPipe.yaml": "apPipe",
43 "ApPipe-legacy.yaml": "apPipe",
44 "ApPipeDaytime.yaml": "apPipe",
45 "ApPipe-DECam-templates.yaml": "apPipe",
46 "ApPipeDaytime-DECam-templates.yaml": "apPipe",
47 "ApPipeWithIsrTaskLSST.yaml": "apPipe",
48 "ApPipeWithPreconvolution.yaml": "apPipe",
49 "ApPipeWithPreconvolution-legacy.yaml": "apPipe",
50 "ApPipeWithFakes.yaml": "apPipe",
51 "SingleFrame.yaml": "singleFrame",
52 "SingleFrameWithIsrTaskLSST.yaml": "singleFrame",
53 "RunIsrForCrosstalkSources.yaml": "runOverscan",
54 "CreateInjectionCatalogs.yaml": "fakeCreation",
55 }
57 def test_graph_build(self):
58 """Test that each pipeline definition file can be
59 used to build a graph.
60 """
61 files = ResourcePath.findFileResources([self.path], file_filter=r".*\.yaml$")
62 for file in files:
63 if "QuickTemplate" in file.path:
64 # Our QuickTemplate definition cannot be tested here because it
65 # depends on drp_tasks, which we cannot make a dependency here.
66 continue
67 if "PromptTemplate" in file.path:
68 # Our PromptTemplate definition cannot be tested here because it
69 # depends on drp_tasks, which we cannot make a dependency here.
70 continue
71 with self.subTest(file=str(file)):
72 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
73 pipeline.addConfigOverride("parameters", "apdb_config", "some/file/path.yaml")
74 # If this fails, it will produce a useful error message.
75 pipeline.to_graph()
77 def test_datasets(self):
78 files = ResourcePath.findFileResources(
79 [self.path.join("_ingredients", forceDirectory=True)], file_filter=r".*\.yaml$"
80 )
81 for file in files:
82 if "QuickTemplate" in file.path:
83 # Our QuickTemplate definition cannot be tested here because it
84 # depends on drp_tasks, which we cannot make a dependency here.
85 continue
86 if "injection/" in file.path:
87 # The source-injection post-processing ingredient is a partial
88 # pipeline merged into full AP pipelines at build time;
89 # it is validated separately by test_injection_ingredient.
90 continue
91 with self.subTest(file=str(file)):
92 expected_inputs = {
93 # ISR
94 "raw", "camera", "crosstalk", "crosstalkSources", "bias", "dark", "flat", "ptc",
95 "fringe", "straylightData", "bfKernel", "newBFKernel", "defects", "linearizer",
96 "opticsTransmission", "filterTransmission", "atmosphereTransmission",
97 "illumMaskedImage", "deferredChargeCalib",
98 # ISR-LSST
99 "bfk", "cti", "dnlLUT", "gain_correction",
100 # Everything else
101 "skyMap", "gaia_dr3_20230707", "gaia_dr2_20200414", "ps1_pv3_3pi_20170110",
102 "template_coadd", "pretrainedModelPackage", "dia_source_apdb"
103 }
104 # Detect source-injection pipelines by task label rather than
105 # relying on filename conventions.
106 temp_pipeline = lsst.pipe.base.Pipeline.from_uri(file)
107 temp_pipeline.addConfigOverride("parameters", "apdb_config", "some/file/path.yaml")
108 if "injectVisit" in temp_pipeline.task_labels:
109 expected_inputs.add("injection_catalog")
110 expected_inputs.add("VisitDetectorFakeSourceCat")
111 tester = PipelineStepTester(
112 filename=file,
113 step_suffixes=[""], # Test full pipeline
114 initial_dataset_types=[("ps1_pv3_3pi_20170110", {"htm7"}, "SimpleCatalog", False),
115 ("gaia_dr2_20200414", {"htm7"}, "SimpleCatalog", False),
116 ("gaia_dr3_20230707", {"htm7"}, "SimpleCatalog", False),
117 ],
118 expected_inputs=expected_inputs,
119 # Pipeline outputs highly in flux, don't test
120 expected_outputs=set(),
121 pipeline_patches={"parameters:apdb_config": "some/file/path.yaml",
122 },
123 )
124 # Tester modifies Butler registry, so need a fresh repo every time
125 with tempfile.TemporaryDirectory() as tempRepo:
126 butler = butlerTests.makeTestRepo(tempRepo)
127 tester.run(butler, self)
129 def test_whole_subset(self):
130 """Test that each pipeline's synonymous subset includes all tasks,
131 including those imported from other files.
132 """
133 files = ResourcePath.findFileResources([self.path], file_filter=r".*\.yaml$")
134 for file in files:
135 if "QuickTemplate" in file.path:
136 # Our QuickTemplate definition cannot be tested here because it
137 # depends on drp_tasks, which we cannot make a dependency here.
138 continue
139 elif "injection/" in file.path:
140 # PostInjectedTasksApPipe is not actually an AP pipeline
141 continue
142 elif "ApdbDeduplication" in file.path:
143 # The task to export catalogs from the APDB and re-run
144 # association is not intended to be part of Prompt Processing
145 # or batch AP pipeline runs.
146 continue
147 elif "PromptTemplate" in file.path:
148 # Our PromptTemplate definition cannot be tested here because it
149 # depends on drp_tasks, which we cannot make a dependency here.
150 continue
151 with self.subTest(file=str(file)):
152 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
153 subset = self.synonyms.get(file.basename(), "<unknown_synonym>")
154 self.assertEqual(pipeline.subsets.get(subset, "<missing>"), set(pipeline.task_labels),
155 msg=f"These tasks are missing from subset '{subset}'")
157 def test_ap_pipe_subsets(self):
158 """Test the unique subsets of ApPipe.
159 """
160 files = ResourcePath.findFileResources([self.path], file_filter=r"^ApPipe.*\.yaml$")
161 required_subsets = {"preload", "prompt", "afterburner"}
162 # getRegionTimeFromVisit is part of no subset besides apPipe. This is a
163 # very deliberate exception; see RFC-997.
164 no_subset_wanted = {"getRegionTimeFromVisit"}
166 for file in files:
167 if "injection/" in file.path: 167 ↛ 169line 167 didn't jump to line 169 because the condition on line 167 was never true
168 # PostInjectedTasksApPipe is not actually an AP pipeline
169 continue
170 with self.subTest(file=str(file)):
171 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
172 # Do all steps exist?
173 self.assertGreaterEqual(pipeline.subsets.keys(), required_subsets,
174 msg="An AP pipeline is missing subsets "
175 f"{required_subsets - pipeline.subsets.keys()}.")
176 # Is each task part of exactly one step?
177 for set1, set2 in itertools.product(required_subsets, required_subsets):
178 if set1 == set2:
179 continue
180 tasks1 = pipeline.subsets[set1]
181 tasks2 = pipeline.subsets[set2]
182 self.assertTrue(tasks1.isdisjoint(tasks2),
183 msg=f"Subsets '{set1}' and '{set2}' share tasks "
184 f"{tasks1.intersection(tasks2)}.")
185 subsetted = set().union(*[pipeline.subsets[s] for s in required_subsets])
186 self.assertEqual(subsetted, set(pipeline.task_labels) - no_subset_wanted,
187 msg=f"These tasks are not in any of the subsets {required_subsets}.")
189 def test_injection_ingredient(self):
190 """Test the source-injection post-processing ingredient pipeline.
192 PostInjectedTasksApPipe is a partial pipeline merged into full AP
193 pipelines at build time by make_injection_pipeline. This test
194 validates that it can build a graph and contains the expected tasks.
195 """
196 ingredient = self.path.join("_ingredients").join("injection").join("PostInjectedTasksApPipe.yaml")
197 with self.subTest(file=str(ingredient)):
198 pipeline = lsst.pipe.base.Pipeline.from_uri(ingredient)
199 expected_tasks = {
200 "injectedMatchDiaSrc",
201 "injectedMatchAssocDiaSrc",
202 "consolidateMatchDiaSrc",
203 "consolidateMatchAssocDiaSrc",
204 }
205 self.assertGreaterEqual(
206 set(pipeline.task_labels),
207 expected_tasks,
208 msg="Source-injection post-processing ingredient is missing expected tasks.",
209 )
211 def test_generated_pipeline_readiness(self):
212 """Test that the generated ApPipeWithFakes ingredient exists and is valid.
214 pipelines/_ingredients/ApPipeWithFakes.yaml is generated at build time
215 by make_injection_pipeline (invoked via scons). This test verifies that
216 generation occurred before pipeline tests run, and that the generated
217 pipeline includes the source-injection task. If this test is skipped,
218 run 'scons' in the ap_pipe root directory first.
219 """
220 generated = self.path.join("_ingredients/ApPipeWithFakes.yaml")
221 if not generated.exists(): 221 ↛ 224line 221 didn't jump to line 224 because the condition on line 221 was never true
222 # fail the test with a message that explains how to fix the problem,
223 # rather than silently skipping it
224 self.fail(
225 f"{generated} has not been generated yet. "
226 "Run 'scons' in the ap_pipe root directory to generate it."
227 )
228 with self.subTest(file=str(generated)):
229 pipeline = lsst.pipe.base.Pipeline.from_uri(generated)
230 pipeline.addConfigOverride("parameters", "apdb_config", "some/file/path.yaml")
231 self.assertIn(
232 "injectVisit",
233 pipeline.task_labels,
234 msg="Generated ApPipeWithFakes.yaml is missing the 'injectVisit' task.",
235 )
237 def test_preconvolution_isr_matches_ap_pipe(self):
238 """Test that, for each instrument, ApPipeWithPreconvolution defines
239 the same isr task (class and config) as the corresponding ApPipe.
241 Preconvolution changes only image subtraction and DIA-source
242 detection; instrument signature removal must be unaffected.
243 """
244 files = [
245 f for f in ResourcePath.findFileResources(
246 [self.path], file_filter=r"^ApPipeWithPreconvolution\.yaml$"
247 )
248 if "_ingredients" not in f.path
249 ]
250 # Sanity-check that this test actually has cameras to compare.
251 self.assertGreater(len(files), 0,
252 msg="No camera-specific ApPipeWithPreconvolution.yaml files found.")
254 for precon_file in files:
255 with self.subTest(file=str(precon_file)):
256 base_file = precon_file.dirname().join("ApPipe.yaml")
257 self.assertTrue(base_file.exists(),
258 msg=f"Expected sibling ApPipe.yaml next to {precon_file}: "
259 f"{base_file} does not exist.")
261 precon = lsst.pipe.base.Pipeline.from_uri(precon_file)
262 base = lsst.pipe.base.Pipeline.from_uri(base_file)
263 # apdb_config has no default and must be set before to_graph().
264 precon.addConfigOverride("parameters", "apdb_config", "some/file/path.yaml")
265 base.addConfigOverride("parameters", "apdb_config", "some/file/path.yaml")
267 precon_isr = precon.to_graph().tasks["isr"]
268 base_isr = base.to_graph().tasks["isr"]
270 self.assertEqual(precon_isr.task_class_name, base_isr.task_class_name,
271 msg=f"isr task class differs between ApPipe.yaml and "
272 f"ApPipeWithPreconvolution.yaml in {precon_file.dirname()}.")
273 # Can't just do `assertEqual(precon_isr, base_isr)` since
274 # Task nodes are intentionally not equality comparable.
275 self.assertTrue(
276 base_isr.config.compare(precon_isr.config, shortcut=False),
277 msg=f"isr task config differs between ApPipe.yaml and "
278 f"ApPipeWithPreconvolution.yaml in {precon_file.dirname()}."
279 )
281 def test_inherited_subsets(self):
282 """Test that instrument-specific pipelines have all the subsets of their
283 generic counterparts.
285 Note that this does not check inheritance *within* `_ingredients`!
286 """
287 files = [
288 f for f in ResourcePath.findFileResources([self.path], file_filter=r".*\.yaml$")
289 if "_ingredients" not in f.path
290 ]
291 for file in files:
292 if "QuickTemplate" in file.path:
293 # Our QuickTemplate definition cannot be tested here because it
294 # depends on drp_tasks, which we cannot make a dependency here.
295 continue
296 with self.subTest(file=str(file)):
297 generic = self.path.join("_ingredients/", forceDirectory=True).join(file.basename())
298 if not generic.exists():
299 continue
300 special_subsets = lsst.pipe.base.Pipeline.from_uri(file).subsets.keys()
301 generic_subsets = lsst.pipe.base.Pipeline.from_uri(generic).subsets.keys()
302 self.assertGreaterEqual(special_subsets, generic_subsets,
303 msg="The instrument-specific pipeline is missing subsets "
304 f"{generic_subsets - special_subsets}.")
307class MemoryTester(lsst.utils.tests.MemoryTestCase):
308 pass
311def setup_module(module):
312 lsst.utils.tests.init()
315if __name__ == "__main__": 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true
316 lsst.utils.tests.init()
317 unittest.main()