Coverage for tests / test_pipelines.py: 38%
84 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-28 09:22 +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_verify pipelines."""
25import glob
26import os
27import unittest
29from lsst.pipe.base import Pipeline, PipelineGraph
30import 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
51class VerifyPipelinesTestCase(lsst.utils.tests.TestCase):
52 """Test case for building the pipelines."""
54 def setUp(self):
55 self.pipeline_path = os.path.join(lsst.utils.getPackageDir("cp_verify"), "pipelines")
57 def _get_pipelines(self, exclude=[]):
58 pipelines = {
59 "verifyBfk.yaml",
60 "verifyBias.yaml",
61 "verifyCrosstalk.yaml",
62 "verifyDark.yaml",
63 "verifyDefectsIndividual.yaml",
64 "verifyDefects.yaml",
65 "verifyFlat.yaml",
66 "verifyScience.yaml",
67 # Old pipeline name.
68 "verifyGain.yaml",
69 # New pipeline name.
70 "verifyGainFromFlatPairs.yaml",
71 "verifyLinearizer.yaml",
72 "verifyPtc.yaml",
73 "verifyIlluminationCorrection.yaml",
74 }
76 for ex in exclude:
77 pipelines.remove(ex)
79 return pipelines
81 def _check_pipeline(self, pipeline_file):
82 # Confirm that the file is there.
83 self.assertTrue(os.path.isfile(pipeline_file), msg=f"Could not find {pipeline_file}")
85 # The following loads the pipeline and confirms that it can parse all
86 # the configs.
87 try:
88 pipeline = Pipeline.fromFile(pipeline_file)
89 graph = pipeline.to_graph()
90 except Exception as e:
91 raise RuntimeError(f"Could not process {pipeline_file}") from e
93 self.assertIsInstance(graph, PipelineGraph)
95 def test_ingredients(self):
96 """Check that all pipelines in pipelines/_ingredients are tested."""
97 glob_str = os.path.join(self.pipeline_path, "_ingredients", "*.yaml")
98 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
99 # and are not tested on their own.
100 ingredients = set(
101 [os.path.basename(pipeline) for pipeline in glob.glob(glob_str) if "LSST.yaml" not in pipeline]
102 )
103 # The _ingredients/verifyGainFromFlatPairs.yaml becomes
104 # verifyGain.yaml in older pipelines for compatibility.
105 expected = self._get_pipelines()
106 # The _ingredients/verifyGainFromFlatPairs.yaml becomes
107 # verifyGain.yaml in older pipelines for compatibility.
108 expected.remove("verifyGain.yaml")
109 # There is no regular verifyScience.yaml (non-LSST)
110 expected.remove("verifyScience.yaml")
111 # The Illumination Correction pipeline is only an "LSST" version.
112 expected.remove("verifyIlluminationCorrection.yaml")
113 self.assertEqual(ingredients, expected)
115 def test_cameras(self):
116 """Check that all the cameras in pipelines are tested."""
117 glob_str = os.path.join(self.pipeline_path, "*")
118 paths = set(
119 [os.path.basename(path) for path in glob.glob(glob_str)]
120 )
121 expected = {
122 "DECam",
123 "HSC",
124 "_ingredients",
125 "LATISS",
126 "LSSTCam",
127 "LSSTCam-imSim",
128 "LSSTComCam",
129 "LSSTComCamSim",
130 "README.md",
131 }
132 self.assertEqual(paths, expected)
134 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
135 def test_latiss_pipelines(self):
136 for pipeline in self._get_pipelines(exclude=[
137 # The old pipeline name should be excluded.
138 "verifyGain.yaml",
139 # The following tasks are not part of the new pipelines.
140 "verifyDefectsIndividual.yaml",
141 # The following tasks will be added in the future.
142 "verifyCrosstalk.yaml",
143 "verifyBfk.yaml",
144 "verifyIlluminationCorrection.yaml",
146 ]):
147 self._check_pipeline(os.path.join(self.pipeline_path, "LATISS", pipeline))
149 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
150 def test_lsstcam_pipelines(self):
151 for pipeline in self._get_pipelines(
152 exclude=[
153 # These are renamed/not used in the new pipelines.
154 "verifyGain.yaml",
155 "verifyDefectsIndividual.yaml",
156 # These are not used yet.
157 "verifyCrosstalk.yaml",
158 "verifyIlluminationCorrection.yaml",
159 ]):
160 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam", pipeline))
162 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
163 def test_lsstcam_imsim_pipelines(self):
164 for pipeline in self._get_pipelines(
165 exclude=[
166 "verifyGainFromFlatPairs.yaml",
167 "verifyScience.yaml",
168 "verifyIlluminationCorrection.yaml",
169 "verifyCrosstalk.yaml",
170 ],
171 ):
172 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam-imSim", pipeline))
174 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
175 def test_lsstcomcam_pipelines(self):
176 for pipeline in self._get_pipelines(
177 exclude=[
178 # These are renamed/not used in the new pipelines.
179 "verifyGain.yaml",
180 "verifyDefectsIndividual.yaml",
181 # These are not used yet.
182 "verifyCrosstalk.yaml",
183 ],
184 ):
185 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCam", pipeline))
187 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
188 def test_lsstcomcamsim_pipelines(self):
189 for pipeline in self._get_pipelines(
190 exclude=[
191 # These are renamed/not used in the new pipelines.
192 "verifyGain.yaml",
193 "verifyDefectsIndividual.yaml",
194 # These are not valid for LSSTComCamSim.
195 "verifyCrosstalk.yaml",
196 "verifyLinearizer.yaml",
197 "verifyIlluminationCorrection.yaml",
199 ],
200 ):
201 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCamSim", pipeline))
203 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
204 def test_decam_pipelines(self):
205 for pipeline in self._get_pipelines(
206 exclude=[
207 "verifyGainFromFlatPairs.yaml",
208 "verifyScience.yaml",
209 "verifyIlluminationCorrection.yaml",
210 "verifyCrosstalk.yaml",
212 ],
213 ):
214 self._check_pipeline(os.path.join(self.pipeline_path, "DECam", pipeline))
216 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
217 def test_hsc_pipelines(self):
218 for pipeline in self._get_pipelines(
219 exclude=[
220 "verifyGainFromFlatPairs.yaml",
221 "verifyScience.yaml",
222 "verifyIlluminationCorrection.yaml",
223 "verifyCrosstalk.yaml",
225 ],
226 ):
227 self._check_pipeline(os.path.join(self.pipeline_path, "HSC", pipeline))
230class TestMemory(lsst.utils.tests.MemoryTestCase):
231 pass
234def setup_module(module):
235 lsst.utils.tests.init()
238if __name__ == "__main__": 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 lsst.utils.tests.init()
240 unittest.main()