Coverage for tests/test_detached_archive.py: 100%

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 

14from collections.abc import Iterator 

15from pathlib import Path 

16from typing import Any, cast 

17 

18import pytest 

19 

20from lsst.images import VisitImage, VisitImageSerializationModel 

21from lsst.images.serialization import ( 

22 ArchiveAccessRequiredError, 

23 ArchiveReadError, 

24 ArchiveTree, 

25 DetachedArchive, 

26 InvalidComponentError, 

27 open_archive, 

28 read_archive, 

29 write_archive, 

30) 

31 

32LOCAL_DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "schema_v1") 

33 

34 

35@pytest.fixture 

36def component_probe_env(tmp_path: Path) -> Iterator[tuple[str, VisitImage, DetachedArchive]]: 

37 """Return (tmpdir, visit_image, archive) for component-probe tests.""" 

38 visit_image = read_archive(os.path.join(LOCAL_DATA_DIR, "visit_image.json")) 

39 yield str(tmp_path), visit_image, DetachedArchive() 

40 

41 

42def test_exception_hierarchy() -> None: 

43 """Verify ArchiveAccessRequiredError is a RuntimeError but not an 

44 ArchiveReadError. 

45 """ 

46 assert issubclass(ArchiveAccessRequiredError, RuntimeError) 

47 # This is a control-flow signal, not a corrupt-file diagnosis: it 

48 # must never be swallowed by 'except ArchiveReadError' handlers 

49 # (e.g. the deferred-PSF handling in VisitImage full reads). 

50 assert not issubclass(ArchiveAccessRequiredError, ArchiveReadError) 

51 

52 

53# TODO[DM-54965]: many of the tests below are fragile, because they pass in 

54# None on the assumption that the archive isn't going to check an argument's 

55# type for correctness before exhibiting the behavior the test is trying to 

56# verify. Running MyPy seems like a good way to flag these. 

57 

58 

59def test_deserialize_pointer_raises() -> None: 

60 """Verify DetachedArchive.deserialize_pointer raises 

61 ArchiveAccessRequiredError. 

62 """ 

63 with pytest.raises(ArchiveAccessRequiredError): 

64 DetachedArchive().deserialize_pointer(None, ArchiveTree, lambda model, archive: None) 

65 

66 

67def test_get_frame_set_raises() -> None: 

68 """Verify DetachedArchive.get_frame_set raises 

69 ArchiveAccessRequiredError. 

70 """ 

71 with pytest.raises(ArchiveAccessRequiredError): 

72 DetachedArchive().get_frame_set(None) 

73 

74 

75def test_get_array_raises() -> None: 

76 """Verify DetachedArchive.get_array raises ArchiveAccessRequiredError.""" 

77 with pytest.raises(ArchiveAccessRequiredError): 

78 DetachedArchive().get_array(None) 

79 

80 

81def test_get_table_raises() -> None: 

82 """Verify DetachedArchive.get_table raises ArchiveAccessRequiredError.""" 

83 with pytest.raises(ArchiveAccessRequiredError): 

84 DetachedArchive().get_table(None) 

85 

86 

87def test_get_structured_array_raises() -> None: 

88 """Verify DetachedArchive.get_structured_array raises 

89 ArchiveAccessRequiredError. 

90 """ 

91 with pytest.raises(ArchiveAccessRequiredError): 

92 DetachedArchive().get_structured_array(None) 

93 

94 

95def test_get_opaque_metadata_is_none() -> None: 

96 """Verify DetachedArchive.get_opaque_metadata returns None.""" 

97 # A detached probe has no file to take opaque metadata from. 

98 assert DetachedArchive().get_opaque_metadata() is None 

99 

100 

101def _get_tree(visit_image: VisitImage, tmpdir: str, extension: str) -> VisitImageSerializationModel[Any]: 

102 """Write the fixture in the given format and return its tree. 

103 

104 The tree is deliberately used after the reader is closed, mirroring 

105 how the formatter cache holds a tree with no open file. 

106 """ 

107 path = os.path.join(tmpdir, "visit_image" + extension) 

108 write_archive(visit_image, path) 

109 with open_archive(path) as reader: 

110 return cast(VisitImageSerializationModel[Any], reader.get_tree()) 

111 

112 

113@pytest.mark.parametrize( 

114 "component", 

115 [ 

116 "sky_projection", 

117 "psf", 

118 "obs_info", 

119 "summary_stats", 

120 "detector", 

121 "aperture_corrections", 

122 "backgrounds", 

123 "band", 

124 "bbox", 

125 ], 

126) 

127def test_free_components( 

128 component: str, component_probe_env: tuple[str, VisitImage, DetachedArchive] 

129) -> None: 

130 """Verify each free component deserializes without file access via 

131 DetachedArchive. 

132 """ 

133 tmpdir, visit_image, archive = component_probe_env 

134 tree = _get_tree(visit_image, tmpdir, ".fits") 

135 value = tree.deserialize_component(component, archive) 

136 assert value is not None 

137 

138 

139@pytest.mark.parametrize( 

140 "component", 

141 ["image", "mask", "variance"], 

142) 

143def test_pixel_components_need_file( 

144 component: str, 

145 component_probe_env: tuple[str, object, DetachedArchive], 

146) -> None: 

147 """Test that pixel components raise ArchiveAccessRequiredError via 

148 DetachedArchive. 

149 """ 

150 tmpdir, visit_image, archive = component_probe_env 

151 tree = _get_tree(visit_image, tmpdir, ".fits") 

152 with pytest.raises(ArchiveAccessRequiredError): 

153 tree.deserialize_component(component, archive) 

154 

155 

156def test_json_pixel_components_need_file( 

157 component_probe_env: tuple[str, object, DetachedArchive], 

158) -> None: 

159 """Test that inline JSON arrays still require file access for pixel 

160 components. 

161 

162 Inline arrays in a JSON tree still go through archive.get_array, 

163 so pixel components fall back to the file even for .json. 

164 """ 

165 tmpdir, visit_image, archive = component_probe_env 

166 tree = _get_tree(visit_image, tmpdir, ".json") 

167 with pytest.raises(ArchiveAccessRequiredError): 

168 tree.deserialize_component("image", archive) 

169 

170 

171def test_invalid_component_propagates( 

172 component_probe_env: tuple[str, object, DetachedArchive], 

173) -> None: 

174 """Test that that deserializing an unknown component raises 

175 InvalidComponentError. 

176 """ 

177 tmpdir, visit_image, archive = component_probe_env 

178 tree = _get_tree(visit_image, tmpdir, ".fits") 

179 with pytest.raises(InvalidComponentError): 

180 tree.deserialize_component("not_a_component", archive)