Coverage for tests/test_serialization_io.py: 30%
115 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-06 08:34 +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 "chebyshev_field.json": "lsst.images.fields.ChebyshevField",
49 "coadd_provenance.json": "lsst.images.cells.CoaddProvenance",
50 "color_image.json": "lsst.images.ColorImage",
51 "detector.json": "lsst.images.cameras.Detector",
52 "gaussian_psf.json": "lsst.images.psfs.GaussianPointSpreadFunction",
53 "image.json": "lsst.images.Image",
54 "mask.json": "lsst.images.Mask",
55 "masked_image.json": "lsst.images.MaskedImage",
56 "piff_psf.json": "lsst.images.psfs.PiffWrapper",
57 "product_field.json": "lsst.images.fields.ProductField",
58 "projection.json": "lsst.images.Projection",
59 "sum_field.json": "lsst.images.fields.SumField",
60 "transform.json": "lsst.images.Transform",
61 "visit_image.json": "lsst.images.VisitImage",
62 "cell_coadd.json": "lsst.images.cells.CellCoadd",
63 "visit_image_dp1.json": "lsst.images.VisitImage",
64 "visit_image_dp2.json": "lsst.images.VisitImage",
65 "difference_image_dp2.json": "lsst.images.DifferenceImage",
66}
69class GenericReadTestCase(unittest.TestCase):
70 """read(path) dispatches by extension and produces the registered type."""
72 def test_visit_image_json(self) -> None:
73 path = os.path.join(DATA_DIR, "visit_image.json")
74 result = read(path)
75 self.assertIsInstance(result, VisitImage)
77 def test_image_json(self) -> None:
78 path = os.path.join(DATA_DIR, "image.json")
79 result = read(path)
80 self.assertIsInstance(result, Image)
83class GenericReadErrorsTestCase(unittest.TestCase):
84 """Unknown schemas and bad extensions raise clean errors."""
86 def setUp(self) -> None:
87 tmp = tempfile.TemporaryDirectory()
88 self.addCleanup(tmp.cleanup)
89 self.tmp = tmp.name
91 def test_unsupported_extension(self) -> None:
92 path = os.path.join(self.tmp, "bogus.txt")
93 with open(path, "w") as f:
94 f.write("nope")
95 # backend_for_path raises ValueError; read() must let it through.
96 with self.assertRaises(ValueError):
97 read(path)
99 def test_unregistered_schema(self) -> None:
100 # Write a JSON file with a fabricated schema name not in the
101 # registry.
102 path = os.path.join(self.tmp, "fake.json")
103 with open(path, "w") as f:
104 f.write(
105 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",'
106 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}'
107 )
108 with self.assertRaises(ArchiveReadError) as ctx:
109 read(path)
110 self.assertIn("no-such-schema", str(ctx.exception))
113class FixtureSweepTestCase(unittest.TestCase):
114 """Every schema_v1 JSON fixture reads through the generic API and
115 produces the Python type pinned in ``EXPECTED_TYPES``.
116 """
118 def test_sweep(self) -> None:
119 # piff is an optional dependency that PiffSerializationModel
120 # imports lazily on deserialise; skip its fixture when missing.
121 skip = set() if PIFF_AVAILABLE else {"piff_psf.json"}
122 seen = set()
123 roots = [DATA_DIR, os.path.join(DATA_DIR, "legacy")]
124 for root in roots:
125 if not os.path.isdir(root):
126 continue
127 for entry in sorted(os.listdir(root)):
128 if not entry.endswith(".json") or entry in skip:
129 continue
130 with self.subTest(entry=entry):
131 self.assertIn(entry, EXPECTED_TYPES, f"no expected type recorded for {entry!r}")
132 result = read(os.path.join(root, entry))
133 self.assertEqual(get_full_type_name(type(result)), EXPECTED_TYPES[entry])
134 seen.add(entry)
135 # Fail if EXPECTED_TYPES drifts from the fixtures actually on disk.
136 self.assertEqual(seen, set(EXPECTED_TYPES) - skip)
139class GenericWriteRoundTripTestCase(unittest.TestCase):
140 """write(obj, path) dispatches by extension and round-trips."""
142 def setUp(self) -> None:
143 tmp = tempfile.TemporaryDirectory()
144 self.addCleanup(tmp.cleanup)
145 self.tmp = tmp.name
146 self.image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
148 def test_round_trip_fits(self) -> None:
149 path = os.path.join(self.tmp, "x.fits")
150 write(self.image, path)
151 result = read(path)
152 self.assertIsInstance(result, Image)
153 np.testing.assert_array_equal(result.array, self.image.array)
155 def test_round_trip_json(self) -> None:
156 path = os.path.join(self.tmp, "x.json")
157 write(self.image, path)
158 result = read(path)
159 self.assertIsInstance(result, Image)
160 np.testing.assert_array_equal(result.array, self.image.array)
162 @unittest.skipUnless(H5PY_AVAILABLE, "h5py not available.")
163 def test_round_trip_ndf(self) -> None:
164 path = os.path.join(self.tmp, "x.sdf")
165 write(self.image, path)
166 result = read(path)
167 self.assertIsInstance(result, Image)
168 np.testing.assert_array_equal(result.array, self.image.array)
171class GenericReadKwargsTestCase(unittest.TestCase):
172 """**kwargs forwarded by read() reach the backend deserialize."""
174 def setUp(self) -> None:
175 tmp = tempfile.TemporaryDirectory()
176 self.addCleanup(tmp.cleanup)
177 self.tmp = tmp.name
179 def test_bbox_subset_fits(self) -> None:
180 img = Image(np.arange(64, dtype=np.float32).reshape(8, 8), bbox=Box.factory[0:8, 0:8])
181 path = os.path.join(self.tmp, "x.fits")
182 write(img, path)
183 # Read a 4x4 subset. bbox is the FITS-specific kwarg understood
184 # by Image.deserialize; the generic read must forward it.
185 sub = read(path, bbox=Box.factory[2:6, 2:6])
186 self.assertEqual(sub.array.shape, (4, 4))
187 np.testing.assert_array_equal(sub.array, img.array[2:6, 2:6])
190class ReadClsTestCase(unittest.TestCase):
191 """read(path, cls=...) validates the deserialized type."""
193 def test_read_cls_match(self) -> None:
194 path = os.path.join(DATA_DIR, "image.json")
195 result = read(path, cls=Image)
196 self.assertIsInstance(result, Image)
198 def test_read_cls_mismatch_raises(self) -> None:
199 from lsst.images import Mask
201 path = os.path.join(DATA_DIR, "image.json")
202 with self.assertRaises(TypeError) as ctx:
203 read(path, cls=Mask)
204 msg = str(ctx.exception)
205 self.assertIn("image", msg) # path / schema name
206 self.assertIn("Image", msg) # actual deserialized type
207 self.assertIn("Mask", msg) # requested cls
210if __name__ == "__main__":
211 unittest.main()