Coverage for tests/test_serialization_io.py: 94%
105 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 09:06 +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.
11from __future__ import annotations
13from pathlib import Path
15import numpy as np
16import pytest
18from lsst.images import Box, Image, VisitImage
19from lsst.images.serialization import ArchiveReadError, read_archive, write_archive
20from lsst.images.tests import current_fixture_path, iter_schema_fixtures
21from lsst.utils.introspection import get_full_type_name
23try:
24 import h5py # noqa: F401 -- detect availability for NDF round-trip skip
26 H5PY_AVAILABLE = True
27except ImportError:
28 H5PY_AVAILABLE = False
30try:
31 import piff # noqa: F401 -- detect availability for piff_psf fixture skip
33 PIFF_AVAILABLE = True
34except ImportError:
35 PIFF_AVAILABLE = False
37skip_no_h5py = pytest.mark.skipif(not H5PY_AVAILABLE, reason="h5py is not installed")
39FIXTURE_DIR = Path(__file__).parent / "data" / "schemas"
41# Full Python type produced when each fixture is read through the generic
42# read_archive() API, keyed by (schema name, variant). These are pinned here
43# rather than derived from the schema registry so the test asserts the
44# externally-observable type instead of re-running read_archive()'s own
45# lookup against itself. Keying by name and variant rather than by file name
46# keeps the table stable across version bumps.
47EXPECTED_TYPES: dict[tuple[str, str | None], str] = {
48 ("aperture_correction_map", None): "dict",
49 ("background_map", None): "lsst.images.BackgroundMap",
50 ("camera_frame_set", None): "lsst.images.cameras.CameraFrameSet",
51 ("cell_aperture_correction_map", None): "dict",
52 ("cell_coadd", "as_shipped"): "lsst.images.cells.CellCoadd",
53 ("cell_coadd", "canonical"): "lsst.images.cells.CellCoadd",
54 ("cell_psf", None): "lsst.images.cells.CellPointSpreadFunction",
55 ("chebyshev_field", None): "lsst.images.fields.ChebyshevField",
56 ("coadd_provenance", None): "lsst.images.cells.CoaddProvenance",
57 ("color_image", None): "lsst.images.ColorImage",
58 ("detector", None): "lsst.images.cameras.Detector",
59 ("difference_image", "dp2"): "lsst.images.DifferenceImage",
60 ("gaussian_psf", None): "lsst.images.psfs.GaussianPointSpreadFunction",
61 ("image", None): "lsst.images.Image",
62 (
63 "image_basis_convolution_kernel",
64 None,
65 ): "lsst.images.convolution_kernels.ImageBasisConvolutionKernel",
66 ("mask", None): "lsst.images.Mask",
67 ("masked_image", None): "lsst.images.MaskedImage",
68 ("observation_summary_stats", None): "lsst.images.ObservationSummaryStats",
69 ("piff_psf", None): "lsst.images.psfs.PiffWrapper",
70 ("product_field", None): "lsst.images.fields.ProductField",
71 ("sky_projection", None): "lsst.images.SkyProjection",
72 ("spline_field", None): "lsst.images.fields.SplineField",
73 ("sum_field", None): "lsst.images.fields.SumField",
74 ("transform", None): "lsst.images.Transform",
75 ("visit_image", None): "lsst.images.VisitImage",
76 ("visit_image", "dp1"): "lsst.images.VisitImage",
77 ("visit_image", "dp2"): "lsst.images.VisitImage",
78}
81def test_generic_read_visit_image_json() -> None:
82 """Verify read_archive() on a visit_image JSON fixture returns a
83 VisitImage.
84 """
85 path = current_fixture_path(FIXTURE_DIR, "visit_image")
86 result = read_archive(path)
87 assert isinstance(result, VisitImage)
90def test_generic_read_image_json() -> None:
91 """Verify read_archive() on an image JSON fixture returns an Image."""
92 path = current_fixture_path(FIXTURE_DIR, "image")
93 result = read_archive(path)
94 assert isinstance(result, Image)
97def test_read_unsupported_extension(tmp_path: Path) -> None:
98 """Verify read_archive() raises ValueError for an unrecognized file
99 extension.
100 """
101 path = tmp_path / "bogus.txt"
102 with open(path, "w") as f:
103 f.write("nope")
104 with pytest.raises(ValueError):
105 read_archive(path)
108def test_read_unregistered_schema(tmp_path: Path) -> None:
109 """Verify read_archive() raises ArchiveReadError for a JSON with an unknown
110 schema.
111 """
112 path = tmp_path / "fake.json"
113 with open(path, "w") as f:
114 f.write(
115 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",'
116 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}'
117 )
118 with pytest.raises(ArchiveReadError) as exc_info:
119 read_archive(path)
120 assert "no-such-schema" in str(exc_info.value)
123def _sweep_sort_key(entry: tuple[str, str | None]) -> tuple[str, str]:
124 """Return a sort key for an ``EXPECTED_TYPES`` entry, ``None`` sorting
125 first within a schema name.
126 """
127 name, variant = entry
128 return (name, variant or "")
131# Assigned to an explicitly annotated variable rather than sorted() inline in
132# the decorator below: parametrize()'s stub types its arguments as `object`,
133# and that context leaks into bidirectional generic inference for a bare
134# sorted() call, defeating the key function's type.
135_SWEEP_ENTRIES: list[tuple[str, str | None]] = sorted(EXPECTED_TYPES, key=_sweep_sort_key)
138def test_expected_types_enumerates_every_committed_fixture_variant() -> None:
139 """Verify EXPECTED_TYPES matches the fixture tree's (name, variant) pairs
140 exactly, so a committed fixture cannot go missing unnoticed.
142 Each rung below may legitimately skip an individual case when an optional
143 dependency (piff) is unavailable, but that must never be confused with a
144 committed fixture file itself going missing: this closes the presence
145 gap for every variant at once, rather than relying on each consumer's own
146 guard against a missing path (which a stray ``git rm`` or an interrupted
147 freeze would otherwise defeat with a silent skip instead of a failure).
149 The two directions fail separately and name their own remedy, because
150 adding a fixture variant and losing one are unrelated mistakes with
151 opposite fixes, and this test is the only thing that catches either.
152 """
153 present = {(f.name, f.variant) for f in iter_schema_fixtures(FIXTURE_DIR) if not f.retired}
154 unlisted = sorted(present - set(EXPECTED_TYPES))
155 absent = sorted(set(EXPECTED_TYPES) - present)
156 assert not unlisted, (
157 f"committed fixture variants missing from EXPECTED_TYPES: {unlisted}; add each with the "
158 "full type name its fixture deserializes to, so test_fixture_sweep covers it"
159 )
160 assert not absent, (
161 f"EXPECTED_TYPES entries with no committed fixture: {absent}; delete each entry, or "
162 "restore the fixture file if it went missing by accident"
163 )
166@pytest.mark.parametrize("entry", _SWEEP_ENTRIES)
167def test_fixture_sweep(entry: tuple[str, str | None]) -> None:
168 """Verify every schema fixture reads to its pinned Python type."""
169 name, variant = entry
170 if name == "piff_psf" and not PIFF_AVAILABLE: 170 ↛ 171line 170 didn't jump to line 171 because the condition on line 170 was never true
171 pytest.skip("piff not available")
172 path = current_fixture_path(FIXTURE_DIR, name, variant=variant)
173 assert path.exists(), f"{path} is a committed fixture and must not go missing"
174 result = read_archive(path)
175 assert get_full_type_name(type(result)) == EXPECTED_TYPES[entry], entry
178def _make_image() -> Image:
179 """Return a small float32 Image for round-trip tests."""
180 return Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
183def test_generic_write_round_trip_fits(tmp_path: Path) -> None:
184 """Verify write_archive() + read_archive() round-trips an Image
185 through FITS.
186 """
187 image = _make_image()
188 path = tmp_path / "x.fits"
189 write_archive(image, path)
190 result = read_archive(path)
191 assert isinstance(result, Image)
192 np.testing.assert_array_equal(result.array, image.array)
195def test_generic_write_round_trip_json(tmp_path: Path) -> None:
196 """Verify write_archive() + read_archive() round-trips an Image
197 through JSON.
198 """
199 image = _make_image()
200 path = tmp_path / "x.json"
201 write_archive(image, path)
202 result = read_archive(path)
203 assert isinstance(result, Image)
204 np.testing.assert_array_equal(result.array, image.array)
207@skip_no_h5py
208def test_generic_write_round_trip_ndf(tmp_path: Path) -> None:
209 """Verify write_archive() + read_archive() round-trips an Image
210 through NDF.
211 """
212 image = _make_image()
213 path = tmp_path / "x.sdf"
214 write_archive(image, path)
215 result = read_archive(path)
216 assert isinstance(result, Image)
217 np.testing.assert_array_equal(result.array, image.array)
220def test_read_bbox_subset_fits(tmp_path: Path) -> None:
221 """Verify read_archive() forwards bbox kwarg to the FITS backend for subset
222 reads.
223 """
224 img = Image(np.arange(64, dtype=np.float32).reshape(8, 8), bbox=Box.factory[0:8, 0:8])
225 path = tmp_path / "x.fits"
226 write_archive(img, path)
227 sub = read_archive(path, bbox=Box.factory[2:6, 2:6])
228 assert sub.array.shape == (4, 4)
229 np.testing.assert_array_equal(sub.array, img.array[2:6, 2:6])
232def test_read_cls_match() -> None:
233 """Verify read_archive() with cls= returns the expected type when it
234 matches.
235 """
236 path = current_fixture_path(FIXTURE_DIR, "image")
237 result = read_archive(path, cls=Image)
238 assert isinstance(result, Image)
241def test_read_cls_mismatch_raises() -> None:
242 """Verify read_archive() raises TypeError when the deserialized type
243 does not match cls.
244 """
245 from lsst.images import Mask
247 path = current_fixture_path(FIXTURE_DIR, "image")
248 with pytest.raises(TypeError) as exc_info:
249 read_archive(path, cls=Mask)
250 msg = str(exc_info.value)
251 assert "image" in msg # path / schema name
252 assert "Image" in msg # actual deserialized type
253 assert "Mask" in msg # requested cls