Coverage for tests/test_serialization_registry.py: 96%

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 unittest 

14from typing import Any, ClassVar 

15from unittest import mock 

16 

17from lsst.images import SkyProjection, VisitImage 

18from lsst.images._image import ImageSerializationModel 

19from lsst.images._visit_image import VisitImageSerializationModel 

20from lsst.images.serialization import ArchiveTree, _io, class_for_schema, public_type_for_schema 

21from lsst.images.serialization._io import _REGISTRY, register_schema_class 

22 

23 

24class ClassForSchemaTestCase(unittest.TestCase): 

25 """class_for_schema returns None for unknown schema names.""" 

26 

27 def test_unknown_returns_none(self) -> None: 

28 self.assertIsNone(class_for_schema("does-not-exist")) 

29 

30 

31class RegistrationTestCase(unittest.TestCase): 

32 """ArchiveTree subclasses register themselves under SCHEMA_NAME.""" 

33 

34 def test_visit_image_registered(self) -> None: 

35 cls = class_for_schema("visit_image") 

36 self.assertIs(cls, VisitImageSerializationModel) 

37 

38 def test_nested_image_registered(self) -> None: 

39 # Nested types are registered too -- "register all" was the spec 

40 # decision so callers can read sub-models directly. 

41 cls = class_for_schema("image") 

42 self.assertIs(cls, ImageSerializationModel) 

43 

44 def test_duplicate_registration_raises(self) -> None: 

45 # Re-registering the same class is a no-op. 

46 register_schema_class(VisitImageSerializationModel) 

47 

48 # Defining a subclass that redeclares SCHEMA_NAME triggers 

49 # __pydantic_init_subclass__, which calls register_schema_class 

50 # with a different class object under an existing name. That 

51 # call raises RuntimeError. 

52 with self.assertRaises(RuntimeError): 

53 

54 class _Imposter(VisitImageSerializationModel): # type: ignore[misc] 

55 SCHEMA_NAME: ClassVar[str] = "visit_image" 

56 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

57 

58 def test_builtin_provider_loaded_on_miss(self) -> None: 

59 schema_names = ("cell_coadd", "cell_psf", "coadd_provenance") 

60 saved = {schema_name: _REGISTRY.pop(schema_name, None) for schema_name in schema_names} 

61 try: 

62 for schema_name in schema_names: 

63 with self.subTest(schema_name=schema_name): 

64 cls = class_for_schema(schema_name) 

65 self.assertIsNotNone(cls) 

66 self.assertEqual(cls.SCHEMA_NAME, schema_name) 

67 finally: 

68 # Preserve any registrations that existed before this test, even 

69 # if an assertion above fails. 

70 _REGISTRY.update({schema_name: cls for schema_name, cls in saved.items() if cls is not None}) 

71 

72 def test_entry_point_provider_loaded_on_miss(self) -> None: 

73 class _EntryPointTree(ArchiveTree): 

74 SCHEMA_NAME: ClassVar[str] = "_entry_point_schema_test" 

75 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

76 MIN_READ_VERSION: ClassVar[int] = 1 

77 

78 def deserialize(self, archive: Any, **kwargs: Any) -> VisitImage: 

79 raise AssertionError("not used") 

80 

81 class _FakeEntryPoint: 

82 value = "tests.test_serialization_registry:_EntryPointTree" 

83 

84 def load(self) -> type[ArchiveTree]: 

85 return _EntryPointTree 

86 

87 _REGISTRY.pop("_entry_point_schema_test", None) 

88 try: 

89 with mock.patch.object( 

90 _io.importlib.metadata, 

91 "entry_points", 

92 return_value=[_FakeEntryPoint()], 

93 ) as entry_points: 

94 cls = class_for_schema("_entry_point_schema_test") 

95 entry_points.assert_called_once_with( 

96 group="lsst.images.schemas", 

97 name="_entry_point_schema_test", 

98 ) 

99 self.assertIs(cls, _EntryPointTree) 

100 finally: 

101 _REGISTRY.pop("_entry_point_schema_test", None) 

102 

103 

104class PublicTypeTestCase(unittest.TestCase): 

105 """public_type_for_schema returns each tree's PUBLIC_TYPE ClassVar.""" 

106 

107 def test_concrete_type(self) -> None: 

108 self.assertIs(public_type_for_schema("visit_image"), VisitImage) 

109 

110 def test_generic_in_memory_type(self) -> None: 

111 # ProjectionSerializationModel produces a Projection (its PUBLIC_TYPE 

112 # is the unparameterised class, not Projection[Any]). 

113 self.assertIs(public_type_for_schema("sky_projection"), SkyProjection) 

114 

115 def test_unregistered_schema_returns_none(self) -> None: 

116 self.assertIsNone(public_type_for_schema("no-such-schema")) 

117 

118 

119def _all_concrete_archive_tree_subclasses() -> list[type]: 

120 """Walk ArchiveTree's subclass tree and return all concrete subclasses 

121 that declare SCHEMA_NAME (i.e. are real schema-bearing leaves). 

122 """ 

123 seen: list[type] = [] 

124 stack: list[type] = list(ArchiveTree.__subclasses__()) 

125 while stack: 

126 cls = stack.pop() 

127 stack.extend(cls.__subclasses__()) 

128 if "SCHEMA_NAME" in cls.__dict__: 

129 seen.append(cls) 

130 return seen 

131 

132 

133class ClassInvariantsTestCase(unittest.TestCase): 

134 """Every ArchiveTree subclass with SCHEMA_NAME is registered and has a 

135 resolvable, concrete deserialize return annotation. 

136 """ 

137 

138 def test_every_subclass_registered(self) -> None: 

139 # Test-local classes from other test methods may linger in 

140 # __subclasses__ after their cleanup runs; restrict the check to 

141 # classes whose modules belong to the package itself. 

142 missing: list[str] = [] 

143 for cls in _all_concrete_archive_tree_subclasses(): 

144 if not cls.__module__.startswith("lsst.images"): 

145 continue 

146 registered = _REGISTRY.get(cls.SCHEMA_NAME) 

147 if registered is None or registered is not cls: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true

148 missing.append(f"{cls.__qualname__} -> {cls.SCHEMA_NAME}") 

149 self.assertEqual(missing, [], f"Unregistered subclasses: {missing}") 

150 

151 def test_every_registered_class_declares_public_type(self) -> None: 

152 # Restrict to package-local classes; test-only ArchiveTree 

153 # subclasses (e.g. tests/test_schema_versioning.py's 

154 # _DummyArchiveTree) may register but are not part of the package. 

155 unresolved: list[str] = [] 

156 for cls in _REGISTRY.values(): 

157 if not cls.__module__.startswith("lsst.images"): 

158 continue 

159 if not isinstance(getattr(cls, "PUBLIC_TYPE", None), type): 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 unresolved.append(cls.__qualname__) 

161 self.assertEqual(unresolved, [], f"No PUBLIC_TYPE declared: {unresolved}") 

162 

163 

164if __name__ == "__main__": 

165 unittest.main()