Coverage for tests / test_pipelines.py: 41%
88 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-21 10:56 +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
28# We need this import here to allow the proj.db to be cleaned up
29# properly.
30# TODO: Remove import after completing DM-54643. Use DM-54656
31try:
32 import pyproj # noqa: F401
33except ImportError:
34 pass
36from lsst.pipe.base import Pipeline, PipelineGraph
37import lsst.utils
39try:
40 import lsst.obs.lsst
41 has_obs_lsst = True
42except ImportError:
43 has_obs_lsst = False
45try:
46 import lsst.obs.subaru
47 has_obs_subaru = True
48except ImportError:
49 has_obs_subaru = False
51try:
52 import lsst.obs.decam
53 has_obs_decam = True
54except ImportError:
55 has_obs_decam = False
58class VerifyPipelinesTestCase(lsst.utils.tests.TestCase):
59 """Test case for building the pipelines."""
61 def setUp(self):
62 self.pipeline_path = os.path.join(lsst.utils.getPackageDir("cp_verify"), "pipelines")
64 def _get_pipelines(self, exclude=[]):
65 pipelines = {
66 "verifyBfk.yaml",
67 "verifyBias.yaml",
68 "verifyCrosstalk.yaml",
69 "verifyDark.yaml",
70 "verifyDefectsIndividual.yaml",
71 "verifyDefects.yaml",
72 "verifyFlat.yaml",
73 "verifyScience.yaml",
74 # Old pipeline name.
75 "verifyGain.yaml",
76 # New pipeline name.
77 "verifyGainFromFlatPairs.yaml",
78 "verifyLinearizer.yaml",
79 "verifyPtc.yaml",
80 "verifyIlluminationCorrection.yaml",
81 }
83 for ex in exclude:
84 pipelines.remove(ex)
86 return pipelines
88 def _check_pipeline(self, pipeline_file):
89 # Confirm that the file is there.
90 self.assertTrue(os.path.isfile(pipeline_file), msg=f"Could not find {pipeline_file}")
92 # The following loads the pipeline and confirms that it can parse all
93 # the configs.
94 try:
95 pipeline = Pipeline.fromFile(pipeline_file)
96 graph = pipeline.to_graph()
97 except Exception as e:
98 raise RuntimeError(f"Could not process {pipeline_file}") from e
100 self.assertIsInstance(graph, PipelineGraph)
102 def test_ingredients(self):
103 """Check that all pipelines in pipelines/_ingredients are tested."""
104 glob_str = os.path.join(self.pipeline_path, "_ingredients", "*.yaml")
105 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
106 # and are not tested on their own.
107 ingredients = set(
108 [os.path.basename(pipeline) for pipeline in glob.glob(glob_str) if "LSST.yaml" not in pipeline]
109 )
110 # The _ingredients/verifyGainFromFlatPairs.yaml becomes
111 # verifyGain.yaml in older pipelines for compatibility.
112 expected = self._get_pipelines()
113 # The _ingredients/verifyGainFromFlatPairs.yaml becomes
114 # verifyGain.yaml in older pipelines for compatibility.
115 expected.remove("verifyGain.yaml")
116 # There is no regular verifyScience.yaml (non-LSST)
117 expected.remove("verifyScience.yaml")
118 # The Illumination Correction pipeline is only an "LSST" version.
119 expected.remove("verifyIlluminationCorrection.yaml")
120 self.assertEqual(ingredients, expected)
122 def test_cameras(self):
123 """Check that all the cameras in pipelines are tested."""
124 glob_str = os.path.join(self.pipeline_path, "*")
125 paths = set(
126 [os.path.basename(path) for path in glob.glob(glob_str)]
127 )
128 expected = {
129 "DECam",
130 "HSC",
131 "_ingredients",
132 "LATISS",
133 "LSSTCam",
134 "LSSTCam-imSim",
135 "LSSTComCam",
136 "LSSTComCamSim",
137 "README.md",
138 }
139 self.assertEqual(paths, expected)
141 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
142 def test_latiss_pipelines(self):
143 for pipeline in self._get_pipelines(exclude=[
144 # The old pipeline name should be excluded.
145 "verifyGain.yaml",
146 # The following tasks are not part of the new pipelines.
147 "verifyDefectsIndividual.yaml",
148 # The following tasks will be added in the future.
149 "verifyCrosstalk.yaml",
150 "verifyBfk.yaml",
151 "verifyIlluminationCorrection.yaml",
153 ]):
154 self._check_pipeline(os.path.join(self.pipeline_path, "LATISS", pipeline))
156 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
157 def test_lsstcam_pipelines(self):
158 for pipeline in self._get_pipelines(
159 exclude=[
160 # These are renamed/not used in the new pipelines.
161 "verifyGain.yaml",
162 "verifyDefectsIndividual.yaml",
163 # These are not used yet.
164 "verifyCrosstalk.yaml",
165 "verifyIlluminationCorrection.yaml",
166 ]):
167 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam", pipeline))
169 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
170 def test_lsstcam_imsim_pipelines(self):
171 for pipeline in self._get_pipelines(
172 exclude=[
173 "verifyGainFromFlatPairs.yaml",
174 "verifyScience.yaml",
175 "verifyIlluminationCorrection.yaml",
176 "verifyCrosstalk.yaml",
177 ],
178 ):
179 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam-imSim", pipeline))
181 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
182 def test_lsstcomcam_pipelines(self):
183 for pipeline in self._get_pipelines(
184 exclude=[
185 # These are renamed/not used in the new pipelines.
186 "verifyGain.yaml",
187 "verifyDefectsIndividual.yaml",
188 # These are not used yet.
189 "verifyCrosstalk.yaml",
190 ],
191 ):
192 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCam", pipeline))
194 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
195 def test_lsstcomcamsim_pipelines(self):
196 for pipeline in self._get_pipelines(
197 exclude=[
198 # These are renamed/not used in the new pipelines.
199 "verifyGain.yaml",
200 "verifyDefectsIndividual.yaml",
201 # These are not valid for LSSTComCamSim.
202 "verifyCrosstalk.yaml",
203 "verifyLinearizer.yaml",
204 "verifyIlluminationCorrection.yaml",
206 ],
207 ):
208 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCamSim", pipeline))
210 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
211 def test_decam_pipelines(self):
212 for pipeline in self._get_pipelines(
213 exclude=[
214 "verifyGainFromFlatPairs.yaml",
215 "verifyScience.yaml",
216 "verifyIlluminationCorrection.yaml",
217 "verifyCrosstalk.yaml",
219 ],
220 ):
221 self._check_pipeline(os.path.join(self.pipeline_path, "DECam", pipeline))
223 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
224 def test_hsc_pipelines(self):
225 for pipeline in self._get_pipelines(
226 exclude=[
227 "verifyGainFromFlatPairs.yaml",
228 "verifyScience.yaml",
229 "verifyIlluminationCorrection.yaml",
230 "verifyCrosstalk.yaml",
232 ],
233 ):
234 self._check_pipeline(os.path.join(self.pipeline_path, "HSC", pipeline))
237class TestMemory(lsst.utils.tests.MemoryTestCase):
238 pass
241def setup_module(module):
242 lsst.utils.tests.init()
245if __name__ == "__main__": 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 lsst.utils.tests.init()
247 unittest.main()