Coverage for tests/test_cli.py: 82%

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 os 

14import subprocess 

15import sys 

16import tempfile 

17import unittest 

18from unittest import mock 

19 

20import astropy.io.fits 

21import click 

22import numpy as np 

23from click.testing import CliRunner 

24 

25from lsst.images import Box, Image 

26from lsst.images import fits as images_fits 

27from lsst.images import json as images_json 

28from lsst.images.cli import main 

29from lsst.images.serialization import backend_for_path, read 

30 

31 

32class CliSkeletonTestCase(unittest.TestCase): 

33 """The root group loads and shows help with core deps only.""" 

34 

35 def test_group_help(self) -> None: 

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

37 self.assertEqual(result.exit_code, 0, result.output) 

38 self.assertIn("convert", result.output) 

39 self.assertIn("inspect", result.output) 

40 

41 def test_python_m_entry_point(self) -> None: 

42 # ``python -m lsst.images`` must run the same CLI group. 

43 result = subprocess.run( 

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

45 capture_output=True, 

46 text=True, 

47 ) 

48 self.assertEqual(result.returncode, 0, result.stderr) 

49 self.assertIn("convert", result.stdout) 

50 self.assertIn("inspect", result.stdout) 

51 

52 

53class InspectTestCase(unittest.TestCase): 

54 """inspect prints schema URL and format version.""" 

55 

56 def setUp(self) -> None: 

57 tmp = tempfile.TemporaryDirectory() 

58 self.addCleanup(tmp.cleanup) 

59 self.tmp = tmp.name 

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

61 

62 def test_inspect_fits(self) -> None: 

63 path = os.path.join(self.tmp, "x.fits") 

64 images_fits.write(self.image, path) 

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

66 self.assertEqual(result.exit_code, 0, result.output) 

67 self.assertIn("https://images.lsst.io/schemas/image-1.0.0", result.output) 

68 self.assertIn("format version: 1", result.output) 

69 

70 def test_inspect_json(self) -> None: 

71 path = os.path.join(self.tmp, "x.json") 

72 images_json.write(self.image, path) 

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

74 self.assertEqual(result.exit_code, 0, result.output) 

75 self.assertIn("image-1.0.0", result.output) 

76 self.assertIn("n/a", result.output) # no container format version for JSON 

77 

78 def test_inspect_fits_python_class(self) -> None: 

79 path = os.path.join(self.tmp, "y.fits") 

80 images_fits.write(self.image, path) 

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

82 self.assertEqual(result.exit_code, 0, result.output) 

83 self.assertIn("python class:", result.output) 

84 self.assertIn("lsst.images.Image", result.output) 

85 

86 def test_inspect_json_python_class(self) -> None: 

87 path = os.path.join(self.tmp, "y.json") 

88 images_json.write(self.image, path) 

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

90 self.assertEqual(result.exit_code, 0, result.output) 

91 self.assertIn("python class:", result.output) 

92 self.assertIn("lsst.images.Image", result.output) 

93 

94 def test_inspect_unregistered_schema(self) -> None: 

95 path = os.path.join(self.tmp, "fake.json") 

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

97 f.write( 

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

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

100 ) 

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

102 # inspect should still succeed: it reports basic info regardless 

103 # of registry membership. 

104 self.assertEqual(result.exit_code, 0, result.output) 

105 self.assertIn("python class:", result.output) 

106 self.assertIn("<unregistered: no-such-schema>", result.output) 

107 

108 def test_inspect_unknown_extension(self) -> None: 

109 path = os.path.join(self.tmp, "x.txt") 

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

111 stream.write("nope") 

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

113 self.assertNotEqual(result.exit_code, 0) 

114 self.assertIn(".fits", result.output) 

115 

116 

117class ReformatTestCase(unittest.TestCase): 

118 """reformat reads a file and writes it back in another container format.""" 

119 

120 def setUp(self) -> None: 

121 tmp = tempfile.TemporaryDirectory() 

122 self.addCleanup(tmp.cleanup) 

123 self.tmp = tmp.name 

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

125 

126 def test_reformat_round_trip_json_fits_json(self) -> None: 

127 # JSON -> FITS -> JSON must preserve the object (NDF is optional, so 

128 # the round trip stays within the always-available backends). 

129 src = os.path.join(self.tmp, "in.json") 

130 mid = os.path.join(self.tmp, "mid.fits") 

131 out = os.path.join(self.tmp, "out.json") 

132 images_json.write(self.image, src) 

133 

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

135 self.assertEqual(result.exit_code, 0, result.output) 

136 self.assertEqual(backend_for_path(mid).input_archive.get_basic_info(mid).schema_name, "image") 

137 

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

139 self.assertEqual(result.exit_code, 0, result.output) 

140 

141 np.testing.assert_array_equal(read(out, Image).array, self.image.array) 

142 

143 def test_reformat_refuses_existing_output(self) -> None: 

144 src = os.path.join(self.tmp, "in.json") 

145 out = os.path.join(self.tmp, "out.fits") 

