Coverage for tests/test_verify_rewrite.py: 70%

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 

16 

17import numpy as np 

18import pytest 

19from click.testing import CliRunner 

20 

21from lsst.images import Background, BackgroundMap, Box 

22from lsst.images.cli._main import main 

23from lsst.images.fields import ChebyshevField 

24from lsst.images.tests import ( 

25 DP2_VISIT_DETECTOR_DATA_ID, 

26 reset_afw_mask_planes, # noqa: F401 

27) 

28from lsst.images.tests.verify_rewrite import ( 

29 RewriteVerifier, 

30 _check_backgrounds, 

31 _check_kernel, 

32 _check_templates, 

33) 

34 

35 

36def _bbox() -> Box: 

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

38 

39 

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

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

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

43 

44 

45def test_check_backgrounds_present_and_finite() -> None: 

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

47 backgrounds = BackgroundMap( 

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

49 subtracted="subtracted", 

50 ) 

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

52 

53 

54def test_check_backgrounds_missing_expected_fails() -> None: 

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

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

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

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

59 

60 

61def test_check_backgrounds_non_finite_fails() -> None: 

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

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

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

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

66 _check_backgrounds(backgrounds, _bbox()) 

67 

68 

69def test_check_backgrounds_no_background_ok() -> None: 

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

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

72 

73 

74def test_check_kernel_none_fails() -> None: 

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

76 with pytest.raises(AssertionError): 

77 _check_kernel(None) 

78 

79 

80def test_check_templates_none_fails() -> None: 

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

82 with pytest.raises(AssertionError): 

83 _check_templates(None, _bbox()) 

84 

85 

86def test_check_templates_empty_fails() -> None: 

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

88 with pytest.raises(AssertionError): 

89 _check_templates([], _bbox()) 

90 

91 

92def test_rewrite_verifier_reports_problems() -> None: 

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

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

95 assert verifier.n_problems == 0 

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

97 assert verifier.n_problems == 1 

98 

99 

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

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

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

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

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

105 

106 

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

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

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

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

111 err.add_note("sky_projection") 

112 verifier.print_error("data_id", err) 

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

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

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

116 ) 

117 

118 

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

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

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

122 err = AssertionError("boom") 

123 err.add_note("visit_image") 

124 err.add_note("sky_projection") 

125 verifier.print_error("data_id", err) 

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

127 

128 

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

130def testdata_dir() -> str: 

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

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

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

134 return result 

135 

136 

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

138 """Run convert then verify-rewrite on a real difference image. 

139 

140 Happy path: asserts the whole flow exits successfully. 

141 """ 

142 try: 

143 from lsst.daf.butler import Butler, DataCoordinate, DatasetRef, DatasetType, FileDataset 

144 except ImportError: 

145 pytest.skip("lsst.daf.butler could not be imported.") 

146 

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

148 converted = str(tmp_path / "difference_image.fits") 

149 result = CliRunner().invoke(main, ["convert", src, converted]) 

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

151 

152 repo = str(tmp_path / "repo") 

153 Butler.makeRepo(repo) 

154 butler = Butler.from_config(repo, run="run1") 

155 reg = butler.registry 

156 reg.insertDimensionData( 

157 "instrument", 

158 { 

159 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

160 "name": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

161 }, 

162 ) 

163 reg.insertDimensionData( 

164 "day_obs", 

165 { 

166 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

167 "day_obs": DP2_VISIT_DETECTOR_DATA_ID["day_obs"], 

168 }, 

169 ) 

170 reg.insertDimensionData( 

171 "physical_filter", 

172 { 

173 "physical_filter": DP2_VISIT_DETECTOR_DATA_ID["physical_filter"], 

174 "band": DP2_VISIT_DETECTOR_DATA_ID["band"], 

175 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

176 }, 

177 ) 

178 reg.insertDimensionData( 

179 "detector", 

180 { 

181 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

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

183 "full_name": "R21_S11", 

184 }, 

185 ) 

186 reg.insertDimensionData( 

187 "visit", 

188 { 

189 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

190 "id": DP2_VISIT_DETECTOR_DATA_ID["visit"], 

191 "physical_filter": DP2_VISIT_DETECTOR_DATA_ID["physical_filter"], 

192 "name": str(DP2_VISIT_DETECTOR_DATA_ID["visit"]), 

193 "day_obs": DP2_VISIT_DETECTOR_DATA_ID["day_obs"], 

194 }, 

195 ) 

196 reg.insertDimensionData( 

197 "visit_detector_region", 

198 { 

199 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

200 "visit": DP2_VISIT_DETECTOR_DATA_ID["visit"], 

201 "detector": DP2_VISIT_DETECTOR_DATA_ID["detector"], 

202 "region": None, 

203 }, 

204 ) 

205 

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

207 data_id = DataCoordinate.standardize( 

208 { 

209 "instrument": DP2_VISIT_DETECTOR_DATA_ID["instrument"], 

210 "visit": DP2_VISIT_DETECTOR_DATA_ID["visit"], 

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.ingest( 

221 FileDataset(src, DatasetRef(legacy_dt, data_id, "run1")), 

222 FileDataset(converted, DatasetRef(new_dt, data_id, "run1")), 

223 ) 

224 

225 result = CliRunner().invoke( 

226 main, 

227 [ 

228 "verify-rewrite", 

229 repo, 

230 "difference_image", 

231 "run1", 

232 "--no-check-kernel", 

233 "--no-check-templates", 

234 "--no-require-compressed", 

235 ], 

236 ) 

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