Coverage for tests/test_cli.py: 91%

368 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-26 09:45 +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 cli_schema_doubles import CliFixtureDouble 

27from click.testing import CliRunner 

28 

29from lsst.images import Box, Image 

30from lsst.images import fits as images_fits 

31from lsst.images import json as images_json 

32from lsst.images.cli import main 

33from lsst.images.serialization import backend_for_path, read_archive 

34from lsst.images.tests import current_fixture_path 

35 

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

37 

38 

39def _copy_fixture_tree(tmp_path: Path, *, source: Path = FIXTURE_DIR, name: str = "schemas") -> Path: 

40 """Copy a fixture tree so a mutating CLI test cannot touch its source.""" 

41 destination = tmp_path / name 

42 shutil.copytree(source, destination) 

43 return destination 

44 

45 

46@pytest.fixture(scope="session") 

47def external_data_dir() -> str: 

48 """Return the external test-data directory path, skipping if unset.""" 

49 if (result := os.environ.get("TESTDATA_IMAGES_DIR")) is None: 49 ↛ 51line 49 didn't jump to line 51 because the condition on line 49 was always true

50 pytest.skip("TESTDATA_IMAGES_DIR is not set.") 

51 return result 

52 

53 

54def _make_cli_input(tmp_path: Path) -> str: 

55 """Return the path to a minimal FITS file written under tmp_path.""" 

56 path = str(tmp_path / "in.fits") 

57 astropy.io.fits.PrimaryHDU().writeto(path) 

58 return path 

59 

60 

61def _make_detect_file(tmp_path: Path, dataset_type: str | None) -> str: 

62 """Return a path to a FITS file with LSST BUTLER DATASETTYPE set to 

63 dataset_type. 

64 """ 

65 name = dataset_type.replace(" ", "_") if dataset_type is not None else "none" 

66 path = str(tmp_path / f"detect_{name}.fits") 

67 hdu = astropy.io.fits.PrimaryHDU() 

68 with images_fits.suppress_fits_card_warnings(): 

69 if dataset_type is not None: 

70 hdu.header["LSST BUTLER DATASETTYPE"] = dataset_type 

71 hdu.writeto(path) 

72 return path 

73 

74 

75def test_group_help() -> None: 

76 """Test that the root CLI group loads and lists core subcommands.""" 

77 result = CliRunner().invoke(main, ["--help"]) 

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

79 assert "convert" in result.output 

80 assert "inspect" in result.output 

81 

82 

83def test_python_m_entry_point() -> None: 

84 """Test that python -m lsst.images runs the same CLI group.""" 

85 result = subprocess.run( 

86 [sys.executable, "-m", "lsst.images", "--help"], 

87 capture_output=True, 

88 text=True, 

89 ) 

90 assert result.returncode == 0, result.stderr 

91 assert "convert" in result.stdout 

92 assert "inspect" in result.stdout 

93 

94 

95def test_inspect_fits(tmp_path: Path) -> None: 

96 """Test 'inspect' on a FITS file.""" 

97 path = str(tmp_path / "x.fits") 

98 image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]) 

99 images_fits.write(image, path) 

100 result = CliRunner().invoke(main, ["inspect", path]) 

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

102 assert "https://images.lsst.io/schemas/image-1.0.0" in result.output 

103 assert "format version: 1" in result.output 

104 assert "python class:" in result.output 

105 assert "lsst.images.Image" in result.output 

106 

107 

108def test_inspect_json(tmp_path: Path) -> None: 

109 """Test 'inspect' on a JSON file.""" 

110 path = str(tmp_path / "x.json") 

111 image = Image(np.zeros((4, 4), dtype=np.float32), bbox=Box.factory[0:4, 0:4]) 

112 images_json.write(image, path) 

113 result = CliRunner().invoke(main, ["inspect", path]) 

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

115 assert "image-1.0.0" in result.output 

116 assert "n/a" in result.output 

117 assert "python class:" in result.output 

118 assert "lsst.images.Image" in result.output 

119 

120 

121def test_inspect_unregistered_schema(tmp_path: Path) -> None: 

122 """Test that 'inspect' succeeds and reports an unregistered schema name.""" 

123 path = str(tmp_path / "fake.json") 

124 with open(path, "w") as f: 

