Coverage for tests/test_schema_versioning.py: 97%
107 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:24 -0700
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
15from pathlib import Path
16from typing import Any, ClassVar
18import pydantic
19import pytest
21from lsst.images import ImageSerializationModel
22from lsst.images.serialization import ArchiveReadError, ArchiveTree, InputArchive
23from lsst.images.serialization._common import _check_compat, _check_format_version
24from lsst.images.tests import check_archive_tree_class_invariants, iter_concrete_archive_tree_subclasses
26SCHEMA_DIR = Path(__file__).parent / "data" / "schema_v1"
29class _DummyArchiveTree(ArchiveTree):
30 """Minimal concrete ArchiveTree for testing the base-class machinery."""
32 SCHEMA_NAME: ClassVar[str] = "dummy"
33 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
34 MIN_READ_VERSION: ClassVar[int] = 1
35 PUBLIC_TYPE: ClassVar[type] = object
37 def deserialize(
38 self, archive: InputArchive[Any], **kwargs: Any
39 ) -> Any: # pragma: no cover - never invoked
40 raise NotImplementedError()
43def test_silent_when_min_read_satisfied() -> None:
44 """Verify no error when min_read_version equals reader major."""
45 _check_compat("foo", "1.0.0", 1, "1.0.0")
48def test_silent_when_on_disk_major_is_lower() -> None:
49 """Verify no error when a 1.0.0 file is read by 2.0.0 code."""
50 _check_compat("foo", "1.0.0", 1, "2.0.0")
53def test_silent_when_on_disk_major_is_higher_but_min_read_low() -> None:
54 """Verify no error when a 2.0.0 file is safe for major-1 readers."""
55 _check_compat("foo", "2.0.0", 1, "1.0.0")
58def test_raises_when_min_read_exceeds_reader_major() -> None:
59 """Verify ArchiveReadError is raised when min_read_version exceeds
60 major.
61 """
62 with pytest.raises(ArchiveReadError) as exc_info:
63 _check_compat("foo", "2.0.0", 2, "1.0.0")
64 assert "foo" in str(exc_info.value)
65 assert ">= 2" in str(exc_info.value)
68def test_format_version_silent_when_equal() -> None:
69 """Verify no error when on-disk format version equals reader version."""
70 _check_format_version("fits", 1, 1)
73def test_format_version_silent_when_on_disk_lower() -> None:
74 """Verify no error when on-disk format version is lower than reader's."""
75 _check_format_version("fits", 1, 2)
78def test_format_version_raises_when_on_disk_higher() -> None:
79 """Verify ArchiveReadError when on-disk format version is too new."""
80 with pytest.raises(ArchiveReadError):
81 _check_format_version("fits", 2, 1)
84def test_default_values_filled_from_classvars() -> None:
85 """Verify default instance values are filled from class-var constants."""
86 instance = _DummyArchiveTree()
87 assert instance.schema_version == "1.0.0"
88 assert instance.min_read_version == 1
91def test_schema_url_is_computed() -> None:
92 """Verify schema_url is computed from SCHEMA_NAME and SCHEMA_VERSION."""
93 instance = _DummyArchiveTree()
94 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
97def test_schema_url_appears_in_dump() -> None:
98 """Verify schema_url, schema_version, and min_read_version are in dumps."""
99 instance = _DummyArchiveTree()
100 dumped = instance.model_dump()
101 assert dumped["schema_url"] == "https://images.lsst.io/schemas/dummy-1.0.0"
102 assert dumped["schema_version"] == "1.0.0"
103 assert dumped["min_read_version"] == 1
106def test_schema_url_ignored_in_input() -> None:
107 """Verify schema_url in input data is ignored; computed value is used."""
108 # Pydantic's default extra='ignore' drops it from inputs.
109 instance = _DummyArchiveTree.model_validate(
110 {"schema_url": "https://example.com/wrong", "schema_version": "1.0.0", "min_read_version": 1}
111 )
112 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
115def test_normalises_to_in_code_values() -> None:
116 """Verify an older file's schema values are normalised to in-code
117 values.
118 """
119 # An older file's values are normalised on load.
120 instance = _DummyArchiveTree.model_validate({"schema_version": "0.9.0", "min_read_version": 1})
121 assert instance.schema_version == "1.0.0"
122 assert instance.min_read_version == 1
125def test_absent_fields_default_to_legacy() -> None:
126 """Verify absent version fields default to the legacy v1 values."""
127 instance = _DummyArchiveTree.model_validate({})
128 assert instance.schema_version == "1.0.0"
129 assert instance.min_read_version == 1
132def test_min_read_version_too_high_rejected() -> None:
133 """Verify min_read_version higher than the reader major is rejected."""
134 # Pydantic mode='after' re-raises ArchiveReadError without
135 # wrapping it in ValidationError.
136 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
137 _DummyArchiveTree.model_validate({"schema_version": "2.0.0", "min_read_version": 2})
140def test_image_schema_has_id_and_title() -> None:
141 """Verify Image's serialization-mode schema has ``$id`` / ``title`` set."""
142 schema = ImageSerializationModel.model_json_schema(mode="serialization")
143 assert schema["$id"] == "https://images.lsst.io/schemas/image-1.0.0"
144 assert schema["title"] == "image"
147def test_constants_are_declared() -> None:
148 """Verify all three ClassVars are declared and well-formed everywhere."""
149 found = list(iter_concrete_archive_tree_subclasses())
150 assert len(found) > 0
151 for sub in found:
152 check_archive_tree_class_invariants(sub)
155def test_schema_names_unique() -> None:
156 """Verify all SCHEMA_NAME values across concrete subclasses are unique."""
157 names: dict[str, type] = {}
158 for sub in iter_concrete_archive_tree_subclasses():
159 # Skip our local _DummyArchiveTree (it lives in a test module).
160 if sub.__module__.startswith("tests."): 160 ↛ 161line 160 didn't jump to line 161 because the condition on line 160 was never true
161 continue
162 # Skip Pydantic generic parametrisations (e.g.
163 # MaskedImageSerializationModel[TypeVar]); only the original
164 # generic class counts. A parametrised form has a non-empty
165 # __pydantic_generic_metadata__["args"].
166 generic_meta = getattr(sub, "__pydantic_generic_metadata__", {})
167 if generic_meta.get("args"):
168 continue
169 existing = names.get(sub.SCHEMA_NAME)
170 if existing is not None: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 pytest.fail(
172 f"Duplicate SCHEMA_NAME {sub.SCHEMA_NAME!r}: {sub.__qualname__} vs {existing.__qualname__}"
173 )
174 names[sub.SCHEMA_NAME] = sub
177@pytest.fixture
178def fixture_path() -> Path:
179 """Return the path to the committed image.json schema fixture."""
180 path = SCHEMA_DIR / "image.json"
181 assert path.exists()
182 return path
185def test_min_read_too_high_raises(fixture_path: Path) -> None:
186 """Verify setting min_read_version above reader major rejects the read."""
187 tree = json_module.loads(fixture_path.read_text())
188 tree["min_read_version"] = 99
189 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
190 ImageSerializationModel.model_validate(tree)
193def test_higher_major_with_low_min_read_succeeds(fixture_path: Path) -> None:
194 """Verify a high schema_version with low min_read_version reads
195 silently.
196 """
197 tree = json_module.loads(fixture_path.read_text())
198 tree["schema_version"] = "99.0.0"
199 tree["min_read_version"] = 1
200 # Asymmetric escape: a 99.0.0 file that declares it's safe for
201 # major-1 readers reads silently.
202 instance = ImageSerializationModel.model_validate(tree)
203 # And gets normalised back to in-code values.
204 assert instance.schema_version == "1.0.0"
205 assert instance.min_read_version == 1
208def test_absent_fields_default_to_legacy_fixture(fixture_path: Path) -> None:
209 """Verify stripping the version fields entirely reads with v1 defaults."""
210 tree = json_module.loads(fixture_path.read_text())
211 del tree["schema_version"]
212 del tree["min_read_version"]
213 del tree["schema_url"]
214 instance = ImageSerializationModel.model_validate(tree)
215 assert instance.schema_version == "1.0.0"
216 assert instance.min_read_version == 1