Coverage for tests/test_serialization_io.py: 95%

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. 

11from __future__ import annotations 

12 

13from pathlib import Path 

14 

15import numpy as np 

16import pytest 

17 

18from lsst.images import Box, Image, VisitImage 

19from lsst.images.serialization import ArchiveReadError, read_archive, write_archive 

20from lsst.images.tests import current_fixture_path, iter_schema_fixtures 

21from lsst.utils.introspection import get_full_type_name 

22 

23try: 

24 import h5py # noqa: F401 -- detect availability for NDF round-trip skip 

25 

26 H5PY_AVAILABLE = True 

27except ImportError: 

28 H5PY_AVAILABLE = False 

29 

30try: 

31 import piff # noqa: F401 -- detect availability for piff_psf fixture skip 

32 

33 PIFF_AVAILABLE = True 

34except ImportError: 

35 PIFF_AVAILABLE = False 

36 

37skip_no_h5py = pytest.mark.skipif(not H5PY_AVAILABLE, reason="h5py is not installed") 

38 

39FIXTURE_DIR = Path(__file__).parent / "data" / "schemas" 

40 

41# Full Python type produced when each fixture is read through the generic 

42# read_archive() API, keyed by (schema name, variant). These are pinned here 

43# rather than derived from the schema registry so the test asserts the 

44# externally-observable type instead of re-running read_archive()'s own 

45# lookup against itself. Keying by name and variant rather than by file name 

46# keeps the table stable across version bumps. 

47EXPECTED_TYPES: dict[tuple[str, str | None], str] = { 

48 ("aperture_correction_map", None): "dict", 

49 ("background_map", None): "lsst.images.BackgroundMap", 

50 ("camera_frame_set", None): "lsst.images.cameras.CameraFrameSet", 

51 ("cell_aperture_correction_map", None): "dict", 

52 ("cell_coadd", "as_shipped"): "lsst.images.cells.CellCoadd", 

53 ("cell_coadd", "canonical"): "lsst.images.cells.CellCoadd", 

54 ("cell_psf", None): "lsst.images.cells.CellPointSpreadFunction", 

55 ("chebyshev_field", None): "lsst.images.fields.ChebyshevField", 

56 ("coadd_provenance", None): "lsst.images.cells.CoaddProvenance", 

57 ("color_image", None): "lsst.images.ColorImage", 

58 ("detector", None): "lsst.images.cameras.Detector", 

59 ("difference_image", "dp2"): "lsst.images.DifferenceImage", 

60 ("difference_image", "template_inputs"): "lsst.images.DifferenceImage", 

61 ("gaussian_psf", None): "lsst.images.psfs.GaussianPointSpreadFunction", 

62 ("image", None): "lsst.images.Image", 

63 ( 

64 "image_basis_convolution_kernel", 

65 None, 

66 ): "lsst.images.convolution_kernels.ImageBasisConvolutionKernel", 

67 ("mask", None): "lsst.images.Mask", 

68 ("masked_image", None): "lsst.images.MaskedImage", 

69 ("observation_summary_stats", None): "lsst.images.ObservationSummaryStats", 

70 ("piff_psf", None): "lsst.images.psfs.PiffWrapper", 

71 ("product_field", None): "lsst.images.fields.ProductField", 

72 ("sky_projection", None): "lsst.images.SkyProjection", 

73 ("spline_field", None): "lsst.images.fields.SplineField", 

74 ("sum_field", None): "lsst.images.fields.SumField", 

75 ("transform", None): "lsst.images.Transform", 

76 ("visit_image", None): "lsst.images.VisitImage", 

77 ("visit_image", "dp1"): "lsst.images.VisitImage", 

78 ("visit_image", "dp2"): "lsst.images.VisitImage", 

79} 

80 

81 

82def test_generic_read_visit_image_json() -> None: 

83 """Verify read_archive() on a visit_image JSON fixture returns a 

84 VisitImage. 

85 """ 

86 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