125 f.write( 

126 '{"schema_url": "https://images.lsst.io/schemas/no-such-schema-99.0.0",' 

127 ' "schema_version": "99.0.0", "min_read_version": 1, "indirect": []}' 

128 ) 

129 result = CliRunner().invoke(main, ["inspect", path]) 

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

131 assert "python class:" in result.output 

132 assert "<unregistered: no-such-schema>" in result.output 

133 

134 

135def test_inspect_unknown_extension(tmp_path: Path) -> None: 

136 """Test that 'inspect' fails with a non-zero exit code for an unsupported 

137 file extension. 

138 """ 

139 path = str(tmp_path / "x.txt") 

140 with open(path, "w") as stream: 

141 stream.write("nope") 

142 result = CliRunner().invoke(main, ["inspect", path]) 

143 assert result.exit_code != 0 

144 assert ".fits" in result.output 

145 

146 

147def test_reformat_round_trip_json_fits_json(tmp_path: Path) -> None: 

148 """Test that reformat JSON→FITS→JSON preserves the image data.""" 

149 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4]) 

150 src = str(tmp_path / "in.json") 

151 mid = str(tmp_path / "mid.fits") 

152 out = str(tmp_path / "out.json") 

153 images_json.write(image, src) 

154 

155 result = CliRunner().invoke(main, ["reformat", src, mid]) 

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

157 assert backend_for_path(mid).input_archive.get_basic_info(mid).schema_name == "image" 

158 

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

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

161 

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

163 

164 

165def test_reformat_refuses_existing_output(tmp_path: Path) -> None: 

166 """Test that reformat refuses to overwrite an existing output file without 

167 --overwrite. 

168 """ 

169 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4]) 

170 src = str(tmp_path / "in.json") 

171 out = str(tmp_path / "out.fits") 

172 images_json.write(image, src) 

173 images_fits.write(image, out) 

174 result = CliRunner().invoke(main, ["reformat", src, out]) 

175 assert result.exit_code != 0 

176 assert "--overwrite" in result.output 

177 

178 

179def test_reformat_unknown_output_extension(tmp_path: Path) -> None: 

180 """Test that reformat fails for an unsupported output file extension.""" 

181 image = Image(np.arange(16, dtype=np.float32).reshape(4, 4), bbox=Box.factory[0:4, 0:4]) 

182 src = str(tmp_path / "in.json") 

183 images_json.write(image, src) 

184 result = CliRunner().invoke(main, ["reformat", src, str(tmp_path / "out.txt")]) 

185 assert result.exit_code != 0 

186 assert ".fits" in result.output 

187 

188 

189def test_detect_visit_image(tmp_path: Path) -> None: 

190 """Test that detect_legacy_type identifies visit_image and 

191 preliminary_visit_image. 

192 """ 

193 from lsst.images.cli._convert import detect_legacy_type 

194 

195 assert detect_legacy_type(_make_detect_file(tmp_path, "visit_image")) == "visit_image" 

196 assert detect_legacy_type(_make_detect_file(tmp_path, "preliminary_visit_image")) == "visit_image" 

197 

198 

199def test_detect_cell_coadd(tmp_path: Path) -> None: 

200 """Test that detect_legacy_type identifies deep_coadd_cell_predetection as 

201 cell_coadd. 

202 """ 

203 from lsst.images.cli._convert import detect_legacy_type 

204 

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

206 

207 

208def test_detect_indeterminate(tmp_path: Path) -> None: 

209 """Test that detect_legacy_type returns None for unknown or absent dataset- 

210 type headers. 

211 """ 

212 from lsst.images.cli._convert import detect_legacy_type 

213 

214 assert detect_legacy_type(_make_detect_file(tmp_path, None)) is None 

215 assert detect_legacy_type(_make_detect_file(tmp_path, "camera")) is None 

216 

217 

218def test_detect_visit_image_fixture(tmp_path: Path, external_data_dir: str) -> None: 

219 """Test that detect_legacy_type detects a real legacy visit-image fixture 

220 file. 

221 """ 

222 from lsst.images.cli._convert import detect_legacy_type 

223 

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

225 assert detect_legacy_type(path) == "visit_image" 

226 

227 

228def test_convert_visit_image_to_json(tmp_path: Path, external_data_dir: str) -> None: 

229 """Test that convert produces a valid visit_image JSON file from a legacy 

230 FITS fixture. 

231 """ 

