Coverage for tests/test_cli.py: 84%
217 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:24 -0700
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 subprocess
15import sys
16from pathlib import Path
17from unittest import mock
19import astropy.io.fits
20import click
21import numpy as np
22import pytest
23from click.testing import CliRunner
25from lsst.images import Box, Image
26from lsst.images import fits as images_fits
27from lsst.images import json as images_json
28from lsst.images.cli import main
29from lsst.images.serialization import backend_for_path, read_archive
32@pytest.fixture(scope="session")
33def external_data_dir() -> str:
34 """Return the external test-data directory path, skipping if unset."""
35 if (result := os.environ.get("TESTDATA_IMAGES_DIR")) is None: 35 ↛ 37line 35 didn't jump to line 37 because the condition on line 35 was always true
36 pytest.skip("TESTDATA_IMAGES_DIR is not set.")
37 return result
40def _make_cli_input(tmp_path: Path) -> str:
41 """Return the path to a minimal FITS file written under tmp_path."""
42 path = str(tmp_path / "in.fits")
43 astropy.io.fits.PrimaryHDU().writeto(path)
44 return path
47def _make_detect_file(tmp_path: Path, dataset_type: str | None) -> str:
48 """Return a path to a FITS file with LSST BUTLER DATASETTYPE set to
49 dataset_type.
50 """
51 name = dataset_type.replace(" ", "_") if dataset_type is not None else "none"
52 path = str(tmp_path / f"detect_{name}.fits")
53 hdu = astropy.io.fits.PrimaryHDU()
54 with images_fits.suppress_fits_card_warnings():
55 if dataset_type is not None:
56 hdu.header["LSST BUTLER DATASETTYPE"] = dataset_type
57 hdu.writeto(path)
58 return path
61def test_group_help() -> None:
62 """Test that the root CLI group loads and lists core subcommands."""
63 result = CliRunner().invoke(main, ["--help"])
64 assert result.exit_code == 0, result.output
65 assert "convert" in result.output
66 assert "inspect" in result.output
69def test_python_m_entry_point() -> None:
70 """Test that python -m lsst.images runs the same CLI group."""
71 result = subprocess.run(
72 [sys.executable, "-m", "lsst.images", "--help"],
73 capture_output=True,
74 text=True,
75 )
76 assert result.returncode == 0, result.stderr
77 assert "convert" in result.stdout
78 assert "inspect" in result.stdout
81def test_inspect_fits(tmp_path: Path) -> None:
82 """Test 'inspect' on a FITS file."""
83 path = str(tmp_path / "x.fits")
84 image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4])
85 images_fits.write(image, path)
86 result = CliRunner().invoke(main, ["inspect", path])
87 assert result.exit_code == 0, result.output
88 assert "https://images.lsst.io/schemas/image-1.0.0" in result.output
89 assert "format version: 1" in result.output
90 assert "python class:" in result.output
91 assert "lsst.images.Image" in result.output
94def test_inspect_json(tmp_path: Path) -> None:
95 """Test 'inspect' on a JSON file."""
96 path = str(tmp_path / "x.json")
97 image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4])
98 images_json.write(image, path)
99 result = CliRunner().invoke(main, ["inspect", path])
100 assert result.exit_code == 0, result.output
101 assert "image-1.0.0" in result.output
102 assert "n/a" in result.output
103 assert "python class:" in result.output
104 assert "lsst.images.Image" in result.output
107def test_inspect_unregistered_schema(tmp_path: Path) -> None:
108 """Test that 'inspect' succeeds and reports an unregistered schema name."""
109 path = str(tmp_path / "fake.json")
110 with open(path, "w") as f:
111 f.write(
112 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",'
113 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}'
114 )
115 result = CliRunner().invoke(main, ["inspect", path])
116 assert result.exit_code == 0, result.output
117 assert "python class:" in result.output
118 assert "<unregistered: no-such-schema>" in result.output
121def test_inspect_unknown_extension(tmp_path: Path) -> None:
122 """Test that 'inspect' fails with a non-zero exit code for an unsupported
123 file extension.
124 """
125 path = str(tmp_path / "x.txt")
126 with open(path, "w") as stream:
127 stream.write("nope")
128 result = CliRunner().invoke(main, ["inspect", path])
129 assert result.exit_code != 0
130 assert ".fits" in result.output
133def test_reformat_round_trip_json_fits_json(tmp_path: Path) -> None:
134 """Test that reformat JSON→FITS→JSON preserves the image data."""
135 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
136 src = str(tmp_path / "in.json")
137 mid = str(tmp_path / "mid.fits")
138 out = str(tmp_path / "out.json")
139 images_json.write(image, src)
141 result = CliRunner().invoke(main, ["reformat", src, mid])
142 assert result.exit_code == 0, result.output
143 assert backend_for_path(mid).input_archive.get_basic_info(mid).schema_name == "image"
145 result = CliRunner().invoke(main, ["reformat", mid, out])
146 assert result.exit_code == 0, result.output
148 np.testing.assert_array_equal(read_archive(out, Image).array, image.array)
151def test_reformat_refuses_existing_output(tmp_path: Path) -> None:
152 """Test that reformat refuses to overwrite an existing output file without
153 --overwrite.
154 """
155 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
156 src = str(tmp_path / "in.json")
157 out = str(tmp_path / "out.fits")
158 images_json.write(image, src)
159 images_fits.write(image, out)
160 result = CliRunner().invoke(main, ["reformat", src, out])
161 assert result.exit_code != 0
162 assert "--overwrite" in result.output
165def test_reformat_unknown_output_extension(tmp_path: Path) -> None:
166 """Test that reformat fails for an unsupported output file extension."""
167 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
168 src = str(tmp_path / "in.json")
169 images_json.write(image, src)
170 result = CliRunner().invoke(main, ["reformat", src, str(tmp_path / "out.txt")])
171 assert result.exit_code != 0
172 assert ".fits" in result.output
175def test_detect_visit_image(tmp_path: Path) -> None:
176 """Test that detect_legacy_type identifies visit_image and
177 preliminary_visit_image.
178 """
179 from lsst.images.cli._convert import detect_legacy_type
181 assert detect_legacy_type(_make_detect_file(tmp_path, "visit_image")) == "visit_image"
182 assert detect_legacy_type(_make_detect_file(tmp_path, "preliminary_visit_image")) == "visit_image"
185def test_detect_cell_coadd(tmp_path: Path) -> None:
186 """Test that detect_legacy_type identifies deep_coadd_cell_predetection as
187 cell_coadd.
188 """
189 from lsst.images.cli._convert import detect_legacy_type
191 assert detect_legacy_type(_make_detect_file(tmp_path, "deep_coadd_cell_predetection")) == "cell_coadd"
194def test_detect_indeterminate(tmp_path: Path) -> None:
195 """Test that detect_legacy_type returns None for unknown or absent dataset-
196 type headers.
197 """
198 from lsst.images.cli._convert import detect_legacy_type
200 assert detect_legacy_type(_make_detect_file(tmp_path, None)) is None
201 assert detect_legacy_type(_make_detect_file(tmp_path, "camera")) is None
204def test_detect_visit_image_fixture(tmp_path: Path, external_data_dir: str) -> None:
205 """Test that detect_legacy_type detects a real legacy visit-image fixture
206 file.
207 """
208 from lsst.images.cli._convert import detect_legacy_type
210 path = os.path.join(external_data_dir, "dp2", "legacy", "visit_image.fits")
211 assert detect_legacy_type(path) == "visit_image"
214def test_convert_visit_image_to_json(tmp_path: Path, external_data_dir: str) -> None:
215 """Test that convert produces a valid visit_image JSON file from a legacy
216 FITS fixture.
217 """
218 pytest.importorskip("lsst.afw.image")
219 src = os.path.join(external_data_dir, "dp2", "legacy", "visit_image.fits")
220 out = str(tmp_path / "converted.json")
221 result = CliRunner().invoke(main, ["convert", src, out])
222 assert result.exit_code == 0, result.output
223 info = backend_for_path(out).input_archive.get_basic_info(out)
224 assert info.schema_name == "visit_image"
227def test_convert_refuses_existing_output(tmp_path: Path, external_data_dir: str) -> None:
228 """Test that convert refuses to overwrite an existing output file without
229 --overwrite.
230 """
231 pytest.importorskip("lsst.afw.image")
232 src = os.path.join(external_data_dir, "dp2", "legacy", "visit_image.fits")
233 out = str(tmp_path / "exists.json")
234 with open(out, "w") as stream:
235 stream.write("{}")
236 result = CliRunner().invoke(main, ["convert", src, out])
237 assert result.exit_code != 0
238 assert "--overwrite" in result.output
241def test_convert_cell_coadd_to_json(tmp_path: Path, external_data_dir: str) -> None:
242 """Test that convert produces a valid cell_coadd JSON file from a legacy
243 FITS MultipleCellCoadd.
244 """
245 pytest.importorskip("lsst.cell_coadds")
246 legacy_dir = os.path.join(external_data_dir, "dp2", "legacy")
247 src = os.path.join(legacy_dir, "deep_coadd_cell_predetection.fits")
248 skymap = os.path.join(legacy_dir, "skyMap.pickle")
249 out = str(tmp_path / "coadd.json")
250 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd", "--skymap", skymap])
251 assert result.exit_code == 0, result.output
252 info = backend_for_path(out).input_archive.get_basic_info(out)
253 assert info.schema_name == "cell_coadd"
256def test_convert_cell_coadd_requires_skymap(tmp_path: Path, external_data_dir: str) -> None:
257 """Test that convert fails with a helpful message when --skymap is missing
258 for cell_coadd.
259 """
260 pytest.importorskip("lsst.cell_coadds")
261 src = os.path.join(external_data_dir, "dp2", "legacy", "deep_coadd_cell_predetection.fits")
262 out = str(tmp_path / "coadd.json")
263 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd"])
264 assert result.exit_code != 0
265 assert "--skymap" in result.output
268def test_preserve_quantization_default_is_true() -> None:
269 """Test that the --preserve-quantization option defaults to True."""
270 from lsst.images.cli._convert import convert
272 option = next(p for p in convert.params if p.name == "preserve_quantization")
273 assert option.default is True
276def test_preserve_quantization_explicit_flag_rejected_for_cell_coadd(tmp_path: Path) -> None:
277 """Test that explicitly passing --preserve-quantization is rejected for
278 cell_coadd conversions.
279 """
280 src = _make_cli_input(tmp_path)
281 out = str(tmp_path / "out.json")
282 result = CliRunner().invoke(
283 main, ["convert", src, out, "--type", "cell_coadd", "--preserve-quantization"]
284 )
285 assert result.exit_code != 0
286 assert "preserve-quantization" in result.output
289def test_preserve_quantization_default_does_not_reject_cell_coadd(tmp_path: Path) -> None:
290 """Test that the --preserve-quantization option default doesn't get in the
291 way of cell-coadd conversion.
292 """
293 src = _make_cli_input(tmp_path)
294 out = str(tmp_path / "out.json")
295 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd"])
296 assert "preserve-quantization" not in result.output
299def test_preserve_quantization_forwarded_to_read_legacy() -> None:
300 """Test that _read_legacy forwards preserve_quantization=True to
301 VisitImage.read_legacy.
302 """
303 from lsst.images.cli._convert import _read_legacy
305 with mock.patch("lsst.images.VisitImage.read_legacy") as read_legacy:
306 _read_legacy("in.fits", "visit_image", None, None, None, True)
307 read_legacy.assert_called_once_with("in.fits", preserve_quantization=True)
310def test_rejects_identical_paths(tmp_path: Path) -> None:
311 """Test that 'convert' rejects identical src and dst paths even with
312 --overwrite.
313 """
314 path = _make_cli_input(tmp_path)
315 result = CliRunner().invoke(main, ["convert", path, path, "--type", "visit_image", "--overwrite"])
316 assert result.exit_code != 0
317 assert "different" in result.output
318 assert os.path.exists(path)
321def test_preserves_existing_output_on_read_failure(tmp_path: Path) -> None:
322 """Test that 'convert' leaves the existing output file intact when
323 read_legacy raises.
324 """
325 src = _make_cli_input(tmp_path)
326 out = str(tmp_path / "out.json")
327 with open(out, "w") as stream:
328 stream.write("ORIGINAL")
329 with mock.patch(
330 "lsst.images.cli._convert._read_legacy",
331 side_effect=click.ClickException("boom"),
332 ):
333 result = CliRunner().invoke(main, ["convert", src, out, "--type", "visit_image", "--overwrite"])
334 assert result.exit_code != 0
335 with open(out) as stream:
336 assert stream.read() == "ORIGINAL"
339def test_subcommands_present() -> None:
340 """Test that the minify, reformat, extract-test-data, verify-rewrite, and
341 fuzz-masked-image are listed by --help.
342 """
343 result = CliRunner().invoke(main, ["--help"])
344 assert result.exit_code == 0, result.output
345 assert "minify" in result.output
346 assert "reformat" in result.output
347 assert "extract-test-data" in result.output
348 assert "verify-rewrite" in result.output
349 assert "fuzz-masked-image" in result.output
352def test_minify_help() -> None:
353 """Verify minify --help exits cleanly."""
354 result = CliRunner().invoke(main, ["minify", "--help"])
355 assert result.exit_code == 0, result.output
358def test_extract_test_data_help() -> None:
359 """Verify extract-test-data --help exits cleanly."""
360 result = CliRunner().invoke(main, ["extract-test-data", "--help"])
361 assert result.exit_code == 0, result.output
364def test_verify_rewrite_help() -> None:
365 """Verify verify-rewrite and its stage4 subcommand load with core deps
366 only.
367 """
368 result = CliRunner().invoke(main, ["verify-rewrite", "--help"])
369 assert result.exit_code == 0, result.output
370 assert "stage4" in result.output
371 result = CliRunner().invoke(main, ["verify-rewrite", "stage4", "--help"])
372 assert result.exit_code == 0, result.output
375@pytest.mark.parametrize(
376 "args",
377 [
378 ["-h"],
379 ["convert", "-h"],
380 ["inspect", "-h"],
381 ["minify", "-h"],
382 ["reformat", "-h"],
383 ["extract-test-data", "-h"],
384 ["extract-test-data", "dp2", "-h"],
385 ["verify-rewrite", "-h"],
386 ["verify-rewrite", "stage4", "-h"],
387 ["fuzz-masked-image", "-h"],
388 ],
389 ids=[
390 "root",
391 "convert",
392 "inspect",
393 "minify",
394 "reformat",
395 "extract-test-data",
396 "extract-test-data-dp2",
397 "verify-rewrite",
398 "verify-rewrite-stage4",
399 "fuzz-masked-image",
400 ],
401)
402def test_short_help_alias(args: list[str]) -> None:
403 """Test that -h is an alias for --help on the group and every
404 subcommand.
405 """
406 result = CliRunner().invoke(main, args)
407 assert result.exit_code == 0, result.output
408 assert "Usage:" in result.output