Coverage for tests/test_ndf_common.py: 97%

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. 

11 

12from __future__ import annotations 

13 

14import unittest 

15 

16try: 

17 from lsst.images.ndf._common import ( 

18 HdsNameShrinker, 

19 NdfPointerModel, 

20 archive_path_to_hdf5_path, 

21 ) 

22 from lsst.images.ndf._hds import DAT__SZNAM 

23 

24 HAVE_H5PY = True 

25except ImportError: 

26 HAVE_H5PY = False 

27 

28 

29@unittest.skipUnless(HAVE_H5PY, "h5py is not installed") 

30class NdfPointerModelTestCase(unittest.TestCase): 

31 """Tests for `NdfPointerModel` and `archive_path_to_hdf5_path`.""" 

32 

33 def setUp(self): 

34 self.shrinker = HdsNameShrinker() 

35 

36 def test_round_trips_through_json(self): 

37 original = NdfPointerModel(path="/MORE/LSST/PSF") 

38 json_bytes = original.model_dump_json().encode() 

39 recovered = NdfPointerModel.model_validate_json(json_bytes) 

40 self.assertEqual(recovered, original) 

41 

42 def test_archive_path_to_hdf5_path(self): 

43 self.assertEqual(archive_path_to_hdf5_path("", self.shrinker), "/MORE/LSST/JSON") 

44 self.assertEqual(archive_path_to_hdf5_path("/psf", self.shrinker), "/MORE/LSST/PSF") 

45 self.assertEqual( 

46 archive_path_to_hdf5_path("/psf/coefficients", self.shrinker), "/MORE/LSST/PSF/COEFFICIENTS" 

47 ) 

48 

49 def test_archive_path_shrinks_long_components(self): 

50 result = archive_path_to_hdf5_path("/psf/this_component_is_too_long", self.shrinker) 

51 self.assertTrue(result.startswith("/MORE/LSST/PSF/")) 

52 leaf = result.rsplit("/", 1)[-1] 

53 self.assertLessEqual(len(leaf), DAT__SZNAM) 

54 # The short parent component is untouched; only the long leaf shrinks. 

55 self.assertEqual(result.split("/")[3], "PSF") 

56 

57 def test_archive_path_shrink_round_trips_to_same_value(self): 

58 self.assertEqual( 

59 archive_path_to_hdf5_path("/noise_realizations/0", self.shrinker), 

60 archive_path_to_hdf5_path("/noise_realizations/0", self.shrinker), 

61 ) 

62 

63 

64@unittest.skipUnless(HAVE_H5PY, "h5py is not installed") 

65class HdsNameShrinkerTestCase(unittest.TestCase): 

66 """Tests for the stateful HDS component shrinker.""" 

67 

68 def setUp(self): 

69 self.shrinker = HdsNameShrinker() 

70 

71 def test_short_names_pass_through_uppercased(self): 

72 self.assertEqual(self.shrinker.shrink("psf"), "PSF") 

73 # A name exactly at the limit passes through unchanged (uppercased). 

74 self.assertEqual(self.shrinker.shrink("a" * DAT__SZNAM), "A" * DAT__SZNAM) 

75 # One character over the limit is shrunk to the limit. 

76 self.assertEqual(len(self.shrinker.shrink("a" * (DAT__SZNAM + 1))), DAT__SZNAM) 

77 

78 def test_long_names_keep_prefix_and_get_counter_token(self): 

79 shrunk = self.shrinker.shrink("noise_realizations") 

80 self.assertEqual(len(shrunk), DAT__SZNAM) 

81 self.assertEqual(shrunk, "NOISE_REALI_001") 

82 

83 def test_shrink_is_deterministic_per_instance(self): 

84 self.assertEqual( 

85 self.shrinker.shrink("noise_realizations"), 

86 self.shrinker.shrink("noise_realizations"), 

87 ) 

88 

89 def test_distinct_long_names_get_distinct_tokens(self): 

90 # Identical truncated prefixes cannot collide because the counter 

91 # increments for each newly assigned name. 

92 self.assertEqual(self.shrinker.shrink("noise_realization_field"), "NOISE_REALI_001") 

93 self.assertEqual(self.shrinker.shrink("noise_realization_other"), "NOISE_REALI_002") 

94 

95 def test_reserve_shortens_the_budget(self): 

96 shrunk = self.shrinker.shrink("noise_realizations", reserve=2) 

97 self.assertEqual(len(shrunk), DAT__SZNAM - 2) 

98 

99 def test_version_one_matches_plain_shrink(self): 

100 self.assertEqual( 

101 self.shrinker.shrink_versioned("noise_realizations", 1), 

102 self.shrinker.shrink("noise_realizations"), 

103 ) 

104 

105 def test_short_versioned_name_keeps_visible_suffix(self): 

106 self.assertEqual(self.shrinker.shrink_versioned("data", 2), "DATA_2") 

107 

108 def test_long_versioned_name_preserves_suffix_within_limit(self): 

109 shrunk = self.shrinker.shrink_versioned("noise_realizations", 99) 

110 self.assertEqual(len(shrunk), DAT__SZNAM) 

111 self.assertTrue(shrunk.endswith("_99")) 

112 

113 def test_same_base_different_versions_are_distinct(self): 

114 self.assertNotEqual( 

115 self.shrinker.shrink_versioned("noise_realizations", 2), 

116 self.shrinker.shrink_versioned("noise_realizations", 3), 

117 )