Coverage for tests/test_serialization_io.py: 95%
115 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-09 09:35 +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_archive, write_archive
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_archive() API, keyed by the fixture's file name. These are pinned
41# here rather than derived from the schema registry so the test asserts
42# the externally-observable type instead of re-running read_archive()'s
43# own lookup against 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_archive(path) dispatches by extension and produces the
72 registered type.
73 """
75 def test_visit_image_json(self) -> None:
76 path = os.path.join(DATA_DIR, "visit_image.json")
77 result = read_archive(path)
78 self.assertIsInstance(result, VisitImage)
80 def test_image_json(self) -> None:
81 path = os.path.join(DATA_DIR, "image.json")
82 result = read_archive(path)
83 self.assertIsInstance(result, Image)
86class GenericReadErrorsTestCase(unittest.TestCase):
87 """Unknown schemas and bad extensions raise clean errors."""
89 def setUp(self) -> None:
90 tmp = tempfile.TemporaryDirectory()
91 self.addCleanup(tmp.cleanup)
92 self.tmp = tmp.name
94 def test_unsupported_extension(self) -> None:
95 path = os.path.join(self.tmp, "bogus.txt")
96 with open(path, "w") as f:
97 f.write("nope")
98 # backend_for_path raises ValueError; read_archive() must let it
99 # through.
100 with self.assertRaises(ValueError):
101 read_archive(path)
103 def test_unregistered_schema(self) -> None:
104 # Write a JSON file with a fabricated schema name not in the
105 # registry.
106 path = os.path.join(self.tmp, "fake.json")
107 with open(path, "w") as f:
108 f.write(
109 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",'
110 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}'
111 )
112 with self.assertRaises(ArchiveReadError) as ctx:
113 read_archive(path)
114 self.assertIn("no-such-schema", str(ctx.exception))
117class FixtureSweepTestCase(unittest.TestCase):
118 """Every schema_v1 JSON fixture reads through the generic API and
119 produces the Python type pinned in ``EXPECTED_TYPES``.
120 """
122 def test_sweep(self) -> None:
123 # piff is an optional dependency that PiffSerializationModel
124 # imports lazily on deserialise; skip its fixture when missing.
125 skip = set() if PIFF_AVAILABLE else {"piff_psf.json"}
126 seen = set()
127 roots = [DATA_DIR, os.path.join(DATA_DIR, "legacy")]
128 for root in roots:
129 if not os.path.isdir(root): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 continue
131 for entry in sorted(os.listdir(root)):
132 if not entry.endswith(".json") or entry in skip:
133 continue
134 with self.subTest(entry=entry):
135 self.assertIn(entry, EXPECTED_TYPES, f"no expected type recorded for {entry!r}")
136 result = read_archive(os.path.join(root, entry))
137 self.assertEqual(get_full_type_name(type(result)), EXPECTED_TYPES[entry])
138 seen.add(entry)
139 # Fail if EXPECTED_TYPES drifts from the fixtures actually on disk.
140 self.assertEqual(seen, set(EXPECTED_TYPES) - skip)
143class GenericWriteRoundTripTestCase(unittest.TestCase):
144 """write_archive(obj, path) dispatches by extension and round-trips."""
146 def setUp(self) -> None:
147 tmp = tempfile.TemporaryDirectory()
148 self.addCleanup(tmp.cleanup)
149 self.tmp = tmp.name
150 self.image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
152 def test_round_trip_fits(self) -> None:
153 path = os.path.join(self.tmp, "x.fits")
154 write_archive(self.image, path)
155 result = read_archive(path)
156 self.assertIsInstance(result, Image)
157 np.testing.assert_array_equal(result.array, self.image.array)
159 def test_round_trip_json(self) -> None:
160 path = os.path.join(self.tmp, "x.json")
161 write_archive(self.image, path)
162 result = read_archive(path)
163 self.assertIsInstance(result, Image)
164 np.testing.assert_array_equal(result.array, self.image.array)
166 @unittest.skipUnless(H5PY_AVAILABLE, "h5py not available.")
167 def test_round_trip_ndf(self) -> None:
168 path = os.path.join(self.tmp, "x.sdf")
169 write_archive(self.image, path)
170 result = read_archive(path)
171 self.assertIsInstance(result, Image)
172 np.testing.assert_array_equal(result.array, self.image.array)
175class GenericReadKwargsTestCase(unittest.TestCase):
176 """**kwargs forwarded by read_archive() reach the backend deserialize."""
178 def setUp(self) -> None:
179 tmp = tempfile.TemporaryDirectory()
180 self.addCleanup(tmp.cleanup)
181 self.tmp = tmp.name
183 def test_bbox_subset_fits(self) -> None:
184 img = Image(np.arange(64, dtype=np.float32).reshape(8, 8), bbox=Box.factory[0:8, 0:8])
185 path = os.path.join(self.tmp, "x.fits")
186 write_archive(img, path)
187 # Read a 4x4 subset. bbox is the FITS-specific kwarg understood
188 # by Image.deserialize; the generic read must forward it.
189 sub = read_archive(path, bbox=Box.factory[2:6, 2:6])
190 self.assertEqual(sub.array.shape, (4, 4))
191 np.testing.assert_array_equal(sub.array, img.array[2:6, 2:6])
194class ReadClsTestCase(unittest.TestCase):
195 """read_archive(path, cls=...) validates the deserialized type."""
197 def test_read_cls_match(self) -> None:
198 path = os.path.join(DATA_DIR, "image.json")
199 result = read_archive(path, cls=Image)
200 self.assertIsInstance(result, Image)
202 def test_read_cls_mismatch_raises(self) -> None:
203 from lsst.images import Mask
205 path = os.path.join(DATA_DIR, "image.json")
206 with self.assertRaises(TypeError) as ctx:
207 read_archive(path, cls=Mask)
208 msg = str(ctx.exception)
209 self.assertIn("image", msg) # path / schema name
210 self.assertIn("Image", msg) # actual deserialized type
211 self.assertIn("Mask", msg) # requested cls
214if __name__ == "__main__":
215 unittest.main()