Coverage for tests/test_pipelines.py: 90%
88 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-08 00: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_verify pipelines."""
25import unittest
27from lsst.pipe.base import Pipeline, PipelineGraph
28from lsst.resources import ResourcePath
29import lsst.utils
31try:
32 import lsst.obs.lsst
33 has_obs_lsst = True
34except ImportError:
35 has_obs_lsst = False
37try:
38 import lsst.obs.subaru
39 has_obs_subaru = True
40except ImportError:
41 has_obs_subaru = False
43try:
44 import lsst.obs.decam
45 has_obs_decam = True
46except ImportError:
47 has_obs_decam = False
49PIPELINE_URI = ResourcePath("eups://cp_verify/pipelines/", forceDirectory=True)
52class VerifyPipelinesTestCase(lsst.utils.tests.TestCase):
53 """Test case for building the pipelines."""
55 def _get_pipelines(self, exclude=[]):
56 pipelines = {
57 "verifyBfk.yaml",
58 "verifyBias.yaml",
59 "verifyCrosstalk.yaml",
60 "verifyDark.yaml",
61 "verifyDefectsIndividual.yaml",
62 "verifyDefects.yaml",
63 "verifyFlat.yaml",
64 "verifyScience.yaml",
65 # Old pipeline name.
66 "verifyGain.yaml",
67 # New pipeline name.
68 "verifyGainFromFlatPairs.yaml",
69 "verifyLinearizer.yaml",
70 "verifyPtc.yaml",
71 "verifyIlluminationCorrection.yaml",
72 }
74 for ex in exclude:
75 pipelines.remove(ex)
77 return pipelines
79 def _check_pipeline(self, pipeline_file: ResourcePath):
80 # Confirm that the file is there.
81 self.assertTrue(pipeline_file.exists(), msg=f"Could not find {pipeline_file}")
83 # The following loads the pipeline and confirms that it can parse all
84 # the configs.
85 try:
86 pipeline = Pipeline.from_uri(pipeline_file)
87 graph = pipeline.to_graph()
88 except Exception as e:
89 raise RuntimeError(f"Could not process {pipeline_file}") from e
91 self.assertIsInstance(graph, PipelineGraph)
93 def test_ingredients(self):
94 """Check that all pipelines in pipelines/_ingredients are tested."""
95 ingredient_files = ResourcePath.findFileResources(
96 [PIPELINE_URI.join("_ingredients")], file_filter=r".*\.yaml$"
97 )
98 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
99 # and are not tested on their own.
100 ingredients = set(
101 [pipeline.basename() for pipeline in ingredient_files if "LSST.yaml" not in pipeline.path]
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 _, paths, _ = next(PIPELINE_URI.walk())
118 expected = {
119 "DECam",
120 "HSC",
121 "_ingredients",
122 "LATISS",
123 "LSSTCam",
124 "LSSTCam-imSim",
125 "LSSTComCam",
126 "LSSTComCamSim",
127 }
128 self.assertEqual(set(paths), expected)
130 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
131 def test_latiss_pipelines(self):
132 latiss_uri = PIPELINE_URI.join("LATISS", forceDirectory=True)
133 for pipeline in self._get_pipelines(exclude=[
134 # The old pipeline name should be excluded.
135 "verifyGain.yaml",
136 # The following tasks are not part of the new pipelines.
137 "verifyDefectsIndividual.yaml",
138 # The following tasks will be added in the future.
139 "verifyCrosstalk.yaml",
140 "verifyBfk.yaml",
141 "verifyIlluminationCorrection.yaml",
143 ]):
144 self._check_pipeline(latiss_uri.join(pipeline))
146 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
147 def test_lsstcam_pipelines(self):
148 lsstcam_uri = PIPELINE_URI.join("LSSTCam", forceDirectory=True)
149 for pipeline in self._get_pipelines(
150 exclude=[
151 # These are renamed/not used in the new pipelines.
152 "verifyGain.yaml",
153 "verifyDefectsIndividual.yaml",
154 # These are not used yet.
155 "verifyCrosstalk.yaml",
156 "verifyIlluminationCorrection.yaml",
157 ]):
158 self._check_pipeline(lsstcam_uri.join(pipeline))
160 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
161 def test_lsstcam_imsim_pipelines(self):
162 sim_uri = PIPELINE_URI.join("LSSTCam-imSim", forceDirectory=True)
163 for pipeline in self._get_pipelines(
164 exclude=[
165 "verifyGainFromFlatPairs.yaml",
166 "verifyScience.yaml",
167 "verifyIlluminationCorrection.yaml",
168 "verifyCrosstalk.yaml",
169 ],
170 ):
171 self._check_pipeline(sim_uri.join(pipeline))
173 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
174 def test_lsstcomcam_pipelines(self):
175 comcam_uri = PIPELINE_URI.join("LSSTComCam", forceDirectory=True)
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(comcam_uri.join(pipeline))
187 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
188 def test_lsstcomcamsim_pipelines(self):
189 comcam_sim_uri = PIPELINE_URI.join("LSSTComCamSim", forceDirectory=True)
190 for pipeline in self._get_pipelines(
191 exclude=[
192 # These are renamed/not used in the new pipelines.
193 "verifyGain.yaml",
194 "verifyDefectsIndividual.yaml",
195 # These are not valid for LSSTComCamSim.
196 "verifyCrosstalk.yaml",
197 "verifyLinearizer.yaml",
198 "verifyIlluminationCorrection.yaml",
200 ],
201 ):
202 self._check_pipeline(comcam_sim_uri.join(pipeline))
204 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
205 def test_decam_pipelines(self):
206 decam_uri = PIPELINE_URI.join("DECam", forceDirectory=True)
207 for pipeline in self._get_pipelines(
208 exclude=[
209 "verifyGainFromFlatPairs.yaml",
210 "verifyScience.yaml",
211 "verifyIlluminationCorrection.yaml",
212 "verifyCrosstalk.yaml",
214 ],
215 ):
216 self._check_pipeline(decam_uri.join(pipeline))
218 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
219 def test_hsc_pipelines(self):
220 hsc_uri = PIPELINE_URI.join("HSC", forceDirectory=True)
221 for pipeline in self._get_pipelines(
222 exclude=[
223 "verifyGainFromFlatPairs.yaml",
224 "verifyScience.yaml",
225 "verifyIlluminationCorrection.yaml",
226 "verifyCrosstalk.yaml",
228 ],
229 ):
230 self._check_pipeline(hsc_uri.join(pipeline))
233class TestMemory(lsst.utils.tests.MemoryTestCase):
234 pass
237def setup_module(module):
238 lsst.utils.tests.init()
241if __name__ == "__main__": 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 lsst.utils.tests.init()
243 unittest.main()