232 pytest.importorskip("lsst.afw.image") 

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

234 out = str(tmp_path / "converted.json") 

235 result = CliRunner().invoke(main, ["convert", src, out]) 

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

237 info = backend_for_path(out).input_archive.get_basic_info(out) 

238 assert info.schema_name == "visit_image" 

239 

240 

241def test_convert_refuses_existing_output(tmp_path: Path, external_data_dir: str) -> None: 

242 """Test that convert refuses to overwrite an existing output file without 

243 --overwrite. 

244 """ 

245 pytest.importorskip("lsst.afw.image") 

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

247 out = str(tmp_path / "exists.json") 

248 with open(out, "w") as stream: 

249 stream.write("{}") 

250 result = CliRunner().invoke(main, ["convert", src, out]) 

251 assert result.exit_code != 0 

252 assert "--overwrite" in result.output 

253 

254 

255def test_convert_cell_coadd_to_json(tmp_path: Path, external_data_dir: str) -> None: 

256 """Test that convert produces a valid cell_coadd JSON file from a legacy 

257 FITS MultipleCellCoadd. 

258 """ 

259 pytest.importorskip("lsst.cell_coadds") 

260 legacy_dir = os.path.join(external_data_dir, "dp2", "legacy") 

261 src = os.path.join(legacy_dir, "deep_coadd_cell_predetection.fits") 

262 skymap = os.path.join(legacy_dir, "skyMap.pickle") 

263 out = str(tmp_path / "coadd.json") 

264 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd", "--skymap", skymap]) 

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

266 info = backend_for_path(out).input_archive.get_basic_info(out) 

267 assert info.schema_name == "cell_coadd" 

268 

269 

270def test_convert_cell_coadd_requires_skymap(tmp_path: Path, external_data_dir: str) -> None: 

271 """Test that convert fails with a helpful message when --skymap is missing 

272 for cell_coadd. 

273 """ 

274 pytest.importorskip("lsst.cell_coadds") 

275 src = os.path.join(external_data_dir, "dp2", "legacy", "deep_coadd_cell_predetection.fits") 

276 out = str(tmp_path / "coadd.json") 

277 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd"]) 

278 assert result.exit_code != 0 

279 assert "--skymap" in result.output 

280 

281 

282def test_preserve_quantization_default_is_true() -> None: 

283 """Test that the --preserve-quantization option defaults to True.""" 

284 from lsst.images.cli._convert import convert 

285 

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

287 assert option.default is True 

288 

289 

290def test_preserve_quantization_explicit_flag_rejected_for_cell_coadd(tmp_path: Path) -> None: 

291 """Test that explicitly passing --preserve-quantization is rejected for 

292 cell_coadd conversions. 

293 """ 

294 src = _make_cli_input(tmp_path) 

295 out = str(tmp_path / "out.json") 

296 result = CliRunner().invoke( 

297 main, ["convert", src, out, "--type", "cell_coadd", "--preserve-quantization"] 

298 ) 

299 assert result.exit_code != 0 

300 assert "preserve-quantization" in result.output 

301 

302 

303def test_preserve_quantization_default_does_not_reject_cell_coadd(tmp_path: Path) -> None: 

304 """Test that the --preserve-quantization option default doesn't get in the 

305 way of cell-coadd conversion. 

306 """ 

307 src = _make_cli_input(tmp_path) 

308 out = str(tmp_path / "out.json") 

309 result = CliRunner().invoke(main, ["convert", src, out, "--type", "cell_coadd"]) 

310 assert "preserve-quantization" not in result.output 

311 

312 

313def test_preserve_quantization_forwarded_to_read_legacy() -> None: 

314 """Test that _read_legacy forwards preserve_quantization=True to 

315 VisitImage.read_legacy. 

316 """ 

317 from lsst.images.cli._convert import _read_legacy 

318 

319 with mock.patch("lsst.images.VisitImage.read_legacy") as read_legacy: 

320 _read_legacy("in.fits", "visit_image", None, None, None, True) 

321 read_legacy.assert_called_once_with("in.fits", preserve_quantization=True) 

322 

323 

324def test_rejects_identical_paths(tmp_path: Path) -> None: 

325 """Test that 'convert' rejects identical src and dst paths even with 

326 --overwrite. 

327 """ 

