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 

13from typing import Any, ClassVar 

14from unittest import mock 

15 

16import pytest 

17 

18from lsst.images import SkyProjection, VisitImage 

19from lsst.images._image import ImageSerializationModel 

20from lsst.images._visit_image import VisitImageSerializationModel 

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

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

23 

24 

25def test_unknown_returns_none() -> None: 

26 """Verify class_for_schema returns None for an unrecognised schema name.""" 

27 assert class_for_schema("does-not-exist") is None 

28 

29 

30def test_visit_image_registered() -> None: 

31 """Verify visit_image is registered and maps to 

32 VisitImageSerializationModel. 

33 """ 

34 cls = class_for_schema("visit_image") 

35 assert cls is VisitImageSerializationModel 

36 

37 

38def test_nested_image_registered() -> None: 

39 """Verify nested types like image are registered so sub-models can 

40 be read directly. 

41 """ 

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

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

44 cls = class_for_schema("image") 

45 assert cls is ImageSerializationModel 

46 

47 

48def test_duplicate_registration_raises() -> None: 

49 """Verify re-registering a class is a no-op but a conflicting class 

50 raises RuntimeError. 

51 """ 

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

53 register_schema_class(VisitImageSerializationModel) 

54 

55 # Defining a subclass that redeclares SCHEMA_NAME triggers 

56 # __pydantic_init_subclass__, which calls register_schema_class 

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

58 # call raises RuntimeError. 

59 with pytest.raises(RuntimeError): 

60 

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

62 SCHEMA_NAME: ClassVar[str] = "visit_image" 

63 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

64 

65 

66def test_builtin_provider_loaded_on_miss() -> None: 

67 """Verify builtin schema providers are loaded lazily on a registry 

68 cache miss. 

69 """ 

70 schema_names = ("cell_aperture_correction_map", "cell_coadd", "cell_psf", "coadd_provenance") 

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

72 try: 

73 for schema_name in schema_names: 

74 cls = class_for_schema(schema_name) 

75 assert cls is not None, f"schema_name={schema_name!r}: expected a registered class" 

76 assert cls.SCHEMA_NAME == schema_name, f"schema_name={schema_name!r}: SCHEMA_NAME mismatch" 

77 finally: 

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

79 # if an assertion above fails. 

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

81 

82 

83def test_entry_point_provider_loaded_on_miss() -> None: 

84 """Verify entry-point providers are loaded lazily on a registry 

85 cache miss. 

86 """ 

87 

88 class _EntryPointTree(ArchiveTree): 

89 SCHEMA_NAME: ClassVar[str] = "_entry_point_schema_test" 

90 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

91 MIN_READ_VERSION: ClassVar[int] = 1 

92 

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

94 raise AssertionError("not used") 

95 

96 class _FakeEntryPoint: 

97 value = "tests.test_serialization_registry:_EntryPointTree" 

98 

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

100 return _EntryPointTree 

101 

102 _REGISTRY.pop("_entry_point_schema_test", None) 

103 try: 

104 with mock.patch.object( 

105 _io.importlib.metadata, 

106 "entry_points", 

107 return_value=[_FakeEntryPoint()], 

108 ) as entry_points: 

109 cls = class_for_schema("_entry_point_schema_test") 

110 entry_points.assert_called_once_with( 

111 group="lsst.images.schemas", 

112 name="_entry_point_schema_test", 

113 ) 

114 assert cls is _EntryPointTree 

115 finally: 

116 _REGISTRY.pop("_entry_point_schema_test", None) 

117 

118 

119def test_concrete_type() -> None: 

120 """Verify public_type_for_schema returns VisitImage for the 

121 visit_image schema. 

122 """ 

123 assert public_type_for_schema("visit_image") is VisitImage 

124 

125 

126def test_generic_in_memory_type() -> None: 

127 """Verify public_type_for_schema returns SkyProjection for the 

128 sky_projection schema. 

129 """ 

130 # ProjectionSerializationModel produces a Projection (its PUBLIC_TYPE 

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

132 assert public_type_for_schema("sky_projection") is SkyProjection 

133 

134 

135def test_unregistered_schema_returns_none() -> None: 

136 """Verify public_type_for_schema returns None for an unregistered 

137 schema name. 

138 """ 

139 assert public_type_for_schema("no-such-schema") is None 

140 

141 

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

143 """Return all concrete ArchiveTree subclasses that declare SCHEMA_NAME.""" 

144 seen: list[type] = [] 

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

146 while stack: 

147 cls = stack.pop() 

148 stack.extend(cls.__subclasses__()) 

149 if "SCHEMA_NAME" in cls.__dict__: 

150 seen.append(cls) 

151 return seen 

152 

153 

154def test_every_subclass_registered() -> None: 

155 """Verify every package-local ArchiveTree subclass with SCHEMA_NAME 

156 is registered. 

157 """ 

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

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

160 # classes whose modules belong to the package itself. 

161 missing: list[str] = [] 

162 for cls in _all_concrete_archive_tree_subclasses(): 

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

164 continue 

165 registered = _REGISTRY.get(cls.SCHEMA_NAME) 

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

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

168 assert missing == [], f"Unregistered subclasses: {missing}" 

169 

170 

171def test_every_registered_class_declares_public_type() -> None: 

172 """Verify every package-local registered ArchiveTree class declares 

173 PUBLIC_TYPE. 

174 """ 

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

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

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

178 unresolved: list[str] = [] 

179 for cls in _REGISTRY.values(): 

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

181 continue 

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

183 unresolved.append(cls.__qualname__) 

184 assert unresolved == [], f"No PUBLIC_TYPE declared: {unresolved}"