Coverage for tests/schema_doubles.py: 94%
81 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-09-30 11:30 +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"""Purpose-built schemas that exist only to exercise the fixture machinery.
13These are defined in the test tree, so ``available_schema_classes`` — which
14filters by defining module — never sees them: they are never frozen, never
15published, and never part of shipped data. Their fixtures live under
16``tests/data/fixture_doubles`` in the same layout as the real fixture tree, so
17the same ladder runs over both.
19This is a plain module rather than a test module so importing it registers the
20schemas deterministically, independent of pytest's collection order.
21"""
23from __future__ import annotations
25__all__ = (
26 "ChainTestModel",
27 "GapTestModel",
28 "MigrationIsolationTestModel",
29 "MigrationTestModel",
30 "ProjectionTestModel",
31 "RetireTestModel",
32 "UnmigratableTestModel",
33)
35from typing import Any, ClassVar
37import pydantic
39from lsst.images.serialization import ArchiveTree, InputArchive, migration
42class ProjectionTestModel(ArchiveTree):
43 """Schema whose 1.1.0 adds an optional field over 1.0.0.
45 The additive common case: the model absorbs a 1.0.0 tree unaided, so the
46 projection oracle must pass on this pair with no escape hatch. This is the
47 oracle's positive control.
48 """
50 SCHEMA_NAME: ClassVar[str] = "projection_test"
51 SCHEMA_VERSION: ClassVar[str] = "1.1.0"
52 MIN_READ_VERSION: ClassVar[int] = 1
53 PUBLIC_TYPE: ClassVar[type] = dict
55 kept: str = pydantic.Field(default="", description="Present at both versions.")
56 added: int = pydantic.Field(default=0, description="Added at 1.1.0.")
58 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
59 return {"kept": self.kept, "added": self.added}
62class RetireTestModel(ArchiveTree):
63 """Schema whose 2.0.0 shape a 1.0.0 tree cannot satisfy.
65 ``required_at_v2`` has no default and no migration is registered, so the
66 1.0.0 fixture can no longer be read. That is what retirement means, and
67 the fixture under ``retired/`` asserts it.
68 """
70 SCHEMA_NAME: ClassVar[str] = "retire_test"
71 SCHEMA_VERSION: ClassVar[str] = "2.0.0"
72 MIN_READ_VERSION: ClassVar[int] = 2
73 PUBLIC_TYPE: ClassVar[type] = dict
75 required_at_v2: str = pydantic.Field(description="Required from 2.0.0 on.")
77 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
78 return {"required_at_v2": self.required_at_v2}
81class MigrationTestModel(ArchiveTree):
82 """Purpose-built schema whose 2.0.0 shape renames a 1.0.0 field.
84 Pydantic alone cannot validate a 1.0.0 tree, because ``renamed`` is
85 required and ``original`` is not a field, so a successful read proves the
86 migration chain did the work.
87 """
89 SCHEMA_NAME: ClassVar[str] = "migration_test"
90 SCHEMA_VERSION: ClassVar[str] = "2.0.0"
91 MIN_READ_VERSION: ClassVar[int] = 2
92 PUBLIC_TYPE: ClassVar[type] = dict
94 renamed: str = pydantic.Field(description="Renamed from 'original' in 2.0.0.")
96 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
97 return {"renamed": self.renamed}
100class ChainTestModel(ArchiveTree):
101 """Purpose-built schema at 3.0.0, to prove steps chain across majors."""
103 SCHEMA_NAME: ClassVar[str] = "chain_test"
104 SCHEMA_VERSION: ClassVar[str] = "3.0.0"
105 MIN_READ_VERSION: ClassVar[int] = 1
106 PUBLIC_TYPE: ClassVar[type] = dict
108 steps: list[str] = pydantic.Field(default_factory=list, description="Migrations applied.")
110 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
111 return {"steps": self.steps}
114class UnmigratableTestModel(ArchiveTree):
115 """Purpose-built schema at 2.0.0 with no migration registered."""
117 SCHEMA_NAME: ClassVar[str] = "unmigratable_test"
118 SCHEMA_VERSION: ClassVar[str] = "2.0.0"
119 MIN_READ_VERSION: ClassVar[int] = 1
120 PUBLIC_TYPE: ClassVar[type] = dict
122 required_at_v2: str = pydantic.Field(description="Required from 2.0.0 on.")
124 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
125 return {"required_at_v2": self.required_at_v2}
128class GapTestModel(ArchiveTree):
129 """Purpose-built schema at 3.0.0 that registers only a non-adjacent step.
131 Only the 2->3 step is registered below, so a tree stamped at major 1 hits
132 a genuine migration gap: nothing is registered for 1->2. This is
133 distinct from a schema with no migrations at all (`UnmigratableTestModel`),
134 which falls through to ordinary validation instead of raising a gap
135 error.
136 """
138 SCHEMA_NAME: ClassVar[str] = "gap_test"
139 SCHEMA_VERSION: ClassVar[str] = "3.0.0"
140 MIN_READ_VERSION: ClassVar[int] = 1
141 PUBLIC_TYPE: ClassVar[type] = dict
143 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
144 return {}
147class MigrationIsolationTestModel(ArchiveTree):
148 """Schema whose migration mutates the tree and then fails validation.
150 ``required`` has no default and the migration below appends to a nested
151 list, so a tree that omits ``required`` is mutated and *then* rejected --
152 the shape a union needs to isolate, since the next variant must see the
153 unmutated input.
154 """
156 SCHEMA_NAME: ClassVar[str] = "migration_isolation_test"
157 SCHEMA_VERSION: ClassVar[str] = "2.0.0"
158 MIN_READ_VERSION: ClassVar[int] = 1
159 PUBLIC_TYPE: ClassVar[type] = dict
161 steps: list[str] = pydantic.Field(description="Migrations applied.")
162 required: str = pydantic.Field(description="Required, so validation fails without it.")
164 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:
165 return {"steps": self.steps, "required": self.required}
168@migration("migration_test", 1)
169def _migration_test_1_to_2(data: dict[str, Any]) -> dict[str, Any]:
170 """Rename 'original' to 'renamed' for the 2.0.0 shape.
172 Parameters
173 ----------
174 data
175 On-disk tree at major 1.
176 """
177 data["renamed"] = data.pop("original")
178 return data
181@migration("chain_test", 1)
182def _chain_test_1_to_2(data: dict[str, Any]) -> dict[str, Any]:
183 """Record that the 1 -> 2 step ran.
185 Parameters
186 ----------
187 data
188 On-disk tree at major 1.
189 """
190 data.setdefault("steps", []).append("1->2")
191 return data
194@migration("chain_test", 2)
195def _chain_test_2_to_3(data: dict[str, Any]) -> dict[str, Any]:
196 """Record that the 2 -> 3 step ran.
198 Parameters
199 ----------
200 data
201 On-disk tree at major 2.
202 """
203 data.setdefault("steps", []).append("2->3")
204 return data
207@migration("migration_isolation_test", 1)
208def _migration_isolation_test_1_to_2(data: dict[str, Any]) -> dict[str, Any]:
209 """Mutate a nested list, so an unisolated candidate is visibly dirty.
211 Parameters
212 ----------
213 data
214 On-disk tree at major 1.
215 """
216 data.setdefault("steps", []).append("migration ran")
217 return data
220@migration("gap_test", 2)
221def _gap_test_2_to_3(data: dict[str, Any]) -> dict[str, Any]:
222 """Advance a major-2 tree to major 3; unreachable from major 1.
224 Deliberately not registered for major 1, so a tree stamped at major 1
225 exercises a genuine migration gap rather than the adjacent case.
227 Parameters
228 ----------
229 data
230 On-disk tree at major 2.
231 """
232 return data