328 path = _make_cli_input(tmp_path) 

329 result = CliRunner().invoke(main, ["convert", path, path, "--type", "visit_image", "--overwrite"]) 

330 assert result.exit_code != 0 

331 assert "different" in result.output 

332 assert os.path.exists(path) 

333 

334 

335def test_preserves_existing_output_on_read_failure(tmp_path: Path) -> None: 

336 """Test that 'convert' leaves the existing output file intact when 

337 read_legacy raises. 

338 """ 

339 src = _make_cli_input(tmp_path) 

340 out = str(tmp_path / "out.json") 

341 with open(out, "w") as stream: 

342 stream.write("ORIGINAL") 

343 with mock.patch( 

344 "lsst.images.cli._convert._read_legacy", 

345 side_effect=click.ClickException("boom"), 

346 ): 

347 result = CliRunner().invoke(main, ["convert", src, out, "--type", "visit_image", "--overwrite"]) 

348 assert result.exit_code != 0 

349 with open(out) as stream: 

350 assert stream.read() == "ORIGINAL" 

351 

352 

353def test_subcommands_present() -> None: 

354 """Test that the minify, reformat, extract-test-data, verify-rewrite, and 

355 fuzz-masked-image are listed by --help. 

356 """ 

357 result = CliRunner().invoke(main, ["--help"]) 

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

359 assert "minify" in result.output 

360 assert "reformat" in result.output 

361 assert "extract-test-data" in result.output 

362 assert "verify-rewrite" in result.output 

363 assert "fuzz-masked-image" in result.output 

364 

365 

366def test_minify_help() -> None: 

367 """Verify minify --help exits cleanly.""" 

368 result = CliRunner().invoke(main, ["minify", "--help"]) 

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

370 

371 

372def test_extract_test_data_help() -> None: 

373 """Verify extract-test-data --help exits cleanly.""" 

374 result = CliRunner().invoke(main, ["extract-test-data", "--help"]) 

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

376 

377 

378def test_verify_rewrite_help() -> None: 

379 """Verify verify-rewrite and its stage4 subcommand load with core deps 

380 only. 

381 """ 

382 result = CliRunner().invoke(main, ["verify-rewrite", "--help"]) 

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

384 assert "stage4" in result.output 

385 result = CliRunner().invoke(main, ["verify-rewrite", "stage4", "--help"]) 

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

387 

388 

389@pytest.mark.parametrize( 

390 "args", 

391 [ 

392 ["-h"], 

393 ["convert", "-h"], 

394 ["inspect", "-h"], 

395 ["minify", "-h"], 

396 ["reformat", "-h"], 

397 ["extract-test-data", "-h"], 

398 ["extract-test-data", "dp2", "-h"], 

399 ["verify-rewrite", "-h"], 

400 ["verify-rewrite", "stage4", "-h"], 

401 ["fuzz-masked-image", "-h"], 

402 ["schemas", "-h"], 

403 ["schemas", "write", "-h"], 

404 ["schemas", "check", "-h"], 

405 ["fixtures", "-h"], 

406 ["fixtures", "check", "-h"], 

407 ["fixtures", "refresh", "-h"], 

408 ["fixtures", "freeze", "-h"], 

409 ["fixtures", "coverage", "-h"], 

410 ], 

411 ids=[ 

412 "root", 

413 "convert", 

414 "inspect", 

415 "minify", 

416 "reformat", 

417 "extract-test-data", 

418 "extract-test-data-dp2", 

419 "verify-rewrite", 

420 "verify-rewrite-stage4", 

421 "fuzz-masked-image", 

422 "schemas", 

423 "schemas-write", 

424 "schemas-check", 

425 "fixtures", 

426 "fixtures-check", 

427 "fixtures-refresh", 

428 "fixtures-freeze", 

429 "fixtures-coverage", 

430 ], 

431) 

432def test_short_help_alias(args: list[str]) -> None: 

433 """Test that -h is an alias for --help on the group and every 

434 subcommand. 

435 """ 

436 result = CliRunner().invoke(main, args) 

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

438 assert "Usage:" in result.output 

439 

440 

441def test_schemas_write_and_check(tmp_path: Path) -> None: 

442 """Verify schemas write populates a directory that schemas check 

443 accepts. 

444 """ 

445 runner = CliRunner() 

