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