Coverage for tests/test_serialization_reader.py: 97%

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 builtins 

14import contextlib 

15import os 

16from pathlib import Path 

17 

18import pytest 

19 

20from lsst.images import VisitImage 

21from lsst.images import fits as images_fits 

22from lsst.images import json as images_json 

23from lsst.images.fits import FitsInputArchive 

24from lsst.images.json import JsonInputArchive 

25from lsst.images.serialization import ArchiveTree, read_archive 

26 

27try: 

28 import h5py # noqa: F401 

29 

30 from lsst.images import ndf as images_ndf 

31 from lsst.images.ndf import NdfInputArchive 

32 

33 HAVE_H5PY = True 

34except ImportError: 

35 HAVE_H5PY = False 

36 

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

38 

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

40 

41 

42@pytest.fixture(scope="session") 

43def visit_image() -> VisitImage: 

44 """Return a VisitImage loaded once from the committed JSON fixture.""" 

45 return read_archive(os.path.join(LOCAL_DATA_DIR, "visit_image.json")) # type: ignore[return-value] 

46 

47 

48@contextlib.contextmanager 

49def count_opens(path: Path | str): 

50 """Return a one-element list that counts how many times ``path`` is 

51 opened. 

52 """ 

53 count = [0] 

54 real_open = builtins.open 

55 

56 def counting_open(file, *args, **kwargs): 

57 if isinstance(file, (str, bytes, os.PathLike)) and os.fspath(file) == os.fspath(path): 57 ↛ 59line 57 didn't jump to line 59 because the condition on line 57 was always true

58 count[0] += 1 

59 return real_open(file, *args, **kwargs) 

60 

61 builtins.open = counting_open 

62 try: 

63 yield count 

64 finally: 

65 builtins.open = real_open 

66 

67 

68def test_fits_open_tree_yields_archive_tree_and_info(visit_image: VisitImage, tmp_path: Path) -> None: 

69 """Verify FitsInputArchive.open_tree yields a live archive/tree/info 

70 triple. 

71 """ 

72 path = tmp_path / "v.fits" 

73 images_fits.write(visit_image, path) 

74 with FitsInputArchive.open_tree(path) as (archive, tree, info): 

75 assert isinstance(tree, ArchiveTree) 

76 assert info.schema_name == "visit_image" 

77 proj = tree.deserialize_component("sky_projection", archive) 

78 assert proj is not None 

79 

80 

81def test_fits_read_still_works(visit_image: VisitImage, tmp_path: Path) -> None: 

82 """Verify read_archive() returns the deserialized object directly 

83 from a FITS file. 

84 """ 

85 path = tmp_path / "v.fits" 

86 images_fits.write(visit_image, path) 

87 result = read_archive(path) 

88 assert type(result).__name__ == "VisitImage" 

89 

90 

91@skip_no_h5py 

92def test_ndf_open_tree_yields_archive_tree_and_info(visit_image: VisitImage, tmp_path: Path) -> None: 

93 """Verify NdfInputArchive.open_tree yields a live archive/tree/info 

94 triple. 

95 """ 

96 path = tmp_path / "v.sdf" 

97 images_ndf.write(visit_image, path) 

98 with NdfInputArchive.open_tree(path) as (archive, tree, info): 

99 assert isinstance(tree, ArchiveTree) 

100 assert info.schema_name == "visit_image" 

101 assert tree.deserialize_component("obs_info", archive) is not None 

102 

103 

104@skip_no_h5py 

105def test_ndf_read_still_works(visit_image: VisitImage, tmp_path: Path) -> None: 

106 """Verify read_archive() returns the deserialized object directly 

107 from an NDF file. 

108 """ 

109 path = tmp_path / "v.sdf" 

110 images_ndf.write(visit_image, path) 

111 assert type(read_archive(path)).__name__ == "VisitImage" 

112 

113 

114def test_json_open_tree_yields_archive_tree_and_info(visit_image: VisitImage, tmp_path: Path) -> None: 

115 """Verify JsonInputArchive.open_tree yields a live archive/tree/info 

116 triple. 

117 """ 

118 path = tmp_path / "v.json" 

119 images_json.write(visit_image, path) 

120 with JsonInputArchive.open_tree(path) as (archive, tree, info): 

121 assert isinstance(tree, ArchiveTree) 

122 assert info.schema_name == "visit_image" 

123 assert tree.deserialize_component("sky_projection", archive) is not None 

124 

125 

126def test_json_read_still_works(visit_image: VisitImage, tmp_path: Path) -> None: 

127 """Verify read_archive() returns the deserialized object directly 

128 from a JSON file. 

129 """ 

130 path = tmp_path / "v.json" 

131 images_json.write(visit_image, path) 

