Coverage for tests/test_serialization_io.py: 95%
115 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-01 09:14 +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
13import os
14import tempfile
15import unittest
17import numpy as np
19from lsst.images import Box, Image, VisitImage
20from lsst.images.serialization import ArchiveReadError, read, write
21from lsst.utils.introspection import get_full_type_name
23try:
24 import h5py # noqa: F401 -- detect availability for NDF round-trip skip
25except ImportError:
26 H5PY_AVAILABLE = False
27else:
28 H5PY_AVAILABLE = True
30try:
31 import piff # noqa: F401 -- detect availability for piff_psf fixture skip
32except ImportError:
33 PIFF_AVAILABLE = False
34else:
35 PIFF_AVAILABLE = True
37DATA_DIR = os.path.join(os.path.dirname(__file__), "data", "schema_v1")
39# Full Python type produced when each fixture is read through the generic
40# read() API, keyed by the fixture's file name. These are pinned here rather
41# than derived from the schema registry so the test asserts the
42# externally-observable type instead of re-running read()'s own lookup against
43# itself.
44EXPECTED_TYPES = {
45 "aperture_correction_map.json": "dict",
46 "background_map.json": "lsst.images.BackgroundMap",
47 "cell_psf.json": "lsst.images.cells.CellPointSpreadFunction",
48 "cell_aperture_correction_map.json": "dict",
49 "chebyshev_field.json": "lsst.images.fields.ChebyshevField",
50 "coadd_provenance.json": "lsst.images.cells.CoaddProvenance",
51 "color_image.json": "lsst.images.ColorImage",
52 "detector.json": "lsst.images.cameras.Detector",
53 "gaussian_psf.json": "lsst.images.psfs.GaussianPointSpreadFunction",
54 "image.json": "lsst.images.Image",
55 "mask.json": "lsst.images.Mask",
56 "masked_image.json": "lsst.images.MaskedImage",
57 "piff_psf.json": "lsst.images.psfs.PiffWrapper",
58 "product_field.json": "lsst.images.fields.ProductField",
59 "sky_projection.json": "lsst.images.SkyProjection",
60 "sum_field.json": "lsst.images.fields.SumField",
61 "transform.json": "lsst.images.Transform",
62 "visit_image.json": "lsst.images.VisitImage",
63 "cell_coadd.json": "lsst.images.cells.CellCoadd",
64 "visit_image_dp1.json": "lsst.images.VisitImage",
65 "visit_image_dp2.json": "lsst.images.VisitImage",
66 "difference_image_dp2.json": "lsst.images.DifferenceImage",
67}
70class GenericReadTestCase(unittest.TestCase):
71 """read(path) dispatches by extension and produces the registered type."""
73 def test_visit_image_json(self) -> None:
74 path = os.path.join(DATA_DIR, "visit_image.json")
75 result = read(path)
76 self.assertIsInstance(result, VisitImage)
78 def test_image_json(self) -> None:
79 path = os.path.join(DATA_DIR, "image.json")
80 result = read(path)
81 self.assertIsInstance(result, Image)
84class GenericReadErrorsTestCase(unittest.TestCase):
85 """Unknown schemas and bad extensions raise clean errors."""
87 def setUp(self) -> None:
88 tmp = tempfile.TemporaryDirectory()
89 self.addCleanup(tmp.cleanup)
90 self.tmp = tmp.name
92 def test_unsupported_extension(self) -> None:
93 path = os.path.join(self.tmp, "bogus.txt")
94 with open(path, "w") as f:
95 f.write("nope")
96 # backend_for_path raises ValueError; read() must let it through.
97 with self.assertRaises(ValueError):
98 read(path)
100 def test_unregistered_schema(self) -> None:
101 # Write a JSON file with a fabricated schema name not in the
102 # registry.
103 path = os.path.join(self.tmp, "fake.json")
104 with open(path, "w") as f:
105 f.write(
106 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",'
107 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}'
108 )
109 with self.assertRaises(ArchiveReadError) as ctx:
110 read(path)
111 self.assertIn("no-such-schema", str(ctx.exception))
114class FixtureSweepTestCase(unittest.TestCase):
115 """Every schema_v1 JSON fixture reads through the generic API and
116 produces the Python type pinned in ``EXPECTED_TYPES``.
117 """
119 def test_sweep(self) -> None:
120 # piff is an optional dependency that PiffSerializationModel
121 # imports lazily on deserialise; skip its fixture when missing.
122 skip = set() if PIFF_AVAILABLE else {"piff_psf.json"}
123 seen = set()
124 roots = [DATA_DIR, os.path.join(DATA_DIR, "legacy")]
125 for root in roots:
126 if not os.path.isdir(root): 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true
127 continue
128 for entry in sorted(os.listdir(root)):
129 if not entry.endswith(".json") or entry in skip:
130 continue
131 with self.subTest(entry=entry):
132 self.assertIn(entry, EXPECTED_TYPES, f"no expected type recorded for {entry!r}")
133 result = read(os.path.join(root, entry))
134 self.assertEqual(get_full_type_name(type(result)), EXPECTED_TYPES[entry])
135 seen.add(entry)
136 # Fail if EXPECTED_TYPES drifts from the fixtures actually on disk.
137 self.assertEqual(seen, set(EXPECTED_TYPES) - skip)
140class GenericWriteRoundTripTestCase(unittest.TestCase):
141 """write(obj, path) dispatches by extension and round-trips."""
143 def setUp(self) -> None:
144 tmp = tempfile.TemporaryDirectory()
145 self.addCleanup(tmp.cleanup)
146 self.tmp = tmp.name
147 self.image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
149 def test_round_trip_fits(self) -> None:
150 path = os.path.join(self.tmp, "x.fits")
151 write(self.image, path)
152 result = read(path)
153 self.assertIsInstance(result, Image)
154 np.testing.assert_array_equal(result.array, self.image.array)
156 def test_round_trip_json(self) -> None:
157 path = os.path.join(self.tmp, "x.json")
158 write(self.image, path)
159 result = read(path)
160 self.assertIsInstance(result, Image)
161 np.testing.assert_array_equal(result.array, self.image.array)
163 @unittest.skipUnless(H5PY_AVAILABLE, "h5py not available.")
164 def test_round_trip_ndf(self) -> None:
165 path = os.path.join(self.tmp, "x.sdf")
166 write(self.image, path)
167 result = read(path)
168 self.assertIsInstance(result, Image)
169 np.testing.assert_array_equal(result.array, self.image.array)
172class GenericReadKwargsTestCase(unittest.TestCase):
173 """**kwargs forwarded by read() reach the backend deserialize."""
175 def setUp(self) -> None:
176 tmp = tempfile.TemporaryDirectory()
177 self.addCleanup(tmp.cleanup)
178 self.tmp = tmp.name
180 def test_bbox_subset_fits(self) -> None:
181 img = Image(np.arange(64, dtype=np.float32).reshape(8, 8), bbox=Box.factory[0:8, 0:8])
182 path = os.path.join(self.tmp, "x.fits")
183 write(img, path)
184 # Read a 4x4 subset. bbox is the FITS-specific kwarg understood
185 # by Image.deserialize; the generic read must forward it.
186 sub = read(path, bbox=Box.factory[2:6, 2:6])
187 self.assertEqual(sub.array.shape, (4, 4))
188 np.testing.assert_array_equal(sub.array, img.array[2:6, 2:6])
191class ReadClsTestCase(unittest.TestCase):
192 """read(path, cls=...) validates the deserialized type."""
194 def test_read_cls_match(self) -> None:
195 path = os.path.join(DATA_DIR, "image.json")
196 result = read(path, cls=Image)
197 self.assertIsInstance(result, Image)
199 def test_read_cls_mismatch_raises(self) -> None:
200 from lsst.images import Mask
202 path = os.path.join(DATA_DIR, "image.json")
203 with self.assertRaises(TypeError) as ctx:
204 read(path, cls=Mask)
205 msg = str(ctx.exception)
206 self.assertIn("image", msg) # path / schema name
207 self.assertIn("Image", msg) # actual deserialized type
208 self.assertIn("Mask", msg) # requested cls
211if __name__ == "__main__":
212 unittest.main()