146 images_json.write(self.image, src) 

147 images_fits.write(self.image, out) 

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

149 self.assertNotEqual(result.exit_code, 0) 

150 self.assertIn("--overwrite", result.output) 

151 

152 def test_reformat_unknown_output_extension(self) -> None: 

153 src = os.path.join(self.tmp, "in.json") 

154 images_json.write(self.image, src) 

155 result = CliRunner().invoke(main, ["reformat", src, os.path.join(self.tmp, "out.txt")]) 

156 self.assertNotEqual(result.exit_code, 0) 

157 self.assertIn(".fits", result.output) 

158 

159 

160EXTERNAL_DATA_DIR = os.environ.get("TESTDATA_IMAGES_DIR", None) 

161 

162 

163class ConvertDetectTestCase(unittest.TestCase): 

164 """Legacy type detection from the LSST BUTLER DATASETTYPE header. 

165 

166 These build small FITS files carrying only the discriminating header so 

167 the test does not depend on a fixture happening to include it. 

168 """ 

169 

170 def _make(self, dataset_type: str | None) -> str: 

171 tmp = tempfile.mkdtemp() 

172 path = os.path.join(tmp, "x.fits") 

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

174 with images_fits.suppress_fits_card_warnings(): 

175 if dataset_type is not None: 

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

177 hdu.writeto(path) 

178 return path 

179 

180 def test_detect_visit_image(self) -> None: 

181 from lsst.images.cli._convert import detect_legacy_type 

182 

183 self.assertEqual(detect_legacy_type(self._make("visit_image")), "visit_image") 

184 self.assertEqual(detect_legacy_type(self._make("preliminary_visit_image")), "visit_image") 

185 

186 def test_detect_cell_coadd(self) -> None: 

187 from lsst.images.cli._convert import detect_legacy_type 

188 

189 self.assertEqual(detect_legacy_type(self._make("deep_coadd_cell_predetection")), "cell_coadd") 

190 

191 def test_detect_indeterminate(self) -> None: 

192 from lsst.images.cli._convert import detect_legacy_type 

193 

194 self.assertIsNone(detect_legacy_type(self._make(None))) 

195 self.assertIsNone(detect_legacy_type(self._make("camera"))) 

196 

