Coverage for tests/test_schema_versioning.py: 98%
168 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-09 10:54 +0000
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.
12from __future__ import annotations
14import json as json_module
15import warnings
16from pathlib import Path
17from typing import Any, ClassVar
19import pydantic
20import pytest
22from lsst.images import ImageSerializationModel
23from lsst.images.json._output_archive import JsonOutputArchive
24from lsst.images.serialization import (
25 ArchiveReadError,
26 ArchiveTree,
27 DevelopmentSchemaWarning,
28 InputArchive,
29 is_development_version,
30 warn_for_development_schemas,
31)
32from lsst.images.serialization._common import (
33 _ARCHIVE_READ_CONTEXT,
34 _check_compat,
35 _check_format_version,
36)
37from lsst.images.tests import (
38 check_archive_tree_class_invariants,
39 current_fixture_path,
40 iter_concrete_archive_tree_subclasses,
41)
43FIXTURE_DIR = Path(__file__).parent / "data" / "schemas"
46class _DummyArchiveTree(ArchiveTree):
47 """Minimal concrete ArchiveTree for testing the base-class machinery."""
49 SCHEMA_NAME: ClassVar[str] = "dummy"
50 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
51 MIN_READ_VERSION: ClassVar[int] = 1
52 PUBLIC_TYPE: ClassVar[type] = object
54 def deserialize(
55 self, archive: InputArchive[Any], **kwargs: Any
56 ) -> Any: # pragma: no cover - never invoked
57 raise NotImplementedError()
60class _BumpedArchiveTree(ArchiveTree):
61 """ArchiveTree whose versions differ from the field defaults, so tests
62 can tell a value taken from the class constants from a default.
63 """
65 SCHEMA_NAME: ClassVar[str] = "dummy_bumped"
66 SCHEMA_VERSION: ClassVar[str] = "3.2.0"
67 MIN_READ_VERSION: ClassVar[int] = 2
68 PUBLIC_TYPE: ClassVar[type] = object
70 def deserialize(
71 self, archive: InputArchive[Any], **kwargs: Any
72 ) -> Any: # pragma: no cover - never invoked
73 raise NotImplementedError()
76class _DevTree(ArchiveTree):
77 """Development-version tree double."""
79 SCHEMA_NAME = "in_dev_helper_test_dev"
80 SCHEMA_VERSION = "1.0.0.dev0"
81 MIN_READ_VERSION = 1
82 PUBLIC_TYPE = object
84 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked
85 raise NotImplementedError()
88class _FinalTree(ArchiveTree):
89 """Finalized-version tree double."""
91 SCHEMA_NAME = "in_dev_helper_test_final"
92 SCHEMA_VERSION = "1.0.0"
93 MIN_READ_VERSION = 1
94 PUBLIC_TYPE = object
96 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked
97 raise NotImplementedError()
100def test_is_development_version() -> None:
101 """Verify is_development_version detects development versions correctly."""
102 assert is_development_version("1.0.0.dev0")
103 assert is_development_version("2.3.4.dev12")
104 assert not is_development_version("1.0.0")
105 assert not is_development_version("2.3.4")
108def test_warn_for_development_schemas_warns() -> None:
109 """Verify warn_for_development_schemas warns for development schemas."""
110 with pytest.warns(DevelopmentSchemaWarning, match="in_dev_helper_test_dev"):
111 warn_for_development_schemas(_DevTree())
114def test_warn_for_development_schemas_silent_when_finalized() -> None:
115 """Verify warn_for_development_schemas is silent for finalized schemas."""
116 with warnings.catch_warnings():
117 warnings.simplefilter("error")
118 warn_for_development_schemas(_FinalTree())
121def test_silent_when_min_read_satisfied() -> None:
122 """Verify no error when min_read_version equals reader major."""
123 _check_compat("foo", "1.0.0", 1, "1.0.0")
126def test_silent_when_on_disk_major_is_lower() -> None:
127 """Verify no error when a 1.0.0 file is read by 2.0.0 code."""
128 _check_compat("foo", "1.0.0", 1, "2.0.0")
131def test_silent_when_on_disk_major_is_higher_but_min_read_low() -> None:
132 """Verify no error when a 2.0.0 file is safe for major-1 readers."""
133 _check_compat("foo", "2.0.0", 1, "1.0.0")
136def test_raises_when_min_read_exceeds_reader_major() -> None:
137 """Verify ArchiveReadError is raised when min_read_version exceeds
138 major.
139 """
140 with pytest.raises(ArchiveReadError) as exc_info:
141 _check_compat("foo", "2.0.0", 2, "1.0.0")
142 assert "foo" in str(exc_info.value)
143 assert ">= 2" in str(exc_info.value)
146def test_format_version_silent_when_equal() -> None:
147 """Verify no error when on-disk format version equals reader version."""
148 _check_format_version("fits", 1, 1)
151def test_format_version_silent_when_on_disk_lower() -> None:
152 """Verify no error when on-disk format version is lower than reader's."""
153 _check_format_version("fits", 1, 2)
156def test_format_version_raises_when_on_disk_higher() -> None:
157 """Verify ArchiveReadError when on-disk format version is too new."""
158 with pytest.raises(ArchiveReadError):
159 _check_format_version("fits", 2, 1)
162def test_default_values_filled_from_classvars() -> None:
163 """Verify default instance values are filled from class-var constants."""
164 instance = _DummyArchiveTree()
165 assert instance.schema_version == "1.0.0"
166 assert instance.min_read_version == 1
169def test_constructed_versions_match_classvars() -> None:
170 """Verify every construction path stamps the class-var versions."""
171 for instance in (
172 _BumpedArchiveTree(),
173 _BumpedArchiveTree.model_validate({}),
174 _BumpedArchiveTree.model_construct(),
175 ):
176 assert instance.schema_version == "3.2.0"
177 assert instance.min_read_version == 2
178 dumped = instance.model_dump()
179 assert dumped["schema_version"] == "3.2.0"
180 assert dumped["min_read_version"] == 2
181 assert dumped["schema_url"] == "https://images.lsst.io/schemas/dummy_bumped-3.2.0"
184def test_model_construct_keeps_explicit_versions() -> None:
185 """Verify model_construct does not override versions it is given."""
186 instance = _BumpedArchiveTree.model_construct(schema_version="3.1.0", min_read_version=1)
187 assert instance.schema_version == "3.1.0"
188 assert instance.min_read_version == 1
191def test_schema_url_is_computed() -> None:
192 """Verify schema_url is computed from SCHEMA_NAME and SCHEMA_VERSION."""
193 instance = _DummyArchiveTree()
194 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
197def test_schema_url_appears_in_dump() -> None:
198 """Verify schema_url, schema_version, and min_read_version are in dumps."""
199 instance = _DummyArchiveTree()
200 dumped = instance.model_dump()
201 assert dumped["schema_url"] == "https://images.lsst.io/schemas/dummy-1.0.0"
202 assert dumped["schema_version"] == "1.0.0"
203 assert dumped["min_read_version"] == 1
206def test_schema_url_ignored_in_input() -> None:
207 """Verify schema_url in input data is ignored; computed value is used."""
208 # Pydantic's default extra='ignore' drops it from inputs.
209 instance = _DummyArchiveTree.model_validate(
210 {"schema_url": "https://example.com/wrong", "schema_version": "1.0.0", "min_read_version": 1}
211 )
212 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
215def test_normalises_to_in_code_values() -> None:
216 """Verify an older file's schema values are normalised to in-code
217 values.
218 """
219 # An older file's values are normalised on load.
220 instance = _DummyArchiveTree.model_validate({"schema_version": "0.9.0", "min_read_version": 1})
221 assert instance.schema_version == "1.0.0"
222 assert instance.min_read_version == 1
225def test_absent_fields_default_for_fresh_construction() -> None:
226 """Verify fresh in-memory construction may omit version fields."""
227 instance = _DummyArchiveTree.model_validate({})
228 assert instance.schema_version == "1.0.0"
229 assert instance.min_read_version == 1
232def test_archive_tree_requires_an_explicit_schema_version() -> None:
233 """Verify archive input cannot silently assume a version."""
234 with pytest.raises(pydantic.ValidationError, match="has no schema_version"):
235 _DummyArchiveTree.model_validate({}, context=_ARCHIVE_READ_CONTEXT)
238def test_archive_context_requires_versions_on_nested_trees() -> None:
239 """Verify the archive-read marker propagates into embedded schemas."""
241 class _ParentTree(ArchiveTree):
242 SCHEMA_NAME: ClassVar[str] = "version_context_parent_test"
243 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
244 MIN_READ_VERSION: ClassVar[int] = 1
245 PUBLIC_TYPE: ClassVar[type] = object
247 child: _DummyArchiveTree
249 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
250 raise NotImplementedError()
252 with pytest.raises(pydantic.ValidationError, match="dummy: archive tree has no schema_version"):
253 _ParentTree.model_validate(
254 {"schema_version": "1.0.0", "min_read_version": 1, "child": {}},
255 context=_ARCHIVE_READ_CONTEXT,
256 )
259def test_min_read_version_too_high_rejected() -> None:
260 """Verify min_read_version higher than the reader major is rejected."""
261 # Pydantic mode='after' re-raises ArchiveReadError without
262 # wrapping it in ValidationError.
263 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
264 _DummyArchiveTree.model_validate({"schema_version": "2.0.0", "min_read_version": 2})
267def test_image_schema_has_id_and_title() -> None:
268 """Verify Image's serialization-mode schema has ``$id`` / ``title`` set."""
269 schema = ImageSerializationModel.model_json_schema(mode="serialization")
270 assert schema["$id"] == "https://images.lsst.io/schemas/image-1.0.0"
271 assert schema["title"] == "image"
274def test_constants_are_declared() -> None:
275 """Verify all three ClassVars are declared and well-formed everywhere."""
276 found = list(iter_concrete_archive_tree_subclasses())
277 assert len(found) > 0
278 for sub in found:
279 check_archive_tree_class_invariants(sub)
282def test_schema_names_unique() -> None:
283 """Verify all SCHEMA_NAME values across concrete subclasses are unique."""
284 names: dict[str, type] = {}
285 for sub in iter_concrete_archive_tree_subclasses():
286 # Skip our local _DummyArchiveTree (it lives in a test module).
287 if sub.__module__.startswith("tests."): 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 continue
289 # Skip Pydantic generic parametrisations (e.g.
290 # MaskedImageSerializationModel[TypeVar]); only the original
291 # generic class counts. A parametrised form has a non-empty
292 # __pydantic_generic_metadata__["args"].
293 generic_meta = getattr(sub, "__pydantic_generic_metadata__", {})
294 if generic_meta.get("args"):
295 continue
296 existing = names.get(sub.SCHEMA_NAME)
297 if existing is not None: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 pytest.fail(
299 f"Duplicate SCHEMA_NAME {sub.SCHEMA_NAME!r}: {sub.__qualname__} vs {existing.__qualname__}"
300 )
301 names[sub.SCHEMA_NAME] = sub
304@pytest.fixture
305def fixture_path() -> Path:
306 """Return the path to the committed image schema fixture."""
307 path = current_fixture_path(FIXTURE_DIR, "image")
308 assert path.exists()
309 return path
312def test_min_read_too_high_raises(fixture_path: Path) -> None:
313 """Verify setting min_read_version above reader major rejects the read."""
314 tree = json_module.loads(fixture_path.read_text())
315 tree["min_read_version"] = 99
316 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
317 ImageSerializationModel.model_validate(tree)
320def test_higher_major_with_low_min_read_succeeds(fixture_path: Path) -> None:
321 """Verify a high schema_version with low min_read_version reads
322 silently.
323 """
324 tree = json_module.loads(fixture_path.read_text())
325 tree["schema_version"] = "99.0.0"
326 tree["min_read_version"] = 1
327 # Asymmetric escape: a 99.0.0 file that declares it's safe for
328 # major-1 readers reads silently.
329 instance = ImageSerializationModel.model_validate(tree)
330 # And gets normalised back to in-code values.
331 assert instance.schema_version == "1.0.0"
332 assert instance.min_read_version == 1
335class _WritableDevObject:
336 _archive_default_name = None
338 def serialize(self, archive):
339 return _DevTree()
342class _WritableFinalObject:
343 _archive_default_name = None
345 def serialize(self, archive):
346 return _FinalTree()
349def test_serialize_root_warns_for_development() -> None:
350 """Verify serialize_root warns when the tree uses a development schema."""
351 with pytest.warns(DevelopmentSchemaWarning, match="in_dev_helper_test_dev"):
352 JsonOutputArchive().serialize_root(_WritableDevObject())
355def test_serialize_root_silent_for_finalized() -> None:
356 """Verify serialize_root does not warn for a finalized schema."""
357 with warnings.catch_warnings():
358 warnings.simplefilter("error")
359 JsonOutputArchive().serialize_root(_WritableFinalObject())