Coverage for tests/test_serialization_reader.py: 34%

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 

14import tempfile 

15import unittest 

16 

17from lsst.images import fits as images_fits 

18from lsst.images import json as images_json 

19from lsst.images.fits import FitsInputArchive 

20from lsst.images.json import JsonInputArchive 

21from lsst.images.serialization import ArchiveTree, backend_for_path, class_for_schema, read 

22 

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

24 

25 

26def _visit_image(): 

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

28 return read(os.path.join(DATA_DIR, "visit_image.json")) 

29 

30 

31class FitsOpenTreeTestCase(unittest.TestCase): 

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

33 

34 def setUp(self) -> None: 

35 tmp = tempfile.TemporaryDirectory() 

36 self.addCleanup(tmp.cleanup) 

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

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

39 

40 def test_open_tree_yields_archive_and_tree(self) -> None: 

41 info = backend_for_path(self.path).input_archive.get_basic_info(self.path) 

42 tree_cls = class_for_schema(info.schema_name) 

43 assert tree_cls is not None 

44 with FitsInputArchive.open_tree(self.path, tree_cls) as (archive, tree): 

45 self.assertIsInstance(tree, ArchiveTree) 

46 proj = tree.deserialize_component("projection", archive) 

47 self.assertIsNotNone(proj) 

48 

49 def test_read_still_works(self) -> None: 

50 # read() returns the deserialized object directly, via open(). 

51 result = read(self.path) 

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

53 

54 

55try: 

56 import h5py # noqa: F401 

57 

58 from lsst.images import ndf as images_ndf 

59 from lsst.images.ndf import NdfInputArchive 

60 

61 HAVE_H5PY = True 

62except ImportError: 

63 HAVE_H5PY = False 

64 

65 

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

67class NdfOpenTreeTestCase(unittest.TestCase): 

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

69 

70 def setUp(self) -> None: 

71 tmp = tempfile.TemporaryDirectory() 

72 self.addCleanup(tmp.cleanup) 

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

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

75 

76 def test_open_tree_yields_archive_and_tree(self) -> None: 

77 info = backend_for_path(self.path).input_archive.get_basic_info(self.path) 

78 tree_cls = class_for_schema(info.schema_name) 

79 assert tree_cls is not None 

80 with NdfInputArchive.open_tree(self.path, tree_cls) as (archive, tree): 

81 self.assertIsInstance(tree, ArchiveTree) 

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

83 

84 def test_read_still_works(self) -> None: 

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

86 

87 

88class JsonOpenTreeTestCase(unittest.TestCase): 

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

90 

91 def setUp(self) -> None: 

92 tmp = tempfile.TemporaryDirectory() 

93 self.addCleanup(tmp.cleanup) 

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

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

96 

97 def test_open_tree_yields_archive_and_tree(self) -> None: 

98 info = backend_for_path(self.path).input_archive.get_basic_info(self.path) 

99 tree_cls = class_for_schema(info.schema_name) 

100 assert tree_cls is not None 

101 with JsonInputArchive.open_tree(self.path, tree_cls) as (archive, tree): 

102 self.assertIsInstance(tree, ArchiveTree) 

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

104 

105 def test_read_still_works(self) -> None: 

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

107 

108 

109class ReaderApiTestCase(unittest.TestCase): 

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

111 

112 def setUp(self) -> None: 

113 tmp = tempfile.TemporaryDirectory() 

114 self.addCleanup(tmp.cleanup) 

115 self.tmp = tmp.name 

116 self.vi = _visit_image() 

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

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

119 

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

121 import lsst.images.serialization as ser 

122 

123 with ser.open(path) as reader: 

124 self.assertIsNotNone(reader.get_component("projection")) 

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

126 full = reader.read() 

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

128 

129 def test_components_and_read_fits(self) -> None: 

130 self._check_components_and_read(self.fits) 

131 

132 def test_components_and_read_json(self) -> None: 

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

134 images_json.write(self.vi, path) 

135 self._check_components_and_read(path) 

136 

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

138 def test_components_and_read_ndf(self) -> None: 

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

140 images_ndf.write(self.vi, path) 

141 self._check_components_and_read(path) 

142 

143 def test_info(self) -> None: 

144 import lsst.images.serialization as ser 

145 

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

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

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

149 self.assertIsInstance(reader.metadata, dict) 

150 

151 def test_cls_match(self) -> None: 

152 import lsst.images.serialization as ser 

153 from lsst.images import VisitImage 

154 

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

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

157 

158 def test_cls_mismatch_raises(self) -> None: 

159 import lsst.images.serialization as ser 

160 from lsst.images import Mask 

161 

162 with self.assertRaises(TypeError): 

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

164 pass 

165 

166 def test_unknown_component(self) -> None: 

167 import lsst.images.serialization as ser 

168 from lsst.images.serialization import InvalidComponentError 

169 

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

171 with self.assertRaises(InvalidComponentError): 

172 reader.get_component("does_not_exist") 

173 

174 def test_use_after_close_raises(self) -> None: 

175 import lsst.images.serialization as ser 

176 

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

178 pass 

179 with self.assertRaises(RuntimeError): 

180 reader.get_component("projection") 

181 

182 

183if __name__ == "__main__": 

184 unittest.main()