Coverage for tests/test_serialization_backends.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. 

11 

12from __future__ import annotations 

13 

14import gzip 

15import io 

16import os 

17import sys 

18import tempfile 

19import unittest 

20from unittest import mock 

21 

22import numpy as np 

23 

24from lsst.images import Box, Image 

25from lsst.images import fits as images_fits 

26from lsst.images.serialization import Backend, backend_for_name, backend_for_path, backend_for_stream 

27from lsst.images.serialization._backends import ( 

28 _decompress_path_to_temp_file, 

29 _is_binary_stream, 

30 _path_is_compressed, 

31) 

32from lsst.resources import ResourcePath 

33 

34try: 

35 from compression import zstd as _stdlib_zstd # noqa: F401 -- detect zstd availability 

36 

37 ZSTD_AVAILABLE = True 

38except ImportError: 

39 try: 

40 import zstandard # noqa: F401 

41 

42 ZSTD_AVAILABLE = True 

43 except ImportError: 

44 ZSTD_AVAILABLE = False 

45 

46 

47def _zstd_compress(data: bytes) -> bytes: 

48 """Compress with whichever zstd library is available.""" 

49 try: 

50 from compression import zstd 

51 except ImportError: 

52 import zstandard 

53 

54 return zstandard.ZstdCompressor().compress(data) 

55 return zstd.compress(data) 

56 

57 

58class BackendForPathTestCase(unittest.TestCase): 

59 """Tests for suffix -> backend resolution.""" 

60 

61 def test_fits(self) -> None: 

62 from lsst.images.fits import FitsInputArchive 

63 

64 b = backend_for_path("a/b/c.fits") 

65 self.assertIsInstance(b, Backend) 

66 self.assertEqual(b.name, "fits") 

67 self.assertIs(b.input_archive, FitsInputArchive) 

68 self.assertTrue(callable(b.write)) 

69 

70 def test_fits_gz(self) -> None: 

71 self.assertEqual(backend_for_path("c.fits.gz").name, "fits") 

72 self.assertEqual(backend_for_path("file://a/b/c.fits.gz?param=2").name, "fits") 

73 

74 def test_json(self) -> None: 

75 from lsst.images.json import JsonInputArchive 

76 

77 b = backend_for_path("c.json") 

78 self.assertEqual(b.name, "json") 

79 self.assertIs(b.input_archive, JsonInputArchive) 

80 

81 def test_ndf(self) -> None: 

82 self.assertEqual(backend_for_path("c.sdf").name, "ndf") 

83 self.assertEqual(backend_for_path("c.h5").name, "ndf") 

84 

85 def test_unknown(self) -> None: 

86 with self.assertRaises(ValueError) as cm: 

87 backend_for_path("c.txt") 

88 self.assertIn(".fits", str(cm.exception)) 

89 

90 

91class MinifyDispatchTestCase(unittest.TestCase): 

92 """minify resolves backend and schema via the shared APIs.""" 

93 

94 def test_minify_unsupported_schema_uses_shared_dispatch(self) -> None: 

95 from lsst.images.tests._minify_for_fixtures import minify 

96 

97 tmp = tempfile.mkdtemp() 

98 src = os.path.join(tmp, "plain.fits") 

99 out = os.path.join(tmp, "plain.json") 

100 images_fits.write(Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]), src) 

101 # Reaching the "no subsetter" error proves backend_for_path and 

102 # get_basic_info ran and detected schema_name "image". 

103 with self.assertRaises(NotImplementedError) as cm: 

104 minify(src, out) 

105 self.assertIn("image", str(cm.exception)) 

106 

107 

108class BackendForNameTestCase(unittest.TestCase): 

109 """Explicit format-name -> backend resolution.""" 

110 

111 def test_names(self) -> None: 

112 self.assertEqual(backend_for_name("fits").name, "fits") 

113 self.assertEqual(backend_for_name("ndf").name, "ndf") 

114 self.assertEqual(backend_for_name("json").name, "json") 

115 

116 def test_unknown(self) -> None: 

117 with self.assertRaises(ValueError) as cm: 

118 backend_for_name("hdf") 

119 self.assertIn("hdf", str(cm.exception)) 

120 

121 

122class BackendForStreamTestCase(unittest.TestCase): 

123 """Content sniffing -> backend resolution.""" 

124 

125 def test_fits_magic(self) -> None: 

126 stream = io.BytesIO(b"SIMPLE = T / conforms") 

127 self.assertEqual(backend_for_stream(stream).name, "fits") 

128 # Sniffing must not consume the stream. 

129 self.assertEqual(stream.tell(), 0) 

130 

131 def test_hdf5_magic(self) -> None: 

132 stream = io.BytesIO(b"\x89HDF\r\n\x1a\n" + b"\x00" * 8) 

133 self.assertEqual(backend_for_stream(stream).name, "ndf") 

134 

135 def test_json(self) -> None: 

136 self.assertEqual(backend_for_stream(io.BytesIO(b'{"schema_url": "x"}')).name, "json") 

137 

138 def test_json_leading_whitespace(self) -> None: 

