Coverage for tests/test_schema_versioning.py: 97%

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. 

11 

12from __future__ import annotations 

13 

14import unittest 

15from pathlib import Path 

16from typing import ClassVar 

17 

18import pydantic 

19 

20from lsst.images.serialization import ArchiveReadError, ArchiveTree 

21from lsst.images.serialization._common import _check_compat, _check_format_version 

22from lsst.images.tests import check_archive_tree_class_invariants, iter_concrete_archive_tree_subclasses 

23 

24SCHEMA_DIR = Path(__file__).parent / "data" / "schema_v1" 

25 

26 

27class _DummyArchiveTree(ArchiveTree): 

28 """Minimal concrete ArchiveTree for testing the base-class machinery.""" 

29 

30 SCHEMA_NAME: ClassVar[str] = "dummy" 

31 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

32 MIN_READ_VERSION: ClassVar[int] = 1 

33 PUBLIC_TYPE: ClassVar[type] = object 

34 

35 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked 

36 raise NotImplementedError() 

37 

38 

39class CheckCompatTestCase(unittest.TestCase): 

40 """Tests for the _check_compat and _check_format_version helpers.""" 

41 

42 def test_silent_when_min_read_satisfied(self): 

43 # min_read_version equals reader major: silent. 

44 _check_compat("foo", "1.0.0", 1, "1.0.0") 

45 

46 def test_silent_when_on_disk_major_is_lower(self): 

47 # 1.0.0 file with min_read_version=1 read by 2.0.0 code: silent. 

48 _check_compat("foo", "1.0.0", 1, "2.0.0") 

49 

50 def test_silent_when_on_disk_major_is_higher_but_min_read_low(self): 

51 # 2.0.0 file declares it is safe for major-1 readers: silent. 

52 _check_compat("foo", "2.0.0", 1, "1.0.0") 

53 

54 def test_raises_when_min_read_exceeds_reader_major(self): 

55 with self.assertRaises(ArchiveReadError) as ctx: 

56 _check_compat("foo", "2.0.0", 2, "1.0.0") 

57 self.assertIn("foo", str(ctx.exception)) 

58 self.assertIn(">= 2", str(ctx.exception)) 

59 

60 def test_format_version_silent_when_equal(self): 

61 _check_format_version("fits", 1, 1) 

62 

63 def test_format_version_silent_when_on_disk_lower(self): 

64 _check_format_version("fits", 1, 2) 

65 

66 def test_format_version_raises_when_on_disk_higher(self): 

67 with self.assertRaises(ArchiveReadError): 

68 _check_format_version("fits", 2, 1) 

69 

70 

71class ArchiveTreeVersionFieldsTestCase(unittest.TestCase): 

72 """Tests for the schema_version / min_read_version / schema_url fields.""" 

73 

74 def test_default_values_filled_from_classvars(self): 

75 instance = _DummyArchiveTree() 

76 self.assertEqual(instance.schema_version, "1.0.0") 

77 self.assertEqual(instance.min_read_version, 1) 

78 

79 def test_schema_url_is_computed(self): 

80 instance = _DummyArchiveTree() 

81 self.assertEqual(instance.schema_url, "https://images.lsst.io/schemas/dummy-1.0.0") 

82 

83 def test_schema_url_appears_in_dump(self): 

84 instance = _DummyArchiveTree() 

85 dumped = instance.model_dump() 

86 self.assertEqual(dumped["schema_url"], "https://images.lsst.io/schemas/dummy-1.0.0") 

87 self.assertEqual(dumped["schema_version"], "1.0.0") 

88 self.assertEqual(dumped["min_read_version"], 1) 

89 

90 def test_schema_url_ignored_in_input(self): 

91 # Pydantic's default extra='ignore' drops it from inputs. 

92 instance = _DummyArchiveTree.model_validate( 

93 {"schema_url": "https://example.com/wrong", "schema_version": "1.0.0", "min_read_version": 1} 

94 ) 

95 self.assertEqual(instance.schema_url, "https://images.lsst.io/schemas/dummy-1.0.0") 

96 

97 def test_normalises_to_in_code_values(self): 

98 # An older file's values are normalised on load. 

99 instance = _DummyArchiveTree.model_validate({"schema_version": "0.9.0", "min_read_version": 1}) 

100 self.assertEqual(instance.schema_version, "1.0.0") 

101 self.assertEqual(instance.min_read_version, 1) 

102 

103 def test_absent_fields_default_to_legacy(self): 

104 instance = _DummyArchiveTree.model_validate({}) 

105 self.assertEqual(instance.schema_version, "1.0.0") 

106 self.assertEqual(instance.min_read_version, 1) 

107 

108 def test_min_read_version_too_high_rejected(self): 

109 # Pydantic mode='after' re-raises ArchiveReadError without 

110 # wrapping it in ValidationError. 

111 with self.assertRaises((ArchiveReadError, pydantic.ValidationError)): 

