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 

14import tempfile 

15import unittest 

16 

17from lsst.images.serialization import ( 

18 ArchiveAccessRequiredError, 

19 ArchiveReadError, 

20 ArchiveTree, 

21 DetachedArchive, 

22 InvalidComponentError, 

23 read, 

24 write, 

25) 

26from lsst.images.serialization import open as ser_open 

27 

28DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "schema_v1") 

29 

30FREE_COMPONENTS = ( 

31 "sky_projection", 

32 "psf", 

33 "obs_info", 

34 "summary_stats", 

35 "detector", 

36 "aperture_corrections", 

37 "backgrounds", 

38 "band", 

39 "bbox", 

40) 

41 

42PIXEL_COMPONENTS = ("image", "mask", "variance") 

43 

44 

45class ArchiveAccessRequiredErrorTestCase(unittest.TestCase): 

46 """Tests for the ArchiveAccessRequiredError hierarchy.""" 

47 

48 def test_exception_hierarchy(self) -> None: 

49 self.assertTrue(issubclass(ArchiveAccessRequiredError, RuntimeError)) 

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

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

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

53 self.assertFalse(issubclass(ArchiveAccessRequiredError, ArchiveReadError)) 

54 

55 

56class DetachedArchiveTestCase(unittest.TestCase): 

57 """Every data-access method of DetachedArchive raises.""" 

58 

59 def setUp(self) -> None: 

60 self.archive = DetachedArchive() 

61 

62 def test_deserialize_pointer_raises(self) -> None: 

63 with self.assertRaises(ArchiveAccessRequiredError): 

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

65 

66 def test_get_frame_set_raises(self) -> None: 

67 with self.assertRaises(ArchiveAccessRequiredError): 

68 self.archive.get_frame_set(None) 

69 

70 def test_get_array_raises(self) -> None: 

71 with self.assertRaises(ArchiveAccessRequiredError): 

72 self.archive.get_array(None) 

73 

74 def test_get_table_raises(self) -> None: 

75 with self.assertRaises(ArchiveAccessRequiredError): 

76 self.archive.get_table(None) 

77 

78 def test_get_structured_array_raises(self) -> None: 

79 with self.assertRaises(ArchiveAccessRequiredError): 

80 self.archive.get_structured_array(None) 

81 

82 def test_get_opaque_metadata_is_none(self) -> None: 

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

84 self.assertIsNone(self.archive.get_opaque_metadata()) 

85 

86 

87class ComponentProbeTestCase(unittest.TestCase): 

88 """Tests for which VisitImage components deserialize without file access. 

89 

90 The fixture has a Gaussian PSF and inline sky-projection mappings, so 

91 both probe as free; PSF models and transforms that use archive pointers 

92 (e.g. Piff PSFs, frame-set references) take the deserialize_pointer / 

93 get_frame_set paths instead and would raise. 

94 """ 

95 

96 def setUp(self) -> None: 

97 tmp = tempfile.TemporaryDirectory() 

98 self.addCleanup(tmp.cleanup) 

99 self.tmpdir = tmp.name 

100 self.visit_image = read(os.path.join(DATA_DIR, "visit_image.json")) 

101 self.archive = DetachedArchive() 

102 

103 def _get_tree(self, extension: str) -> ArchiveTree: 

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

105 

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

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

108 """ 

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

110 write(self.visit_image, path) 

111 with ser_open(path) as reader: 

112 return reader.get_tree() 

113 

114 def test_free_components(self) -> None: 

115 tree = self._get_tree(".fits") 

116 for component in FREE_COMPONENTS: 

117 with self.subTest(component=component): 

118 value = tree.deserialize_component(component, self.archive) 

119 self.assertIsNotNone(value) 

120 

121 def test_pixel_components_need_file(self) -> None: 

122 tree = self._get_tree(".fits") 

123 for component in PIXEL_COMPONENTS: 

124 with self.subTest(component=component): 

125 with self.assertRaises(ArchiveAccessRequiredError): 

126 tree.deserialize_component(component, self.archive) 

127 

128 def test_json_pixel_components_need_file(self) -> None: 

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

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

131 tree = self._get_tree(".json") 

132 with self.assertRaises(ArchiveAccessRequiredError): 

133 tree.deserialize_component("image", self.archive) 

134 

135 def test_invalid_component_propagates(self) -> None: 

136 tree = self._get_tree(".fits") 

137 with self.assertRaises(InvalidComponentError): 

138 tree.deserialize_component("not_a_component", self.archive) 

139 

140 

141if __name__ == "__main__": 

142 unittest.main()