Coverage for tests/test_fuzz.py: 100%

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 

14from pathlib import Path 

15 

16import astropy.units as u 

17import numpy as np 

18from click.testing import CliRunner 

19 

20from lsst.images import Image, MaskedImage, MaskPlane, MaskSchema 

21from lsst.images.cli import main 

22from lsst.images.cli._fuzz import shuffle_blocks 

23from lsst.images.serialization import read_archive 

24 

25 

26def test_consistency_even_blocks() -> None: 

27 """Verify a shared per-block permutation keeps image/mask/variance planes 

28 aligned. 

29 """ 

30 ny, nx = 4, 4 

31 codes = np.arange(ny * nx, dtype=np.float64).reshape(ny, nx) 

32 image = codes.copy() 

33 variance = codes.copy() 

34 mask = codes.astype(np.uint8).reshape(ny, nx, 1).copy() 

35 

36 shuffle_blocks(image, mask, variance, (2, 2), np.random.default_rng(7)) 

37 

38 np.testing.assert_array_equal(image, variance) 

39 np.testing.assert_array_equal(image.astype(np.uint8), mask[..., 0]) 

40 # Every 2x2 block keeps its original multiset of values. 

41 for y0 in (0, 2): 

42 for x0 in (0, 2): 

43 np.testing.assert_array_equal( 

44 np.sort(image[y0 : y0 + 2, x0 : x0 + 2], axis=None), 

45 np.sort(codes[y0 : y0 + 2, x0 : x0 + 2], axis=None), 

46 ) 

47 

48 

49def test_partial_edge_blocks() -> None: 

50 """Verify edge blocks smaller than block_size shuffle without error or 

51 value loss. 

52 """ 

53 ny, nx = 5, 5 

54 codes = np.arange(ny * nx, dtype=np.float64).reshape(ny, nx) 

55 image = codes.copy() 

56 variance = codes.copy() 

57 mask = codes.astype(np.uint8).reshape(ny, nx, 1).copy() 

58 

59 shuffle_blocks(image, mask, variance, (2, 2), np.random.default_rng(3)) 

60 

61 np.testing.assert_array_equal(image, variance) 

62 np.testing.assert_array_equal(image.astype(np.uint8), mask[..., 0]) 

63 np.testing.assert_array_equal(np.sort(image, axis=None), np.sort(codes, axis=None)) 

64 

65 

66def _make_masked_image() -> MaskedImage: 

67 """Return a small MaskedImage with noisy pixels and some mask bits set.""" 

68 rng = np.random.default_rng(500) 

69 mi = MaskedImage( 

70 Image(rng.normal(100.0, 8.0, size=(64, 64)), dtype=np.float32, unit=u.nJy), 

71 mask_schema=MaskSchema([MaskPlane("BAD", "Bad pixel.")]), 

72 metadata={"hello": "world"}, 

73 ) 

74 mi.mask.array |= np.multiply.outer(mi.image.array < 100.0, mi.mask.schema.bitmask("BAD")) 

75 mi.variance.array = rng.normal(64.0, 0.5, size=mi.bbox.shape).astype(np.float32) 

76 return mi 

77 

78 

79def test_fuzz_help() -> None: 

80 """Verify the fuzz-masked-image CLI command exposes help text.""" 

81 result = CliRunner().invoke(main, ["fuzz-masked-image", "--help"]) 

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

83 

84 

85def test_fuzz_no_files_errors() -> None: 

86 """Verify fuzz-masked-image fails when no files are given.""" 

87 result = CliRunner().invoke(main, ["fuzz-masked-image"]) 

88 assert result.exit_code != 0 

89 

90 

91def test_fuzz_round_trip_fits(tmp_path: Path) -> None: 

92 """Verify fuzz-masked-image produces a modified but structurally valid 

93 FITS file. 

94 """ 

95 src = tmp_path / "mi.fits" 

96 mi = _make_masked_image() 

97 mi.write(src) 

98 original_image = mi.image.array.copy() 

99 original_mask = mi.mask.array.copy() 

100 

101 result = CliRunner().invoke(main, ["fuzz-masked-image", str(src)]) 

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

103 

104 out = tmp_path / "mi.fuzzed.fits" 

105 assert os.path.exists(out) 

106 check = read_archive(out) 

107 finite = np.isfinite(original_image) 

108 changed = float(np.mean(check.image.array[finite] != original_image[finite])) 

109 assert changed >= 0.5 

110 assert not np.array_equal(check.mask.array, original_mask) 

111 # An untouched part of the object survives the round trip. 

112 assert check.metadata.get("hello") == "world" 

113 

114 

115def test_fuzz_skips_existing_without_overwrite(tmp_path: Path) -> None: 

116 """Verify fuzz-masked-image leaves an existing output file untouched when 

117 not overwriting. 

118 """ 

119 src = tmp_path / "mi.fits" 

120 _make_masked_image().write(src) 

121 out = tmp_path / "mi.fuzzed.fits" 

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

123 stream.write("EXISTING") 

124 result = CliRunner().invoke(main, ["fuzz-masked-image", str(src)]) 

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

126 with open(out) as stream: 

127 assert stream.read() == "EXISTING"