112 _DummyArchiveTree.model_validate({"schema_version": "2.0.0", "min_read_version": 2}) 

113 

114 

115class JsonSchemaInjectionTestCase(unittest.TestCase): 

116 """ArchiveTree injects $id and title into each subclass's JSON Schema.""" 

117 

118 def test_image_schema_has_id_and_title(self): 

119 """Image's serialization-mode schema has ``$id`` / ``title`` set.""" 

120 from lsst.images._image import ImageSerializationModel 

121 

122 schema = ImageSerializationModel.model_json_schema(mode="serialization") 

123 self.assertEqual(schema["$id"], "https://images.lsst.io/schemas/image-1.0.0") 

124 self.assertEqual(schema["title"], "image") 

125 

126 

127class ArchiveTreeClassInvariantsTestCase(unittest.TestCase): 

128 """Concrete ArchiveTree subclasses must declare the version ClassVars. 

129 

130 The reusable discovery and per-class check live in ``lsst.images.tests`` 

131 (`iter_concrete_archive_tree_subclasses` and 

132 `check_archive_tree_class_invariants`) so the latter can be invoked 

133 manually on a single ``ArchiveTree`` independent of the metaprogramming. 

134 """ 

135 

136 def test_constants_are_declared(self): 

137 """All three ClassVars are declared and well-formed everywhere.""" 

138 found = list(iter_concrete_archive_tree_subclasses()) 

139 self.assertGreater(len(found), 0) 

140 for sub in found: 

141 with self.subTest(cls=sub.__name__): 

142 check_archive_tree_class_invariants(self, sub) 

143 

144 def test_schema_names_unique(self): 

145 """All SCHEMA_NAME values across concrete subclasses are unique.""" 

146 names: dict[str, type] = {} 

147 for sub in iter_concrete_archive_tree_subclasses(): 

148 # Skip our local _DummyArchiveTree (it lives in a test module). 

149 if sub.__module__.startswith("tests."): 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true

150 continue 

151 # Skip Pydantic generic parametrisations (e.g. 

152 # MaskedImageSerializationModel[TypeVar]); only the original 

153 # generic class counts. A parametrised form has a non-empty 

154 # __pydantic_generic_metadata__["args"]. 

155 generic_meta = getattr(sub, "__pydantic_generic_metadata__", {}) 

156 if generic_meta.get("args"): 

157 continue 

158 existing = names.get(sub.SCHEMA_NAME) 

159 if existing is not None: 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 self.fail( 

161 f"Duplicate SCHEMA_NAME {sub.SCHEMA_NAME!r}: " 

162 f"{sub.__qualname__} vs {existing.__qualname__}" 

163 ) 

164 names[sub.SCHEMA_NAME] = sub 

165 

166 

167class FixtureMutationTestCase(unittest.TestCase): 

168 """Mutate a fixture in-memory and verify the read behavior.""" 

169 

170 def setUp(self): 

171 self.fixture_path = SCHEMA_DIR / "image.json" 

172 self.assertTrue(self.fixture_path.exists()) 

173 

174 def test_min_read_too_high_raises(self): 

175 """Setting min_read_version above reader major rejects the read.""" 

176 import json as json_module 

177 

178 from lsst.images._image import ImageSerializationModel 

179 

180 tree = json_module.loads(self.fixture_path.read_text()) 

181 tree["min_read_version"] = 99 

182 with self.assertRaises((ArchiveReadError, pydantic.ValidationError)): 

183 ImageSerializationModel.model_validate(tree) 

184 

185 def test_higher_major_with_low_min_read_succeeds(self): 

186 """A higher schema_version with low min_read_version reads silently.""" 

187 import json as json_module 

188 

189 from lsst.images._image import ImageSerializationModel 

190 

191 tree = json_module.loads(self.fixture_path.read_text()) 

192 tree["schema_version"] = "99.0.0" 

193 tree["min_read_version"] = 1 

194 # Asymmetric escape: a 99.0.0 file that declares it's safe for 

195 # major-1 readers reads silently. 

196 instance = ImageSerializationModel.model_validate(tree) 

197 # And gets normalised back to in-code values. 

198 self.assertEqual(instance.schema_version, "1.0.0") 

199 self.assertEqual(instance.min_read_version, 1) 

200 

201 def test_absent_fields_default_to_legacy(self): 

202 """Stripping the version fields entirely reads with v1 defaults.""" 

203 import json as json_module 

204 

205 from lsst.images._image import ImageSerializationModel 

206 

207 tree = json_module.loads(self.fixture_path.read_text()) 

208 del tree["schema_version"] 

209 del tree["min_read_version"] 

210 del tree["schema_url"] 

211 instance = ImageSerializationModel.model_validate(tree) 

212 self.assertEqual(instance.schema_version, "1.0.0") 

213 self.assertEqual(instance.min_read_version, 1) 

214 

215 

216if __name__ == "__main__": 

217 unittest.main()