87 result = read_archive(path) 

88 assert isinstance(result, VisitImage) 

89 

90 

91def test_generic_read_image_json() -> None: 

92 """Verify read_archive() on an image JSON fixture returns an Image.""" 

93 path = current_fixture_path(FIXTURE_DIR, "image") 

94 result = read_archive(path) 

95 assert isinstance(result, Image) 

96 

97 

98def test_read_unsupported_extension(tmp_path: Path) -> None: 

99 """Verify read_archive() raises ValueError for an unrecognized file 

100 extension. 

101 """ 

102 path = tmp_path / "bogus.txt" 

103 with open(path, "w") as f: 

104 f.write("nope") 

105 with pytest.raises(ValueError, match="Unrecognized file extension"): 

106 read_archive(path) 

107 

108 

109def test_read_unregistered_schema(tmp_path: Path) -> None: 

110 """Verify read_archive() raises ArchiveReadError for a JSON with an unknown 

111 schema. 

112 """ 

113 path = tmp_path / "fake.json" 

114 with open(path, "w") as f: 

115 f.write( 

116 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",' 

117 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}' 

118 ) 

119 with pytest.raises(ArchiveReadError) as exc_info: 

120 read_archive(path) 

121 assert "no-such-schema" in str(exc_info.value) 

122 

123 

124def _sweep_sort_key(entry: tuple[str, str | None]) -> tuple[str, str]: 

125 """Return a sort key for an ``EXPECTED_TYPES`` entry, ``None`` sorting 

126 first within a schema name. 

127 """ 

128 name, variant = entry 

129 return (name, variant or "") 

130 

131 

132# Assigned to an explicitly annotated variable rather than sorted() inline in 

133# the decorator below: parametrize()'s stub types its arguments as `object`, 

134# and that context leaks into bidirectional generic inference for a bare 

135# sorted() call, defeating the key function's type. 

136_SWEEP_ENTRIES: list[tuple[str, str | None]] = sorted(EXPECTED_TYPES, key=_sweep_sort_key) 

137 

138 

139def test_expected_types_enumerates_every_committed_fixture_variant() -> None: 

140 """Verify EXPECTED_TYPES matches the fixture tree's (name, variant) pairs 

141 exactly, so a committed fixture cannot go missing unnoticed. 

142 

143 Each rung below may legitimately skip an individual case when an optional 

144 dependency (piff) is unavailable, but that must never be confused with a 

145 committed fixture file itself going missing: this closes the presence 

146 gap for every variant at once, rather than relying on each consumer's own 

147 guard against a missing path (which a stray ``git rm`` or an interrupted 

148 freeze would otherwise defeat with a silent skip instead of a failure). 

149 

150 The two directions fail separately and name their own remedy, because 

151 adding a fixture variant and losing one are unrelated mistakes with 

152 opposite fixes, and this test is the only thing that catches either. 

153 """ 

154 present = {(f.name, f.variant) for f in iter_schema_fixtures(FIXTURE_DIR) if not f.retired} 

155 unlisted = sorted(present - set(EXPECTED_TYPES)) 

156 absent = sorted(set(EXPECTED_TYPES) - present) 

157 assert not unlisted, ( 

158 f"committed fixture variants missing from EXPECTED_TYPES: {unlisted}; add each with the " 

159 "full type name its fixture deserializes to, so test_fixture_sweep covers it" 

160 ) 

161 assert not absent, ( 

162 f"EXPECTED_TYPES entries with no committed fixture: {absent}; delete each entry, or " 

163 "restore the fixture file if it went missing by accident" 

164 ) 

165 

166 

167@pytest.mark.parametrize("entry", _SWEEP_ENTRIES) 

168def test_fixture_sweep(entry: tuple[str, str | None]) -> None: 

169 """Verify every schema fixture reads to its pinned Python type.""" 

170 name, variant = entry 

171 if name == "piff_psf" and not PIFF_AVAILABLE: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true

172 pytest.skip("piff not available") 

