Coverage for tests/test_verify_rewrite.py: 65%

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14import os 

15from pathlib import Path 

16from typing import Any 

17 

18import numpy as np 

19import pytest 

20from click.testing import CliRunner 

21 

22from lsst.images import Background, BackgroundMap, Box, DifferenceImage 

23from lsst.images.cli._main import main 

24from lsst.images.fields import ChebyshevField 

25from lsst.images.tests import ( 

26 DP2_VISIT_DETECTOR_DATA_ID, 

27 get_dp2_exposure_record, 

28 reset_afw_mask_planes, # noqa: F401 

29) 

30from lsst.images.tests.verify_rewrite import ( 

31 RewriteVerifier, 

32 _check_backgrounds, 

33 _check_kernel, 

34 _check_templates, 

35) 

36 

37 

38def _bbox() -> Box: 

39 return Box.factory[0:10, 0:10] 

40 

41 

42def _constant_background(name: str = "subtracted") -> Background: 

43 field = ChebyshevField(_bbox(), np.array([[5.0]])) 

44 return Background(name=name, field=field, description="test") 

45 

46 

47def test_check_backgrounds_present_and_finite() -> None: 

48 """Check that present and finite backgrounds pass.""" 

49 backgrounds = BackgroundMap( 

50 [_constant_background("subtracted"), _constant_background("skyCorr")], 

51 subtracted="subtracted", 

52 ) 

53 _check_backgrounds(backgrounds, _bbox(), expected=("subtracted", "skyCorr")) 

54 

55 

56def test_check_backgrounds_missing_expected_fails() -> None: 

57 """Check that a missing expected background name fails.""" 

58 backgrounds = BackgroundMap([_constant_background("subtracted")], subtracted="subtracted") 

59 with pytest.raises(AssertionError, match="expected background 'skyCorr' not attached"): 

60 _check_backgrounds(backgrounds, _bbox(), expected=("subtracted", "skyCorr")) 

61 

62 

63def test_check_backgrounds_non_finite_fails() -> None: 

64 """Check that a non-finite background field fails.""" 

65 field = ChebyshevField(_bbox(), np.array([[np.nan]])) 

66 backgrounds = BackgroundMap([Background(name="subtracted", field=field)], subtracted="subtracted") 

67 with pytest.raises(AssertionError, match="non-finite"): 

68 _check_backgrounds(backgrounds, _bbox()) 

69 

70 

71def test_check_backgrounds_no_background_ok() -> None: 

72 """Check that an empty background map with no expectations passes.""" 

73 _check_backgrounds(BackgroundMap(), _bbox(), expected=()) 

74 

75 

76def test_check_kernel_none_fails() -> None: 

77 """Check that a None kernel fails.""" 

78 with pytest.raises(AssertionError): 

79 _check_kernel(None) 

80 

81 

82def test_check_templates_none_fails() -> None: 

83 """Check that a None templates list fails.""" 

84 with pytest.raises(AssertionError): 

85 _check_templates(None, _bbox()) 

86 

87 

88def test_check_templates_empty_fails() -> None: 

89 """Check that an empty templates list fails.""" 

90 with pytest.raises(AssertionError): 

91 _check_templates([], _bbox()) 

92 

93 

94def test_rewrite_verifier_reports_problems() -> None: 

95 """Assert that print_error increments the problem count.""" 

96 verifier = RewriteVerifier(None, "difference_image", old_prefix="legacy_", new_prefix="") 

97 assert verifier.n_problems == 0 

98 verifier.print_error("data_id", AssertionError("boom")) 

99 assert verifier.n_problems == 1 

100 

101 

102def test_print_error_no_note(capsys: pytest.CaptureFixture[str]) -> None: 

103 """An error without a component note is printed on one line.""" 

104 verifier = RewriteVerifier(None, "difference_image", old_prefix="legacy_", new_prefix="") 

105 verifier.print_error("data_id", AssertionError("boom")) 

106 assert capsys.readouterr().out == "data_id: boom (AssertionError)\n" 

107 

108 

109def test_print_error_component_note(capsys: pytest.CaptureFixture[str]) -> None: 

110 """A component note is printed as a prefix on its own indented line.""" 

111 verifier = RewriteVerifier(None, "difference_image", old_prefix="legacy_", new_prefix="") 

112 err = AssertionError("81/6642 values differ; max abs diff 1e-9 at index (np.int64(0),)") 

113 err.add_note("sky_projection") 

114 verifier.print_error("data_id", err) 

115 assert capsys.readouterr().out == ( 

116 "data_id\n sky_projection: 81/6642 values differ; max abs diff 1e-9 at index" 

117 " (np.int64(0),) (AssertionError)\n" 

118 ) 

119 

120 

121def test_print_error_nested_notes(capsys: pytest.CaptureFixture[str]) -> None: 

122 """Nested component notes are rendered as a path, not repeated messages.""" 

123 verifier = RewriteVerifier(None, "difference_image", old_prefix="legacy_", new_prefix="") 

124 err = AssertionError("boom") 

125 err.add_note("visit_image") 

126 err.add_note("sky_projection") 

127 verifier.print_error("data_id", err) 

128 assert capsys.readouterr().out == "data_id\n visit_image -> sky_projection: boom (AssertionError)\n" 

129 

130 

