Coverage for tests / test_pipelines.py: 20%
85 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-16 00:52 -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 glob
23import itertools
24import os.path
25import tempfile
26import unittest
28# need to import pyproj to prevent file handle leakage
29import pyproj # noqa: F401
31import lsst.daf.butler.tests as butlerTests
32import lsst.pipe.base
33from lsst.pipe.base.tests.pipelineStepTester import PipelineStepTester # Can't use fully-qualified name
34import lsst.utils
35import lsst.utils.tests
38class PipelineDefintionsTestSuite(lsst.utils.tests.TestCase):
39 """Tests of the self-consistency of our pipeline definitions.
40 """
41 def setUp(self):
42 self.path = os.path.join(lsst.utils.getPackageDir("ap_pipe"), "pipelines")
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 = glob.glob(os.path.join(self.path, "**", "*.yaml"))
59 for file in files:
60 if "QuickTemplate" in file:
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:
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):
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 = glob.glob(os.path.join(self.path, "_ingredients", "*.yaml"))
76 for file in files:
77 if "QuickTemplate" in file:
78 # Our QuickTemplate definition cannot be tested here because it
79 # depends on drp_tasks, which we cannot make a dependency here.
80 continue
81 with self.subTest(file):
82 expected_inputs = {
83 # ISR
84 "raw", "camera", "crosstalk", "crosstalkSources", "bias", "dark", "flat", "ptc",
85 "fringe", "straylightData", "bfKernel", "newBFKernel", "defects", "linearizer",
86 "opticsTransmission", "filterTransmission", "atmosphereTransmission",
87 "illumMaskedImage", "deferredChargeCalib",
88 # ISR-LSST
89 "bfk", "cti", "dnlLUT", "gain_correction",
90 # Everything else
91 "skyMap", "gaia_dr3_20230707", "gaia_dr2_20200414", "ps1_pv3_3pi_20170110",
92 "template_coadd", "pretrainedModelPackage", "dia_source_apdb"
93 }
94 if "WithFakes" in file:
95 expected_inputs.add("injection_catalog")
96 tester = PipelineStepTester(
97 filename=file,
98 step_suffixes=[""], # Test full pipeline
99 initial_dataset_types=[("ps1_pv3_3pi_20170110", {"htm7"}, "SimpleCatalog", False),
100 ("gaia_dr2_20200414", {"htm7"}, "SimpleCatalog", False),
101 ("gaia_dr3_20230707", {"htm7"}, "SimpleCatalog", False),
102 ],
103 expected_inputs=expected_inputs,
104 # Pipeline outputs highly in flux, don't test
105 expected_outputs=set(),
106 pipeline_patches={"parameters:apdb_config": "some/file/path.yaml",
107 },
108 )
109 # Tester modifies Butler registry, so need a fresh repo every time
110 with tempfile.TemporaryDirectory() as tempRepo:
111 butler = butlerTests.makeTestRepo(tempRepo)
112 tester.run(butler, self)
114 def test_whole_subset(self):
115 """Test that each pipeline's synonymous subset includes all tasks,
116 including those imported from other files.
117 """
118 files = glob.glob(os.path.join(self.path, "**", "*.yaml"))
119 for file in files:
120 if "QuickTemplate" in file:
121 # Our QuickTemplate definition cannot be tested here because it
122 # depends on drp_tasks, which we cannot make a dependency here.
123 continue
124 elif "ApdbDeduplication" in file:
125 # The task to export catalogs from the APDB and re-run
126 # association is not intended to be part of Prompt Processing
127 # or batch AP pipeline runs.
128 continue
129 elif "PromptTemplate" in file:
130 # Our PromptTemplate definition cannot be tested here because it
131 # depends on drp_tasks, which we cannot make a dependency here.
132 continue
133 with self.subTest(file):
134 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
135 subset = self.synonyms[os.path.basename(file)]
136 self.assertEqual(pipeline.subsets[subset], set(pipeline.task_labels),
137 msg=f"These tasks are missing from subset '{subset}'.")
139 def test_ap_pipe_subsets(self):
140 """Test the unique subsets of ApPipe.
141 """
142 files = glob.glob(os.path.join(self.path, "**", "ApPipe*.yaml"))
143 required_subsets = {"preload", "prompt", "afterburner"}
144 # getRegionTimeFromVisit is part of no subset besides apPipe. This is a
145 # very deliberate exception; see RFC-997.
146 no_subset_wanted = {"getRegionTimeFromVisit"}
148 for file in files:
149 with self.subTest(file):
150 pipeline = lsst.pipe.base.Pipeline.from_uri(file)
151 # Do all steps exist?
152 self.assertGreaterEqual(pipeline.subsets.keys(), required_subsets,
153 msg="An AP pipeline is missing subsets "
154 f"{required_subsets - pipeline.subsets.keys()}.")
155 # Is each task part of exactly one step?
156 for set1, set2 in itertools.product(required_subsets, required_subsets):
157 if set1 == set2:
158 continue
159 tasks1 = pipeline.subsets[set1]
160 tasks2 = pipeline.subsets[set2]
161 self.assertTrue(tasks1.isdisjoint(tasks2),
162 msg=f"Subsets '{set1}' and '{set2}' share tasks "
163 f"{tasks1.intersection(tasks2)}.")
164 subsetted = set().union(*[pipeline.subsets[s] for s in required_subsets])
165 self.assertEqual(subsetted, set(pipeline.task_labels) - no_subset_wanted,
166 msg=f"These tasks are not in any of the subsets {required_subsets}.")
168 def test_inherited_subsets(self):
169 """Test that instrument-specific pipelines have all the subsets of their
170 generic counterparts.
172 Note that this does not check inheritance *within* `_ingredients`!
173 """
174 files = [f for f in glob.glob(os.path.join(self.path, "**", "*.yaml"))
175 if "_ingredients" not in f]
176 for file in files:
177 with self.subTest(file):
178 generic = os.path.join(self.path, "_ingredients", os.path.basename(file))
179 if not os.path.exists(generic):
180 continue
181 special_subsets = lsst.pipe.base.Pipeline.from_uri(file).subsets.keys()
182 generic_subsets = lsst.pipe.base.Pipeline.from_uri(generic).subsets.keys()
183 self.assertGreaterEqual(special_subsets, generic_subsets,
184 msg="The instrument-specific pipeline is missing subsets "
185 f"{generic_subsets - special_subsets}.")
188class MemoryTester(lsst.utils.tests.MemoryTestCase):
189 pass
192def setup_module(module):
193 lsst.utils.tests.init()
196if __name__ == "__main__": 196 ↛ 197line 196 didn't jump to line 197 because the condition on line 196 was never true
197 lsst.utils.tests.init()
198 unittest.main()