446 result = runner.invoke(main, ["schemas", "write", "--dir", str(tmp_path)]) 

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

448 assert list(tmp_path.glob("image/image-*.json")) 

449 result = runner.invoke(main, ["schemas", "check", "--dir", str(tmp_path)]) 

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

451 

452 

453def test_schemas_write_package_option(tmp_path: Path) -> None: 

454 """Verify --package freezes only schemas defined under that package.""" 

455 runner = CliRunner() 

456 result = runner.invoke( 

457 main, ["schemas", "write", "--dir", str(tmp_path), "--package", "lsst.images.cells"] 

458 ) 

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

460 names = sorted(p.name for p in tmp_path.rglob("*.json")) 

461 assert any(n.startswith("cell_coadd-") for n in names) 

462 assert not any(n.startswith("image-") for n in names) 

463 

464 

465def test_schemas_check_fails_when_stale(tmp_path: Path) -> None: 

466 """Verify schemas check exits nonzero and names the fix when stale.""" 

467 runner = CliRunner() 

468 runner.invoke(main, ["schemas", "write", "--dir", str(tmp_path)]) 

469 (path,) = tmp_path.glob("image/image-*.json") 

470 path.unlink() 

471 result = runner.invoke(main, ["schemas", "check", "--dir", str(tmp_path)]) 

472 assert result.exit_code != 0 

473 assert "schemas write" in result.output 

474 

475 

476def test_cli_describe_visit_image() -> None: 

477 """The describe command renders a deserialized VisitImage.""" 

478 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

479 result = CliRunner().invoke(main, ["describe", str(path)]) 

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

481 assert "VisitImage" in result.output 

482 

483 

484def test_cli_describe_coadd_provenance() -> None: 

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

486 

487 Provenance is describable in its own right, not only as part of a coadd, 

488 so the command must not fall back to the default object repr for it. 

489 """ 

490 path = current_fixture_path(FIXTURE_DIR, "coadd_provenance") 

491 result = CliRunner().invoke(main, ["describe", str(path)]) 

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

493 assert "CoaddProvenance" in result.output 

494 assert "input images" in result.output 

495 

496 

497def test_fixtures_check_reports_a_clean_tree() -> None: 

498 """Verify 'fixtures check' exits zero on the committed tree.""" 

499 runner = click.testing.CliRunner() 

500 result = runner.invoke( 

501 main, 

502 [ 

503 "fixtures", 

504 "check", 

505 "--dir", 

506 str(Path(__file__).parent / "data" / "schemas"), 

507 "--schema-dir", 

508 str(Path(__file__).parent.parent / "schemas"), 

509 "--exempt", 

510 "psfex_psf", 

511 ], 

512 ) 

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

514 

515 

516def test_fixtures_coverage_reports_positions_for_a_composite() -> None: 

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

518 

519 Asserted on structure rather than on any particular gap, so adding a model 

520 or widening a fixture cannot turn this into a failure. 

521 """ 

522 result = click.testing.CliRunner().invoke( 

523 main, ["fixtures", "coverage", "--dir", str(FIXTURE_DIR), "--schema", "cell_coadd"] 

524 ) 

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

526 assert result.output.startswith("cell_coadd ") 

527 assert "holds .psf [cell_psf]" in result.output 

528 

529 

530def test_fixtures_coverage_emits_parseable_json() -> None: 

531 """Verify 'fixtures coverage --format json' emits a JSON object.""" 

532 result = click.testing.CliRunner().invoke( 

533 main, 

534 ["fixtures", "coverage", "--dir", str(FIXTURE_DIR), "--schema", "cell_coadd", "--format", "json"], 

535 ) 

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

537 parsed = json.loads(result.output) 

538 (entry,) = parsed.values() 

539 assert {"sources", "expressed", "absent", "positions"} == set(entry) 

540 assert any(position["path"] == ".psf" for position in entry["positions"]) 

541 

542 

543def test_fixtures_coverage_reports_an_unknown_schema_cleanly() -> None: 

544 """Verify a schema name that matches nothing says so and exits zero.""" 

545 result = click.testing.CliRunner().invoke( 

546 main, ["fixtures", "coverage", "--dir", str(FIXTURE_DIR), "--schema", "no_such_schema"] 

547 ) 

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

549 assert "no schema matches 'no_such_schema'" in result.output 

