Coverage for tests / test_pipelines.py: 32%
100 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-23 08:59 +0000
1#!/usr/bin/env python
3#
4# LSST Data Management System
5#
6# This product includes software developed by the
7# LSST Project (http://www.lsst.org/).
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 LSST License Statement and
20# the GNU General Public License along with this program. If not,
21# see <https://www.lsstcorp.org/LegalNotices/>.
22#
23"""Test cases for cp_pipe pipelines."""
25import unittest
27from lsst.pipe.base import Pipeline, PipelineGraph
28from lsst.resources import ResourcePath
29import lsst.utils
32try:
33 import lsst.obs.lsst
34 has_obs_lsst = True
35except ImportError:
36 has_obs_lsst = False
38try:
39 import lsst.obs.subaru
40 has_obs_subaru = True
41except ImportError:
42 has_obs_subaru = False
44try:
45 import lsst.obs.decam
46 has_obs_decam = True
47except ImportError:
48 has_obs_decam = False
50PIPELINE_URI = ResourcePath("eups://cp_pipe/pipelines/", forceDirectory=True)
53class CalibrationPipelinesTestCase(lsst.utils.tests.TestCase):
54 """Test case for building the pipelines."""
56 def _get_pipelines(self, exclude=[]):
57 pipelines = {
58 "cpBfk.yaml",
59 "cpBias.yaml",
60 "cpCrosstalk.yaml",
61 "cpCti.yaml",
62 "cpDarkForDefects.yaml",
63 "cpDark.yaml",
64 "cpDefectsIndividual.yaml",
65 "cpDefects.yaml",
66 "cpFilterScan.yaml",
67 "cpFlatSingleChip.yaml",
68 "cpFlat.yaml",
69 "cpFringe.yaml",
70 "cpLinearizer.yaml",
71 "cpMonochromatorScan.yaml",
72 "cpPlotPtc.yaml",
73 "cpPtc.yaml",
74 "cpSky.yaml",
75 "cpBiasBootstrap.yaml",
76 "cpDarkBootstrap.yaml",
77 "cpFlatBootstrap.yaml",
78 "cpSpectroFlat.yaml",
79 # TODO DM-52883: Remove cpPtcFixupGainRatios and cpPtcRename.
80 "cpPtcFixupGainRatios.yaml",
81 "cpPtcRename.yaml",
82 "cpIlluminationCorrection.yaml",
83 "cpFlatAnaglyph.yaml",
84 "cpFlatGradientReference.yaml",
85 "cpQuadNotch.yaml",
86 "cpGainCorrection.yaml",
87 }
89 for ex in exclude:
90 pipelines.remove(ex)
92 return pipelines
94 def _check_pipeline(self, pipeline_file: ResourcePath, overrides={}):
95 # Confirm that the file is there.
96 self.assertTrue(pipeline_file.exists(), msg=f"Could not find {pipeline_file}")
98 # The following loads the pipeline and confirms that it can parse all
99 # the configs.
100 try:
101 pipeline = Pipeline.from_uri(pipeline_file)
103 if overrides:
104 for label, value in overrides.items():
105 pipeline.addConfigOverride(label, value[0], value[1])
107 graph = pipeline.to_graph()
108 except Exception as e:
109 raise RuntimeError(f"Could not process {pipeline_file} {e}") from e
111 self.assertIsInstance(graph, PipelineGraph)
113 def test_ingredients(self):
114 """Check that all pipelines in pipelines/_ingredients are tested."""
115 ingredient_files = ResourcePath.findFileResources(
116 [PIPELINE_URI.join("_ingredients")], file_filter=r".*\.yaml$"
117 )
118 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
119 # and are not to be tested on their own.
120 ingredients = set(
121 [pipeline.basename() for pipeline in ingredient_files if "LSST.yaml" not in pipeline.path]
122 )
123 # The *Bootstrap* pipelines are used by LATISS/LSSTComCam/LSSTCam
124 # but are renamed on import.
125 expected = set([pipeline for pipeline in self._get_pipelines() if "Bootstrap" not in pipeline])
126 # These pipelines have only an "LSST" version.
127 expected.discard("cpIlluminationCorrection.yaml")
128 expected.discard("cpFlatAnaglyph.yaml")
129 expected.discard("cpFlatGradientReference.yaml")
130 expected.discard("cpGainCorrection.yaml")
131 self.assertEqual(ingredients, expected)
133 def test_cameras(self):
134 """Check that all the cameras in pipelines are tested."""
135 _, paths, _ = next(PIPELINE_URI.walk())
136 expected = {
137 "DECam",
138 "HSC",
139 "_ingredients",
140 "LATISS",
141 "LSSTCam",
142 "LSSTCam-imSim",
143 "LSSTComCam",
144 "LSSTComCamSim",
145 "LSST-TS8",
146 }
147 self.assertEqual(set(paths), expected)
149 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
150 def test_latiss_pipelines(self):
151 latiss_uri = PIPELINE_URI.join("LATISS", forceDirectory=True)
152 for pipeline in self._get_pipelines(exclude=[
153 # The following two tasks are not part of the new pipelines.
154 "cpDarkForDefects.yaml",
155 "cpDefectsIndividual.yaml",
156 # The following tasks are not defined for LATISS.
157 "cpMonochromatorScan.yaml",
158 "cpIlluminationCorrection.yaml",
159 "cpFlatAnaglyph.yaml",
160 "cpFlatGradientReference.yaml",
161 "cpGainCorrection.yaml",
162 # The following tasks will be added in the future.
163 "cpCrosstalk.yaml",
164 "cpFringe.yaml",
165 # TODO: DM-46426
166 "cpCti.yaml",
167 ]):
168 self._check_pipeline(latiss_uri.join(pipeline))
170 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
171 def test_lsstcam_pipelines(self):
172 lsstcam_uri = PIPELINE_URI.join("LSSTCam", forceDirectory=True)
173 for pipeline in self._get_pipelines(exclude=[
174 "cpFilterScan.yaml",
175 "cpMonochromatorScan.yaml",
176 "cpSpectroFlat.yaml",
177 "cpDarkForDefects.yaml",
178 "cpDefectsIndividual.yaml",
179 "cpIlluminationCorrection.yaml",
180 # Unsupported pipelines.
181 "cpCrosstalk.yaml",
182 "cpFringe.yaml",
183 "cpQuadNotch.yaml",
184 ]):
185 if pipeline == "cpFlatAnaglyph.yaml":
186 overrides = {
187 "cpFlatBlueNormalize": ("downSelectionValue", "test1"),
188 "cpFlatRedNormalize": ("downSelectionValue", "test2"),
189 }
190 else:
191 overrides = {}
192 self._check_pipeline(
193 lsstcam_uri.join(pipeline),
194 overrides=overrides,
195 )
197 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
198 def test_lsstcam_imsim_pipelines(self):
199 sim_uri = PIPELINE_URI.join("LSSTCam-imSim", forceDirectory=True)
200 for pipeline in self._get_pipelines(exclude=[
201 "cpDarkForDefects.yaml",
202 "cpFilterScan.yaml",
203 "cpMonochromatorScan.yaml",
204 "cpSpectroFlat.yaml",
205 "cpBiasBootstrap.yaml",
206 "cpDarkBootstrap.yaml",
207 "cpFlatBootstrap.yaml",
208 "cpPtcFixupGainRatios.yaml",
209 "cpPtcRename.yaml",
210 "cpIlluminationCorrection.yaml",
211 "cpFlatAnaglyph.yaml",
212 "cpFlatGradientReference.yaml",
213 "cpQuadNotch.yaml",
214 "cpGainCorrection.yaml",
215 ]):
216 self._check_pipeline(sim_uri.join(pipeline))
218 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
219 def test_lsstcomcam_pipelines(self):
220 comcam_uri = PIPELINE_URI.join("LSSTComCam", forceDirectory=True)
221 for pipeline in self._get_pipelines(exclude=[
222 # The following tasks are not part of the new pipelines.
223 "cpDarkForDefects.yaml",
224 "cpDefectsIndividual.yaml",
225 # The following tasks are not for ComCam.
226 "cpFilterScan.yaml",
227 "cpMonochromatorScan.yaml",
228 "cpSpectroFlat.yaml",
229 "cpCrosstalk.yaml",
230 "cpFringe.yaml",
231 "cpFlatAnaglyph.yaml",
232 "cpFlatGradientReference.yaml",
233 "cpQuadNotch.yaml",
234 "cpGainCorrection.yaml",
235 # TODO: DM-46426
236 "cpCti.yaml",
237 ]):
238 self._check_pipeline(comcam_uri.join(pipeline))
240 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
241 def test_lsstcomcamsim_pipelines(self):
242 comcam_sim_uri = PIPELINE_URI.join("LSSTComCamSim", forceDirectory=True)
243 for pipeline in self._get_pipelines(exclude=[
244 # The following tasks are not part of the new pipelines.
245 "cpDarkForDefects.yaml",
246 "cpDefectsIndividual.yaml",
247 # The following tasks are not for ComCamSim.
248 "cpFilterScan.yaml",
249 "cpMonochromatorScan.yaml",
250 "cpSpectroFlat.yaml",
251 "cpFringe.yaml",
252 "cpLinearizer.yaml",
253 "cpCrosstalk.yaml",
254 "cpCti.yaml",
255 "cpPtcFixupGainRatios.yaml",
256 "cpPtcRename.yaml",
257 "cpIlluminationCorrection.yaml",
258 "cpFlatAnaglyph.yaml",
259 "cpFlatGradientReference.yaml",
260 "cpQuadNotch.yaml",
261 "cpGainCorrection.yaml",
262 ]):
263 self._check_pipeline(comcam_sim_uri.join(pipeline))
265 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSST-TS8 pipelines without obs_lsst")
266 def test_lsst_ts8_pipelines(self):
267 ts8_uri = PIPELINE_URI.join("LSST-TS8", forceDirectory=True)
268 for pipeline in self._get_pipelines(exclude=[
269 "cpFilterScan.yaml",
270 "cpMonochromatorScan.yaml",
271 "cpSpectroFlat.yaml",
272 "cpBiasBootstrap.yaml",
273 "cpDarkBootstrap.yaml",
274 "cpFlatBootstrap.yaml",
275 "cpPtcFixupGainRatios.yaml",
276 "cpPtcRename.yaml",
277 "cpIlluminationCorrection.yaml",
278 "cpFlatAnaglyph.yaml",
279 "cpFlatGradientReference.yaml",
280 "cpQuadNotch.yaml",
281 "cpGainCorrection.yaml",
282 ]):
283 self._check_pipeline(ts8_uri.join(pipeline))
285 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
286 def test_decam_pipelines(self):
287 decam_uri = PIPELINE_URI.join("DECam", forceDirectory=True)
288 for pipeline in self._get_pipelines(exclude=[
289 "cpDarkForDefects.yaml",
290 "cpFilterScan.yaml",
291 "cpMonochromatorScan.yaml",
292 "cpSpectroFlat.yaml",
293 "cpBiasBootstrap.yaml",
294 "cpDarkBootstrap.yaml",
295 "cpFlatBootstrap.yaml",
296 "cpPtcFixupGainRatios.yaml",
297 "cpPtcRename.yaml",
298 "cpIlluminationCorrection.yaml",
299 "cpFlatAnaglyph.yaml",
300 "cpFlatGradientReference.yaml",
301 "cpQuadNotch.yaml",
302 "cpGainCorrection.yaml",
303 ]):
304 self._check_pipeline(decam_uri.join(pipeline))
306 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
307 def test_hsc_pipelines(self):
308 hsc_uri = PIPELINE_URI.join("HSC", forceDirectory=True)
309 for pipeline in self._get_pipelines(exclude=[
310 "cpDarkForDefects.yaml",
311 "cpFilterScan.yaml",
312 "cpMonochromatorScan.yaml",
313 "cpSpectroFlat.yaml",
314 "cpBiasBootstrap.yaml",
315 "cpDarkBootstrap.yaml",
316 "cpFlatBootstrap.yaml",
317 "cpPtcFixupGainRatios.yaml",
318 "cpPtcRename.yaml",
319 "cpIlluminationCorrection.yaml",
320 "cpFlatAnaglyph.yaml",
321 "cpFlatGradientReference.yaml",
322 "cpQuadNotch.yaml",
323 "cpGainCorrection.yaml",
324 ]):
325 self._check_pipeline(hsc_uri.join(pipeline))
328class TestMemory(lsst.utils.tests.MemoryTestCase):
329 pass
332def setup_module(module):
333 lsst.utils.tests.init()
336if __name__ == "__main__": 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 lsst.utils.tests.init()
338 unittest.main()