Coverage for tests/test_frozen_schemas.py: 98%
169 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 09:32 +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 importlib.metadata
15import json
16from pathlib import Path
17from typing import Any, ClassVar
19import pydantic
20import pytest
22from lsst.images.serialization import (
23 ArchiveTree,
24 FrozenSchemaError,
25 InputArchive,
26 available_schema_classes,
27 check_frozen_schemas,
28 dump_schema,
29 frozen_schema_filename,
30 frozen_schema_path,
31 is_development_version,
32 write_frozen_schemas,
33)
34from lsst.images.serialization._frozen_schemas import _summary_description
36REPO_SCHEMA_DIR = Path(__file__).parent.parent / "schemas"
39class _ForeignArchiveTree(ArchiveTree):
40 """Concrete ArchiveTree defined outside lsst.images; registers itself on
41 class creation, like the test doubles in other test modules.
42 """
44 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_foreign"
45 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
46 MIN_READ_VERSION: ClassVar[int] = 1
47 PUBLIC_TYPE: ClassVar[type] = object
49 def deserialize(
50 self, archive: InputArchive[Any], **kwargs: Any
51 ) -> Any: # pragma: no cover - never invoked
52 raise NotImplementedError()
55class _CustomBaseArchiveTree(ArchiveTree):
56 """Concrete ArchiveTree with a third-party schema URL base."""
58 SCHEMA_URL_BASE: ClassVar[str] = "https://example.org/schemas"
59 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_custom_base"
60 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
61 MIN_READ_VERSION: ClassVar[int] = 1
62 PUBLIC_TYPE: ClassVar[type] = object
64 def deserialize(
65 self, archive: InputArchive[Any], **kwargs: Any
66 ) -> Any: # pragma: no cover - never invoked
67 raise NotImplementedError()
70class _DevDoubleArchiveTree(ArchiveTree):
71 """Development-version double, defined in the test package."""
73 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_development"
74 SCHEMA_VERSION: ClassVar[str] = "1.0.0.dev0"
75 MIN_READ_VERSION: ClassVar[int] = 1
76 PUBLIC_TYPE: ClassVar[type] = object
78 def deserialize(
79 self, archive: InputArchive[Any], **kwargs: Any
80 ) -> Any: # pragma: no cover - never invoked
81 raise NotImplementedError()
84def test_custom_schema_url_base() -> None:
85 """Verify a subclass can override SCHEMA_URL_BASE so its schema URL and
86 $id are minted under its own documentation site.
87 """
88 expected = "https://example.org/schemas/frozen_schemas_test_custom_base-1.0.0"
89 assert _CustomBaseArchiveTree().schema_url == expected
90 schema = dump_schema(_CustomBaseArchiveTree)
91 assert schema["$id"] == expected
92 assert schema["title"] == "frozen_schemas_test_custom_base"
95def test_available_schema_classes() -> None:
96 """Verify enumeration finds the package's schemas, sorted by name."""
97 classes = available_schema_classes()
98 names = [cls.SCHEMA_NAME for cls in classes]
99 assert names == sorted(names)
100 assert "image" in names
101 assert "cell_coadd" in names # lazily-loaded built-in provider
104def test_available_schema_classes_excludes_foreign_modules() -> None:
105 """Verify classes registered from outside lsst.images (e.g. test
106 doubles) are not treated as package-owned schemas.
107 """
108 names = [cls.SCHEMA_NAME for cls in available_schema_classes()]
109 assert "frozen_schemas_test_foreign" not in names
112def test_available_schema_classes_package_filter() -> None:
113 """Verify the package argument selects schemas by defining module, so an
114 external package can freeze its own schemas.
115 """
116 classes = available_schema_classes(package=_ForeignArchiveTree.__module__)
117 names = [cls.SCHEMA_NAME for cls in classes]
118 assert "frozen_schemas_test_foreign" in names
119 assert "image" not in names
122def test_available_schema_classes_discovers_entry_points(monkeypatch: pytest.MonkeyPatch) -> None:
123 """Verify schemas provided only through the ``lsst.images.schemas``
124 entry point group are discovered without prior registration.
125 """
126 state: dict[str, type[ArchiveTree]] = {}
128 def _load() -> type[ArchiveTree]:
129 if "cls" not in state: 129 ↛ 143line 129 didn't jump to line 143 because the condition on line 129 was always true
131 class _EntryPointArchiveTree(ArchiveTree):
132 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_entry_point"
133 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
134 MIN_READ_VERSION: ClassVar[int] = 1
135 PUBLIC_TYPE: ClassVar[type] = object
137 def deserialize(
138 self, archive: InputArchive[Any], **kwargs: Any
139 ) -> Any: # pragma: no cover - never invoked
140 raise NotImplementedError()
142 state["cls"] = _EntryPointArchiveTree
143 return state["cls"]
145 class _FakeEntryPoint:
146 name = "frozen_schemas_test_entry_point"
147 value = "fake.module:FakeClass"
149 def load(self) -> type[ArchiveTree]:
150 return _load()
152 real_entry_points = importlib.metadata.entry_points
154 def _fake_entry_points(**kwargs: Any) -> Any:
155 if kwargs.get("group") == "lsst.images.schemas": 155 ↛ 160line 155 didn't jump to line 160 because the condition on line 155 was always true
156 eps = [_FakeEntryPoint()]
157 if (name := kwargs.get("name")) is not None:
158 eps = [ep for ep in eps if ep.name == name]
159 return eps
160 return real_entry_points(**kwargs)
162 monkeypatch.setattr(importlib.metadata, "entry_points", _fake_entry_points)
163 classes = available_schema_classes(package=_ForeignArchiveTree.__module__)
164 assert "frozen_schemas_test_entry_point" in [cls.SCHEMA_NAME for cls in classes]
167def test_write_skips_development_schemas(tmp_path: Path) -> None:
168 """Verify a development schema is never frozen."""
169 write_frozen_schemas(tmp_path, package=_DevDoubleArchiveTree.__module__)
170 assert not frozen_schema_path(tmp_path, _DevDoubleArchiveTree).exists()
173def test_check_skips_development_schemas(tmp_path: Path) -> None:
174 """Verify check does not report a missing file for a development schema."""
175 problems = check_frozen_schemas(tmp_path, package=_DevDoubleArchiveTree.__module__)
176 assert all("development" not in p for p in problems)
179def test_write_refuses_to_change_finalized(tmp_path: Path) -> None:
180 """Verify a finalized frozen file cannot be overwritten."""
181 write_frozen_schemas(tmp_path)
182 (cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"]
183 path = frozen_schema_path(tmp_path, cls)
184 stale = json.loads(path.read_text())
185 stale["description"] = "changed by hand"
186 path.write_text(json.dumps(stale, indent=2, sort_keys=True) + "\n")
187 with pytest.raises(FrozenSchemaError, match="bump"):
188 write_frozen_schemas(tmp_path)
191def test_dump_schema_has_id_and_title() -> None:
192 """Verify the dumped schema carries the canonical $id and title."""
193 (cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"]
194 schema = dump_schema(cls)
195 assert schema["$id"] == f"https://images.lsst.io/schemas/image-{cls.SCHEMA_VERSION}"
196 assert schema["title"] == "image"
197 assert frozen_schema_filename(cls) == f"image-{cls.SCHEMA_VERSION}.json"
200def test_dumped_schema_ids_match_declaring_class() -> None:
201 """Verify every dumped schema's $id and title come from the declaring
202 class, not an inherited concrete schema (e.g. visit_image, which
203 subclasses masked_image).
204 """
205 for cls in available_schema_classes():
206 schema = dump_schema(cls)
207 expected = f"https://images.lsst.io/schemas/{cls.SCHEMA_NAME}-{cls.SCHEMA_VERSION}"
208 assert schema["$id"] == expected, cls.SCHEMA_NAME
209 assert schema["title"] == cls.SCHEMA_NAME
212def test_write_and_check_round_trip(tmp_path: Path) -> None:
213 """Verify write produces files that check accepts, and rewrite is a
214 no-op.
215 """
216 changed = write_frozen_schemas(tmp_path)
217 assert changed
218 assert check_frozen_schemas(tmp_path) == []
219 assert write_frozen_schemas(tmp_path) == []
222def test_frozen_layout_per_schema_subdirectory(tmp_path: Path) -> None:
223 """Verify frozen files are laid out as ``{name}/{name}-{version}.json``
224 so the directory scales as versions accumulate.
225 """
226 write_frozen_schemas(tmp_path)
227 (image_cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"]
228 path = frozen_schema_path(tmp_path, image_cls)
229 assert path == tmp_path / "image" / frozen_schema_filename(image_cls)
230 assert path.exists()
231 assert not list(tmp_path.glob("*.json"))
234def test_check_reports_missing_and_stale(tmp_path: Path) -> None:
235 """Verify check flags a missing file and a stale file."""
236 write_frozen_schemas(tmp_path)
237 (cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"]
238 path = frozen_schema_path(tmp_path, cls)
239 stale = json.loads(path.read_text())
240 stale["description"] = "something else"
241 path.write_text(json.dumps(stale, indent=2, sort_keys=True) + "\n")
242 problems = check_frozen_schemas(tmp_path)
243 assert any("bump SCHEMA_VERSION" in p for p in problems)
244 path.unlink()
245 problems = check_frozen_schemas(tmp_path)
246 assert any("missing" in p for p in problems)
249def test_write_preserves_superseded_versions(tmp_path: Path) -> None:
250 """Verify write never deletes a frozen file for an old schema version."""
251 old = tmp_path / "image" / "image-0.9.9.json"
252 old.parent.mkdir()
253 old.write_text("{}\n")
254 write_frozen_schemas(tmp_path)
255 assert old.exists()
258def test_fixtures_validate_against_frozen_schemas() -> None:
259 """Verify representative archive fixtures validate against the frozen
260 schemas with a strict draft 2020-12 validator, which also proves every
261 reference inside the published documents is resolvable.
262 """
263 jsonschema = pytest.importorskip("jsonschema")
264 checked = 0
265 for fixture_path in sorted((Path(__file__).parent / "data" / "schema_v1").glob("*.json")):
266 instance = json.loads(fixture_path.read_text())
267 name, _, version = instance["schema_url"].rsplit("/", 1)[-1].rpartition("-")
268 schema_file = REPO_SCHEMA_DIR / name / f"{name}-{version}.json"
269 if not schema_file.exists():
270 continue # development schema: no frozen document to validate against
271 schema = json.loads(schema_file.read_text())
272 jsonschema.Draft202012Validator(schema).validate(instance)
273 checked += 1
274 assert checked
277_DEVELOPMENT_SCHEMAS = {
278 "aperture_correction_map",
279 "camera_frame_set",
280 "color_image",
281 "detector",
282 "difference_image",
283 "gaussian_psf",
284 "image_basis_convolution_kernel",
285 "masked_image",
286 "observation_summary_stats",
287 "piff_psf",
288 "psfex_psf",
289 "visit_image",
290}
293def test_development_schemas_are_unfrozen() -> None:
294 """Verify exactly the intended schemas are development (unfrozen) and the
295 rest are finalized (frozen).
296 """
297 for cls in available_schema_classes():
298 frozen = frozen_schema_path(REPO_SCHEMA_DIR, cls)
299 if cls.SCHEMA_NAME in _DEVELOPMENT_SCHEMAS:
300 assert is_development_version(cls.SCHEMA_VERSION), cls.SCHEMA_NAME
301 assert not frozen.exists(), f"{cls.SCHEMA_NAME} should be unfrozen"
302 else:
303 assert not is_development_version(cls.SCHEMA_VERSION), cls.SCHEMA_NAME
304 assert frozen.exists(), f"{cls.SCHEMA_NAME} should be frozen"
307def test_committed_frozen_schemas_are_current() -> None:
308 """Verify the git-committed frozen schema files match the models.
310 A failure here means a serialization model changed; run
311 ``lsst-images-admin schemas write`` and commit the result.
312 """
313 problems = check_frozen_schemas(REPO_SCHEMA_DIR)
314 assert not problems, (
315 "Frozen schema files are stale; run 'lsst-images-admin schemas write' "
316 "and commit the result: " + ", ".join(problems)
317 )
320class _NotesArchiveTree(ArchiveTree):
321 """Summary line for the notes tree.
323 Extended summary paragraph that should be kept.
325 Notes
326 -----
327 Implementation detail that must not appear in the schema description.
328 """
330 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_notes"
331 SCHEMA_VERSION: ClassVar[str] = "1.0.0"
332 MIN_READ_VERSION: ClassVar[int] = 1
333 PUBLIC_TYPE: ClassVar[type] = object
335 label: str = pydantic.Field(default="", description="A one-line field description with no sections.")
337 def deserialize(
338 self, archive: InputArchive[Any], **kwargs: Any
339 ) -> Any: # pragma: no cover - never invoked
340 raise NotImplementedError()
343@pytest.mark.parametrize(
344 ("text", "expected"),
345 [
346 ("Just a summary.", "Just a summary."),
347 ("Summary.\n\nExtended paragraph.", "Summary.\n\nExtended paragraph."),
348 ("Summary.\n\nNotes\n-----\nDetail.", "Summary."),
349 ("Summary.\n\nMore.\n\nParameters\n----------\nx\n A thing.", "Summary.\n\nMore."),
350 ("A dash - not a section header.", "A dash - not a section header."),
351 ],
352)
353def test_summary_description(text: str, expected: str) -> None:
354 """Verify the summary extends up to (not into) the first numpydoc section
355 header, and text without a section header is returned unchanged.
356 """
357 assert _summary_description(text) == expected
360def test_dump_schema_description_drops_numpydoc_sections() -> None:
361 """Verify a class docstring's numpydoc sections are dropped from the
362 schema description, while a one-line field description is untouched.
363 """
364 schema = dump_schema(_NotesArchiveTree)
365 assert schema["description"] == (
366 "Summary line for the notes tree.\n\nExtended summary paragraph that should be kept."
367 )
368 assert "Notes" not in schema["description"]
369 assert schema["properties"]["label"]["description"] == "A one-line field description with no sections."