Coverage for tests/test_serialization_basic_info.py: 98%

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 json 

14from pathlib import Path 

15 

16import numpy as np 

17import pydantic 

18import pytest 

19 

20from lsst.images import Box, Image 

21from lsst.images import fits as images_fits 

22from lsst.images import json as images_json 

23from lsst.images.fits import FitsInputArchive 

24from lsst.images.json import JsonInputArchive 

25from lsst.images.serialization import ArchiveInfo, InputArchive 

26 

27try: 

28 import h5py 

29 

30 from lsst.images import ndf as images_ndf 

31 from lsst.images.ndf import NdfInputArchive 

32 

33 HAVE_H5PY = True 

34except ImportError: 

35 HAVE_H5PY = False 

36 

37skip_no_h5py = pytest.mark.skipif(not HAVE_H5PY, reason="h5py is not installed") 

38 

39 

40def _make_image() -> Image: 

41 """Return a small float32 Image for serialization tests.""" 

42 return Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]) 

43 

44 

45def test_archive_info_from_schema_url() -> None: 

46 """Verify ArchiveInfo.from_schema_url parses schema name, version, and 

47 format correctly. 

48 """ 

49 info = ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/visit_image-1.2.3", format_version=1) 

50 assert info.schema_name == "visit_image" 

51 assert info.schema_version == "1.2.3" 

52 assert info.schema_url == "https://images.lsst.io/schemas/visit_image-1.2.3" 

53 assert info.format_version == 1 

54 

55 

56def test_archive_info_from_schema_url_none_format() -> None: 

57 """Verify ArchiveInfo.from_schema_url accepts None for format_version.""" 

58 info = ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/image-1.0.0", format_version=None) 

59 assert info.schema_name == "image" 

60 assert info.format_version is None 

61 

62 

63def test_archive_info_frozen() -> None: 

64 """Verify ArchiveInfo is frozen (schema_name cannot be reassigned).""" 

65 info = ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/image-1.0.0", format_version=None) 

66 with pytest.raises(pydantic.ValidationError): 

67 info.schema_name = "other" # type: ignore[misc] 

68 

69 

70def test_archive_info_from_schema_url_invalid() -> None: 

71 """Verify ArchiveInfo.from_schema_url raises ValueError for URLs without 

72 a version. 

73 """ 

74 with pytest.raises(ValueError): 

75 ArchiveInfo.from_schema_url("https://images.lsst.io/schemas/noversion", format_version=None) 

76 

77 

78def test_archive_info_from_schema_url_foreign_host() -> None: 

79 """Verify ArchiveInfo.from_schema_url accepts third-party hosts whose 

80 URLs follow the ``.../schemas/{name}-{version}`` shape, since external 

81 packages mint schema URLs under their own documentation sites. 

82 """ 

83 info = ArchiveInfo.from_schema_url( 

84 "https://example.org/products/schemas/extended_psf-1.2.0", format_version=None 

85 ) 

86 assert info.schema_name == "extended_psf" 

87 assert info.schema_version == "1.2.0" 

88 

89 

90def test_archive_info_from_schema_url_not_schema_shaped() -> None: 

91 """Verify ArchiveInfo.from_schema_url rejects values that do not have 

92 the schema URL shape, such as a DATAMODL header written by an unrelated 

93 tool. 

94 """ 

95 with pytest.raises(ValueError): 

96 # No "schemas" parent path segment. 

97 ArchiveInfo.from_schema_url("https://example.org/image-1.0.0", format_version=None) 

98 with pytest.raises(ValueError): 

99 # Not an http(s) URL at all. 

100 ArchiveInfo.from_schema_url("IMAGE", format_version=None) 

101 with pytest.raises(ValueError): 

102 # No host. 

103 ArchiveInfo.from_schema_url("https:///schemas/image-1.0.0", format_version=None) 

104 

105 

106def test_input_archive_get_basic_info_base_raises() -> None: 

107 """Verify InputArchive.get_basic_info raises NotImplementedError on the 

108 base class. 

109 """ 

110 with pytest.raises(NotImplementedError): 

111 InputArchive.get_basic_info("x.fits") 

112 

113 

114def test_json_basic_info(tmp_path: Path) -> None: 

115 """Verify JsonInputArchive.get_basic_info returns correct schema 

116 metadata. 

117 """ 

118 path = tmp_path / "x.json" 

119 images_json.write(_make_image(), path) 

120 info = JsonInputArchive.get_basic_info(path) 

121 assert info.schema_name == "image" 

122 assert info.schema_version == "1.0.0" 

123 assert info.schema_url == "https://images.lsst.io/schemas/image-1.0.0" 

124 assert info.format_version is None 

125 

126 

127def test_fits_basic_info(tmp_path: Path) -> None: 

128 """Verify FitsInputArchive.get_basic_info returns correct schema 

129 metadata. 

130 """ 

131 path = tmp_path / "x.fits" 

132 images_fits.write(_make_image(), path) 

133 info = FitsInputArchive.get_basic_info(path) 

134 assert info.schema_name == "image" 

135 assert info.schema_version == "1.0.0" 

136 assert info.schema_url == "https://images.lsst.io/schemas/image-1.0.0" 

137 assert info.format_version == 1 

138 

139 

140@skip_no_h5py 

141def test_ndf_basic_info(tmp_path: Path) -> None: 

142 """Verify NdfInputArchive.get_basic_info returns correct schema 

143 metadata. 

144 """ 

145 path = tmp_path / "x.sdf" 

146 images_ndf.write(_make_image(), path) 

147 info = NdfInputArchive.get_basic_info(path) 

148 assert info.schema_name == "image" 

149 assert info.schema_version == "1.0.0" 

150 assert info.schema_url == "https://images.lsst.io/schemas/image-1.0.0" 

151 assert info.format_version == 1 

152 

153 

154@skip_no_h5py 

155def test_ndf_basic_info_ignores_nested_json(tmp_path: Path) -> None: 

156 """Verify NdfInputArchive.get_basic_info reads the top-level schema, not 

157 nested pointer trees. 

158 """ 

159 path = tmp_path / "nested.sdf" 

160 images_ndf.write(_make_image(), path) 

161 # Inject a nested pointer-tree JSON alongside the top-level one; it 

162 # sorts before "JSON" so a depth-first scan would hit it first. 

163 payload = json.dumps({"schema_url": "https://images.lsst.io/schemas/aaa_child-9.9.9"}) 

164 with h5py.File(path, "a") as handle: 

165 child = handle.require_group("MORE/LSST/AAA") 

166 child.create_dataset("JSON", data=np.frombuffer(payload.encode("utf-8"), dtype=np.uint8)) 

167 info = NdfInputArchive.get_basic_info(path) 

168 assert info.schema_name == "image" 

169 assert info.schema_url == "https://images.lsst.io/schemas/image-1.0.0"