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 

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(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() returns the deserialized object directly, via open(). 

74 result = read(self.path) 

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

76 

77 

78try: 

79 import h5py # noqa: F401 

80 

81 from lsst.images import ndf as images_ndf 

82 from lsst.images.ndf import NdfInputArchive 

83 

84 HAVE_H5PY = True 

85except ImportError: 

86 HAVE_H5PY = False 

87 

88 

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

90class NdfOpenTreeTestCase(unittest.TestCase): 

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

92 

93 def setUp(self) -> None: 

94 tmp = tempfile.TemporaryDirectory() 

95 self.addCleanup(tmp.cleanup) 

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

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

98 

99 def test_open_tree_yields_archive_tree_and_info(self) -> None: 

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

101 self.assertIsInstance(tree, ArchiveTree) 

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

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

104 

105 def test_read_still_works(self) -> None: 

106 self.assertEqual(type(read(self.path)).__name__, "VisitImage") 

107 

108 

109class JsonOpenTreeTestCase(unittest.TestCase): 

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

111 

112 def setUp(self) -> None: 

113 tmp = tempfile.TemporaryDirectory() 

114 self.addCleanup(tmp.cleanup) 

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

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

117 

118 def test_open_tree_yields_archive_tree_and_info(self) -> None: 

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

120 self.assertIsInstance(tree, ArchiveTree) 

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

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

123 

124 def test_read_still_works(self) -> None: 

125 self.assertEqual(type(read(self.path)).__name__, "VisitImage") 

126 

127 

128class ReaderApiTestCase(unittest.TestCase): 

129 """The user-facing serialization.open() / Reader interface.""" 

130 

131 def setUp(self) -> None: 

132 tmp = tempfile.TemporaryDirectory() 

133 self.addCleanup(tmp.cleanup) 

134 self.tmp = tmp.name 

135 self.vi = _visit_image() 

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

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

138 

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

140 import lsst.images.serialization as ser 

141 

142 with ser.open(path) as reader: 

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

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

145 full = reader.read() 

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

147 

148 def test_components_and_read_fits(self) -> None: 

149 self._check_components_and_read(self.fits) 

150 

151 def test_components_and_read_json(self) -> None: 

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

153 images_json.write(self.vi, path) 

154 self._check_components_and_read(path) 

155 

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

157 def test_components_and_read_ndf(self) -> None: 

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

159 images_ndf.write(self.vi, path) 

160 self._check_components_and_read(path) 

161 

162 def test_info(self) -> None: 

163 import lsst.images.serialization as ser 

164 

165 with ser.open(self.fits) as reader: 

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

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

168 self.assertIsInstance(reader.metadata, dict) 

169 

170 def test_cls_match(self) -> None: 

171 import lsst.images.serialization as ser 

172 from lsst.images import VisitImage 

173 

174 with ser.open(self.fits, cls=VisitImage) as reader: 

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

176 

177 def test_cls_mismatch_raises(self) -> None: 

178 import lsst.images.serialization as ser 

179 from lsst.images import Mask 

180 

181 with self.assertRaises(TypeError): 

182 with ser.open(self.fits, cls=Mask): 

183 pass 

184 

185 def test_unknown_component(self) -> None: 

186 import lsst.images.serialization as ser 

187 from lsst.images.serialization import InvalidComponentError 

188 

189 with ser.open(self.fits) as reader: 

190 with self.assertRaises(InvalidComponentError): 

191 reader.get_component("does_not_exist") 

192 

193 def test_use_after_close_raises(self) -> None: 

194 import lsst.images.serialization as ser 

195 

196 with ser.open(self.fits) as reader: 

197 pass 

198 with self.assertRaises(RuntimeError): 

199 reader.get_component("sky_projection") 

200 

201 def test_fits_open_reads_file_once(self) -> None: 

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

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

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

205 import lsst.images.serialization as ser 

206 

207 with count_opens(self.fits) as count: 

208 with ser.open(self.fits) as reader: 

209 reader.get_component("sky_projection") 

210 reader.get_component("obs_info") 

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

212 

213 

214if __name__ == "__main__": 

215 unittest.main()