Coverage for tests / test_pipelines.py: 34%
104 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-18 09:18 +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
27# Need to import pyproj to prevent file handle leakage since importing
28# pyproj automatically opens proj.db and never closes it. We can not wait
29# for some dependent code to import it whilst the test is running since then
30# the leak checker will think it is a leak.
31# TODO: Remove import after completing DM-54643. Use DM-54656
32try:
33 import pyproj # noqa: F401
34except ImportError:
35 pass
37from lsst.pipe.base import Pipeline, PipelineGraph
38from lsst.resources import ResourcePath
39import lsst.utils
42try:
43 import lsst.obs.lsst
44 has_obs_lsst = True
45except ImportError:
46 has_obs_lsst = False
48try:
49 import lsst.obs.subaru
50 has_obs_subaru = True
51except ImportError:
52 has_obs_subaru = False
54try:
55 import lsst.obs.decam
56 has_obs_decam = True
57except ImportError:
58 has_obs_decam = False
60PIPELINE_URI = ResourcePath("eups://cp_pipe/pipelines/", forceDirectory=True)
63class CalibrationPipelinesTestCase(lsst.utils.tests.TestCase):
64 """Test case for building the pipelines."""
66 def _get_pipelines(self, exclude=[]):
67 pipelines = {
68 "cpBfk.yaml",
69 "cpBias.yaml",
70 "cpCrosstalk.yaml",
71 "cpCti.yaml",
72 "cpDarkForDefects.yaml",
73 "cpDark.yaml",
74 "cpDefectsIndividual.yaml",
75 "cpDefects.yaml",
76 "cpFilterScan.yaml",
77 "cpFlatSingleChip.yaml",
78 "cpFlat.yaml",
79 "cpFringe.yaml",
80 "cpLinearizer.yaml",
81 "cpMonochromatorScan.yaml",
82 "cpPlotPtc.yaml",
83 "cpPtc.yaml",
84 "cpSky.yaml",
85 "cpBiasBootstrap.yaml",
86 "cpDarkBootstrap.yaml",
87 "cpFlatBootstrap.yaml",
88 "cpSpectroFlat.yaml",
89 # TODO DM-52883: Remove cpPtcFixupGainRatios and cpPtcRename.
90 "cpPtcFixupGainRatios.yaml",
91 "cpPtcRename.yaml",
92 "cpIlluminationCorrection.yaml",
93 "cpFlatAnaglyph.yaml",
94 "cpFlatGradientReference.yaml",
95 "cpQuadNotch.yaml",
96 "cpGainCorrection.yaml",
97 }
99 for ex in exclude:
100 pipelines.remove(ex)
102 return pipelines
104 def _check_pipeline(self, pipeline_file: ResourcePath, overrides={}):
105 # Confirm that the file is there.
106 self.assertTrue(pipeline_file.exists(), msg=f"Could not find {pipeline_file}")
108 # The following loads the pipeline and confirms that it can parse all
109 # the configs.
110 try:
111 pipeline = Pipeline.from_uri(pipeline_file)
113 if overrides:
114 for label, value in overrides.items():
115 pipeline.addConfigOverride(label, value[0], value[1])
117 graph = pipeline.to_graph()
118 except Exception as e:
119 raise RuntimeError(f"Could not process {pipeline_file} {e}") from e
121 self.assertIsInstance(graph, PipelineGraph)
123 def test_ingredients(self):
124 """Check that all pipelines in pipelines/_ingredients are tested."""
125 ingredient_files = ResourcePath.findFileResources(
126 [PIPELINE_URI.join("_ingredients")], file_filter=r".*\.yaml$"
127 )
128 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
129 # and are not to be tested on their own.
130 ingredients = set(
131 [pipeline.basename() for pipeline in ingredient_files if "LSST.yaml" not in pipeline.path]
132 )
133 # The *Bootstrap* pipelines are used by LATISS/LSSTComCam/LSSTCam
134 # but are renamed on import.
135 expected = set([pipeline for pipeline in self._get_pipelines() if "Bootstrap" not in pipeline])
136 # These pipelines have only an "LSST" version.
137 expected.discard("cpIlluminationCorrection.yaml")
138 expected.discard("cpFlatAnaglyph.yaml")
139 expected.discard("cpFlatGradientReference.yaml")
140 expected.discard("cpGainCorrection.yaml")
141 self.assertEqual(ingredients, expected)
143 def test_cameras(self):
144 """Check that all the cameras in pipelines are tested."""
145 _, paths, _ = next(PIPELINE_URI.walk())
146 expected = {
147 "DECam",
148 "HSC",
149 "_ingredients",
150 "LATISS",
151 "LSSTCam",
152 "LSSTCam-imSim",
153 "LSSTComCam",
154 "LSSTComCamSim",
155 "LSST-TS8",
156 }
157 self.assertEqual(set(paths), expected)
159 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
160 def test_latiss_pipelines(self):
161 latiss_uri = PIPELINE_URI.join("LATISS", forceDirectory=True)
162 for pipeline in self._get_pipelines(exclude=[
163 # The following two tasks are not part of the new pipelines.
164 "cpDarkForDefects.yaml",
165 "cpDefectsIndividual.yaml",
166 # The following tasks are not defined for LATISS.
167 "cpMonochromatorScan.yaml",
168 "cpIlluminationCorrection.yaml",
169 "cpFlatAnaglyph.yaml",
170 "cpFlatGradientReference.yaml",
171 "cpGainCorrection.yaml",
172 # The following tasks will be added in the future.
173 "cpCrosstalk.yaml",
174 "cpFringe.yaml",
175 # TODO: DM-46426
176 "cpCti.yaml",
177 ]):
178 self._check_pipeline(latiss_uri.join(pipeline))
180 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
181 def test_lsstcam_pipelines(self):
182 lsstcam_uri = PIPELINE_URI.join("LSSTCam", forceDirectory=True)
183 for pipeline in self._get_pipelines(exclude=[
184 "cpFilterScan.yaml",
185 "cpMonochromatorScan.yaml",
186 "cpSpectroFlat.yaml",
187 "cpDarkForDefects.yaml",
188 "cpDefectsIndividual.yaml",
189 "cpIlluminationCorrection.yaml",
190 # Unsupported pipelines.
191 "cpCrosstalk.yaml",
192 "cpFringe.yaml",
193 "cpQuadNotch.yaml",
194 ]):
195 if pipeline == "cpFlatAnaglyph.yaml":
196 overrides = {
197 "cpFlatBlueNormalize": ("downSelectionValue", "test1"),
198 "cpFlatRedNormalize": ("downSelectionValue", "test2"),
199 }
200 else:
201 overrides = {}
202 self._check_pipeline(
203 lsstcam_uri.join(pipeline),
204 overrides=overrides,
205 )
207 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
208 def test_lsstcam_imsim_pipelines(self):
209 sim_uri = PIPELINE_URI.join("LSSTCam-imSim", forceDirectory=True)
210 for pipeline in self._get_pipelines(exclude=[
211 "cpDarkForDefects.yaml",
212 "cpFilterScan.yaml",
213 "cpMonochromatorScan.yaml",
214 "cpSpectroFlat.yaml",
215 "cpBiasBootstrap.yaml",
216 "cpDarkBootstrap.yaml",
217 "cpFlatBootstrap.yaml",
218 "cpPtcFixupGainRatios.yaml",
219 "cpPtcRename.yaml",
220 "cpIlluminationCorrection.yaml",
221 "cpFlatAnaglyph.yaml",
222 "cpFlatGradientReference.yaml",
223 "cpQuadNotch.yaml",
224 "cpGainCorrection.yaml",
225 ]):
226 self._check_pipeline(sim_uri.join(pipeline))
228 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
229 def test_lsstcomcam_pipelines(self):
230 comcam_uri = PIPELINE_URI.join("LSSTComCam", forceDirectory=True)
231 for pipeline in self._get_pipelines(exclude=[
232 # The following tasks are not part of the new pipelines.
233 "cpDarkForDefects.yaml",
234 "cpDefectsIndividual.yaml",
235 # The following tasks are not for ComCam.
236 "cpFilterScan.yaml",
237 "cpMonochromatorScan.yaml",
238 "cpSpectroFlat.yaml",
239 "cpCrosstalk.yaml",
240 "cpFringe.yaml",
241 "cpFlatAnaglyph.yaml",
242 "cpFlatGradientReference.yaml",
243 "cpQuadNotch.yaml",
244 "cpGainCorrection.yaml",
245 # TODO: DM-46426
246 "cpCti.yaml",
247 ]):
248 self._check_pipeline(comcam_uri.join(pipeline))
250 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
251 def test_lsstcomcamsim_pipelines(self):
252 comcam_sim_uri = PIPELINE_URI.join("LSSTComCamSim", forceDirectory=True)
253 for pipeline in self._get_pipelines(exclude=[
254 # The following tasks are not part of the new pipelines.
255 "cpDarkForDefects.yaml",
256 "cpDefectsIndividual.yaml",
257 # The following tasks are not for ComCamSim.
258 "cpFilterScan.yaml",
259 "cpMonochromatorScan.yaml",
260 "cpSpectroFlat.yaml",
261 "cpFringe.yaml",
262 "cpLinearizer.yaml",
263 "cpCrosstalk.yaml",
264 "cpCti.yaml",
265 "cpPtcFixupGainRatios.yaml",
266 "cpPtcRename.yaml",
267 "cpIlluminationCorrection.yaml",
268 "cpFlatAnaglyph.yaml",
269 "cpFlatGradientReference.yaml",
270 "cpQuadNotch.yaml",
271 "cpGainCorrection.yaml",
272 ]):
273 self._check_pipeline(comcam_sim_uri.join(pipeline))
275 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSST-TS8 pipelines without obs_lsst")
276 def test_lsst_ts8_pipelines(self):
277 ts8_uri = PIPELINE_URI.join("LSST-TS8", forceDirectory=True)
278 for pipeline in self._get_pipelines(exclude=[
279 "cpFilterScan.yaml",
280 "cpMonochromatorScan.yaml",
281 "cpSpectroFlat.yaml",
282 "cpBiasBootstrap.yaml",
283 "cpDarkBootstrap.yaml",
284 "cpFlatBootstrap.yaml",
285 "cpPtcFixupGainRatios.yaml",
286 "cpPtcRename.yaml",
287 "cpIlluminationCorrection.yaml",
288 "cpFlatAnaglyph.yaml",
289 "cpFlatGradientReference.yaml",
290 "cpQuadNotch.yaml",
291 "cpGainCorrection.yaml",
292 ]):
293 self._check_pipeline(ts8_uri.join(pipeline))
295 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
296 def test_decam_pipelines(self):
297 decam_uri = PIPELINE_URI.join("DECam", forceDirectory=True)
298 for pipeline in self._get_pipelines(exclude=[
299 "cpDarkForDefects.yaml",
300 "cpFilterScan.yaml",
301 "cpMonochromatorScan.yaml",
302 "cpSpectroFlat.yaml",
303 "cpBiasBootstrap.yaml",
304 "cpDarkBootstrap.yaml",
305 "cpFlatBootstrap.yaml",
306 "cpPtcFixupGainRatios.yaml",
307 "cpPtcRename.yaml",
308 "cpIlluminationCorrection.yaml",
309 "cpFlatAnaglyph.yaml",
310 "cpFlatGradientReference.yaml",
311 "cpQuadNotch.yaml",
312 "cpGainCorrection.yaml",
313 ]):
314 self._check_pipeline(decam_uri.join(pipeline))
316 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
317 def test_hsc_pipelines(self):
318 hsc_uri = PIPELINE_URI.join("HSC", forceDirectory=True)
319 for pipeline in self._get_pipelines(exclude=[
320 "cpDarkForDefects.yaml",
321 "cpFilterScan.yaml",
322 "cpMonochromatorScan.yaml",
323 "cpSpectroFlat.yaml",
324 "cpBiasBootstrap.yaml",
325 "cpDarkBootstrap.yaml",
326 "cpFlatBootstrap.yaml",
327 "cpPtcFixupGainRatios.yaml",
328 "cpPtcRename.yaml",
329 "cpIlluminationCorrection.yaml",
330 "cpFlatAnaglyph.yaml",
331 "cpFlatGradientReference.yaml",
332 "cpQuadNotch.yaml",
333 "cpGainCorrection.yaml",
334 ]):
335 self._check_pipeline(hsc_uri.join(pipeline))
338class TestMemory(lsst.utils.tests.MemoryTestCase):
339 pass
342def setup_module(module):
343 lsst.utils.tests.init()
346if __name__ == "__main__": 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true
347 lsst.utils.tests.init()
348 unittest.main()