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 

26from lsst.images.tests import current_fixture_path 

27 

28try: 

29 import h5py # noqa: F401 

30 

31 from lsst.images import ndf as images_ndf 

32 from lsst.images.ndf import NdfInputArchive 

33 

34 HAVE_H5PY = True 

35except ImportError: 

36 HAVE_H5PY = False 

37 

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

39 

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

41 

42 

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

44def visit_image() -> VisitImage: 

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

46 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

47 return read_archive(path) # type: ignore[return-value] 

48 

49 

50@contextlib.contextmanager 

51def count_opens(path: Path | str): 

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

53 opened. 

54 """ 

55 count = [0] 

56 real_open = builtins.open 

57 

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

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

60 count[0] += 1 

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

62 

63 builtins.open = counting_open 

64 try: 

65 yield count 

66 finally: 

67 builtins.open = real_open 

68 

69 

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

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

72 triple. 

73 """ 

74 path = tmp_path / "v.fits" 

75 images_fits.write(visit_image, path) 

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

77 assert isinstance(tree, ArchiveTree) 

78 assert info.schema_name == "visit_image" 

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

80 assert proj is not None 

81 

82 

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

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

85 from a FITS file. 

86 """ 

87 path = tmp_path / "v.fits" 

88 images_fits.write(visit_image, path) 

89 result = read_archive(path) 

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

91 

92 

93@skip_no_h5py 

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

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

96 triple. 

97 """ 

98 path = tmp_path / "v.sdf" 

99 images_ndf.write(visit_image, path) 

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

101 assert isinstance(tree, ArchiveTree) 

102 assert info.schema_name == "visit_image" 

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

104 

105 

106@skip_no_h5py 

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

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

109 from an NDF file. 

110 """ 

111 path = tmp_path / "v.sdf" 

112 images_ndf.write(visit_image, path) 

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

114 

115 

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

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

118 triple. 

119 """ 

120 path = tmp_path / "v.json" 

121 images_json.write(visit_image, path) 

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

123 assert isinstance(tree, ArchiveTree) 

124 assert info.schema_name == "visit_image" 

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

126 

127 

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

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

130 from a JSON file. 

131 """ 

132 path = tmp_path / "v.json" 

133 images_json.write(visit_image, path) 

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

135 

136 

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

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

139 full read on ``path``. 

140 """ 

141 import lsst.images.serialization as ser 

142 

143 with ser.open_archive(path) as reader: 

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

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

146 full = reader.read() 

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

148 

149 

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

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

152 path = tmp_path / "v.fits" 

153 images_fits.write(visit_image, path) 

154 _check_components_and_read(path) 

155 

156 

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

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

159 path = tmp_path / "v.json" 

160 images_json.write(visit_image, path) 

161 _check_components_and_read(path) 

162 

163 

164@skip_no_h5py 

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

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

167 path = tmp_path / "v.sdf" 

168 images_ndf.write(visit_image, path) 

169 _check_components_and_read(path) 

170 

171 

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

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

174 import lsst.images.serialization as ser 

175 

176 path = tmp_path / "v.fits" 

177 images_fits.write(visit_image, path) 

178 with ser.open_archive(path) as reader: 

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

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

181 assert isinstance(reader.metadata, dict) 

182 

183 

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

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

186 import lsst.images.serialization as ser 

187 from lsst.images import VisitImage 

188 

189 path = tmp_path / "v.fits" 

190 images_fits.write(visit_image, path) 

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

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

193 

194 

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

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

197 schema. 

198 """ 

199 import lsst.images.serialization as ser 

200 from lsst.images import Mask 

201 

202 path = tmp_path / "v.fits" 

203 images_fits.write(visit_image, path) 

204 with pytest.raises(TypeError): 

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

206 pass 

207 

208 

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

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

211 component name. 

212 """ 

213 import lsst.images.serialization as ser 

214 from lsst.images.serialization import InvalidComponentError 

215 

216 path = tmp_path / "v.fits" 

217 images_fits.write(visit_image, path) 

218 with ser.open_archive(path) as reader: 

219 with pytest.raises(InvalidComponentError): 

220 reader.get_component("does_not_exist") 

221 

222 

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

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

225 import lsst.images.serialization as ser 

226 

227 path = tmp_path / "v.fits" 

228 images_fits.write(visit_image, path) 

229 with ser.open_archive(path) as reader: 

230 pass 

231 with pytest.raises(RuntimeError): 

232 reader.get_component("sky_projection") 

233 

234 

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

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

237 regardless of component reads. 

238 """ 

239 import lsst.images.serialization as ser 

240 

241 path = tmp_path / "v.fits" 

242 images_fits.write(visit_image, path) 

243 with count_opens(path) as count: 

244 with ser.open_archive(path) as reader: 

245 reader.get_component("sky_projection") 

246 reader.get_component("obs_info") 

247 assert count[0] == 1