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 

18from pathlib import Path 

19from unittest import mock 

20 

21import numpy as np 

22import pytest 

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 

58def test_backend_for_path_fits() -> None: 

59 """Verify that a .fits path resolves to the FITS backend.""" 

60 from lsst.images.fits import FitsInputArchive 

61 

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

63 assert isinstance(b, Backend) 

64 assert b.name == "fits" 

65 assert b.input_archive is FitsInputArchive 

66 assert callable(b.write) 

67 

68 

69def test_backend_for_path_fits_gz() -> None: 

70 """Verify that .fits.gz and URL-form .fits.gz both resolve to the 

71 FITS backend. 

72 """ 

73 assert backend_for_path("c.fits.gz").name == "fits" 

74 assert backend_for_path("file://a/b/c.fits.gz?param=2").name == "fits" 

75 

76 

77def test_backend_for_path_json() -> None: 

78 """Verify that a .json path resolves to the JSON backend.""" 

79 from lsst.images.json import JsonInputArchive 

80 

81 b = backend_for_path("c.json") 

82 assert b.name == "json" 

83 assert b.input_archive is JsonInputArchive 

84 

85 

86def test_backend_for_path_ndf() -> None: 

87 """Verify that .sdf and .h5 paths both resolve to the NDF backend.""" 

88 assert backend_for_path("c.sdf").name == "ndf" 

89 assert backend_for_path("c.h5").name == "ndf" 

90 

91 

92def test_backend_for_path_unknown_raises() -> None: 

93 """Verify that an unrecognised suffix raises ValueError naming known 

94 formats. 

95 """ 

96 with pytest.raises(ValueError) as exc_info: 

97 backend_for_path("c.txt") 

98 assert ".fits" in str(exc_info.value) 

99 

100 

101def test_minify_unsupported_schema_uses_shared_dispatch(tmp_path: Path) -> None: 

102 """Verify minify resolves backend and schema via the shared APIs. 

103 

104 Reaching the 'no subsetter' error proves backend_for_path and 

105 get_basic_info ran and detected schema_name 'image'. 

106 """ 

107 from lsst.images.tests._minify_for_fixtures import minify 

108 

109 src = os.path.join(tmp_path, "plain.fits") 

110 out = os.path.join(tmp_path, "plain.json") 

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

112 with pytest.raises(NotImplementedError) as exc_info: 

113 minify(src, out) 

114 assert "image" in str(exc_info.value) 

115 

116 

117def test_backend_for_name() -> None: 

118 """Verify explicit format names resolve to their backends.""" 

119 assert backend_for_name("fits").name == "fits" 

120 assert backend_for_name("ndf").name == "ndf" 

121 assert backend_for_name("json").name == "json" 

122 

123 

124def test_backend_for_name_unknown_raises() -> None: 

125 """Verify an unknown format name raises ValueError naming it.""" 

126 with pytest.raises(ValueError) as exc_info: 

127 backend_for_name("hdf") 

128 assert "hdf" in str(exc_info.value) 

129 

130 

131def test_backend_for_stream_fits_magic() -> None: 

132 """Verify FITS magic bytes resolve to the FITS backend without consuming 

133 the stream. 

134 """ 

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

136 assert backend_for_stream(stream).name == "fits" 

137 # Sniffing must not consume the stream. 

138 assert stream.tell() == 0 

139 

140 

141def test_backend_for_stream_hdf5_magic() -> None: 

142 """Verify the HDF5 signature resolves to the NDF backend.""" 

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

144 assert backend_for_stream(stream).name == "ndf" 

145 

146 

147def test_backend_for_stream_json() -> None: 

148 """Verify JSON content resolves to the JSON backend.""" 

149 assert backend_for_stream(io.BytesIO(b'{"schema_url": "x"}')).name == "json" 

150 

151 

152def test_backend_for_stream_json_leading_whitespace() -> None: 

153 """Verify JSON detection tolerates leading whitespace.""" 

154 assert backend_for_stream(io.BytesIO(b' \n\t {"a": 1}')).name == "json" 

155 

156 

157def test_backend_for_stream_unknown_raises() -> None: 

158 """Verify unrecognized content raises ValueError naming known formats.""" 

159 with pytest.raises(ValueError) as exc_info: 

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

161 msg = str(exc_info.value) 

162 assert "FITS" in msg 

163 assert "format" in msg 

164 

165 

166def test_backend_for_stream_gzip_compressed_raises() -> None: 

