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