Coverage for tests/test_serialization.py: 100%

1# This file is part of astro_metadata_translator. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the LICENSE 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 

12"""Test that archival ObservationInfo JSON files can be read back. 

13 

14The JSON files in ``tests/data`` are stored long term, so we need to 

15guarantee that future versions of the code can still deserialize them. 

16""" 

17 

18import os 

19import unittest 

20from collections.abc import Mapping 

21from typing import Any 

22 

23import astropy.coordinates 

24import astropy.time 

25import astropy.units as u 

26import pydantic 

27 

28from astro_metadata_translator import ( 

29 ObservationInfo, 

30 PropertyDefinition, 

31 StubTranslator, 

32) 

33 

34TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

35DATADIR = os.path.join(TESTDIR, "data") 

36 

37 

38# Auto-registers via StubTranslator.__init_subclass__ when this module is 

39# imported, so the "_translator": "fixture" reference in the archival JSON 

40# resolves and its extension definitions are available. 

41class _FixtureTranslator(StubTranslator): 

42 name = "fixture" 

43 supported_instrument = "fixture" 

44 extensions = dict( 

45 number=PropertyDefinition("A number", int), 

46 foo=PropertyDefinition("A string", str), 

47 ) 

48 

49 @classmethod 

50 def can_translate(cls, header: Mapping[str, Any], filename: str | None = None) -> bool: 

51 # Never used for header-based auto-detection; this translator only 

52 # exists so the "_translator": "fixture" reference in the archival 

53 # JSON can be resolved. 

54 return False 

55 

56 

57class ObservationInfoSerializationTestCase(unittest.TestCase): 

58 """Read archival ObservationInfo JSON fixtures and check the values.""" 

59 

60 def _check_core_properties(self, obsinfo: ObservationInfo) -> None: 

61 self.assertEqual(obsinfo.telescope, "Test Telescope") 

62 self.assertEqual(obsinfo.instrument, "TestCam") 

63 self.assertEqual(obsinfo.exposure_id, 987654321) 

64 self.assertEqual(obsinfo.visit_id, 987654000) 

65 self.assertEqual(obsinfo.physical_filter, "r") 

66 self.assertEqual(obsinfo.detector_num, 42) 

67 self.assertEqual(obsinfo.detector_name, "S00") 

68 self.assertEqual(obsinfo.detector_unique_name, "R22_S00") 

69 self.assertEqual(obsinfo.detector_serial, "ITL-3800C-145") 

70 self.assertEqual(obsinfo.detector_group, "R22") 

71 self.assertEqual(obsinfo.detector_exposure_id, 98765432142) 

72 self.assertEqual(obsinfo.object, "HD12345") 

73 self.assertEqual(obsinfo.science_program, "2025A-001") 

74 self.assertEqual(obsinfo.observation_type, "science") 

75 self.assertEqual(obsinfo.observation_id, "TC_O_20250521_000123") 

76 self.assertEqual(obsinfo.observation_reason, "science") 

77 self.assertEqual(obsinfo.exposure_group, "20250521_000123") 

78 self.assertEqual(obsinfo.observing_day, 20250520) 

79 self.assertEqual(obsinfo.observation_counter, 123) 

80 self.assertEqual(obsinfo.group_counter_start, 123) 

81 self.assertEqual(obsinfo.group_counter_end, 123) 

82 self.assertEqual(obsinfo.boresight_rotation_coord, "sky") 

83 self.assertEqual(obsinfo.has_simulated_content, False) 

84 self.assertEqual(obsinfo.can_see_sky, True) 

85 

86 self.assertAlmostEqual(obsinfo.boresight_airmass, 1.1037) 

87 self.assertAlmostEqual(obsinfo.relative_humidity, 42.5) 

88 

89 self.assertIsInstance(obsinfo.location, astropy.coordinates.EarthLocation) 

