Coverage for tests/test_schema_migration.py: 98%
89 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-08 11:10 +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.
11"""The schema migration chain.
13Migration is proven by purpose-built schemas from ``schema_doubles``, which is
14never frozen, published, or part of shipped data.
15"""
17from __future__ import annotations
19from pathlib import Path
20from typing import Any, ClassVar
22import pydantic
23import pytest
24from schema_doubles import (
25 ChainTestModel,
26 GapTestModel,
27 MigrationIsolationTestModel,
28 MigrationTestModel,
29 ProjectionTestModel,
30 UnmigratableTestModel,
31)
33from lsst.images.serialization import ArchiveReadError, ArchiveTree, InputArchive, migration
34from lsst.images.serialization._migrations import _MIGRATIONS
35from lsst.images.tests import iter_schema_fixtures, read_fixture_tree
37DOUBLE_DIR = Path(__file__).parent / "data" / "fixture_doubles"
40def test_registration() -> None:
41 """Verify the decorator registers one entry per (name, from_major)."""
42 assert ("migration_test", 1) in _MIGRATIONS
43 assert ("migration_isolation_test", 1) in _MIGRATIONS
44 assert ("chain_test", 1) in _MIGRATIONS
45 assert ("chain_test", 2) in _MIGRATIONS
46 assert ("gap_test", 2) in _MIGRATIONS
49def test_registering_twice_raises() -> None:
50 """Verify a duplicate registration is refused rather than silently won."""
51 with pytest.raises(RuntimeError, match="already registered"):
53 @migration("migration_test", 1)
54 def _duplicate(data: dict[str, Any]) -> dict[str, Any]:
55 return data
58def test_migration_morphs_an_old_tree() -> None:
59 """Verify a 1.0.0 tree reads through the chain into the 2.0.0 shape."""
60 tree = MigrationTestModel.model_validate(
61 {"schema_version": "1.0.0", "min_read_version": 1, "original": "hello"}
62 )
63 assert tree.renamed == "hello"
64 assert tree.schema_version == "2.0.0"
65 assert tree.min_read_version == 2
68def test_migration_leaves_a_current_tree_alone() -> None:
69 """Verify a current-version tree is not passed through any step."""
70 tree = MigrationTestModel.model_validate(
71 {"schema_version": "2.0.0", "min_read_version": 2, "renamed": "hello"}
72 )
73 assert tree.renamed == "hello"
76def test_migration_chains_across_two_majors() -> None:
77 """Verify a 1.0.0 tree runs both registered steps, in order."""
78 tree = ChainTestModel.model_validate({"schema_version": "1.0.0", "min_read_version": 1})
79 assert tree.steps == ["1->2", "2->3"]
82def test_failed_union_migration_does_not_mutate_the_fallback_input() -> None:
83 """Verify a failed migration candidate is isolated from later variants.
85 ``MigrationIsolationTestModel`` and its migration live in
86 ``schema_doubles`` rather than here: registration is process-global and
87 refuses to be replaced, so registering from inside a test body would make
88 the test fail on any second run in the same process.
89 """
91 class _Fallback(pydantic.BaseModel):
92 schema_version: str
93 min_read_version: int
94 steps: list[str]
96 class _Wrapper(pydantic.BaseModel):
97 member: MigrationIsolationTestModel | _Fallback = pydantic.Field(union_mode="left_to_right")
99 payload = {"member": {"schema_version": "1.0.0", "min_read_version": 1, "steps": ["original"]}}
100 wrapper = _Wrapper.model_validate(payload)
101 assert isinstance(wrapper.member, _Fallback)
102 assert wrapper.member.steps == ["original"]
103 assert payload["member"]["steps"] == ["original"]
106def test_missing_step_raises_a_clean_error() -> None:
107 """Verify a genuine migration gap names the missing step, not a stray
108 error.
110 GapTestModel registers only its 2->3 step, so a tree stamped at major 1
111 has nothing registered for 1->2 -- a real gap, unlike a schema that has
112 registered no migration at all (see the fallthrough tests below).
114 The gap is raised from a ``mode="before"`` validator as an exception that
115 is deliberately both `ArchiveReadError` and `ValueError` (see
116 ``_MigrationGapError``), so a ``left_to_right`` union can treat it as "try
117 the next variant" (`test_migration_gap_falls_through_a_left_to_right_union`
118 below). That same dual nature means Pydantic -- which wraps any
119 validator-raised `ValueError` into its own `pydantic.ValidationError`,
120 indistinguishably from a nested union trial or a standalone top-level
121 call -- reports this one as a `pydantic.ValidationError` here rather than
122 a bare `ArchiveReadError`, exactly as it already does for other schema
123 incompatibilities (see ``test_retire_double_fixture_is_rejected``); the
124 message naming the missing step survives that wrapping unchanged.
125 """
126 with pytest.raises((ArchiveReadError, pydantic.ValidationError), match="no migration from major 1 to 2"):
127 GapTestModel.model_validate({"schema_version": "1.0.0", "min_read_version": 1})
130def test_schemas_without_migrations_are_untouched() -> None:
131 """Verify the fast path leaves a migration-free schema alone.
133 UnmigratableTestModel has no registered migration despite being at major
134 2, and reading a tree stamped at an older version must not raise a
135 missing-step error: only a schema that has itself registered a migration
136 should ever reach the chaining loop, regardless of what some unrelated
137 schema (e.g. GapTestModel, registered above) has registered. Before the
138 gate was fixed to check only this schema's own registrations, this exact
139 read raised the gap error as soon as any other schema in the process had
140 registered a migration.
141 """
142 assert UnmigratableTestModel.SCHEMA_NAME not in {name for name, _ in _MIGRATIONS}
143 tree = UnmigratableTestModel.model_validate(
144 {"schema_version": "1.0.0", "min_read_version": 1, "required_at_v2": "present"}
145 )
146 assert tree.required_at_v2 == "present" # type: ignore[attr-defined]
147 assert tree.schema_version == "2.0.0"
150def test_no_migration_schema_falls_through_to_a_validation_error() -> None:
151 """Verify a migration-free schema falls through to ordinary Pydantic
152 validation rather than a migration-gap error when the older tree
153 genuinely cannot satisfy the current shape.
155 UnmigratableTestModel's ``required_at_v2`` has no default, so a tree that
156 omits it fails Pydantic's own validation -- the other half of what
157 :ref:`lsst.images-schema-versioning` promises for a schema with no
158 migration: either in-model backfill handles the older tree (the test
159 above), or Pydantic reports a validation error, but never the clean
160 migration-gap error, which is reserved for a schema that has actually
161 opted into migrations.
162 """
163 with pytest.raises(pydantic.ValidationError):
164 UnmigratableTestModel.model_validate({"schema_version": "1.0.0", "min_read_version": 1})
167def test_migration_gap_falls_through_a_left_to_right_union() -> None:
168 """Verify a migration gap in one union member does not abort the union.
170 ``ArchiveReadError`` derives from `RuntimeError`, which Pydantic does not
171 treat as a failed union candidate; the gap raises a narrower exception
172 that is also a `ValueError`, so Pydantic moves on to the next variant
173 instead of the whole union attempt failing. This is the shape of
174 `~lsst.images._visit_image.VisitImageSerializationModel.psf`: a
175 ``left_to_right`` union where an earlier variant may no longer accept an
176 older tree while a later one still matches it directly.
177 """
179 class _Wrapper(pydantic.BaseModel):
180 member: GapTestModel | ProjectionTestModel = pydantic.Field(union_mode="left_to_right")
182 wrapper = _Wrapper.model_validate({"member": {"schema_version": "1.0.0", "min_read_version": 1}})
183 assert isinstance(wrapper.member, ProjectionTestModel)
186@pytest.mark.parametrize("stamp", ["not.a.version", "", 2, None])
187def test_malformed_stamp_falls_through_a_left_to_right_union(stamp: object) -> None:
188 """Verify an unparsable stamp fails one variant, not the whole union.
190 The on-disk major is parsed in the same ``mode="before"`` validator as the
191 migration chain, ahead of any field validation, so it runs for every
192 migratable variant a union tries -- including variants the payload does
193 not belong to. A stamp the reader cannot parse must therefore fail that
194 one candidate the way a migration gap does, rather than abort the union.
195 """
197 class _Fallback(pydantic.BaseModel):
198 schema_version: Any
199 min_read_version: int
201 class _Wrapper(pydantic.BaseModel):
202 member: MigrationTestModel | _Fallback = pydantic.Field(union_mode="left_to_right")
204 wrapper = _Wrapper.model_validate({"member": {"schema_version": stamp, "min_read_version": 1}})
205 assert isinstance(wrapper.member, _Fallback)
208def test_malformed_stamp_names_the_unparsable_version() -> None:
209 """Verify a standalone read reports which stamp could not be parsed.
211 Wrapped by Pydantic for the reason
212 ``test_missing_step_raises_a_clean_error`` describes, with the message
213 preserved.
214 """
215 with pytest.raises((ArchiveReadError, pydantic.ValidationError), match="has non-integer major"):
216 MigrationTestModel.model_validate({"schema_version": "x.0.0", "min_read_version": 1})
219def test_nested_subtree_migrates() -> None:
220 """Verify a nested old sub-tree is morphed by its own registered step."""
222 class _Parent(ArchiveTree):
223 SCHEMA_NAME: ClassVar[str] = "migration_parent_test"
224 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
225 MIN_READ_VERSION: ClassVar[int] = 1
226 PUBLIC_TYPE: ClassVar[type] = dict
228 child: MigrationTestModel = pydantic.Field(description="An embedded sub-tree.")
230 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
231 return {"child": self.child.deserialize(archive)}
233 parent = _Parent.model_validate(
234 {
235 "schema_version": "1.0.0",
236 "min_read_version": 1,
237 "child": {"schema_version": "1.0.0", "min_read_version": 1, "original": "nested"},
238 }
239 )
240 assert parent.child.renamed == "nested"
243def test_migration_test_fixture_reads_from_disk() -> None:
244 """Verify the declared projection divergence transforms correctly."""
245 fixtures = {f.version: f for f in iter_schema_fixtures(DOUBLE_DIR) if f.name == "migration_test"}
246 tree = read_fixture_tree(fixtures["1.0.0"])
247 assert tree.renamed == "exemplar" # type: ignore[attr-defined]
248 current = read_fixture_tree(fixtures["2.0.0"])
249 assert current.renamed == "exemplar" # type: ignore[attr-defined]