Coverage for tests/test_serialization_io.py: 92%

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 

13import os 

14from pathlib import Path 

15 

16import numpy as np 

17import pytest 

18 

19from lsst.images import Box, Image, VisitImage 

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

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 

39LOCAL_DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "schema_v1") 

40 

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

42# read_archive() API, keyed by the fixture's file name. These are pinned 

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

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

45# own lookup against itself. 

46EXPECTED_TYPES = { 

47 "aperture_correction_map.json": "dict", 

48 "background_map.json": "lsst.images.BackgroundMap", 

49 "cell_psf.json": "lsst.images.cells.CellPointSpreadFunction", 

50 "cell_aperture_correction_map.json": "dict", 

51 "chebyshev_field.json": "lsst.images.fields.ChebyshevField", 

52 "coadd_provenance.json": "lsst.images.cells.CoaddProvenance", 

53 "color_image.json": "lsst.images.ColorImage", 

54 "detector.json": "lsst.images.cameras.Detector", 

55 "gaussian_psf.json": "lsst.images.psfs.GaussianPointSpreadFunction", 

56 "image.json": "lsst.images.Image", 

57 "mask.json": "lsst.images.Mask", 

58 "masked_image.json": "lsst.images.MaskedImage", 

59 "piff_psf.json": "lsst.images.psfs.PiffWrapper", 

60 "product_field.json": "lsst.images.fields.ProductField", 

61 "sky_projection.json": "lsst.images.SkyProjection", 

62 "sum_field.json": "lsst.images.fields.SumField", 

63 "transform.json": "lsst.images.Transform", 

64 "visit_image.json": "lsst.images.VisitImage", 

65 "cell_coadd.json": "lsst.images.cells.CellCoadd", 

66 "visit_image_dp1.json": "lsst.images.VisitImage", 

67 "visit_image_dp2.json": "lsst.images.VisitImage", 

68 "difference_image_dp2.json": "lsst.images.DifferenceImage", 

69} 

70 

71 

72def test_generic_read_visit_image_json() -> None: 

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

74 VisitImage. 

75 """ 

76 path = os.path.join(LOCAL_DATA_DIR, "visit_image.json") 

77 result = read_archive(path) 

78 assert isinstance(result, VisitImage) 

79 

80 

81def test_generic_read_image_json() -> None: 

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

83 path = os.path.join(LOCAL_DATA_DIR, "image.json") 

84 result = read_archive(path) 

85 assert isinstance(result, Image) 

86 

87 

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

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

90 extension. 

91 """ 

92 path = tmp_path / "bogus.txt" 

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

94 f.write("nope") 

95 with pytest.raises(ValueError): 

96 read_archive(path) 

97 

98 

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

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

101 schema. 

102 """ 

103 path = tmp_path / "fake.json" 

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

105 f.write( 

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

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

108 ) 

109 with pytest.raises(ArchiveReadError) as exc_info: 

110 read_archive(path) 

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

112 

113 

114@pytest.mark.parametrize("entry", sorted(EXPECTED_TYPES)) 

115def test_fixture_sweep(entry: str) -> None: 

116 """Verify every schema_v1 JSON fixture reads to the pinned Python type.""" 

117 if entry == "piff_psf.json" and not PIFF_AVAILABLE: 117 ↛ 118line 117 didn't jump to line 118 because the condition on line 117 was never true

118 pytest.skip("piff not available") 

119 roots = [LOCAL_DATA_DIR, os.path.join(LOCAL_DATA_DIR, "legacy")] 

120 for root in roots: 120 ↛ 126line 120 didn't jump to line 126 because the loop on line 120 didn't complete

121 path = os.path.join(root, entry) 

122 if os.path.exists(path): 

123 result = read_archive(path) 

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

125 return 

126 pytest.skip(f"fixture {entry!r} not found on disk") 

127 

128 

129def _make_image() -> Image: 

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

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

132 

133 

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

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

136 through FITS. 

137 """ 

138 image = _make_image() 

139 path = tmp_path / "x.fits" 

140 write_archive(image, path) 

141 result = read_archive(path) 

142 assert isinstance(result, Image) 

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

144 

145 

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

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

148 through JSON. 

149 """ 

150 image = _make_image() 

151 path = tmp_path / "x.json" 

152 write_archive(image, path) 

153 result = read_archive(path) 

154 assert isinstance(result, Image) 

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

156 

157 

158@skip_no_h5py 

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

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

161 through NDF. 

162 """ 

163 image = _make_image() 

164 path = tmp_path / "x.sdf" 

165 write_archive(image, path) 

166 result = read_archive(path) 

167 assert isinstance(result, Image) 

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

169 

170 

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

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

173 reads. 

174 """ 

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

176 path = tmp_path / "x.fits" 

177 write_archive(img, path) 

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

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

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

181 

182 

183def test_read_cls_match() -> None: 

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

185 matches. 

186 """ 

187 path = os.path.join(LOCAL_DATA_DIR, "image.json") 

188 result = read_archive(path, cls=Image) 

189 assert isinstance(result, Image) 

190 

191 

192def test_read_cls_mismatch_raises() -> None: 

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

194 does not match cls. 

195 """ 

196 from lsst.images import Mask 

197 

198 path = os.path.join(LOCAL_DATA_DIR, "image.json") 

199 with pytest.raises(TypeError) as exc_info: 

200 read_archive(path, cls=Mask) 

201 msg = str(exc_info.value) 

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

203 assert "Image" in msg # actual deserialized type 

204 assert "Mask" in msg # requested cls