167 """Verify a gzip-compressed stream raises ValueError naming gzip.""" 

168 with pytest.raises(ValueError) as exc_info: 

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

170 assert "gzip" in str(exc_info.value) 

171 

172 

173def test_backend_for_stream_zstd_compressed_raises() -> None: 

174 """Verify a zstd-compressed stream raises ValueError naming zstd.""" 

175 with pytest.raises(ValueError) as exc_info: 

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

177 assert "zstd" in str(exc_info.value) 

178 

179 

180def test_backend_for_path_strips_gz() -> None: 

181 """Verify .gz compression suffixes are stripped before extension 

182 dispatch. 

183 """ 

184 assert backend_for_path("c.json.gz").name == "json" 

185 assert backend_for_path("c.h5.gz").name == "ndf" 

186 

187 

188def test_backend_for_path_strips_zst() -> None: 

189 """Verify .zst compression suffixes are stripped before extension 

190 dispatch. 

191 """ 

192 assert backend_for_path("c.fits.zst").name == "fits" 

193 assert backend_for_path("c.sdf.zst").name == "ndf" 

194 

195 

196def test_backend_for_path_bare_compression_suffix_raises() -> None: 

197 """Verify a bare compression suffix with no format extension raises 

198 ValueError. 

199 """ 

200 with pytest.raises(ValueError): 

201 backend_for_path("c.gz") 

202 

203 

204def test_is_binary_stream() -> None: 

205 """Verify _is_binary_stream accepts streams and rejects path-like 

206 inputs. 

207 """ 

208 assert _is_binary_stream(io.BytesIO(b"")) 

209 assert not _is_binary_stream("a.fits") 

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

211 # stream. 

212 assert not _is_binary_stream(ResourcePath("a.fits", forceAbsolute=False)) 

213 

214 

215def test_path_is_compressed() -> None: 

216 """Verify _path_is_compressed recognizes .gz and .zst suffixes only.""" 

217 assert _path_is_compressed("a/b.fits.gz") 

218 assert _path_is_compressed("a/b.json.zst") 

219 assert not _path_is_compressed("a/b.fits") 

220 

221 

222_DECOMPRESS_PAYLOAD = b"SIMPLE = fake fits payload " * 1000 

223 

224 

225def test_decompress_path_gzip(tmp_path: Path) -> None: 

226 """Verify a .gz path stream-decompresses into a real temporary file.""" 

227 path = tmp_path / "x.fits.gz" 

228 path.write_bytes(gzip.compress(_DECOMPRESS_PAYLOAD)) 

229 with _decompress_path_to_temp_file(str(path)) as handle: 

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

231 assert not isinstance(handle, io.BytesIO) 

232 assert hasattr(handle, "fileno") 

233 assert handle.read() == _DECOMPRESS_PAYLOAD 

234 

235 

236@pytest.mark.skipif(not ZSTD_AVAILABLE, reason="no zstd decompressor available") 

237def test_decompress_path_zstd(tmp_path: Path) -> None: 

238 """Verify a .zst path decompresses through the available zstd library.""" 

239 path = tmp_path / "x.fits.zst" 

240 path.write_bytes(_zstd_compress(_DECOMPRESS_PAYLOAD)) 

241 with _decompress_path_to_temp_file(str(path)) as handle: 

242 assert handle.read() == _DECOMPRESS_PAYLOAD 

243 

244 

245def test_decompress_suffix_without_compression_raises(tmp_path: Path) -> None: 

246 """Verify a .gz name whose content is not gzip fails honestly.""" 

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

248 # is not gzip fails honestly. 

249 path = tmp_path / "x.fits.gz" 

250 path.write_bytes(_DECOMPRESS_PAYLOAD) 

251 with pytest.raises(gzip.BadGzipFile): 

252 _decompress_path_to_temp_file(str(path)) 

253 

254 

255def test_decompress_zstd_no_decompressor(tmp_path: Path) -> None: 

256 """Verify a helpful ValueError when no zstd decompressor can be 

257 imported. 

258 """ 

259 path = tmp_path / "x.fits.zst" 

260 path.write_bytes(b"\x28\xb5\x2f\xfd" + b"pretend zstd frame") 

261 # None entries make both import routes raise ImportError. 

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

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

264 with pytest.raises(ValueError) as exc_info: 

265 _decompress_path_to_temp_file(str(path)) 

266 assert "zstd" in str(exc_info.value)