Coverage for tests/test_cli.py: 91%
367 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-08 04:08 -0700
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-08 04:08 -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 json
14import os
15import shutil
16import subprocess
17import sys
18from pathlib import Path
19from unittest import mock
21import astropy.io.fits
22import click
23import click.testing
24import numpy as np
25import pytest
26from click.testing import CliRunner
28from lsst.images import Box, Image
29from lsst.images import fits as images_fits
30from lsst.images import json as images_json
31from lsst.images.cli import main
32from lsst.images.serialization import backend_for_path, read_archive
33from lsst.images.tests import current_fixture_path
35FIXTURE_DIR = Path(__file__).parent / "data" / "schemas"
38def _copy_fixture_tree(tmp_path: Path, *, source: Path = FIXTURE_DIR, name: str = "schemas") -> Path:
39 """Copy a fixture tree so a mutating CLI test cannot touch its source."""
40 destination = tmp_path / name
41 shutil.copytree(source, destination)
42 return destination
45@pytest.fixture(scope="session")
46def external_data_dir() -> str:
47 """Return the external test-data directory path, skipping if unset."""
48 if (result := os.environ.get("TESTDATA_IMAGES_DIR")) is None: 48 ↛ 50line 48 didn't jump to line 50 because the condition on line 48 was always true
49 pytest.skip("TESTDATA_IMAGES_DIR is not set.")
50 return result
53def _make_cli_input(tmp_path: Path) -> str:
54 """Return the path to a minimal FITS file written under tmp_path."""
55 path = str(tmp_path / "in.fits")
56 astropy.io.fits.PrimaryHDU().writeto(path)
57 return path
60def _make_detect_file(tmp_path: Path, dataset_type: str | None) -> str:
61 """Return a path to a FITS file with LSST BUTLER DATASETTYPE set to
62 dataset_type.
63 """
64 name = dataset_type.replace(" ", "_") if dataset_type is not None else "none"
65 path = str(tmp_path / f"detect_{name}.fits")
66 hdu = astropy.io.fits.PrimaryHDU()
67 with images_fits.suppress_fits_card_warnings():
68 if dataset_type is not None:
69 hdu.header["LSST BUTLER DATASETTYPE"] = dataset_type
70 hdu.writeto(path)
71 return path
74def test_group_help() -> None:
75 """Test that the root CLI group loads and lists core subcommands."""
76 result = CliRunner().invoke(main, ["--help"])
77 assert result.exit_code == 0, result.output
78 assert "convert" in result.output
79 assert "inspect" in result.output
82def test_python_m_entry_point() -> None:
83 """Test that python -m lsst.images runs the same CLI group."""
84 result = subprocess.run(
85 [sys.executable, "-m", "lsst.images", "--help"],
86 capture_output=True,
87 text=True,
88 )
89 assert result.returncode == 0, result.stderr
90 assert "convert" in result.stdout
91 assert "inspect" in result.stdout
94def test_inspect_fits(tmp_path: Path) -> None:
95 """Test 'inspect' on a FITS file."""
96 path = str(tmp_path / "x.fits")
97 image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4])
98 images_fits.write(image, path)
99 result = CliRunner().invoke(main, ["inspect", path])
100 assert result.exit_code == 0, result.output
101 assert "https://images.lsst.io/schemas/image-1.0.0" in result.output
102 assert "format version: 1" in result.output
103 assert "python class:" in result.output
104 assert "lsst.images.Image" in result.output
107def test_inspect_json(tmp_path: Path) -> None:
108 """Test 'inspect' on a JSON file."""
109 path = str(tmp_path / "x.json")
110 image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4])
111 images_json.write(image, path)
112 result = CliRunner().invoke(main, ["inspect", path])
113 assert result.exit_code == 0, result.output
114 assert "image-1.0.0" in result.output
115 assert "n/a" in result.output
116 assert "python class:" in result.output
117 assert "lsst.images.Image" in result.output
120def test_inspect_unregistered_schema(tmp_path: Path) -> None:
121 """Test that 'inspect' succeeds and reports an unregistered schema name."""
122 path = str(tmp_path / "fake.json")
123 with open(path, "w") as f:
124 f.write(
125 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",'
126 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}'
127 )
128 result = CliRunner().invoke(main, ["inspect", path])
129 assert result.exit_code == 0, result.output
130 assert "python class:" in result.output
131 assert "<unregistered: no-such-schema>" in result.output
134def test_inspect_unknown_extension(tmp_path: Path) -> None:
135 """Test that 'inspect' fails with a non-zero exit code for an unsupported
136 file extension.
137 """
138 path = str(tmp_path / "x.txt")
139 with open(path, "w") as stream:
140 stream.write("nope")
141 result = CliRunner().invoke(main, ["inspect", path])
142 assert result.exit_code != 0
143 assert ".fits" in result.output
146def test_reformat_round_trip_json_fits_json(tmp_path: Path) -> None:
147 """Test that reformat JSON→FITS→JSON preserves the image data."""
148 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
149 src = str(tmp_path / "in.json")
150 mid = str(tmp_path / "mid.fits")
151 out = str(tmp_path / "out.json")
152 images_json.write(image, src)
154 result = CliRunner().invoke(main, ["reformat", src, mid])
155 assert result.exit_code == 0, result.output
156 assert backend_for_path(mid).input_archive.get_basic_info(mid).schema_name == "image"
158 result = CliRunner().invoke(main, ["reformat", mid, out])
159 assert result.exit_code == 0, result.output
161 np.testing.assert_array_equal(read_archive(out, Image).array, image.array)
164def test_reformat_refuses_existing_output(tmp_path: Path) -> None:
165 """Test that reformat refuses to overwrite an existing output file without
166 --overwrite.
167 """
168 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
169 src = str(tmp_path / "in.json")
170 out = str(tmp_path / "out.fits")
171 images_json.write(image, src)
172 images_fits.write(image, out)
173 result = CliRunner().invoke(main, ["reformat", src, out])
174 assert result.exit_code != 0
175 assert "--overwrite" in result.output
178def test_reformat_unknown_output_extension(tmp_path: Path) -> None:
179 """Test that reformat fails for an unsupported output file extension."""
180 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4])
181 src = str(tmp_path / "in.json")
182 images_json.write(image, src)
183 result = CliRunner().invoke(main, ["reformat", src, str(tmp_path / "out.txt")])
184 assert result.exit_code != 0
185 assert ".fits" in result.output
188def test_detect_visit_image(tmp_path: Path) -> None:
189 """Test that detect_legacy_type identifies visit_image and
190 preliminary_visit_image.
191 """
192 from lsst.images.cli._convert import detect_legacy_type
194 assert detect_legacy_type(_make_detect_file(tmp_path, "visit_image")) == "visit_image"
195 assert detect_legacy_type(_make_detect_file(tmp_path, "preliminary_visit_image")) == "visit_image"
198def test_detect_cell_coadd(tmp_path: Path) -> None:
199 """Test that detect_legacy_type identifies deep_coadd_cell_predetection as
200 cell_coadd.
201 """
202 from lsst.images.cli._convert import detect_legacy_type
204 assert detect_legacy_type(_make_detect_file(tmp_path, "deep_coadd_cell_predetection")) == "cell_coadd"
207def test_detect_indeterminate(tmp_path: Path) -> None:
208 """Test that detect_legacy_type returns None for unknown or absent dataset-
209 type headers.
210 """
211 from lsst.images.cli._convert import detect_legacy_type
213 assert detect_legacy_type(_make_detect_file(tmp_path, None)) is None
214 assert detect_legacy_type(_make_detect_file(tmp_path, "camera")) is None
217def test_detect_visit_image_fixture(tmp_path: Path, external_data_dir: str) -> None:
218 """Test that detect_legacy_type detects a real legacy visit-image fixture
219 file.
220 """
221 from lsst.images.cli._convert import detect_legacy_type
223 path = os.path.join(external_data_dir, "dp2", "legacy", "visit_image.fits")
224 assert detect_legacy_type(path) == "visit_image"
227def test_convert_visit_image_to_json(tmp_path: Path, external_data_dir: str) -> None:
228 """Test that convert produces a valid visit_image JSON file from a legacy
229 FITS fixture.
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 / "converted.json")
234 result = CliRunner().invoke(main, ["convert", src, out])
235 assert result.exit_code == 0, result.output
236 info = backend_for_path(out).input_archive.get_basic_info(out)
237 assert info.schema_name == "visit_image"
240def test_convert_refuses_existing_output(tmp_path: Path, external_data_dir: str) -> None:
241 """Test that convert refuses to overwrite an existing output file without
242 --overwrite.
243 """
244 pytest.importorskip("lsst.afw.image")
245 src = os.path.join(external_data_dir, "dp2", "legacy", "visit_image.fits")
246 out = str(tmp_path / "exists.json")
247 with open(out, "w") as stream:
248 stream.write("{}")
249 result = CliRunner().invoke(main, ["convert", src, out])
250 assert result.exit_code != 0
251 assert "--overwrite" in result.output
254def test_convert_cell_coadd_to_json(tmp_path: Path, external_data_dir: str) -> None:
255 """Test that convert produces a valid cell_coadd JSON file from a legacy
256 FITS MultipleCellCoadd.
257 """
258 pytest.importorskip("lsst.cell_coadds")
259 legacy_dir = os.path.join(external_data_dir, "dp2", "legacy")
260 src = os.path.join(legacy_dir, "deep_coadd_cell_predetection.fits")
261 skymap = os.path.join(legacy_dir, "skyMap.pickle")
262 out = str(tmp_path / "coadd.json")
263 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd", "--skymap", skymap])
264 assert result.exit_code == 0, result.output
265 info = backend_for_path(out).input_archive.get_basic_info(out)
266 assert info.schema_name == "cell_coadd"
269def test_convert_cell_coadd_requires_skymap(tmp_path: Path, external_data_dir: str) -> None:
270 """Test that convert fails with a helpful message when --skymap is missing
271 for cell_coadd.
272 """
273 pytest.importorskip("lsst.cell_coadds")
274 src = os.path.join(external_data_dir, "dp2", "legacy", "deep_coadd_cell_predetection.fits")
275 out = str(tmp_path / "coadd.json")
276 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd"])
277 assert result.exit_code != 0
278 assert "--skymap" in result.output
281def test_preserve_quantization_default_is_true() -> None:
282 """Test that the --preserve-quantization option defaults to True."""
283 from lsst.images.cli._convert import convert
285 option = next(p for p in convert.params if p.name == "preserve_quantization")
286 assert option.default is True
289def test_preserve_quantization_explicit_flag_rejected_for_cell_coadd(tmp_path: Path) -> None:
290 """Test that explicitly passing --preserve-quantization is rejected for
291 cell_coadd conversions.
292 """
293 src = _make_cli_input(tmp_path)
294 out = str(tmp_path / "out.json")
295 result = CliRunner().invoke(
296 main, ["convert", src, out, "--type", "cell_coadd", "--preserve-quantization"]
297 )
298 assert result.exit_code != 0
299 assert "preserve-quantization" in result.output
302def test_preserve_quantization_default_does_not_reject_cell_coadd(tmp_path: Path) -> None:
303 """Test that the --preserve-quantization option default doesn't get in the
304 way of cell-coadd conversion.
305 """
306 src = _make_cli_input(tmp_path)
307 out = str(tmp_path / "out.json")
308 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd"])
309 assert "preserve-quantization" not in result.output
312def test_preserve_quantization_forwarded_to_read_legacy() -> None:
313 """Test that _read_legacy forwards preserve_quantization=True to
314 VisitImage.read_legacy.
315 """
316 from lsst.images.cli._convert import _read_legacy
318 with mock.patch("lsst.images.VisitImage.read_legacy") as read_legacy:
319 _read_legacy("in.fits", "visit_image", None, None, None, True)
320 read_legacy.assert_called_once_with("in.fits", preserve_quantization=True)
323def test_rejects_identical_paths(tmp_path: Path) -> None:
324 """Test that 'convert' rejects identical src and dst paths even with
325 --overwrite.
326 """
327 path = _make_cli_input(tmp_path)
328 result = CliRunner().invoke(main, ["convert", path, path, "--type", "visit_image", "--overwrite"])
329 assert result.exit_code != 0
330 assert "different" in result.output
331 assert os.path.exists(path)
334def test_preserves_existing_output_on_read_failure(tmp_path: Path) -> None:
335 """Test that 'convert' leaves the existing output file intact when
336 read_legacy raises.
337 """
338 src = _make_cli_input(tmp_path)
339 out = str(tmp_path / "out.json")
340 with open(out, "w") as stream:
341 stream.write("ORIGINAL")
342 with mock.patch(
343 "lsst.images.cli._convert._read_legacy",
344 side_effect=click.ClickException("boom"),
345 ):
346 result = CliRunner().invoke(main, ["convert", src, out, "--type", "visit_image", "--overwrite"])
347 assert result.exit_code != 0
348 with open(out) as stream:
349 assert stream.read() == "ORIGINAL"
352def test_subcommands_present() -> None:
353 """Test that the minify, reformat, extract-test-data, verify-rewrite, and
354 fuzz-masked-image are listed by --help.
355 """
356 result = CliRunner().invoke(main, ["--help"])
357 assert result.exit_code == 0, result.output
358 assert "minify" in result.output
359 assert "reformat" in result.output
360 assert "extract-test-data" in result.output
361 assert "verify-rewrite" in result.output
362 assert "fuzz-masked-image" in result.output
365def test_minify_help() -> None:
366 """Verify minify --help exits cleanly."""
367 result = CliRunner().invoke(main, ["minify", "--help"])
368 assert result.exit_code == 0, result.output
371def test_extract_test_data_help() -> None:
372 """Verify extract-test-data --help exits cleanly."""
373 result = CliRunner().invoke(main, ["extract-test-data", "--help"])
374 assert result.exit_code == 0, result.output
377def test_verify_rewrite_help() -> None:
378 """Verify verify-rewrite and its stage4 subcommand load with core deps
379 only.
380 """
381 result = CliRunner().invoke(main, ["verify-rewrite", "--help"])
382 assert result.exit_code == 0, result.output
383 assert "stage4" in result.output
384 result = CliRunner().invoke(main, ["verify-rewrite", "stage4", "--help"])
385 assert result.exit_code == 0, result.output
388@pytest.mark.parametrize(
389 "args",
390 [
391 ["-h"],
392 ["convert", "-h"],
393 ["inspect", "-h"],
394 ["minify", "-h"],
395 ["reformat", "-h"],
396 ["extract-test-data", "-h"],
397 ["extract-test-data", "dp2", "-h"],
398 ["verify-rewrite", "-h"],
399 ["verify-rewrite", "stage4", "-h"],
400 ["fuzz-masked-image", "-h"],
401 ["schemas", "-h"],
402 ["schemas", "write", "-h"],
403 ["schemas", "check", "-h"],
404 ["fixtures", "-h"],
405 ["fixtures", "check", "-h"],
406 ["fixtures", "refresh", "-h"],
407 ["fixtures", "freeze", "-h"],
408 ["fixtures", "coverage", "-h"],
409 ],
410 ids=[
411 "root",
412 "convert",
413 "inspect",
414 "minify",
415 "reformat",
416 "extract-test-data",
417 "extract-test-data-dp2",
418 "verify-rewrite",
419 "verify-rewrite-stage4",
420 "fuzz-masked-image",
421 "schemas",
422 "schemas-write",
423 "schemas-check",
424 "fixtures",
425 "fixtures-check",
426 "fixtures-refresh",
427 "fixtures-freeze",
428 "fixtures-coverage",
429 ],
430)
431def test_short_help_alias(args: list[str]) -> None:
432 """Test that -h is an alias for --help on the group and every
433 subcommand.
434 """
435 result = CliRunner().invoke(main, args)
436 assert result.exit_code == 0, result.output
437 assert "Usage:" in result.output
440def test_schemas_write_and_check(tmp_path: Path) -> None:
441 """Verify schemas write populates a directory that schemas check
442 accepts.
443 """
444 runner = CliRunner()
445 result = runner.invoke(main, ["schemas", "write", "--dir", str(tmp_path)])
446 assert result.exit_code == 0, result.output
447 assert list(tmp_path.glob("image/image-*.json"))
448 result = runner.invoke(main, ["schemas", "check", "--dir", str(tmp_path)])
449 assert result.exit_code == 0, result.output
452def test_schemas_write_package_option(tmp_path: Path) -> None:
453 """Verify --package freezes only schemas defined under that package."""
454 runner = CliRunner()
455 result = runner.invoke(
456 main, ["schemas", "write", "--dir", str(tmp_path), "--package", "lsst.images.cells"]
457 )
458 assert result.exit_code == 0, result.output
459 names = sorted(p.name for p in tmp_path.rglob("*.json"))
460 assert any(n.startswith("cell_coadd-") for n in names)
461 assert not any(n.startswith("image-") for n in names)
464def test_schemas_check_fails_when_stale(tmp_path: Path) -> None:
465 """Verify schemas check exits nonzero and names the fix when stale."""
466 runner = CliRunner()
467 runner.invoke(main, ["schemas", "write", "--dir", str(tmp_path)])
468 (path,) = tmp_path.glob("image/image-*.json")
469 path.unlink()
470 result = runner.invoke(main, ["schemas", "check", "--dir", str(tmp_path)])
471 assert result.exit_code != 0
472 assert "schemas write" in result.output
475def test_cli_describe_visit_image() -> None:
476 """The describe command renders a deserialized VisitImage."""
477 path = current_fixture_path(FIXTURE_DIR, "visit_image")
478 result = CliRunner().invoke(main, ["describe", str(path)])
479 assert result.exit_code == 0, result.output
480 assert "VisitImage" in result.output
483def test_cli_describe_coadd_provenance() -> None:
484 """The describe command renders a deserialized CoaddProvenance.
486 Provenance is describable in its own right, not only as part of a coadd,
487 so the command must not fall back to the default object repr for it.
488 """
489 path = current_fixture_path(FIXTURE_DIR, "coadd_provenance")
490 result = CliRunner().invoke(main, ["describe", str(path)])
491 assert result.exit_code == 0, result.output
492 assert "CoaddProvenance" in result.output
493 assert "input images" in result.output
496def test_fixtures_check_reports_a_clean_tree() -> None:
497 """Verify 'fixtures check' exits zero on the committed tree."""
498 runner = click.testing.CliRunner()
499 result = runner.invoke(
500 main,
501 [
502 "fixtures",
503 "check",
504 "--dir",
505 str(Path(__file__).parent / "data" / "schemas"),
506 "--schema-dir",
507 str(Path(__file__).parent.parent / "schemas"),
508 "--exempt",
509 "psfex_psf",
510 ],
511 )
512 assert result.exit_code == 0, result.output
515def test_fixtures_coverage_reports_positions_for_a_composite() -> None:
516 """Verify 'fixtures coverage --schema' reports a composite's positions.
518 Asserted on structure rather than on any particular gap, so adding a model
519 or widening a fixture cannot turn this into a failure.
520 """
521 result = click.testing.CliRunner().invoke(
522 main, ["fixtures", "coverage", "--dir", str(FIXTURE_DIR), "--schema", "cell_coadd"]
523 )
524 assert result.exit_code == 0, result.output
525 assert result.output.startswith("cell_coadd ")
526 assert "holds .psf [cell_psf]" in result.output
529def test_fixtures_coverage_emits_parseable_json() -> None:
530 """Verify 'fixtures coverage --format json' emits a JSON object."""
531 result = click.testing.CliRunner().invoke(
532 main,
533 ["fixtures", "coverage", "--dir", str(FIXTURE_DIR), "--schema", "cell_coadd", "--format", "json"],
534 )
535 assert result.exit_code == 0, result.output
536 parsed = json.loads(result.output)
537 (entry,) = parsed.values()
538 assert {"sources", "expressed", "absent", "positions"} == set(entry)
539 assert any(position["path"] == ".psf" for position in entry["positions"])
542def test_fixtures_coverage_reports_an_unknown_schema_cleanly() -> None:
543 """Verify a schema name that matches nothing says so and exits zero."""
544 result = click.testing.CliRunner().invoke(
545 main, ["fixtures", "coverage", "--dir", str(FIXTURE_DIR), "--schema", "no_such_schema"]
546 )
547 assert result.exit_code == 0, result.output
548 assert "no schema matches 'no_such_schema'" in result.output
551def test_fixtures_check_requires_a_schema_directory() -> None:
552 """Verify pairing cannot be accidentally omitted from the CLI check."""
553 result = click.testing.CliRunner().invoke(main, ["fixtures", "check"])
554 assert result.exit_code != 0
555 assert "Missing option '--schema-dir'" in result.output
558def test_fixtures_check_reports_a_problem(tmp_path: Path) -> None:
559 """Verify 'fixtures check' reports why a malformed fixture fails.
561 The empty object fails model validation, which must be reported as a
562 validation failure on this exact file, not merely as some problem
563 somewhere; a fixture tree with no ``image`` file at all would already
564 report that file as missing, so the assertion has to name the failure
565 mode to tell the two apart.
566 """
567 name = current_fixture_path(FIXTURE_DIR, "image").name
568 directory = tmp_path / "image"
569 directory.mkdir(parents=True)
570 (directory / name).write_text("{}\n")
571 runner = click.testing.CliRunner()
572 result = runner.invoke(
573 main,
574 [
575 "fixtures",
576 "check",
577 "--dir",
578 str(tmp_path),
579 "--schema-dir",
580 str(Path(__file__).parent.parent / "schemas"),
581 ],
582 )
583 assert result.exit_code != 0
584 assert f"{name}: does not validate" in result.output
587def test_fixtures_refresh_reports_no_change(tmp_path: Path) -> None:
588 """Verify 'fixtures refresh' would be a no-op on the committed tree.
590 The command is intentionally mutating, so run it on a fresh copy. If a
591 committed development fixture is dirty, every test run must copy and
592 detect that same dirty input rather than repairing the repository on the
593 first run and passing on the second.
594 """
595 directory = _copy_fixture_tree(tmp_path)
596 runner = click.testing.CliRunner()
597 result = runner.invoke(main, ["fixtures", "refresh", "--dir", str(directory)])
598 assert result.exit_code == 0, result.output
599 assert "already up to date" in result.output
602def test_fixtures_freeze_reports_nothing_to_freeze(tmp_path: Path) -> None:
603 """Verify 'fixtures freeze' would be a no-op on the committed tree.
605 Run the destructive command on a copy so a newly finalized schema makes
606 this test fail on every run without moving or deleting the developer's
607 fixture.
608 """
609 directory = _copy_fixture_tree(tmp_path)
610 runner = click.testing.CliRunner()
611 result = runner.invoke(main, ["fixtures", "freeze", "--dir", str(directory)])
612 assert result.exit_code == 0, result.output
613 assert "nothing to freeze" in result.output
616def test_fixtures_refresh_detection_is_repeatable_without_mutating_source(tmp_path: Path) -> None:
617 """Verify a dirty fixture is detected repeatedly without being repaired."""
618 source = _copy_fixture_tree(tmp_path, name="dirty-source")
619 fixture = current_fixture_path(source, "camera_frame_set")
620 data = json.loads(fixture.read_text())
621 data["schema_version"] = "2.0.0"
622 data["schema_url"] = "https://images.lsst.io/schemas/camera_frame_set-2.0.0"
623 dirty_text = json.dumps(data, indent=2) + "\n"
624 fixture.write_text(dirty_text)
626 for run in range(2):
627 directory = _copy_fixture_tree(tmp_path, source=source, name=f"run-{run}")
628 result = click.testing.CliRunner().invoke(main, ["fixtures", "refresh", "--dir", str(directory)])
629 assert result.exit_code == 0, result.output
630 assert "already up to date" not in result.output
631 assert "wrote" in result.output
632 assert fixture.read_text() == dirty_text
635def test_fixtures_refresh_reports_finalized_conflict_cleanly(tmp_path: Path) -> None:
636 """Verify 'fixtures refresh' turns a finalized-fixture conflict into a
637 clean error instead of a raw traceback.
638 """
639 source = current_fixture_path(FIXTURE_DIR, "image")
640 name = source.name
641 data = json.loads(source.read_text())
642 directory = tmp_path / "image"
643 directory.mkdir(parents=True)
644 # Valid but not canonically formatted, so it differs from the text
645 # refresh would regenerate for this already-finalized version.
646 (directory / name).write_text(json.dumps(data))
647 runner = click.testing.CliRunner()
648 result = runner.invoke(main, ["fixtures", "refresh", "--dir", str(tmp_path)])
649 assert result.exit_code != 0
650 assert name in result.output
651 assert "bump SCHEMA_VERSION" in result.output
652 assert "Traceback" not in result.output
655def test_fixtures_freeze_reports_existing_target_cleanly(tmp_path: Path) -> None:
656 """Verify 'fixtures freeze' turns a pre-existing target conflict into a
657 clean error instead of a raw traceback.
658 """
659 target = current_fixture_path(FIXTURE_DIR, "image")
660 target_name = target.name
661 dev_name = target_name.removesuffix(".json") + ".dev.json"
662 text = target.read_text()
663 directory = tmp_path / "image"
664 directory.mkdir(parents=True)
665 (directory / dev_name).write_text(text)
666 (directory / target_name).write_text(text)
667 runner = click.testing.CliRunner()
668 result = runner.invoke(main, ["fixtures", "freeze", "--dir", str(tmp_path)])
669 assert result.exit_code != 0
670 assert target_name in result.output
671 assert "already exists" in result.output
672 assert "Traceback" not in result.output
675def test_fixtures_freeze_reports_a_validation_failure_cleanly(tmp_path: Path) -> None:
676 """Verify 'fixtures freeze' turns a fixture that no longer validates into
677 a clean error instead of a raw traceback.
679 freeze_schema_fixtures reads each fixture through its live model
680 (read_fixture_tree) as it freezes it; a fixture that fails that read
681 raises pydantic.ValidationError or ArchiveReadError, neither of which is
682 SchemaFixtureError, so the CLI must catch them too rather than let a
683 traceback escape.
684 """
685 directory = tmp_path / "image"
686 directory.mkdir(parents=True)
687 (directory / "image-1.0.0.dev.json").write_text("{}\n")
688 runner = click.testing.CliRunner()
689 result = runner.invoke(main, ["fixtures", "freeze", "--dir", str(tmp_path)])
690 assert result.exit_code != 0
691 assert "Traceback" not in result.output
692 # A raw pydantic.ValidationError escaping uncaught propagates out of
693 # CliRunner.invoke() as result.exception with no "Error: ..." line ever
694 # written to result.output; only a caught-and-reraised
695 # click.ClickException produces that line and a clean SystemExit.
696 assert "Error:" in result.output
697 assert isinstance(result.exception, SystemExit)
700def test_fixtures_refresh_reports_a_validation_failure_cleanly(tmp_path: Path) -> None:
701 """Verify 'fixtures refresh' turns a fixture that no longer validates
702 into a clean error instead of a raw traceback.
704 refresh_schema_fixtures reads a development fixture through
705 read_fixture_tree to canonicalize it; a fixture that fails that read
706 raises pydantic.ValidationError or ArchiveReadError, neither of which is
707 SchemaFixtureError, so the CLI must catch them too rather than let a
708 traceback escape. camera_frame_set is a real, still-developing
709 lsst.images schema, so this needs no test double.
710 """
711 directory = tmp_path / "camera_frame_set"
712 directory.mkdir(parents=True)
713 (directory / "camera_frame_set-1.0.0.dev.json").write_text("{}\n")
714 runner = click.testing.CliRunner()
715 result = runner.invoke(main, ["fixtures", "refresh", "--dir", str(tmp_path)])
716 assert result.exit_code != 0
717 assert "Traceback" not in result.output
718 # A raw pydantic.ValidationError escaping uncaught propagates out of
719 # CliRunner.invoke() as result.exception with no "Error: ..." line ever
720 # written to result.output; only a caught-and-reraised
721 # click.ClickException produces that line and a clean SystemExit.
722 assert "Error:" in result.output
723 assert isinstance(result.exception, SystemExit)