Coverage for tests/test_cli.py: 85%

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, get_dp2_exposure_record 

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 from lsst.daf.butler.tests import makeTestRepo 

234 

235 repo_dir = tmp_path / "repo" 

236 with makeTestRepo(repo_dir) as butler: 

237 universe = butler.dimensions 

238 exposure_record = get_dp2_exposure_record(universe) 

239 instrument = exposure_record.instrument 

240 # Records for dimensions that exposure requires or implies must exist 

241 # first. 

242 butler.registry.insertDimensionData(universe["instrument"], {"name": instrument}) 

243 butler.registry.insertDimensionData( 

244 universe["day_obs"], {"instrument": instrument, "day_obs": exposure_record.day_obs} 

245 ) 

246 butler.registry.insertDimensionData( 

247 universe["group"], {"instrument": instrument, "name": exposure_record.group} 

248 ) 

249 butler.registry.insertDimensionData( 

250 universe["physical_filter"], 

251 {"instrument": instrument, "name": exposure_record.physical_filter, "band": "r"}, 

252 ) 

253 butler.registry.insertDimensionData(universe["exposure"], exposure_record.toDict()) 

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

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

256 result = CliRunner().invoke( 

257 main, 

258 [ 

259 "convert", 

260 src, 

261 out, 

262 "--butler", 

263 str(repo_dir), 

264 "--instrument", 

265 instrument, 

266 "--exposure", 

267 str(exposure_record.id), 

268 ], 

269 ) 

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

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

272 assert info.schema_name == "visit_image" 

273 

274 

275def test_convert_visit_image_without_record(tmp_path: Path, external_data_dir: str) -> None: 

276 """Test that convert accepts a visit image without --butler/--exposure, 

277 producing a valid file with a degraded (record-less) ObservationInfo. 

278 """ 

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

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

281 out = str(tmp_path / "norecord.json") 

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

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

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

285 assert info.schema_name == "visit_image" 

286 

287 

288@pytest.mark.parametrize( 

289 ("extra_args", "expected_error"), 

290 [ 

291 (["--butler", "/nonexistent/repo"], "must be given together"), 

292 (["--butler", "/nonexistent/repo", "--exposure", "123"], "require --instrument"), 

293 (["--instrument", "LSSTCam"], "no effect"), 

294 ], 

295) 

296def test_convert_record_option_combinations( 

297 tmp_path: Path, external_data_dir: str, extra_args: list[str], expected_error: str 

298) -> None: 

299 """Test that convert rejects incoherent exposure-record option 

300 combinations before touching any repository. 

301 """ 

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

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

304 out = str(tmp_path / "combos.json") 

305 result = CliRunner().invoke(main, ["convert", src, out, *extra_args]) 

306 assert result.exit_code != 0 

307 assert expected_error in result.output 

308 

309 

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

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

312 --overwrite. 

313 """ 

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

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

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

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

318 stream.write("{}") 

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

320 assert result.exit_code != 0 

321 assert "--overwrite" in result.output 

322 

323 

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

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

326 FITS MultipleCellCoadd. 

327 """ 

328 pytest.importorskip("lsst.cell_coadds") 

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

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

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

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

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

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

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

336 assert info.schema_name == "cell_coadd" 

337 

338 

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

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

341 for cell_coadd. 

342 """ 

343 pytest.importorskip("lsst.cell_coadds") 

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

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

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

347 assert result.exit_code != 0 

348 assert "--skymap" in result.output 

349 

350 

351def test_preserve_quantization_default_is_true() -> None: 

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

353 from lsst.images.cli._convert import convert 

354 

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

356 assert option.default is True 

357 

358 

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

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

361 cell_coadd conversions. 

362 """ 

363 src = _make_cli_input(tmp_path) 

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

365 result = CliRunner().invoke( 

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

367 ) 

368 assert result.exit_code != 0 

369 assert "preserve-quantization" in result.output 

370 

371 

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

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

374 way of cell-coadd conversion. 

375 """ 

376 src = _make_cli_input(tmp_path) 

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

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

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

380 

381 

382def test_preserve_quantization_forwarded_to_read_legacy() -> None: 

383 """Test that _read_legacy forwards preserve_quantization=True and the 

384 resolved exposure record to VisitImage.read_legacy. 

385 """ 

386 from lsst.images.cli._convert import _read_legacy 

387 

388 with ( 

389 mock.patch("lsst.images.cli._convert._fetch_exposure_record") as resolve, 

390 mock.patch("lsst.images.VisitImage.read_legacy") as read_legacy, 

391 ): 

392 resolve.return_value = mock.sentinel.record 

393 _read_legacy("in.fits", "visit_image", None, "repo", None, True, instrument="LSSTCam", exposure=42) 

394 read_legacy.assert_called_once_with( 

395 "in.fits", preserve_quantization=True, exposure_record=mock.sentinel.record 

396 ) 

397 

398 

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

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

401 --overwrite. 

402 """ 

403 path = _make_cli_input(tmp_path) 

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

405 assert result.exit_code != 0 

406 assert "different" in result.output 

407 assert os.path.exists(path) 

408 

409 

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

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

412 read_legacy raises. 

413 """ 

414 src = _make_cli_input(tmp_path) 

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

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

417 stream.write("ORIGINAL") 

418 with mock.patch( 

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

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

421 ): 

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

423 assert result.exit_code != 0 

424 with open(out) as stream: 

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

426 

427 

428def test_subcommands_present() -> None: 

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

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

431 """ 

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

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

434 assert "minify" in result.output 

435 assert "reformat" in result.output 

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

