Coverage for tests/test_cli.py: 25%

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 if dataset_type is not None: 

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

176 hdu.writeto(path) 

177 return path 

178 

179 def test_detect_visit_image(self) -> None: 

180 from lsst.images.cli._convert import detect_legacy_type 

181 

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

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

184 

185 def test_detect_cell_coadd(self) -> None: 

186 from lsst.images.cli._convert import detect_legacy_type 

187 

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

189 

190 def test_detect_indeterminate(self) -> None: 

191 from lsst.images.cli._convert import detect_legacy_type 

192 

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

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

195 

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

197 def test_detect_visit_image_fixture(self) -> None: 

198 from lsst.images.cli._convert import detect_legacy_type 

199 

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

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

202 

203 

204class ConvertVisitImageTestCase(unittest.TestCase): 

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

206 

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

208 def test_convert_visit_image_to_json(self) -> None: 

209 try: 

210 import lsst.afw.image # noqa: F401 

211 except ImportError: 

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

213 tmp = tempfile.mkdtemp() 

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

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

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

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

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

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

220 

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

222 def test_convert_refuses_existing_output(self) -> None: 

223 try: 

224 import lsst.afw.image # noqa: F401 

225 except ImportError: 

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

227 tmp = tempfile.mkdtemp() 

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

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

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

231 stream.write("{}") 

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

233 self.assertNotEqual(result.exit_code, 0) 

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

235 

236 

237class ConvertCellCoaddTestCase(unittest.TestCase): 

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

239 

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

241 def test_convert_cell_coadd_to_json(self) -> None: 

242 try: 

243 import lsst.cell_coadds # noqa: F401 

244 except ImportError: 

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

246 tmp = tempfile.mkdtemp() 

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

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

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

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

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

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

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

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

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

256 

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

258 def test_convert_cell_coadd_requires_skymap(self) -> None: 

259 try: 

260 import lsst.cell_coadds # noqa: F401 

261 except ImportError: 

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

263 tmp = tempfile.mkdtemp() 

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

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

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

267 self.assertNotEqual(result.exit_code, 0) 

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

269 

270 

271class ConvertPreserveQuantizationTestCase(unittest.TestCase): 

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

273 is rejected only when set explicitly for other types. 

274 """ 

275 

276 def setUp(self) -> None: 

277 tmp = tempfile.TemporaryDirectory() 

278 self.addCleanup(tmp.cleanup) 

279 self.tmp = tmp.name 

280 

281 def _make_input(self) -> str: 

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

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

284 return path 

285 

286 def test_default_is_true(self) -> None: 

287 from lsst.images.cli._convert import convert 

288 

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

290 self.assertIs(option.default, True) 

291 

292 def test_explicit_flag_rejected_for_cell_coadd(self) -> None: 

293 src = self._make_input() 

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

295 result = CliRunner().invoke( 

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

297 ) 

298 self.assertNotEqual(result.exit_code, 0) 

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

300 

301 def test_default_does_not_reject_cell_coadd(self) -> None: 

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

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

304 # preserve-quantization gate. 

305 src = self._make_input() 

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

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

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

309 

310 def test_forwarded_to_read_legacy(self) -> None: 

311 from lsst.images.cli._convert import _read_legacy 

312 

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

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

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

316 

317 

318class CliRegistrationTestCase(unittest.TestCase): 

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

320 

321 def test_subcommands_present(self) -> None: 

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

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

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

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

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

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

328 

329 def test_minify_help(self) -> None: 

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

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

332 

333 def test_extract_test_data_help(self) -> None: 

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

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

336 

337 def test_verify_rewrite_help(self) -> None: 

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

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

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

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

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

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

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

345 

346 def test_short_help_alias(self) -> None: 

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

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

349 for args in ( 

350 ["-h"], 

351 ["convert", "-h"], 

352 ["inspect", "-h"], 

353 ["minify", "-h"], 

354 ["reformat", "-h"], 

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

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

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

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

359 ): 

360 with self.subTest(args=args): 

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

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

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

364 

365 

366class ConvertSafetyTestCase(unittest.TestCase): 

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

368 

369 def setUp(self) -> None: 

370 tmp = tempfile.TemporaryDirectory() 

371 self.addCleanup(tmp.cleanup) 

372 self.tmp = tmp.name 

373 

374 def _make_input(self) -> str: 

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

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

377 return path 

378 

379 def test_rejects_identical_paths(self) -> None: 

380 path = self._make_input() 

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

382 self.assertNotEqual(result.exit_code, 0) 

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

384 # The file is left untouched. 

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

386 

387 def test_preserves_existing_output_on_read_failure(self) -> None: 

388 src = self._make_input() 

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

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

391 stream.write("ORIGINAL") 

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

393 with mock.patch( 

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

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

396 ): 

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

398 self.assertNotEqual(result.exit_code, 0) 

399 with open(out) as stream: 

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