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 

16import tempfile 

17import unittest 

18 

19from lsst.images import fits as images_fits 

20from lsst.images import json as images_json 

21from lsst.images.fits import FitsInputArchive 

22from lsst.images.json import JsonInputArchive 

23from lsst.images.serialization import ArchiveTree, read_archive 

24 

25 

26@contextlib.contextmanager 

27def count_opens(path: str): 

28 """Count how many times ``path`` is physically opened for reading. 

29 

30 Yields a one-element list whose single entry is the running open count; 

31 read it after the ``with`` block. 

32 """ 

33 count = [0] 

34 real_open = builtins.open 

35 

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

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

38 count[0] += 1 

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

40 

41 builtins.open = counting_open 

42 try: 

43 yield count 

44 finally: 

45 builtins.open = real_open 

46 

47 

48DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "schema_v1") 

49 

50 

51def _visit_image(): 

52 """Load a VisitImage from the committed JSON fixture.""" 

53 return read_archive(os.path.join(DATA_DIR, "visit_image.json")) 

54 

55 

56class FitsOpenTreeTestCase(unittest.TestCase): 

57 """InputArchive.open_tree yields a live (archive, tree) pair.""" 

58 

59 def setUp(self) -> None: 

60 tmp = tempfile.TemporaryDirectory() 

61 self.addCleanup(tmp.cleanup) 

62 self.path = os.path.join(tmp.name, "v.fits") 

63 images_fits.write(_visit_image(), self.path) 

64 

65 def test_open_tree_yields_archive_tree_and_info(self) -> None: 

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

67 self.assertIsInstance(tree, ArchiveTree) 

68 self.assertEqual(info.schema_name, "visit_image") 

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

70 self.assertIsNotNone(proj) 

71 

72 def test_read_still_works(self) -> None: 

73 # read_archive() returns the deserialized object directly, via 

74 # open_archive(). 

75 result = read_archive(self.path) 

76 self.assertEqual(type(result).__name__, "VisitImage") 

77 

78 

79try: 

80 import h5py # noqa: F401 

81 

82 from lsst.images import ndf as images_ndf 

83 from lsst.images.ndf import NdfInputArchive 

84 

85 HAVE_H5PY = True 

86except ImportError: 

87 HAVE_H5PY = False 

88 

89 

90@unittest.skipUnless(HAVE_H5PY, "h5py is not available.") 

91class NdfOpenTreeTestCase(unittest.TestCase): 

92 """open_tree works for the NDF backend.""" 

93 

94 def setUp(self) -> None: 

95 tmp = tempfile.TemporaryDirectory() 

96 self.addCleanup(tmp.cleanup) 

97 self.path = os.path.join(tmp.name, "v.sdf") 

98 images_ndf.write(_visit_image(), self.path) 

99 

100 def test_open_tree_yields_archive_tree_and_info(self) -> None: 

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

102 self.assertIsInstance(tree, ArchiveTree) 

103 self.assertEqual(info.schema_name, "visit_image") 

104 self.assertIsNotNone(tree.deserialize_component("obs_info", archive)) 

105 

106 def test_read_still_works(self) -> None: 

107 self.assertEqual(type(read_archive(self.path)).__name__, "VisitImage") 

108 

109 

110class JsonOpenTreeTestCase(unittest.TestCase): 

111 """open_tree works for the JSON backend.""" 

112 

113 def setUp(self) -> None: 

114 tmp = tempfile.TemporaryDirectory() 

115 self.addCleanup(tmp.cleanup) 

116 self.path = os.path.join(tmp.name, "v.json") 

117 images_json.write(_visit_image(), self.path) 

118 

119 def test_open_tree_yields_archive_tree_and_info(self) -> None: 

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

121 self.assertIsInstance(tree, ArchiveTree) 

122 self.assertEqual(info.schema_name, "visit_image") 

123 self.assertIsNotNone(tree.deserialize_component("sky_projection", archive)) 

124 

125 def test_read_still_works(self) -> None: 

126 self.assertEqual(type(read_archive(self.path)).__name__, "VisitImage") 

127 

128 

129class ReaderApiTestCase(unittest.TestCase): 

130 """The user-facing serialization.open_archive() / Reader interface.""" 

131 

132 def setUp(self) -> None: 

133 tmp = tempfile.TemporaryDirectory() 

134 self.addCleanup(tmp.cleanup) 

135 self.tmp = tmp.name 

136 self.vi = _visit_image() 

137 self.fits = os.path.join(self.tmp, "v.fits") 

138 images_fits.write(self.vi, self.fits) 

139 

140 def _check_components_and_read(self, path: str) -> None: 

141 import lsst.images.serialization as ser 

142 

143 with ser.open_archive(path) as reader: 

144 self.assertIsNotNone(reader.get_component("sky_projection")) 

145 self.assertIsNotNone(reader.get_component("obs_info")) 

146 full = reader.read() 

147 self.assertEqual(type(full).__name__, "VisitImage") 

148 

149 def test_components_and_read_fits(self) -> None: 

150 self._check_components_and_read(self.fits) 

151 

152 def test_components_and_read_json(self) -> None: 

153 path = os.path.join(self.tmp, "v.json") 

154 images_json.write(self.vi, path) 

155 self._check_components_and_read(path) 

156 

157 @unittest.skipUnless(HAVE_H5PY, "h5py is not available.") 

158 def test_components_and_read_ndf(self) -> None: 

159 path = os.path.join(self.tmp, "v.sdf") 

160 images_ndf.write(self.vi, path) 

161 self._check_components_and_read(path) 

162 

163 def test_info(self) -> None: 

164 import lsst.images.serialization as ser 

165 

166 with ser.open_archive(self.fits) as reader: 

167 self.assertEqual(reader.info.schema_name, "visit_image") 

168 self.assertEqual(reader.info.schema_version, "1.0.0") 

169 self.assertIsInstance(reader.metadata, dict) 

170 

171 def test_cls_match(self) -> None: 

172 import lsst.images.serialization as ser 

173 from lsst.images import VisitImage 

174 

175 with ser.open_archive(self.fits, cls=VisitImage) as reader: 

176 self.assertIsInstance(reader.read(), VisitImage) 

177 

178 def test_cls_mismatch_raises(self) -> None: 

179 import lsst.images.serialization as ser 

180 from lsst.images import Mask 

181 

182 with self.assertRaises(TypeError): 

183 with ser.open_archive(self.fits, cls=Mask): 

184 pass 

185 

186 def test_unknown_component(self) -> None: 

187 import lsst.images.serialization as ser 

188 from lsst.images.serialization import InvalidComponentError 

189 

190 with ser.open_archive(self.fits) as reader: 

191 with self.assertRaises(InvalidComponentError): 

192 reader.get_component("does_not_exist") 

193 

194 def test_use_after_close_raises(self) -> None: 

195 import lsst.images.serialization as ser 

196 

197 with ser.open_archive(self.fits) as reader: 

198 pass 

199 with self.assertRaises(RuntimeError): 

200 reader.get_component("sky_projection") 

201 

202 def test_fits_open_reads_file_once(self) -> None: 

203 # open() must not open the file a second time just to read the schema 

204 # from the primary header: the archive it opens already parses that 

205 # header, so the schema is identified from a single open. 

206 import lsst.images.serialization as ser 

207 

208 with count_opens(self.fits) as count: 

209 with ser.open_archive(self.fits) as reader: 

210 reader.get_component("sky_projection") 

211 reader.get_component("obs_info") 

212 self.assertEqual(count[0], 1) 

213 

214 

215if __name__ == "__main__": 

216 unittest.main()