437 assert "verify-rewrite" in result.output 

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

439 

440 

441def test_minify_help() -> None: 

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

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

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

445 

446 

447def test_extract_test_data_help() -> None: 

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

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

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

451 

452 

453def test_verify_rewrite_help() -> None: 

454 """Verify verify-rewrite loads with core deps only.""" 

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

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

457 

458 

459@pytest.mark.parametrize( 

460 "args", 

461 [ 

462 ["-h"], 

463 ["convert", "-h"], 

464 ["inspect", "-h"], 

465 ["minify", "-h"], 

466 ["reformat", "-h"], 

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

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

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

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

471 ["schemas", "-h"], 

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

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

474 ["fixtures", "-h"], 

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

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

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

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

479 ], 

480 ids=[ 

481 "root", 

482 "convert", 

483 "inspect", 

484 "minify", 

485 "reformat", 

486 "extract-test-data", 

487 "extract-test-data-dp2", 

488 "verify-rewrite", 

489 "fuzz-masked-image", 

490 "schemas", 

491 "schemas-write", 

492 "schemas-check", 

493 "fixtures", 

494 "fixtures-check", 

495 "fixtures-refresh", 

496 "fixtures-freeze", 

497 "fixtures-coverage", 

498 ], 

499) 

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

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

502 subcommand. 

503 """ 

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

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

506 assert "Usage:" in result.output 

507 

508 

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

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

511 accepts. 

512 """ 

513 runner = CliRunner() 

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

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

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

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

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

519 

520 

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

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

523 runner = CliRunner() 

524 result = runner.invoke( 

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

526 ) 

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

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

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

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

531 

532 

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

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

535 runner = CliRunner() 

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

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

538 path.unlink() 

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

540 assert result.exit_code != 0 

541 assert "schemas write" in result.output 

542 

543 

544def test_cli_describe_visit_image() -> None: 

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

546 path = current_fixture_path(FIXTURE_DIR, "visit_image") 

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

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

549 assert "VisitImage" in result.output 

550 

551 

552def test_cli_describe_coadd_provenance() -> None: 

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

554 

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

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

557 """ 

558 path = current_fixture_path(FIXTURE_DIR, "coadd_provenance") 

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

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

561 assert "CoaddProvenance" in result.output 

562 assert "input images" in result.output 

563 

564 

565def test_fixtures_check_reports_a_clean_tree() -> None: 

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

567 runner = click.testing.CliRunner() 

568 result = runner.invoke( 

569 main, 

570 [ 

571 "fixtures", 

572 "check", 

573 "--dir", 

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

575 "--schema-dir", 

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

577 "--exempt", 

578 "psfex_psf", 

579 ], 

580 ) 

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

582 

583 

584def test_fixtures_coverage_reports_positions_for_a_composite() -> None: 

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

586 

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

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

589 """ 

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

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

592 ) 

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

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

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

596 

597 

598def test_fixtures_coverage_emits_parseable_json() -> None: 

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

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

601 main, 

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

603 ) 

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

605 parsed = json.loads(result.output) 

606 (entry,) = parsed.values() 

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

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

609 

610 

611def test_fixtures_coverage_reports_an_unknown_schema_cleanly() -> None: 

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

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

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

615 ) 

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

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

618 

619 

620def test_fixtures_check_requires_a_schema_directory() -> None: 

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

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

623 assert result.exit_code != 0 

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

625 

626 

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

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

629 

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

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

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

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

634 mode to tell the two apart. 

635 """ 

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

637 directory = tmp_path / "image" 

638 directory.mkdir(parents=True) 

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

640 runner = click.testing.CliRunner() 

641 result = runner.invoke( 

642 main, 

643 [ 

644 "fixtures", 

645 "check", 

646 "--dir", 

647 str(tmp_path), 

648 "--schema-dir", 

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

650 ], 

651 ) 

652 assert result.exit_code != 0 

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

654 

655 

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

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

658 

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

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

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

662 first run and passing on the second. 

663 """ 

664 directory = _copy_fixture_tree(tmp_path) 

665 runner = click.testing.CliRunner() 

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

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

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

669 

670 

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

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

673 

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

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

676 fixture. 

677 """ 

678 directory = _copy_fixture_tree(tmp_path) 

679 runner = click.testing.CliRunner() 

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

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

682 assert "nothing to freeze" in result.output 

683 

684 

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

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

687 

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

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

690 refresh acts only on a schema still in development. 

691 """ 

692 name = CliFixtureDouble.SCHEMA_NAME 

693 source = tmp_path / "dirty-source" 

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

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

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

697 fixture.write_text(dirty_text) 

698 

699 for run in range(2): 

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

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

702 main, 

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

704 ) 

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

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

707 assert "wrote" in result.output 

708 assert fixture.read_text() == dirty_text 

709 

710 

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

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

713 clean error instead of a raw traceback. 

714 """ 

715 source = current_fixture_path(FIXTURE_DIR, "image") 

716 name = source.name 

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

718 directory = tmp_path / "image" 

719 directory.mkdir(parents=True) 

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

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

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

723 runner = click.testing.CliRunner() 

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

725 assert result.exit_code != 0 

726 assert name in result.output 

727 assert "bump SCHEMA_VERSION" in result.output 

728 assert "Traceback" not in result.output 

729 

730 

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

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

733 clean error instead of a raw traceback. 

734 """ 

735 target = current_fixture_path(FIXTURE_DIR, "image") 

736 target_name = target.name 

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

738 text = target.read_text() 

739 directory = tmp_path / "image" 

740 directory.mkdir(parents=True) 

741 (directory / dev_name).write_text(text)