550 

551 

552def test_fixtures_check_requires_a_schema_directory() -> None: 

553 """Verify pairing cannot be accidentally omitted from the CLI check.""" 

554 result = click.testing.CliRunner().invoke(main, ["fixtures", "check"]) 

555 assert result.exit_code != 0 

556 assert "Missing option '--schema-dir'" in result.output 

557 

558 

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

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

561 

562 The empty object fails model validation, which must be reported as a 

563 validation failure on this exact file, not merely as some problem 

564 somewhere; a fixture tree with no ``image`` file at all would already 

565 report that file as missing, so the assertion has to name the failure 

566 mode to tell the two apart. 

567 """ 

568 name = current_fixture_path(FIXTURE_DIR, "image").name 

569 directory = tmp_path / "image" 

570 directory.mkdir(parents=True) 

571 (directory / name).write_text("{}\n") 

572 runner = click.testing.CliRunner() 

573 result = runner.invoke( 

574 main, 

575 [ 

576 "fixtures", 

577 "check", 

578 "--dir", 

579 str(tmp_path), 

580 "--schema-dir", 

581 str(Path(__file__).parent.parent / "schemas"), 

582 ], 

583 ) 

584 assert result.exit_code != 0 

585 assert f"{name}: does not validate" in result.output 

586 

587 

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

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

590 

591 The command is intentionally mutating, so run it on a fresh copy. If a 

592 committed development fixture is dirty, every test run must copy and 

593 detect that same dirty input rather than repairing the repository on the 

594 first run and passing on the second. 

595 """ 

596 directory = _copy_fixture_tree(tmp_path) 

597 runner = click.testing.CliRunner() 

598 result = runner.invoke(main, ["fixtures", "refresh", "--dir", str(directory)]) 

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

600 assert "already up to date" in result.output 

601 

602 

603def test_fixtures_freeze_reports_nothing_to_freeze(tmp_path: Path) -> None: 

604 """Verify 'fixtures freeze' would be a no-op on the committed tree. 

605 

606 Run the destructive command on a copy so a newly finalized schema makes 

607 this test fail on every run without moving or deleting the developer's 

608 fixture. 

609 """ 

610 directory = _copy_fixture_tree(tmp_path) 

611 runner = click.testing.CliRunner() 

612 result = runner.invoke(main, ["fixtures", "freeze", "--dir", str(directory)]) 

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

614 assert "nothing to freeze" in result.output 

615 

616 

617def test_fixtures_refresh_detection_is_repeatable_without_mutating_source(tmp_path: Path) -> None: 

618 """Verify a dirty fixture is detected repeatedly without being repaired. 

619 

620 Dirty here means valid but not canonically formatted, which is what 

621 refresh regenerates. The fixture belongs to `CliFixtureDouble` because 

622 refresh acts only on a schema still in development. 

623 """ 

624 name = CliFixtureDouble.SCHEMA_NAME 

625 source = tmp_path / "dirty-source" 

626 (source / name).mkdir(parents=True) 

627 fixture = source / name / f"{name}-1.0.0.dev.json" 

628 dirty_text = json.dumps({"schema_version": "1.0.0.dev0", "min_read_version": 1, "value": "x"}) 

629 fixture.write_text(dirty_text) 

630 

631 for run in range(2): 

632 directory = _copy_fixture_tree(tmp_path, source=source, name=f"run-{run}") 

633 result = click.testing.CliRunner().invoke( 

634 main, 

635 ["fixtures", "refresh", "--dir", str(directory), "--package", CliFixtureDouble.__module__], 

636 ) 

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

638 assert "already up to date" not in result.output 

639 assert "wrote" in result.output 

640 assert fixture.read_text() == dirty_text 

641 

642 

643def test_fixtures_refresh_reports_finalized_conflict_cleanly(tmp_path: Path) -> None: 

644 """Verify 'fixtures refresh' turns a finalized-fixture conflict into a 

645 clean error instead of a raw traceback. 

646 """ 

647 source = current_fixture_path(FIXTURE_DIR, "image") 

648 name = source.name 

649 data = json.loads(source.read_text()) 

650 directory = tmp_path / "image" 

651 directory.mkdir(parents=True) 

652 # Valid but not canonically formatted, so it differs from the text 

653 # refresh would regenerate for this already-finalized version. 

