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 rejects URLs from non-lsst hosts.""" 

80 with pytest.raises(ValueError): 

81 ArchiveInfo.from_schema_url("https://evil.example.com/schemas/image-1.0.0", format_version=None) 

82 

83 

84def test_input_archive_get_basic_info_base_raises() -> None: 

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

86 base class. 

87 """ 

88 with pytest.raises(NotImplementedError): 

89 InputArchive.get_basic_info("x.fits") 

90 

91 

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

93 """Verify JsonInputArchive.get_basic_info returns correct schema 

94 metadata. 

95 """ 

96 path = tmp_path / "x.json" 

97 images_json.write(_make_image(), path) 

98 info = JsonInputArchive.get_basic_info(path) 

99 assert info.schema_name == "image" 

100 assert info.schema_version == "1.0.0" 

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

102 assert info.format_version is None 

103 

104 

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

106 """Verify FitsInputArchive.get_basic_info returns correct schema 

107 metadata. 

108 """ 

109 path = tmp_path / "x.fits" 

110 images_fits.write(_make_image(), path) 

111 info = FitsInputArchive.get_basic_info(path) 

112 assert info.schema_name == "image" 

113 assert info.schema_version == "1.0.0" 

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

115 assert info.format_version == 1 

116 

117 

118@skip_no_h5py 

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

120 """Verify NdfInputArchive.get_basic_info returns correct schema 

121 metadata. 

122 """ 

123 path = tmp_path / "x.sdf" 

124 images_ndf.write(_make_image(), path) 

125 info = NdfInputArchive.get_basic_info(path) 

126 assert info.schema_name == "image" 

127 assert info.schema_version == "1.0.0" 

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

129 assert info.format_version == 1 

130 

131 

132@skip_no_h5py 

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

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

135 nested pointer trees. 

136 """ 

137 path = tmp_path / "nested.sdf" 

138 images_ndf.write(_make_image(), path) 

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

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

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

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

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

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

145 info = NdfInputArchive.get_basic_info(path) 

146 assert info.schema_name == "image" 

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