139 self.assertEqual(backend_for_stream(io.BytesIO(b' \n\t {"a": 1}')).name, "json") 

140 

141 def test_unknown(self) -> None: 

142 with self.assertRaises(ValueError) as cm: 

143 backend_for_stream(io.BytesIO(b"not any known format")) 

144 msg = str(cm.exception) 

145 self.assertIn("FITS", msg) 

146 self.assertIn("format", msg) 

147 

148 def test_gzip_compressed_stream_raises(self) -> None: 

149 with self.assertRaises(ValueError) as cm: 

150 backend_for_stream(io.BytesIO(gzip.compress(b"SIMPLE = whatever"))) 

151 self.assertIn("gzip", str(cm.exception)) 

152 

153 def test_zstd_compressed_stream_raises(self) -> None: 

154 with self.assertRaises(ValueError) as cm: 

155 backend_for_stream(io.BytesIO(b"\x28\xb5\x2f\xfd" + b"frame")) 

156 self.assertIn("zstd", str(cm.exception)) 

157 

158 

159class BackendForPathCompressionTestCase(unittest.TestCase): 

160 """Compression suffixes are stripped before extension dispatch.""" 

161 

162 def test_gz(self) -> None: 

163 self.assertEqual(backend_for_path("c.json.gz").name, "json") 

164 self.assertEqual(backend_for_path("c.h5.gz").name, "ndf") 

165 

166 def test_zst(self) -> None: 

167 self.assertEqual(backend_for_path("c.fits.zst").name, "fits") 

168 self.assertEqual(backend_for_path("c.sdf.zst").name, "ndf") 

169 

170 def test_bare_compression_suffix(self) -> None: 

171 with self.assertRaises(ValueError): 

172 backend_for_path("c.gz") 

173 

174 

175class StreamHelpersTestCase(unittest.TestCase): 

176 """Stream detection and compression-suffix helpers.""" 

177 

178 def test_is_binary_stream(self) -> None: 

179 self.assertTrue(_is_binary_stream(io.BytesIO(b""))) 

180 self.assertFalse(_is_binary_stream("a.fits")) 

181 # ResourcePath has read() but no seek(); it must not look like a 

182 # stream. 

183 self.assertFalse(_is_binary_stream(ResourcePath("a.fits", forceAbsolute=False))) 

184 

185 def test_path_is_compressed(self) -> None: 

186 self.assertTrue(_path_is_compressed("a/b.fits.gz")) 

187 self.assertTrue(_path_is_compressed("a/b.json.zst")) 

188 self.assertFalse(_path_is_compressed("a/b.fits")) 

189 

190 

191class DecompressPathToTempFileTestCase(unittest.TestCase): 

192 """Compressed paths stream-decompress into a temporary file on disk.""" 

193 

194 def setUp(self) -> None: 

195 tmp = tempfile.TemporaryDirectory() 

196 self.addCleanup(tmp.cleanup) 

197 self.tmp = tmp.name 

198 self.payload = b"SIMPLE = fake fits payload " * 1000 

199 

200 def test_gzip(self) -> None: 

201 path = os.path.join(self.tmp, "x.fits.gz") 

202 with open(path, "wb") as f: 

203 f.write(gzip.compress(self.payload)) 

204 with _decompress_path_to_temp_file(path) as handle: 

205 # The decompressed data lives in a real file, not in memory. 

206 self.assertNotIsInstance(handle, io.BytesIO) 

207 self.assertTrue(hasattr(handle, "fileno")) 

208 self.assertEqual(handle.read(), self.payload) 

209 

210 @unittest.skipUnless(ZSTD_AVAILABLE, "no zstd decompressor available.") 

211 def test_zstd(self) -> None: 

212 path = os.path.join(self.tmp, "x.fits.zst") 

213 with open(path, "wb") as f: 

214 f.write(_zstd_compress(self.payload)) 

215 with _decompress_path_to_temp_file(path) as handle: 

216 self.assertEqual(handle.read(), self.payload) 

217 

218 def test_suffix_without_compression_raises(self) -> None: 

219 # The suffix selects the decompressor, so a .gz name whose content 

220 # is not gzip fails honestly. 

221 path = os.path.join(self.tmp, "x.fits.gz") 

222 with open(path, "wb") as f: 

223 f.write(self.payload) 

224 with self.assertRaises(gzip.BadGzipFile): 

225 _decompress_path_to_temp_file(path) 

226 

227 def test_zstd_no_decompressor(self) -> None: 

228 path = os.path.join(self.tmp, "x.fits.zst") 

229 with open(path, "wb") as f: 

230 f.write(b"\x28\xb5\x2f\xfd" + b"pretend zstd frame") 

231 # None entries make both import routes raise ImportError. 

232 blocked = {"compression": None, "compression.zstd": None, "zstandard": None} 

233 with mock.patch.dict(sys.modules, blocked): 

234 with self.assertRaises(ValueError) as cm: 

235 _decompress_path_to_temp_file(path) 

236 self.assertIn("zstd", str(cm.exception)) 

237 

238 

239if __name__ == "__main__": 

240 unittest.main()