Coverage for tests / test_pipelines.py: 39%
85 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-05 18:54 +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.
30import pyproj # noqa: F401
32from lsst.pipe.base import Pipeline, PipelineGraph
33import lsst.utils
35try:
36 import lsst.obs.lsst
37 has_obs_lsst = True
38except ImportError:
39 has_obs_lsst = False
41try:
42 import lsst.obs.subaru
43 has_obs_subaru = True
44except ImportError:
45 has_obs_subaru = False
47try:
48 import lsst.obs.decam
49 has_obs_decam = True
50except ImportError:
51 has_obs_decam = False
54class VerifyPipelinesTestCase(lsst.utils.tests.TestCase):
55 """Test case for building the pipelines."""
57 def setUp(self):
58 self.pipeline_path = os.path.join(lsst.utils.getPackageDir("cp_verify"), "pipelines")
60 def _get_pipelines(self, exclude=[]):
61 pipelines = {
62 "verifyBfk.yaml",
63 "verifyBias.yaml",
64 "verifyCrosstalk.yaml",
65 "verifyDark.yaml",
66 "verifyDefectsIndividual.yaml",
67 "verifyDefects.yaml",
68 "verifyFlat.yaml",
69 "verifyScience.yaml",
70 # Old pipeline name.
71 "verifyGain.yaml",
72 # New pipeline name.
73 "verifyGainFromFlatPairs.yaml",
74 "verifyLinearizer.yaml",
75 "verifyPtc.yaml",
76 "verifyIlluminationCorrection.yaml",
77 }
79 for ex in exclude:
80 pipelines.remove(ex)
82 return pipelines
84 def _check_pipeline(self, pipeline_file):
85 # Confirm that the file is there.
86 self.assertTrue(os.path.isfile(pipeline_file), msg=f"Could not find {pipeline_file}")
88 # The following loads the pipeline and confirms that it can parse all
89 # the configs.
90 try:
91 pipeline = Pipeline.fromFile(pipeline_file)
92 graph = pipeline.to_graph()
93 except Exception as e:
94 raise RuntimeError(f"Could not process {pipeline_file}") from e
96 self.assertIsInstance(graph, PipelineGraph)
98 def test_ingredients(self):
99 """Check that all pipelines in pipelines/_ingredients are tested."""
100 glob_str = os.path.join(self.pipeline_path, "_ingredients", "*.yaml")
101 # The *LSST.yaml pipelines are imported by LATISS/LSSTComCam/LSSTCam
102 # and are not tested on their own.
103 ingredients = set(
104 [os.path.basename(pipeline) for pipeline in glob.glob(glob_str) if "LSST.yaml" not in pipeline]
105 )
106 # The _ingredients/verifyGainFromFlatPairs.yaml becomes
107 # verifyGain.yaml in older pipelines for compatibility.
108 expected = self._get_pipelines()
109 # The _ingredients/verifyGainFromFlatPairs.yaml becomes
110 # verifyGain.yaml in older pipelines for compatibility.
111 expected.remove("verifyGain.yaml")
112 # There is no regular verifyScience.yaml (non-LSST)
113 expected.remove("verifyScience.yaml")
114 # The Illumination Correction pipeline is only an "LSST" version.
115 expected.remove("verifyIlluminationCorrection.yaml")
116 self.assertEqual(ingredients, expected)
118 def test_cameras(self):
119 """Check that all the cameras in pipelines are tested."""
120 glob_str = os.path.join(self.pipeline_path, "*")
121 paths = set(
122 [os.path.basename(path) for path in glob.glob(glob_str)]
123 )
124 expected = {
125 "DECam",
126 "HSC",
127 "_ingredients",
128 "LATISS",
129 "LSSTCam",
130 "LSSTCam-imSim",
131 "LSSTComCam",
132 "LSSTComCamSim",
133 "README.md",
134 }
135 self.assertEqual(paths, expected)
137 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LATISS pipelines without obs_lsst")
138 def test_latiss_pipelines(self):
139 for pipeline in self._get_pipelines(exclude=[
140 # The old pipeline name should be excluded.
141 "verifyGain.yaml",
142 # The following tasks are not part of the new pipelines.
143 "verifyDefectsIndividual.yaml",
144 # The following tasks will be added in the future.
145 "verifyCrosstalk.yaml",
146 "verifyBfk.yaml",
147 "verifyIlluminationCorrection.yaml",
149 ]):
150 self._check_pipeline(os.path.join(self.pipeline_path, "LATISS", pipeline))
152 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam pipelines without obs_lsst")
153 def test_lsstcam_pipelines(self):
154 for pipeline in self._get_pipelines(
155 exclude=[
156 # These are renamed/not used in the new pipelines.
157 "verifyGain.yaml",
158 "verifyDefectsIndividual.yaml",
159 # These are not used yet.
160 "verifyCrosstalk.yaml",
161 "verifyIlluminationCorrection.yaml",
162 ]):
163 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam", pipeline))
165 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTCam-imSim pipelines without obs_lsst")
166 def test_lsstcam_imsim_pipelines(self):
167 for pipeline in self._get_pipelines(
168 exclude=[
169 "verifyGainFromFlatPairs.yaml",
170 "verifyScience.yaml",
171 "verifyIlluminationCorrection.yaml",
172 ],
173 ):
174 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTCam-imSim", pipeline))
176 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCam pipelines without obs_lsst")
177 def test_lsstcomcam_pipelines(self):
178 for pipeline in self._get_pipelines(
179 exclude=[
180 # These are renamed/not used in the new pipelines.
181 "verifyGain.yaml",
182 "verifyDefectsIndividual.yaml",
183 # These are not used yet.
184 "verifyCrosstalk.yaml",
185 ],
186 ):
187 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCam", pipeline))
189 @unittest.skipIf(not has_obs_lsst, reason="Cannot test LSSTComCamSim pipelines without obs_lsst")
190 def test_lsstcomcamsim_pipelines(self):
191 for pipeline in self._get_pipelines(
192 exclude=[
193 # These are renamed/not used in the new pipelines.
194 "verifyGain.yaml",
195 "verifyDefectsIndividual.yaml",
196 # These are not valid for LSSTComCamSim.
197 "verifyCrosstalk.yaml",
198 "verifyLinearizer.yaml",
199 "verifyIlluminationCorrection.yaml",
201 ],
202 ):
203 self._check_pipeline(os.path.join(self.pipeline_path, "LSSTComCamSim", pipeline))
205 @unittest.skipIf(not has_obs_decam, reason="Cannot test DECam pipelines without obs_decam")
206 def test_decam_pipelines(self):
207 for pipeline in self._get_pipelines(
208 exclude=[
209 "verifyGainFromFlatPairs.yaml",
210 "verifyScience.yaml",
211 "verifyIlluminationCorrection.yaml",
213 ],
214 ):
215 self._check_pipeline(os.path.join(self.pipeline_path, "DECam", pipeline))
217 @unittest.skipIf(not has_obs_subaru, reason="Cannot test HSC pipelines without obs_subaru")
218 def test_hsc_pipelines(self):
219 for pipeline in self._get_pipelines(
220 exclude=[
221 "verifyGainFromFlatPairs.yaml",
222 "verifyScience.yaml",
223 "verifyIlluminationCorrection.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()