131@pytest.fixture(scope="module") 

132def testdata_dir() -> str: 

133 """Return the external test-data directory, skipping if unset.""" 

134 if (result := os.environ.get("TESTDATA_IMAGES_DIR")) is None: 134 ↛ 136line 134 didn't jump to line 136 because the condition on line 134 was always true

135 pytest.skip("TESTDATA_IMAGES_DIR is not set.") 

136 return result 

137 

138 

139def _make_dp2_rewrite_repo(repo: str, src: str) -> None: 

140 """Build a repo holding a legacy DP2 difference image and its rewritten 

141 counterpart, ready for ``verify-rewrite``. 

142 

143 The rewritten dataset is converted from the source file with the given 

144 (or the true DP2) exposure record; passing a tampered record produces a 

145 product that ``verify-rewrite``'s observation-info check should reject. 

146 """ 

147 from lsst.afw.image import ExposureF 

148 from lsst.daf.butler import Butler, DataCoordinate, DatasetType 

149 

150 Butler.makeRepo(repo) 

151 with Butler.from_config(repo, run="run1") as butler: 

152 exposure_record = get_dp2_exposure_record(butler.dimensions) 

153 reg = butler.registry 

154 reg.insertDimensionData("instrument", {"instrument": exposure_record.instrument}) 

155 reg.insertDimensionData( 

156 "day_obs", {"instrument": exposure_record.instrument, "day_obs": exposure_record.day_obs} 

157 ) 

158 reg.insertDimensionData( 

159 "physical_filter", 

160 { 

161 "physical_filter": exposure_record.physical_filter, 

162 "band": DP2_VISIT_DETECTOR_DATA_ID["band"], 

163 "instrument": exposure_record.instrument, 

164 }, 

165 ) 

166 reg.insertDimensionData( 

167 "detector", 

168 { 

169 "instrument": exposure_record.instrument, 

170 "id": DP2_VISIT_DETECTOR_DATA_ID["detector"], 

171 "full_name": "R21_S11", 

172 }, 

173 ) 

174 reg.insertDimensionData( 

175 "visit", 

176 { 

177 "instrument": exposure_record.instrument, 

178 "id": exposure_record.id, 

179 "physical_filter": exposure_record.physical_filter, 

180 "name": str(exposure_record.id), 

181 "day_obs": exposure_record.day_obs, 

182 }, 

183 ) 

184 reg.insertDimensionData( 

185 "visit_detector_region", 

186 { 

187 "instrument": exposure_record.instrument, 

188 "visit": exposure_record.id, 

189 "detector": DP2_VISIT_DETECTOR_DATA_ID["detector"], 

190 "region": None, 

191 }, 

192 ) 

193 reg.insertDimensionData( 

194 "group", {"instrument": exposure_record.instrument, "name": exposure_record.group} 

195 ) 

196 reg.insertDimensionData("exposure", exposure_record) 

197 reg.insertDimensionData( 

198 "visit_definition", 

199 { 

200 "instrument": exposure_record.instrument, 

201 "visit": exposure_record.id, 

202 "exposure": exposure_record.id, 

203 }, 

204 ) 

205 

206 dims = ("instrument", "visit", "detector") 

207 data_id = DataCoordinate.standardize( 

208 { 

209 "instrument": exposure_record.instrument, 

210 "visit": exposure_record.id, 

211 "detector": DP2_VISIT_DETECTOR_DATA_ID["detector"], 

212 }, 

213 universe=butler.dimensions, 

214 ) 

215 legacy_dt = DatasetType("legacy_difference_image", dims, "ExposureF", universe=butler.dimensions) 

216 new_dt = DatasetType("difference_image", dims, "DifferenceImage", universe=butler.dimensions) 

217 reg.registerDatasetType(legacy_dt) 

218 reg.registerDatasetType(new_dt) 

219 

220 butler.put(ExposureF(src), legacy_dt, data_id) 

221 # Convert with a second read of the file, not from_legacy() on the 

222 # exposure above: the conversion shares pixel data with, and mutates 

223 # the metadata of, the exposure it is given. 

224 new = DifferenceImage.read_legacy(src, exposure_record=exposure_record) 

225 butler.put(new, new_dt, data_id) 

226 

227 

228def _invoke_verify_rewrite(repo: str) -> Any: 

229 return CliRunner().invoke( 

230 main, 

231 [ 

232 "verify-rewrite", 

233 repo, 

234 "difference_image", 

235 "run1", 

236 "--no-check-kernel", 

237 "--no-check-templates", 

238 "--no-require-compressed", 

239 ], 

240 ) 

241 

242 

243def test_verify_rewrite_end_to_end(tmp_path: Path, testdata_dir: str, reset_afw_mask_planes) -> None: # noqa: F811 

244 """Run verify-rewrite on a real difference image.""" 

245 try: 

246 import lsst.afw.image 

247 import lsst.daf.butler # noqa: F401 

248 except ImportError: 

249 pytest.skip("lsst.daf.butler and lsst.afw could not be imported.") 

250 

251 src = os.path.join(testdata_dir, "dp2", "legacy", "difference_image.fits") 

252 repo = str(tmp_path / "repo") 

253 _make_dp2_rewrite_repo(repo, src) 

254 result = _invoke_verify_rewrite(repo) 

255 assert result.exit_code == 0, result.output