Coverage for tests/test_schema_versioning.py: 97%
146 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 02:46 +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 _check_compat, _check_format_version
33from lsst.images.tests import check_archive_tree_class_invariants, iter_concrete_archive_tree_subclasses
35SCHEMA_DIR = Path(__file__).parent / "data" / "schema_v1"
38class _DummyArchiveTree(ArchiveTree):
39 """Minimal concrete ArchiveTree for testing the base-class machinery."""
41 SCHEMA_NAME: ClassVar[str] = "dummy"
42 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
43 MIN_READ_VERSION: ClassVar[int] = 1
44 PUBLIC_TYPE: ClassVar[type] = object
46 def deserialize(
47 self, archive: InputArchive[Any], **kwargs: Any
48 ) -> Any: # pragma: no cover - never invoked
49 raise NotImplementedError()
52class _DevTree(ArchiveTree):
53 """Development-version tree double."""
55 SCHEMA_NAME = "in_dev_helper_test_dev"
56 SCHEMA_VERSION = "1.0.0.dev0"
57 MIN_READ_VERSION = 1
58 PUBLIC_TYPE = object
60 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked
61 raise NotImplementedError()
64class _FinalTree(ArchiveTree):
65 """Finalized-version tree double."""
67 SCHEMA_NAME = "in_dev_helper_test_final"
68 SCHEMA_VERSION = "1.0.0"
69 MIN_READ_VERSION = 1
70 PUBLIC_TYPE = object
72 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked
73 raise NotImplementedError()
76def test_is_development_version() -> None:
77 """Verify is_development_version detects development versions correctly."""
78 assert is_development_version("1.0.0.dev0")
79 assert is_development_version("2.3.4.dev12")
80 assert not is_development_version("1.0.0")
81 assert not is_development_version("2.3.4")
84def test_warn_for_development_schemas_warns() -> None:
85 """Verify warn_for_development_schemas warns for development schemas."""
86 with pytest.warns(DevelopmentSchemaWarning, match="in_dev_helper_test_dev"):
87 warn_for_development_schemas(_DevTree())
90def test_warn_for_development_schemas_silent_when_finalized() -> None:
91 """Verify warn_for_development_schemas is silent for finalized schemas."""
92 with warnings.catch_warnings():
93 warnings.simplefilter("error")
94 warn_for_development_schemas(_FinalTree())
97def test_silent_when_min_read_satisfied() -> None:
98 """Verify no error when min_read_version equals reader major."""
99 _check_compat("foo", "1.0.0", 1, "1.0.0")
102def test_silent_when_on_disk_major_is_lower() -> None:
103 """Verify no error when a 1.0.0 file is read by 2.0.0 code."""
104 _check_compat("foo", "1.0.0", 1, "2.0.0")
107def test_silent_when_on_disk_major_is_higher_but_min_read_low() -> None:
108 """Verify no error when a 2.0.0 file is safe for major-1 readers."""
109 _check_compat("foo", "2.0.0", 1, "1.0.0")
112def test_raises_when_min_read_exceeds_reader_major() -> None:
113 """Verify ArchiveReadError is raised when min_read_version exceeds
114 major.
115 """
116 with pytest.raises(ArchiveReadError) as exc_info:
117 _check_compat("foo", "2.0.0", 2, "1.0.0")
118 assert "foo" in str(exc_info.value)
119 assert ">= 2" in str(exc_info.value)
122def test_format_version_silent_when_equal() -> None:
123 """Verify no error when on-disk format version equals reader version."""
124 _check_format_version("fits", 1, 1)
127def test_format_version_silent_when_on_disk_lower() -> None:
128 """Verify no error when on-disk format version is lower than reader's."""
129 _check_format_version("fits", 1, 2)
132def test_format_version_raises_when_on_disk_higher() -> None:
133 """Verify ArchiveReadError when on-disk format version is too new."""
134 with pytest.raises(ArchiveReadError):
135 _check_format_version("fits", 2, 1)
138def test_default_values_filled_from_classvars() -> None:
139 """Verify default instance values are filled from class-var constants."""
140 instance = _DummyArchiveTree()
141 assert instance.schema_version == "1.0.0"
142 assert instance.min_read_version == 1
145def test_schema_url_is_computed() -> None:
146 """Verify schema_url is computed from SCHEMA_NAME and SCHEMA_VERSION."""
147 instance = _DummyArchiveTree()
148 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
151def test_schema_url_appears_in_dump() -> None:
152 """Verify schema_url, schema_version, and min_read_version are in dumps."""
153 instance = _DummyArchiveTree()
154 dumped = instance.model_dump()
155 assert dumped["schema_url"] == "https://images.lsst.io/schemas/dummy-1.0.0"
156 assert dumped["schema_version"] == "1.0.0"
157 assert dumped["min_read_version"] == 1
160def test_schema_url_ignored_in_input() -> None:
161 """Verify schema_url in input data is ignored; computed value is used."""
162 # Pydantic's default extra='ignore' drops it from inputs.
163 instance = _DummyArchiveTree.model_validate(
164 {"schema_url": "https://example.com/wrong", "schema_version": "1.0.0", "min_read_version": 1}
165 )
166 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
169def test_normalises_to_in_code_values() -> None:
170 """Verify an older file's schema values are normalised to in-code
171 values.
172 """
173 # An older file's values are normalised on load.
174 instance = _DummyArchiveTree.model_validate({"schema_version": "0.9.0", "min_read_version": 1})
175 assert instance.schema_version == "1.0.0"
176 assert instance.min_read_version == 1
179def test_absent_fields_default_to_legacy() -> None:
180 """Verify absent version fields default to the legacy v1 values."""
181 instance = _DummyArchiveTree.model_validate({})
182 assert instance.schema_version == "1.0.0"
183 assert instance.min_read_version == 1
186def test_min_read_version_too_high_rejected() -> None:
187 """Verify min_read_version higher than the reader major is rejected."""
188 # Pydantic mode='after' re-raises ArchiveReadError without
189 # wrapping it in ValidationError.
190 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
191 _DummyArchiveTree.model_validate({"schema_version": "2.0.0", "min_read_version": 2})
194def test_image_schema_has_id_and_title() -> None:
195 """Verify Image's serialization-mode schema has ``$id`` / ``title`` set."""
196 schema = ImageSerializationModel.model_json_schema(mode="serialization")
197 assert schema["$id"] == "https://images.lsst.io/schemas/image-1.0.0"
198 assert schema["title"] == "image"
201def test_constants_are_declared() -> None:
202 """Verify all three ClassVars are declared and well-formed everywhere."""
203 found = list(iter_concrete_archive_tree_subclasses())
204 assert len(found) > 0
205 for sub in found:
206 check_archive_tree_class_invariants(sub)
209def test_schema_names_unique() -> None:
210 """Verify all SCHEMA_NAME values across concrete subclasses are unique."""
211 names: dict[str, type] = {}
212 for sub in iter_concrete_archive_tree_subclasses():
213 # Skip our local _DummyArchiveTree (it lives in a test module).
214 if sub.__module__.startswith("tests."): 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true
215 continue
216 # Skip Pydantic generic parametrisations (e.g.
217 # MaskedImageSerializationModel[TypeVar]); only the original
218 # generic class counts. A parametrised form has a non-empty
219 # __pydantic_generic_metadata__["args"].
220 generic_meta = getattr(sub, "__pydantic_generic_metadata__", {})
221 if generic_meta.get("args"):
222 continue
223 existing = names.get(sub.SCHEMA_NAME)
224 if existing is not None: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 pytest.fail(
226 f"Duplicate SCHEMA_NAME {sub.SCHEMA_NAME!r}: {sub.__qualname__} vs {existing.__qualname__}"
227 )
228 names[sub.SCHEMA_NAME] = sub
231@pytest.fixture
232def fixture_path() -> Path:
233 """Return the path to the committed image.json schema fixture."""
234 path = SCHEMA_DIR / "image.json"
235 assert path.exists()
236 return path
239def test_min_read_too_high_raises(fixture_path: Path) -> None:
240 """Verify setting min_read_version above reader major rejects the read."""
241 tree = json_module.loads(fixture_path.read_text())
242 tree["min_read_version"] = 99
243 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
244 ImageSerializationModel.model_validate(tree)
247def test_higher_major_with_low_min_read_succeeds(fixture_path: Path) -> None:
248 """Verify a high schema_version with low min_read_version reads
249 silently.
250 """
251 tree = json_module.loads(fixture_path.read_text())
252 tree["schema_version"] = "99.0.0"
253 tree["min_read_version"] = 1
254 # Asymmetric escape: a 99.0.0 file that declares it's safe for
255 # major-1 readers reads silently.
256 instance = ImageSerializationModel.model_validate(tree)
257 # And gets normalised back to in-code values.
258 assert instance.schema_version == "1.0.0"
259 assert instance.min_read_version == 1
262def test_absent_fields_default_to_legacy_fixture(fixture_path: Path) -> None:
263 """Verify stripping the version fields entirely reads with v1 defaults."""
264 tree = json_module.loads(fixture_path.read_text())
265 del tree["schema_version"]
266 del tree["min_read_version"]
267 del tree["schema_url"]
268 instance = ImageSerializationModel.model_validate(tree)
269 assert instance.schema_version == "1.0.0"
270 assert instance.min_read_version == 1
273class _WritableDevObject:
274 _archive_default_name = None
276 def serialize(self, archive):
277 return _DevTree()
280class _WritableFinalObject:
281 _archive_default_name = None
283 def serialize(self, archive):
284 return _FinalTree()
287def test_serialize_root_warns_for_development() -> None:
288 """Verify serialize_root warns when the tree uses a development schema."""
289 with pytest.warns(DevelopmentSchemaWarning, match="in_dev_helper_test_dev"):
290 JsonOutputArchive().serialize_root(_WritableDevObject())
293def test_serialize_root_silent_for_finalized() -> None:
294 """Verify serialize_root does not warn for a finalized schema."""
295 with warnings.catch_warnings():
296 warnings.simplefilter("error")
297 JsonOutputArchive().serialize_root(_WritableFinalObject())