90 self.assertAlmostEqual(obsinfo.location.height.to_value(u.m), 2663.0, places=3) 

91 

92 self.assertIsInstance(obsinfo.datetime_begin, astropy.time.Time) 

93 self.assertEqual(obsinfo.datetime_begin.scale, "tai") 

94 self.assertEqual( 

95 obsinfo.datetime_begin.tai.isot, 

96 "2025-05-21T01:23:45.000", 

97 ) 

98 self.assertEqual( 

99 obsinfo.datetime_end.tai.isot, 

100 "2025-05-21T01:24:15.000", 

101 ) 

102 

103 self.assertAlmostEqual(obsinfo.exposure_time.to_value(u.s), 30.0) 

104 self.assertAlmostEqual(obsinfo.exposure_time_requested.to_value(u.s), 30.0) 

105 self.assertAlmostEqual(obsinfo.dark_time.to_value(u.s), 31.0) 

106 self.assertAlmostEqual(obsinfo.focus_z.to_value(u.m), 0.001) 

107 self.assertAlmostEqual(obsinfo.temperature.to_value(u.K), 283.15) 

108 self.assertAlmostEqual(obsinfo.pressure.to_value(u.hPa), 750.0) 

109 self.assertAlmostEqual(obsinfo.boresight_rotation_angle.to_value(u.deg), 45.0) 

110 self.assertEqual(round(obsinfo.observing_day_offset.to_value(u.s)), 43200) 

111 

112 self.assertIsInstance(obsinfo.tracking_radec, astropy.coordinates.SkyCoord) 

113 self.assertAlmostEqual(obsinfo.tracking_radec.icrs.ra.to_value(u.deg), 123.456) 

114 self.assertAlmostEqual(obsinfo.tracking_radec.icrs.dec.to_value(u.deg), -45.678) 

115 

116 self.assertIsInstance(obsinfo.altaz_begin, astropy.coordinates.AltAz) 

117 self.assertAlmostEqual(obsinfo.altaz_begin.az.to_value(u.deg), 110.0) 

118 self.assertAlmostEqual(obsinfo.altaz_begin.alt.to_value(u.deg), 65.0) 

119 self.assertAlmostEqual(obsinfo.altaz_end.az.to_value(u.deg), 110.5) 

120 self.assertAlmostEqual(obsinfo.altaz_end.alt.to_value(u.deg), 64.5) 

121 

122 def test_read_full(self) -> None: 

123 """Read the fixture with every core property populated.""" 

124 path = os.path.join(DATADIR, "obsinfo-full.json") 

125 with open(path) as fh: 

126 obsinfo = ObservationInfo.from_json(fh.read()) 

127 

128 self.assertIsInstance(obsinfo, ObservationInfo) 

129 self._check_core_properties(obsinfo) 

130 # No extensions present in this fixture. 

131 self.assertEqual(obsinfo.extensions, {}) 

132 

133 # Translator class is empty. 

134 self.assertIsNone(obsinfo._translator) 

135 self.assertEqual(obsinfo.translator_class_name, "<None>") 

136 

137 # Round-tripping through JSON should be stable. 

138 round_tripped = ObservationInfo.from_json(obsinfo.to_json()) 

139 self.assertEqual(round_tripped, obsinfo) 

140 

141 # Should not include a spurious translator name. 

142 self.assertNotIn("_translator", obsinfo.model_dump()) 

143 

144 def test_read_full_with_extensions(self) -> None: 

145 """Read the fixture that also contains extension keys.""" 

146 path = os.path.join(DATADIR, "obsinfo-full-extensions.json") 

147 with open(path) as fh: 

148 obsinfo = ObservationInfo.from_json(fh.read()) 

149 

150 self.assertIsInstance(obsinfo, ObservationInfo) 

151 self._check_core_properties(obsinfo) 

152 

153 # Extension properties round-trip via the "fixture" translator name 

154 # stored in the JSON. 