197 @unittest.skipUnless(EXTERNAL_DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not set.") 

198 def test_detect_visit_image_fixture(self) -> None: 

199 from lsst.images.cli._convert import detect_legacy_type 

200 

201 path = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_image.fits") 

202 self.assertEqual(detect_legacy_type(path), "visit_image") 

203 

204 

205class ConvertVisitImageTestCase(unittest.TestCase): 

206 """convert of a legacy visit image (needs afw + testdata).""" 

207 

208 @unittest.skipUnless(EXTERNAL_DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not set.") 

209 def test_convert_visit_image_to_json(self) -> None: 

210 try: 

211 import lsst.afw.image # noqa: F401 

212 except ImportError: 

213 self.skipTest("afw not available.") 

214 tmp = tempfile.mkdtemp() 

215 src = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_image.fits") 

216 out = os.path.join(tmp, "converted.json") 

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

218 self.assertEqual(result.exit_code, 0, result.output) 

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

220 self.assertEqual(info.schema_name, "visit_image") 

221 

222 @unittest.skipUnless(EXTERNAL_DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not set.") 

223 def test_convert_refuses_existing_output(self) -> None: 

224 try: 

225 import lsst.afw.image # noqa: F401 

226 except ImportError: 

227 self.skipTest("afw not available.") 

228 tmp = tempfile.mkdtemp() 

229 src = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "visit_image.fits") 

230 out = os.path.join(tmp, "exists.json") 

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

232 stream.write("{}") 

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

234 self.assertNotEqual(result.exit_code, 0) 

235 self.assertIn("--overwrite", result.output) 

236 

237 

238class ConvertCellCoaddTestCase(unittest.TestCase): 

239 """convert of a legacy cell coadd (needs cell_coadds + testdata).""" 

240 

241 @unittest.skipUnless(EXTERNAL_DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not set.") 

242 def test_convert_cell_coadd_to_json(self) -> None: 

243 try: 

244 import lsst.cell_coadds # noqa: F401 

245 except ImportError: 

246 self.skipTest("cell_coadds not available.") 

247 tmp = tempfile.mkdtemp() 

248 legacy_dir = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy") 

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

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

251 out = os.path.join(tmp, "coadd.json") 

252 # This fixture has no LSST BUTLER DATASETTYPE header, so pass --type. 

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

254 self.assertEqual(result.exit_code, 0, result.output) 

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

256 self.assertEqual(info.schema_name, "cell_coadd") 

257 

258 @unittest.skipUnless(EXTERNAL_DATA_DIR is not None, "TESTDATA_IMAGES_DIR is not set.") 

259 def test_convert_cell_coadd_requires_skymap(self) -> None: 

260 try: 

261 import lsst.cell_coadds # noqa: F401 

262 except ImportError: 

263 self.skipTest("cell_coadds not available.") 

264 tmp = tempfile.mkdtemp() 

265 src = os.path.join(EXTERNAL_DATA_DIR, "dp2", "legacy", "deep_coadd_cell_predetection.fits") 

266 out = os.path.join(tmp, "coadd.json") 

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

268 self.assertNotEqual(result.exit_code, 0) 

269 self.assertIn("--skymap", result.output) 

270 

271 

272class ConvertPreserveQuantizationTestCase(unittest.TestCase): 

273 """--preserve-quantization defaults on, forwards for visit images, and 

274 is rejected only when set explicitly for other types. 

275 """ 

276 

277 def setUp(self) -> None: 

278 tmp = tempfile.TemporaryDirectory() 

279 self.addCleanup(tmp.cleanup) 

280 self.tmp = tmp.name 

281 

282 def _make_input(self) -> str: 

283 path = os.path.join(self.tmp, "in.fits") 

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

285 return path 

286 

287 def test_default_is_true(self) -> None: 

288 from lsst.images.cli._convert import convert 

289 

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

291 self.assertIs(option.default, True) 

292 

293 def test_explicit_flag_rejected_for_cell_coadd(self) -> None: 

294 src = self._make_input() 

295 out = os.path.join(self.tmp, "out.json") 

296 result = CliRunner().invoke( 

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

298 ) 

299 self.assertNotEqual(result.exit_code, 0) 

300 self.assertIn("preserve-quantization", result.output) 

301 

302 def test_default_does_not_reject_cell_coadd(self) -> None: 

303 # The on-by-default flag must not block cell-coadd conversion; any 

304 # failure here is about later steps (skymap / missing deps), never the 

305 # preserve-quantization gate. 

306 src = self._make_input() 

307 out = os.path.join(self.tmp, "out.json") 

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

309 self.assertNotIn("preserve-quantization", result.output) 

310 

311 def test_forwarded_to_read_legacy(self) -> None: 

312 from lsst.images.cli._convert import _read_legacy 

313 

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

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

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

317 

318 

319class CliRegistrationTestCase(unittest.TestCase): 

320 """minify, extract-test-data, and verify-rewrite register and load.""" 

321 

322 def test_subcommands_present(self) -> None: 

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

324 self.assertEqual(result.exit_code, 0, result.output) 

325 self.assertIn("minify", result.output) 

326 self.assertIn("reformat", result.output) 

327 self.assertIn("extract-test-data", result.output) 

328 self.assertIn("verify-rewrite", result.output) 

329 self.assertIn("fuzz-masked-image", result.output) 

330 

331 def test_minify_help(self) -> None: 

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

333 self.assertEqual(result.exit_code, 0, result.output) 

334 

335 def test_extract_test_data_help(self) -> None: 

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

337 self.assertEqual(result.exit_code, 0, result.output) 

338 

339 def test_verify_rewrite_help(self) -> None: 

340 # The group and its stage4 command load with core deps only; the 

341 # full Rubin environment is only required when stage4 actually runs. 

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

343 self.assertEqual(result.exit_code, 0, result.output) 

344 self.assertIn("stage4", result.output) 

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

346 self.assertEqual(result.exit_code, 0, result.output) 

347 

348 def test_short_help_alias(self) -> None: 

349 # -h is an alias for --help on the group and every subcommand, 

350 # including the nested extract-test-data sub-group. 

351 for args in ( 

352 ["-h"], 

353 ["convert", "-h"], 

354 ["inspect", "-h"], 

355 ["minify", "-h"], 

356 ["reformat", "-h"], 

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

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

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

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

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

362 ): 

363 with self.subTest(args=args): 

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

365 self.assertEqual(result.exit_code, 0, result.output) 

366 self.assertIn("Usage:", result.output) 

367 

368 

369class ConvertSafetyTestCase(unittest.TestCase): 

370 """convert must not destroy data on identical paths or on failure.""" 

371 

372 def setUp(self) -> None: 

373 tmp = tempfile.TemporaryDirectory() 

374 self.addCleanup(tmp.cleanup) 

375 self.tmp = tmp.name 

376 

377 def _make_input(self) -> str: 

378 path = os.path.join(self.tmp, "in.fits") 

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

380 return path 

381 

382 def test_rejects_identical_paths(self) -> None: 

383 path = self._make_input() 

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

385 self.assertNotEqual(result.exit_code, 0) 

386 self.assertIn("different", result.output) 

387 # The file is left untouched. 

388 self.assertTrue(os.path.exists(path)) 

389 

390 def test_preserves_existing_output_on_read_failure(self) -> None: 

391 src = self._make_input() 

392 out = os.path.join(self.tmp, "out.json") 

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

394 stream.write("ORIGINAL") 

395 # A read failure after the overwrite gate must leave OUTPUT intact. 

396 with mock.patch( 

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

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

399 ): 

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

401 self.assertNotEqual(result.exit_code, 0) 

402 with open(out) as stream: 

403 self.assertEqual(stream.read(), "ORIGINAL")