Coverage for tests/test_schema_versioning.py: 31%

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 

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

35 raise NotImplementedError() 

36 

37 

38class CheckCompatTestCase(unittest.TestCase): 

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

40 

41 def test_silent_when_min_read_satisfied(self): 

42 # min_read_version equals reader major: silent. 

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

44 

45 def test_silent_when_on_disk_major_is_lower(self): 

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

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

48 

49 def test_silent_when_on_disk_major_is_higher_but_min_read_low(self): 

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

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

52 

53 def test_raises_when_min_read_exceeds_reader_major(self): 

54 with self.assertRaises(ArchiveReadError) as ctx: 

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

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

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

58 

59 def test_format_version_silent_when_equal(self): 

60 _check_format_version("fits", 1, 1) 

61 

62 def test_format_version_silent_when_on_disk_lower(self): 

63 _check_format_version("fits", 1, 2) 

64 

65 def test_format_version_raises_when_on_disk_higher(self): 

66 with self.assertRaises(ArchiveReadError): 

67 _check_format_version("fits", 2, 1) 

68 

69 

70class ArchiveTreeVersionFieldsTestCase(unittest.TestCase): 

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

72 

73 def test_default_values_filled_from_classvars(self): 

74 instance = _DummyArchiveTree() 

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

76 self.assertEqual(instance.min_read_version, 1) 

77 

78 def test_schema_url_is_computed(self): 

79 instance = _DummyArchiveTree() 

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

81 

82 def test_schema_url_appears_in_dump(self): 

83 instance = _DummyArchiveTree() 

84 dumped = instance.model_dump() 

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

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

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

88 

89 def test_schema_url_ignored_in_input(self): 

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

91 instance = _DummyArchiveTree.model_validate( 

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

93 ) 

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

95 

96 def test_normalises_to_in_code_values(self): 

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

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

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

100 self.assertEqual(instance.min_read_version, 1) 

101 

102 def test_absent_fields_default_to_legacy(self): 

103 instance = _DummyArchiveTree.model_validate({}) 

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

105 self.assertEqual(instance.min_read_version, 1) 

106 

107 def test_min_read_version_too_high_rejected(self): 

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

109 # wrapping it in ValidationError. 

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

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

112 

113 

114class JsonSchemaInjectionTestCase(unittest.TestCase): 

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

116 

117 def test_image_schema_has_id_and_title(self): 

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

119 from lsst.images._image import ImageSerializationModel 

120 

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

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

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

124 

125 

126class ArchiveTreeClassInvariantsTestCase(unittest.TestCase): 

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

128 

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

130 (`iter_concrete_archive_tree_subclasses` and 

131 `check_archive_tree_class_invariants`) so the latter can be invoked 

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

133 """ 

134 

135 def test_constants_are_declared(self): 

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

137 found = list(iter_concrete_archive_tree_subclasses()) 

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

139 for sub in found: 

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

141 check_archive_tree_class_invariants(self, sub) 

142 

143 def test_schema_names_unique(self): 

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

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

146 for sub in iter_concrete_archive_tree_subclasses(): 

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

148 if sub.__module__.startswith("tests."): 

149 continue 

150 # Skip Pydantic generic parametrisations (e.g. 

151 # MaskedImageSerializationModel[TypeVar]); only the original 

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

153 # __pydantic_generic_metadata__["args"]. 

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

155 if generic_meta.get("args"): 

156 continue 

157 existing = names.get(sub.SCHEMA_NAME) 

158 if existing is not None: 

159 self.fail( 

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

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

162 ) 

163 names[sub.SCHEMA_NAME] = sub 

164 

165 

166class FixtureMutationTestCase(unittest.TestCase): 

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

168 

169 def setUp(self): 

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

171 self.assertTrue(self.fixture_path.exists()) 

172 

173 def test_min_read_too_high_raises(self): 

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

175 import json as json_module 

176 

177 from lsst.images._image import ImageSerializationModel 

178 

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

180 tree["min_read_version"] = 99 

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

182 ImageSerializationModel.model_validate(tree) 

183 

184 def test_higher_major_with_low_min_read_succeeds(self): 

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

186 import json as json_module 

187 

188 from lsst.images._image import ImageSerializationModel 

189 

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

191 tree["schema_version"] = "99.0.0" 

192 tree["min_read_version"] = 1 

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

194 # major-1 readers reads silently. 

195 instance = ImageSerializationModel.model_validate(tree) 

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

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

198 self.assertEqual(instance.min_read_version, 1) 

199 

200 def test_absent_fields_default_to_legacy(self): 

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

202 import json as json_module 

203 

204 from lsst.images._image import ImageSerializationModel 

205 

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

207 del tree["schema_version"] 

208 del tree["min_read_version"] 

209 del tree["schema_url"] 

210 instance = ImageSerializationModel.model_validate(tree) 

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

212 self.assertEqual(instance.min_read_version, 1) 

213 

214 

215if __name__ == "__main__": 

216 unittest.main()