173 path = current_fixture_path(FIXTURE_DIR, name, variant=variant) 

174 assert path.exists(), f"{path} is a committed fixture and must not go missing" 

175 result = read_archive(path) 

176 assert get_full_type_name(type(result)) == EXPECTED_TYPES[entry], entry 

177 

178 

179def _make_image() -> Image: 

180 """Return a small float32 Image for round-trip tests.""" 

181 return Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4]) 

182 

183 

184def test_generic_write_round_trip_fits(tmp_path: Path) -> None: 

185 """Verify write_archive() + read_archive() round-trips an Image 

186 through FITS. 

187 """ 

188 image = _make_image() 

189 path = tmp_path / "x.fits" 

190 write_archive(image, path) 

191 result = read_archive(path) 

192 assert isinstance(result, Image) 

193 np.testing.assert_array_equal(result.array, image.array) 

194 

195 

196def test_generic_write_round_trip_json(tmp_path: Path) -> None: 

197 """Verify write_archive() + read_archive() round-trips an Image 

198 through JSON. 

199 """ 

200 image = _make_image() 

201 path = tmp_path / "x.json" 

202 write_archive(image, path) 

203 result = read_archive(path) 

204 assert isinstance(result, Image) 

205 np.testing.assert_array_equal(result.array, image.array) 

206 

207 

208def test_write_metadata_override_does_not_mutate_source(tmp_path: Path) -> None: 

209 """Write-only metadata is detached from the source object's mapping.""" 

210 image = _make_image() 

211 image.metadata["shared"] = "source" 

212 path = tmp_path / "metadata.json" 

213 tree = write_archive(image, path, metadata={"shared": "file", "extra": 5}) 

214 

215 assert image.metadata == {"shared": "source"} 

216 assert tree.metadata == {"shared": "file", "extra": 5} 

217 assert tree.metadata is not image.metadata 

218 result = read_archive(path, Image) 

219 assert result.metadata == tree.metadata 

220 

221 

222@skip_no_h5py 

223def test_generic_write_round_trip_ndf(tmp_path: Path) -> None: 

224 """Verify write_archive() + read_archive() round-trips an Image 

225 through NDF. 

226 """ 

227 image = _make_image() 

228 path = tmp_path / "x.sdf" 

229 write_archive(image, path) 

230 result = read_archive(path) 

231 assert isinstance(result, Image) 

232 np.testing.assert_array_equal(result.array, image.array) 

233 

234 

235def test_read_bbox_subset_fits(tmp_path: Path) -> None: 

236 """Verify read_archive() forwards bbox kwarg to the FITS backend for subset 

237 reads. 

238 """ 

239 img = Image(np.arange(64, dtype=np.float32).reshape(8, 8), bbox=Box.factory[0:8, 0:8]) 

240 path = tmp_path / "x.fits" 

241 write_archive(img, path) 

242 sub = read_archive(path, bbox=Box.factory[2:6, 2:6]) 

243 assert sub.array.shape == (4, 4) 

244 np.testing.assert_array_equal(sub.array, img.array[2:6, 2:6]) 

245 

246 

247def test_read_cls_match() -> None: 

248 """Verify read_archive() with cls= returns the expected type when it 

249 matches. 

250 """ 

251 path = current_fixture_path(FIXTURE_DIR, "image") 

252 result = read_archive(path, cls=Image) 

253 assert isinstance(result, Image) 

254 

255 

256def test_read_cls_mismatch_raises() -> None: 

257 """Verify read_archive() raises TypeError when the deserialized type 

258 does not match cls. 

259 """ 

260 from lsst.images import Mask 

261 

262 path = current_fixture_path(FIXTURE_DIR, "image") 

263 with pytest.raises(TypeError) as exc_info: 

264 read_archive(path, cls=Mask) 

265 msg = str(exc_info.value) 

266 assert "image" in msg # path / schema name 

267 assert "Image" in msg # actual deserialized type 

268 assert "Mask" in msg # requested cls