Coverage for tests / test_pipelines.py: 19%
84 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-23 01:43 -0800
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
26# need to import pyproj to prevent file handle leakage
27import pyproj # noqa: F401
29import lsst.daf.butler.tests as butlerTests
30import lsst.pipe.base
31from lsst.pipe.base.tests.pipelineStepTester import PipelineStepTester # Can't use fully-qualified name
32import lsst.utils
33import lsst.utils.tests
35from lsst.resources import ResourcePath
38class PipelineDefintionsTestSuite(lsst.utils.tests.TestCase):
39 """Tests of the self-consistency of our pipeline definitions.
40 """
41 def setUp(self):
42 self.path = ResourcePath("eups://ap_pipe/pipelines/", forceDirectory=True)
43 # Each pipeline file should have a subset that represents it in
44 # higher-level pipelines.
45 self.synonyms = {"ApPipe.yaml": "apPipe",
46 "ApPipeWithIsrTaskLSST.yaml": "apPipe",
47 "ApPipeWithFakes.yaml": "apPipe",
48 "SingleFrame.yaml": "singleFrame",
49 "SingleFrameWithIsrTaskLSST.yaml": "singleFrame",
50 "RunIsrWithoutInterChipCrosstalk.yaml": "runIsr",
51 "RunIsrForCrosstalkSources.yaml": "runOverscan",
52 }
54 def test_graph_build(self):
55 """Test that each pipeline definition file can be
56 used to build a graph.
57 """
58 files = ResourcePath.findFileResources([self.path], file_filter=r".*\.yaml$")
59 for file in files:
60 if "QuickTemplate" in file.path:
61 # Our QuickTemplate definition cannot be tested here because it
62 # depends on drp_tasks, which we cannot make a dependency here.
63 continue
64 if "PromptTemplate" in file.path:
65 # Our PromptTemplate definition cannot be tested here because it
66 # depends on drp_tasks, which we cannot make a dependency here.
67 continue
68 with self.subTest(file=str(file)):
69 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
70 pipeline.addConfigOverride("parameters", "apdb_config", "some/file/path.yaml")
71 # If this fails, it will produce a useful error message.
72 pipeline.to_graph()
74 def test_datasets(self):
75 files = [
76 f for f in ResourcePath.findFileResources(
77 [self.path.join("_ingredients", forceDirectory=True)], file_filter=r".*\.yaml$"
78 )
79 # Validation currently broken for injection pipelines.
80 # TODO: DM-54077
81 if "injection/" not in f.path
82 ]
83 for file in files:
84 if "QuickTemplate" in file.path:
85 # Our QuickTemplate definition cannot be tested here because it
86 # depends on drp_tasks, which we cannot make a dependency here.
87 continue
88 with self.subTest(file=str(file)):
89 expected_inputs = {
90 # ISR
91 "raw", "camera", "crosstalk", "crosstalkSources", "bias", "dark", "flat", "ptc",
92 "fringe", "straylightData", "bfKernel", "newBFKernel", "defects", "linearizer",
93 "opticsTransmission", "filterTransmission", "atmosphereTransmission",
94 "illumMaskedImage", "deferredChargeCalib",
95 # ISR-LSST
96 "bfk", "cti", "dnlLUT", "gain_correction",
97 # Everything else
98 "skyMap", "gaia_dr3_20230707", "gaia_dr2_20200414", "ps1_pv3_3pi_20170110",
99 "template_coadd", "pretrainedModelPackage", "dia_source_apdb"
100 }
101 if "WithFakes" in file.path:
102 expected_inputs.add("injection_catalog")
103 tester = PipelineStepTester(
104 filename=file,
105 step_suffixes=[""], # Test full pipeline
106 initial_dataset_types=[("ps1_pv3_3pi_20170110", {"htm7"}, "SimpleCatalog", False),
107 ("gaia_dr2_20200414", {"htm7"}, "SimpleCatalog", False),
108 ("gaia_dr3_20230707", {"htm7"}, "SimpleCatalog", False),
109 ],
110 expected_inputs=expected_inputs,
111 # Pipeline outputs highly in flux, don't test
112 expected_outputs=set(),
113 pipeline_patches={"parameters:apdb_config": "some/file/path.yaml",
114 },
115 )
116 # Tester modifies Butler registry, so need a fresh repo every time
117 with tempfile.TemporaryDirectory() as tempRepo:
118 butler = butlerTests.makeTestRepo(tempRepo)
119 tester.run(butler, self)
121 def test_whole_subset(self):
122 """Test that each pipeline's synonymous subset includes all tasks,
123 including those imported from other files.
124 """
125 files = [
126 f for f in ResourcePath.findFileResources([self.path], file_filter=r".*\.yaml$")
127 # Validation currently broken for injection pipelines.
128 # TODO: DM-54077
129 if "injection/" not in f.path
130 ]
131 for file in files:
132 if "QuickTemplate" in file.path:
133 # Our QuickTemplate definition cannot be tested here because it
134 # depends on drp_tasks, which we cannot make a dependency here.
135 continue
136 elif "ApdbDeduplication" in file.path:
137 # The task to export catalogs from the APDB and re-run
138 # association is not intended to be part of Prompt Processing
139 # or batch AP pipeline runs.
140 continue
141 elif "PromptTemplate" in file.path:
142 # Our PromptTemplate definition cannot be tested here because it
143 # depends on drp_tasks, which we cannot make a dependency here.
144 continue
145 with self.subTest(file=str(file)):
146 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
147 subset = self.synonyms.get(file.basename(), "<unknown_synonym>")
148 self.assertEqual(pipeline.subsets.get(subset, "<missing>"), set(pipeline.task_labels),
149 msg=f"These tasks are missing from subset '{subset}'")
151 def test_ap_pipe_subsets(self):
152 """Test the unique subsets of ApPipe.
153 """
154 files = [
155 f for f in ResourcePath.findFileResources([self.path], file_filter=r"^ApPipe.*\.yaml$")
156 # Validation currently broken for injection pipelines.
157 # TODO: DM-54077
158 if "injection/" not in f.path
159 ]
160 required_subsets = {"preload", "prompt", "afterburner"}
161 # getRegionTimeFromVisit is part of no subset besides apPipe. This is a
162 # very deliberate exception; see RFC-997.
163 no_subset_wanted = {"getRegionTimeFromVisit"}
165 for file in files:
166 with self.subTest(file=str(file)):
167 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
168 # Do all steps exist?
169 self.assertGreaterEqual(pipeline.subsets.keys(), required_subsets,
170 msg="An AP pipeline is missing subsets "
171 f"{required_subsets - pipeline.subsets.keys()}.")
172 # Is each task part of exactly one step?
173 for set1, set2 in itertools.product(required_subsets, required_subsets):
174 if set1 == set2:
175 continue
176 tasks1 = pipeline.subsets[set1]
177 tasks2 = pipeline.subsets[set2]
178 self.assertTrue(tasks1.isdisjoint(tasks2),
179 msg=f"Subsets '{set1}' and '{set2}' share tasks "
180 f"{tasks1.intersection(tasks2)}.")
181 subsetted = set().union(*[pipeline.subsets[s] for s in required_subsets])
182 self.assertEqual(subsetted, set(pipeline.task_labels) - no_subset_wanted,
183 msg=f"These tasks are not in any of the subsets {required_subsets}.")
185 def test_inherited_subsets(self):
186 """Test that instrument-specific pipelines have all the subsets of their
187 generic counterparts.
189 Note that this does not check inheritance *within* `_ingredients`!
190 """
191 files = [
192 f for f in ResourcePath.findFileResources([self.path], file_filter=r".*\.yaml$")
193 if "_ingredients" not in f.path
194 ]
195 for file in files:
196 with self.subTest(file=str(file)):
197 generic = self.path.join("_ingredients/", forceDirectory=True).join(file.basename())
198 if not generic.exists():
199 continue
200 special_subsets = lsst.pipe.base.Pipeline.from_uri(file).subsets.keys()
201 generic_subsets = lsst.pipe.base.Pipeline.from_uri(generic).subsets.keys()
202 self.assertGreaterEqual(special_subsets, generic_subsets,
203 msg="The instrument-specific pipeline is missing subsets "
204 f"{generic_subsets - special_subsets}.")
207class MemoryTester(lsst.utils.tests.MemoryTestCase):
208 pass
211def setup_module(module):
212 lsst.utils.tests.init()
215if __name__ == "__main__": 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true
216 lsst.utils.tests.init()
217 unittest.main()