Coverage for tests/test_cli.py: 91%

367 statements  

« prev     ^ index     » next       coverage.py v7.15.3, created at 2026-08-08 11:10 +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 

12 

13import json 

14import os 

15import shutil 

16import subprocess 

17import sys 

18from pathlib import Path 

19from unittest import mock 

20 

21import astropy.io.fits 

22import click 

23import click.testing 

24import numpy as np 

25import pytest 

26from click.testing import CliRunner 

27 

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 

34 

35FIXTURE_DIR = Path(__file__).parent / "data" / "schemas" 

36 

37 

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 

43 

44 

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 

51 

52 

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 

58 

59 

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 

72 

73 

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 

80 

81 

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 

92 

93 

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 

105 

106 

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 

118 

119 

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 

132 

133 

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 

144 

145 

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) 

153 

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" 

157 

158 result = CliRunner().invoke(main, ["reformat", mid, out]) 

159 assert result.exit_code == 0, result.output 

160 

161 np.testing.assert_array_equal(read_archive(out, Image).array, image.array) 

162 

163 

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 

176 

177 

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 

186 

187 

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 

193 

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" 

196 

197 

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 

203 

204 assert detect_legacy_type(_make_detect_file(tmp_path, "deep_coadd_cell_predetection")) == "cell_coadd" 

205 

206 

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 

212 

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 

215 

216 

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 

222 

223 path = os.path.join(external_data_dir, "dp2", "legacy", "visit_image.fits") 

224 assert detect_legacy_type(path) == "visit_image" 

225 

226 

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" 

238 

239 

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 

252 

253 

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" 

267 

268 

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 

279 

280 

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 

284 

285 option = next(p for p in convert.params if p.name == "preserve_quantization") 

286 assert option.default is True 

287 

288 

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 

300 

301 

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 

310 

311 

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 

317 

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) 

321 

322 

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) 

332 

333 

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" 

350 

351 

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 

363 

364 

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 

369 

370 

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 

375 

376 

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 

386 

387 

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 

438 

439 

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 

450 

451 

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) 

462 

463 

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 

473 

474 

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 

481 

482 

483def test_cli_describe_coadd_provenance() -> None: 

484 """The describe command renders a deserialized CoaddProvenance. 

485 

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 

494 

495 

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 

513 

514 

515def test_fixtures_coverage_reports_positions_for_a_composite() -> None: 

516 """Verify 'fixtures coverage --schema' reports a composite's positions. 

517 

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 

527 

528 

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"]) 

540 

541 

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 

549 

550 

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 

556 

557 

558def test_fixtures_check_reports_a_problem(tmp_path: Path) -> None: 

559 """Verify 'fixtures check' reports why a malformed fixture fails. 

560 

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 

585 

586 

587def test_fixtures_refresh_reports_no_change(tmp_path: Path) -> None: 

588 """Verify 'fixtures refresh' would be a no-op on the committed tree. 

589 

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 

600 

601 

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. 

604 

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 

614 

615 

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) 

625 

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 

633 

634 

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 

653 

654 

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 

673 

674 

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. 

678 

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) 

698 

699 

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. 

703 

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)