654 (directory / name).write_text(json.dumps(data)) 

655 runner = click.testing.CliRunner() 

656 result = runner.invoke(main, ["fixtures", "refresh", "--dir", str(tmp_path)]) 

657 assert result.exit_code != 0 

658 assert name in result.output 

659 assert "bump SCHEMA_VERSION" in result.output 

660 assert "Traceback" not in result.output 

661 

662 

663def test_fixtures_freeze_reports_existing_target_cleanly(tmp_path: Path) -> None: 

664 """Verify 'fixtures freeze' turns a pre-existing target conflict into a 

665 clean error instead of a raw traceback. 

666 """ 

667 target = current_fixture_path(FIXTURE_DIR, "image") 

668 target_name = target.name 

669 dev_name = target_name.removesuffix(".json") + ".dev.json" 

670 text = target.read_text() 

671 directory = tmp_path / "image" 

672 directory.mkdir(parents=True) 

673 (directory / dev_name).write_text(text) 

674 (directory / target_name).write_text(text) 

675 runner = click.testing.CliRunner() 

676 result = runner.invoke(main, ["fixtures", "freeze", "--dir", str(tmp_path)]) 

677 assert result.exit_code != 0 

678 assert target_name in result.output 

679 assert "already exists" in result.output 

680 assert "Traceback" not in result.output 

681 

682 

683def test_fixtures_freeze_reports_a_validation_failure_cleanly(tmp_path: Path) -> None: 

684 """Verify 'fixtures freeze' turns a fixture that no longer validates into 

685 a clean error instead of a raw traceback. 

686 

687 freeze_schema_fixtures reads each fixture through its live model 

688 (read_fixture_tree) as it freezes it; a fixture that fails that read 

689 raises pydantic.ValidationError or ArchiveReadError, neither of which is 

690 SchemaFixtureError, so the CLI must catch them too rather than let a 

691 traceback escape. 

692 """ 

693 directory = tmp_path / "image" 

694 directory.mkdir(parents=True) 

695 (directory / "image-1.0.0.dev.json").write_text("{}\n") 

696 runner = click.testing.CliRunner() 

697 result = runner.invoke(main, ["fixtures", "freeze", "--dir", str(tmp_path)]) 

698 assert result.exit_code != 0 

699 assert "Traceback" not in result.output 

700 # A raw pydantic.ValidationError escaping uncaught propagates out of 

701 # CliRunner.invoke() as result.exception with no "Error: ..." line ever 

702 # written to result.output; only a caught-and-reraised 

703 # click.ClickException produces that line and a clean SystemExit. 

704 assert "Error:" in result.output 

705 assert isinstance(result.exception, SystemExit) 

706 

707 

708def test_fixtures_refresh_reports_a_validation_failure_cleanly(tmp_path: Path) -> None: 

709 """Verify 'fixtures refresh' turns a fixture that no longer validates 

710 into a clean error instead of a raw traceback. 

711 

712 refresh_schema_fixtures reads a development fixture through 

713 read_fixture_tree to canonicalize it; a fixture that fails that read 

714 raises pydantic.ValidationError or ArchiveReadError, neither of which is 

715 SchemaFixtureError, so the CLI must catch them too rather than let a 

716 traceback escape. Refresh acts only on a schema still in development, so 

717 this uses `CliFixtureDouble` rather than a real schema that would be 

718 finalized eventually, and is finalized during every release. 

719 """ 

720 name = CliFixtureDouble.SCHEMA_NAME 

721 directory = tmp_path / name 

722 directory.mkdir(parents=True) 

723 (directory / f"{name}-1.0.0.dev.json").write_text("{}\n") 

724 runner = click.testing.CliRunner() 

725 result = runner.invoke( 

726 main, ["fixtures", "refresh", "--dir", str(tmp_path), "--package", CliFixtureDouble.__module__] 

727 ) 

728 assert result.exit_code != 0 

729 assert "Traceback" not in result.output 

730 # A raw pydantic.ValidationError escaping uncaught propagates out of 

731 # CliRunner.invoke() as result.exception with no "Error: ..." line ever 

732 # written to result.output; only a caught-and-reraised 

733 # click.ClickException produces that line and a clean SystemExit. 

734 assert "Error:" in result.output 

735 assert isinstance(result.exception, SystemExit)