155 self.assertEqual(set(obsinfo.extensions), {"number", "foo"}) 

156 self.assertEqual(obsinfo.ext_foo, "bar") 

157 self.assertEqual(obsinfo.ext_number, 12345) 

158 

159 self.assertIsInstance(obsinfo._translator, _FixtureTranslator) 

160 self.assertEqual(obsinfo.translator_name, "fixture") 

161 

162 round_tripped = ObservationInfo.from_json(obsinfo.to_json()) 

163 self.assertEqual(round_tripped, obsinfo) 

164 self.assertEqual(round_tripped.ext_foo, "bar") 

165 self.assertEqual(round_tripped.ext_number, 12345) 

166 

167 # Should dump the translator name. 

168 self.assertIsInstance(round_tripped._translator, _FixtureTranslator) 

169 self.assertEqual(round_tripped.translator_name, "fixture") 

170 self.assertIn("_translator", obsinfo.model_dump()) 

171 

172 # If there are extension properties the translator must be registered 

173 # so that the extension definitions are available. The error must 

174 # identify the missing translator as the cause rather than simply 

175 # reporting an unknown property. 

176 path = os.path.join(DATADIR, "obsinfo-extensions-unknown-translator.json") 

177 with open(path) as fh: 

178 json_str = fh.read() 

179 with self.assertRaises(KeyError) as cm: 

180 ObservationInfo.from_json(json_str) 

181 message = str(cm.exception) 

182 self.assertIn("fixture_unknown", message) 

183 self.assertIn("not registered", message) 

184 self.assertIn("ext_foo", message) 

185 

186 def test_read_unregistered(self) -> None: 

187 """Read a serialization with unregistered translator name.""" 

188 path = os.path.join(DATADIR, "obsinfo-unregistered.json") 

189 with open(path) as fh: 

190 obsinfo = ObservationInfo.from_json(fh.read()) 

191 

192 self.assertIsInstance(obsinfo, ObservationInfo) 

193 self._check_core_properties(obsinfo) 

194 # No extensions present in this fixture. 

195 self.assertEqual(obsinfo.extensions, {}) 

196 

197 self.assertIsNone(obsinfo._translator, None) 

198 self.assertEqual(obsinfo.translator_name, "Unregistered") 

199 

200 # Round-tripping through JSON should be stable. 

201 round_tripped = ObservationInfo.from_json(obsinfo.to_json()) 

202 self.assertEqual(round_tripped, obsinfo) 

203 self.assertEqual(round_tripped.translator_name, "Unregistered") 

204 

205 def test_nested_in_pydantic_model(self) -> None: 

206 """Nesting in another Pydantic model preserves the wire format. 

207 

208 Nested serialization goes through Pydantic directly and does not 

209 call ``to_simple`` / ``to_json``, so the alias behavior must come 

210 from ``serialize_by_alias`` in the model config rather than from 

211 the call-site kwargs. 

212 """ 

213 

214 class Wrapper(pydantic.BaseModel): 

215 obs: ObservationInfo 

216 

217 path = os.path.join(DATADIR, "obsinfo-full-extensions.json") 

218 with open(path) as fh: 

219 obsinfo = ObservationInfo.from_json(fh.read()) 

220 

221 wrapper = Wrapper(obs=obsinfo) 

222 

223 nested = wrapper.model_dump()["obs"] 

224 self.assertIn("_translator", nested) 

225 self.assertNotIn("translator_name", nested) 

226 # Extension keys remain flat under the ext_ prefix. 

227 self.assertEqual(nested["ext_foo"], "bar") 

228 self.assertEqual(nested["ext_number"], 12345) 

229 

230 nested_json = wrapper.model_dump_json() 

231 round_tripped = Wrapper.model_validate_json(nested_json) 

232 self.assertEqual(round_tripped.obs, obsinfo) 

233 

234 

235if __name__ == "__main__": 

236 unittest.main()