132 assert type(read_archive(path)).__name__ == "VisitImage" 

133 

134 

135def _check_components_and_read(path: Path | str) -> None: 

136 """Assert that serialization.open_archive() exposes components and a 

137 full read on ``path``. 

138 """ 

139 import lsst.images.serialization as ser 

140 

141 with ser.open_archive(path) as reader: 

142 assert reader.get_component("sky_projection") is not None 

143 assert reader.get_component("obs_info") is not None 

144 full = reader.read() 

145 assert type(full).__name__ == "VisitImage" 

146 

147 

148def test_reader_api_components_and_read_fits(visit_image: VisitImage, tmp_path: Path) -> None: 

149 """Verify the Reader API exposes components and a full read for FITS.""" 

150 path = tmp_path / "v.fits" 

151 images_fits.write(visit_image, path) 

152 _check_components_and_read(path) 

153 

154 

155def test_reader_api_components_and_read_json(visit_image: VisitImage, tmp_path: Path) -> None: 

156 """Verify the Reader API exposes components and a full read for JSON.""" 

157 path = tmp_path / "v.json" 

158 images_json.write(visit_image, path) 

159 _check_components_and_read(path) 

160 

161 

162@skip_no_h5py 

163def test_reader_api_components_and_read_ndf(visit_image: VisitImage, tmp_path: Path) -> None: 

164 """Verify the Reader API exposes components and a full read for NDF.""" 

165 path = tmp_path / "v.sdf" 

166 images_ndf.write(visit_image, path) 

167 _check_components_and_read(path) 

168 

169 

170def test_reader_api_info(visit_image: VisitImage, tmp_path: Path) -> None: 

171 """Verify the Reader API exposes correct schema info for a FITS file.""" 

172 import lsst.images.serialization as ser 

173 

174 path = tmp_path / "v.fits" 

175 images_fits.write(visit_image, path) 

176 with ser.open_archive(path) as reader: 

177 assert reader.info.schema_name == "visit_image" 

178 assert reader.info.schema_version == "1.0.0" 

179 assert isinstance(reader.metadata, dict) 

180 

181 

182def test_reader_api_cls_match(visit_image: VisitImage, tmp_path: Path) -> None: 

183 """Verify the Reader API accepts cls= when the type matches.""" 

184 import lsst.images.serialization as ser 

185 from lsst.images import VisitImage 

186 

187 path = tmp_path / "v.fits" 

188 images_fits.write(visit_image, path) 

189 with ser.open_archive(path, cls=VisitImage) as reader: 

190 assert isinstance(reader.read(), VisitImage) 

191 

192 

193def test_reader_api_cls_mismatch_raises(visit_image: VisitImage, tmp_path: Path) -> None: 

194 """Verify the Reader API raises TypeError when cls= does not match the 

195 schema. 

196 """ 

197 import lsst.images.serialization as ser 

198 from lsst.images import Mask 

199 

200 path = tmp_path / "v.fits" 

201 images_fits.write(visit_image, path) 

202 with pytest.raises(TypeError): 

203 with ser.open_archive(path, cls=Mask): 

204 pass 

205 

206 

207def test_reader_api_unknown_component(visit_image: VisitImage, tmp_path: Path) -> None: 

208 """Verify the Reader API raises InvalidComponentError for an unknown 

209 component name. 

210 """ 

211 import lsst.images.serialization as ser 

212 from lsst.images.serialization import InvalidComponentError 

213 

214 path = tmp_path / "v.fits" 

215 images_fits.write(visit_image, path) 

216 with ser.open_archive(path) as reader: 

217 with pytest.raises(InvalidComponentError): 

218 reader.get_component("does_not_exist") 

219 

220 

221def test_reader_api_use_after_close_raises(visit_image: VisitImage, tmp_path: Path) -> None: 

222 """Verify the Reader API raises RuntimeError when used after closing.""" 

223 import lsst.images.serialization as ser 

224 

225 path = tmp_path / "v.fits" 

226 images_fits.write(visit_image, path) 

227 with ser.open_archive(path) as reader: 

228 pass 

229 with pytest.raises(RuntimeError): 

230 reader.get_component("sky_projection") 

231 

232 

233def test_fits_open_reads_file_once(visit_image: VisitImage, tmp_path: Path) -> None: 

234 """Verify serialization.open_archive() opens a FITS file exactly once 

235 regardless of component reads. 

236 """ 

237 import lsst.images.serialization as ser 

238 

239 path = tmp_path / "v.fits" 

240 images_fits.write(visit_image, path) 

241 with count_opens(path) as count: 

242 with ser.open_archive(path) as reader: 

243 reader.get_component("sky_projection") 

244 reader.get_component("obs_info") 

245 assert count[0] == 1