Coverage for tests/test_schema_versioning.py: 98%
151 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-27 02:32 -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
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 _DevTree(ArchiveTree):
61 """Development-version tree double."""
63 SCHEMA_NAME = "in_dev_helper_test_dev"
64 SCHEMA_VERSION = "1.0.0.dev0"
65 MIN_READ_VERSION = 1
66 PUBLIC_TYPE = object
68 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked
69 raise NotImplementedError()
72class _FinalTree(ArchiveTree):
73 """Finalized-version tree double."""
75 SCHEMA_NAME = "in_dev_helper_test_final"
76 SCHEMA_VERSION = "1.0.0"
77 MIN_READ_VERSION = 1
78 PUBLIC_TYPE = object
80 def deserialize(self, archive, **kwargs): # pragma: no cover - never invoked
81 raise NotImplementedError()
84def test_is_development_version() -> None:
85 """Verify is_development_version detects development versions correctly."""
86 assert is_development_version("1.0.0.dev0")
87 assert is_development_version("2.3.4.dev12")
88 assert not is_development_version("1.0.0")
89 assert not is_development_version("2.3.4")
92def test_warn_for_development_schemas_warns() -> None:
93 """Verify warn_for_development_schemas warns for development schemas."""
94 with pytest.warns(DevelopmentSchemaWarning, match="in_dev_helper_test_dev"):
95 warn_for_development_schemas(_DevTree())
98def test_warn_for_development_schemas_silent_when_finalized() -> None:
99 """Verify warn_for_development_schemas is silent for finalized schemas."""
100 with warnings.catch_warnings():
101 warnings.simplefilter("error")
102 warn_for_development_schemas(_FinalTree())
105def test_silent_when_min_read_satisfied() -> None:
106 """Verify no error when min_read_version equals reader major."""
107 _check_compat("foo", "1.0.0", 1, "1.0.0")
110def test_silent_when_on_disk_major_is_lower() -> None:
111 """Verify no error when a 1.0.0 file is read by 2.0.0 code."""
112 _check_compat("foo", "1.0.0", 1, "2.0.0")
115def test_silent_when_on_disk_major_is_higher_but_min_read_low() -> None:
116 """Verify no error when a 2.0.0 file is safe for major-1 readers."""
117 _check_compat("foo", "2.0.0", 1, "1.0.0")
120def test_raises_when_min_read_exceeds_reader_major() -> None:
121 """Verify ArchiveReadError is raised when min_read_version exceeds
122 major.
123 """
124 with pytest.raises(ArchiveReadError) as exc_info:
125 _check_compat("foo", "2.0.0", 2, "1.0.0")
126 assert "foo" in str(exc_info.value)
127 assert ">= 2" in str(exc_info.value)
130def test_format_version_silent_when_equal() -> None:
131 """Verify no error when on-disk format version equals reader version."""
132 _check_format_version("fits", 1, 1)
135def test_format_version_silent_when_on_disk_lower() -> None:
136 """Verify no error when on-disk format version is lower than reader's."""
137 _check_format_version("fits", 1, 2)
140def test_format_version_raises_when_on_disk_higher() -> None:
141 """Verify ArchiveReadError when on-disk format version is too new."""
142 with pytest.raises(ArchiveReadError):
143 _check_format_version("fits", 2, 1)
146def test_default_values_filled_from_classvars() -> None:
147 """Verify default instance values are filled from class-var constants."""
148 instance = _DummyArchiveTree()
149 assert instance.schema_version == "1.0.0"
150 assert instance.min_read_version == 1
153def test_schema_url_is_computed() -> None:
154 """Verify schema_url is computed from SCHEMA_NAME and SCHEMA_VERSION."""
155 instance = _DummyArchiveTree()
156 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
159def test_schema_url_appears_in_dump() -> None:
160 """Verify schema_url, schema_version, and min_read_version are in dumps."""
161 instance = _DummyArchiveTree()
162 dumped = instance.model_dump()
163 assert dumped["schema_url"] == "https://images.lsst.io/schemas/dummy-1.0.0"
164 assert dumped["schema_version"] == "1.0.0"
165 assert dumped["min_read_version"] == 1
168def test_schema_url_ignored_in_input() -> None:
169 """Verify schema_url in input data is ignored; computed value is used."""
170 # Pydantic's default extra='ignore' drops it from inputs.
171 instance = _DummyArchiveTree.model_validate(
172 {"schema_url": "https://example.com/wrong", "schema_version": "1.0.0", "min_read_version": 1}
173 )
174 assert instance.schema_url == "https://images.lsst.io/schemas/dummy-1.0.0"
177def test_normalises_to_in_code_values() -> None:
178 """Verify an older file's schema values are normalised to in-code
179 values.
180 """
181 # An older file's values are normalised on load.
182 instance = _DummyArchiveTree.model_validate({"schema_version": "0.9.0", "min_read_version": 1})
183 assert instance.schema_version == "1.0.0"
184 assert instance.min_read_version == 1
187def test_absent_fields_default_for_fresh_construction() -> None:
188 """Verify fresh in-memory construction may omit version fields."""
189 instance = _DummyArchiveTree.model_validate({})
190 assert instance.schema_version == "1.0.0"
191 assert instance.min_read_version == 1
194def test_archive_tree_requires_an_explicit_schema_version() -> None:
195 """Verify archive input cannot silently assume a version."""
196 with pytest.raises(pydantic.ValidationError, match="has no schema_version"):
197 _DummyArchiveTree.model_validate({}, context=_ARCHIVE_READ_CONTEXT)
200def test_archive_context_requires_versions_on_nested_trees() -> None:
201 """Verify the archive-read marker propagates into embedded schemas."""
203 class _ParentTree(ArchiveTree):
204 SCHEMA_NAME: ClassVar[str] = "version_context_parent_test"
205 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
206 MIN_READ_VERSION: ClassVar[int] = 1
207 PUBLIC_TYPE: ClassVar[type] = object
209 child: _DummyArchiveTree
211 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
212 raise NotImplementedError()
214 with pytest.raises(pydantic.ValidationError, match="dummy: archive tree has no schema_version"):
215 _ParentTree.model_validate(
216 {"schema_version": "1.0.0", "min_read_version": 1, "child": {}},
217 context=_ARCHIVE_READ_CONTEXT,
218 )
221def test_min_read_version_too_high_rejected() -> None:
222 """Verify min_read_version higher than the reader major is rejected."""
223 # Pydantic mode='after' re-raises ArchiveReadError without
224 # wrapping it in ValidationError.
225 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
226 _DummyArchiveTree.model_validate({"schema_version": "2.0.0", "min_read_version": 2})
229def test_image_schema_has_id_and_title() -> None:
230 """Verify Image's serialization-mode schema has ``$id`` / ``title`` set."""
231 schema = ImageSerializationModel.model_json_schema(mode="serialization")
232 assert schema["$id"] == "https://images.lsst.io/schemas/image-1.0.0"
233 assert schema["title"] == "image"
236def test_constants_are_declared() -> None:
237 """Verify all three ClassVars are declared and well-formed everywhere."""
238 found = list(iter_concrete_archive_tree_subclasses())
239 assert len(found) > 0
240 for sub in found:
241 check_archive_tree_class_invariants(sub)
244def test_schema_names_unique() -> None:
245 """Verify all SCHEMA_NAME values across concrete subclasses are unique."""
246 names: dict[str, type] = {}
247 for sub in iter_concrete_archive_tree_subclasses():
248 # Skip our local _DummyArchiveTree (it lives in a test module).
249 if sub.__module__.startswith("tests."): 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 continue
251 # Skip Pydantic generic parametrisations (e.g.
252 # MaskedImageSerializationModel[TypeVar]); only the original
253 # generic class counts. A parametrised form has a non-empty
254 # __pydantic_generic_metadata__["args"].
255 generic_meta = getattr(sub, "__pydantic_generic_metadata__", {})
256 if generic_meta.get("args"):
257 continue
258 existing = names.get(sub.SCHEMA_NAME)
259 if existing is not None: 259 ↛ 260line 259 didn't jump to line 260 because the condition on line 259 was never true
260 pytest.fail(
261 f"Duplicate SCHEMA_NAME {sub.SCHEMA_NAME!r}: {sub.__qualname__} vs {existing.__qualname__}"
262 )
263 names[sub.SCHEMA_NAME] = sub
266@pytest.fixture
267def fixture_path() -> Path:
268 """Return the path to the committed image schema fixture."""
269 path = current_fixture_path(FIXTURE_DIR, "image")
270 assert path.exists()
271 return path
274def test_min_read_too_high_raises(fixture_path: Path) -> None:
275 """Verify setting min_read_version above reader major rejects the read."""
276 tree = json_module.loads(fixture_path.read_text())
277 tree["min_read_version"] = 99
278 with pytest.raises((ArchiveReadError, pydantic.ValidationError)):
279 ImageSerializationModel.model_validate(tree)
282def test_higher_major_with_low_min_read_succeeds(fixture_path: Path) -> None:
283 """Verify a high schema_version with low min_read_version reads
284 silently.
285 """
286 tree = json_module.loads(fixture_path.read_text())
287 tree["schema_version"] = "99.0.0"
288 tree["min_read_version"] = 1
289 # Asymmetric escape: a 99.0.0 file that declares it's safe for
290 # major-1 readers reads silently.
291 instance = ImageSerializationModel.model_validate(tree)
292 # And gets normalised back to in-code values.
293 assert instance.schema_version == "1.0.0"
294 assert instance.min_read_version == 1
297class _WritableDevObject:
298 _archive_default_name = None
300 def serialize(self, archive):
301 return _DevTree()
304class _WritableFinalObject:
305 _archive_default_name = None
307 def serialize(self, archive):
308 return _FinalTree()
311def test_serialize_root_warns_for_development() -> None:
312 """Verify serialize_root warns when the tree uses a development schema."""
313 with pytest.warns(DevelopmentSchemaWarning, match="in_dev_helper_test_dev"):
314 JsonOutputArchive().serialize_root(_WritableDevObject())
317def test_serialize_root_silent_for_finalized() -> None:
318 """Verify serialize_root does not warn for a finalized schema."""
319 with warnings.catch_warnings():
320 warnings.simplefilter("error")
321 JsonOutputArchive().serialize_root(_WritableFinalObject())