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) 

31from lsst.images.tests import current_fixture_path 

32 

33FIXTURE_DIR = Path(__file__).parent / "data" / "schemas" 

34 

35 

36@pytest.fixture 

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

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

39 visit_image = read_archive(current_fixture_path(FIXTURE_DIR, "visit_image")) 

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

41 

42 

43def test_exception_hierarchy() -> None: 

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

45 ArchiveReadError. 

46 """ 

47 assert issubclass(ArchiveAccessRequiredError, RuntimeError) 

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

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

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

51 assert not issubclass(ArchiveAccessRequiredError, ArchiveReadError) 

52 

53 

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

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

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

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

58 

59 

60def test_deserialize_pointer_raises() -> None: 

61 """Verify DetachedArchive.deserialize_pointer raises 

62 ArchiveAccessRequiredError. 

63 """ 

64 with pytest.raises(ArchiveAccessRequiredError): 

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

66 

67 

68def test_get_frame_set_raises() -> None: 

69 """Verify DetachedArchive.get_frame_set raises 

70 ArchiveAccessRequiredError. 

71 """ 

72 with pytest.raises(ArchiveAccessRequiredError): 

73 DetachedArchive().get_frame_set(None) 

74 

75 

76def test_get_array_raises() -> None: 

77 """Verify DetachedArchive.get_array raises ArchiveAccessRequiredError.""" 

78 with pytest.raises(ArchiveAccessRequiredError): 

79 DetachedArchive().get_array(None) 

80 

81 

82def test_get_table_raises() -> None: 

83 """Verify DetachedArchive.get_table raises ArchiveAccessRequiredError.""" 

84 with pytest.raises(ArchiveAccessRequiredError): 

85 DetachedArchive().get_table(None) 

86 

87 

88def test_get_structured_array_raises() -> None: 

89 """Verify DetachedArchive.get_structured_array raises 

90 ArchiveAccessRequiredError. 

91 """ 

92 with pytest.raises(ArchiveAccessRequiredError): 

93 DetachedArchive().get_structured_array(None) 

94 

95 

96def test_get_opaque_metadata_is_none() -> None: 

97 """Verify DetachedArchive.get_opaque_metadata returns None.""" 

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

99 assert DetachedArchive().get_opaque_metadata() is None 

100 

101 

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

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

104 

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

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

107 """ 

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

109 write_archive(visit_image, path) 

110 with open_archive(path) as reader: 

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

112 

113 

114@pytest.mark.parametrize( 

115 "component", 

116 [ 

117 "sky_projection", 

118 "psf", 

119 "obs_info", 

120 "summary_stats", 

121 "detector", 

122 "aperture_corrections", 

123 "backgrounds", 

124 "band", 

125 "bbox", 

126 ], 

127) 

128def test_free_components( 

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

130) -> None: 

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

132 DetachedArchive. 

133 """ 

134 tmpdir, visit_image, archive = component_probe_env 

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

136 value = tree.deserialize_component(component, archive) 

137 assert value is not None 

138 

139 

140@pytest.mark.parametrize( 

141 "component", 

142 ["image", "mask", "variance"], 

143) 

144def test_pixel_components_need_file( 

145 component: str, 

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

147) -> None: 

148 """Test that pixel components raise ArchiveAccessRequiredError via 

149 DetachedArchive. 

150 """ 

151 tmpdir, visit_image, archive = component_probe_env 

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

153 with pytest.raises(ArchiveAccessRequiredError): 

154 tree.deserialize_component(component, archive) 

155 

156 

157def test_json_pixel_components_need_file( 

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

159) -> None: 

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

161 components. 

162 

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

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

165 """ 

166 tmpdir, visit_image, archive = component_probe_env 

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

168 with pytest.raises(ArchiveAccessRequiredError): 

169 tree.deserialize_component("image", archive) 

170 

171 

172def test_invalid_component_propagates( 

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

174) -> None: 

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

176 InvalidComponentError. 

177 """ 

178 tmpdir, visit_image, archive = component_probe_env 

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

180 with pytest.raises(InvalidComponentError): 

181 tree.deserialize_component("not_a_component", archive)