Coverage for tests / test_schema.py: 34%

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"""Tests for JSON Schema generation from the ObservationInfo model.""" 

13 

14import json 

15import os.path 

16import unittest 

17from collections.abc import Mapping 

18from typing import Any 

19 

20import astropy.time 

21import astropy.units as u 

22 

23try: 

24 import jsonschema 

25 

26 HAS_JSONSCHEMA = True 

27except ImportError: 

28 HAS_JSONSCHEMA = False 

29 

30from astro_metadata_translator import ( 

31 ObservationInfo, 

32 PropertyDefinition, 

33 StubTranslator, 

34 makeObservationInfo, 

35) 

36 

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

38 

39 

40class _SchemaTranslator(StubTranslator): 

41 """Test translator that defines extension properties.""" 

42 

43 name = "schema_dummy" 

44 supported_instrument = "schema_dummy" 

45 extensions = dict( 

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

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

48 ) 

49 _const_map = { 

50 "ext_foo": "bar", 

51 "ext_number": 17, 

52 "detector_name": "detA", 

53 } 

54 _trivial_map = { 

55 "instrument": "INSTRUME", 

56 } 

57 

58 @classmethod 

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

60 return "INSTRUME" in header and header["INSTRUME"] == "schema_dummy" 

61 

62 

63# Mapping of internal Field-style names that must NOT appear as keys in the 

64# generated JSON schema (these are runtime state, not part of the wire format). 

65_INTERNAL_FIELDS = frozenset({"filename", "translator_class_name", "extensions", "all_properties"}) 

66 

67 

68class JsonSchemaTestCase(unittest.TestCase): 

69 """Test JSON Schema generation.""" 

70 

71 def test_schema_serialization_mode_generatable(self) -> None: 

72 """Serialization-mode schema must be generatable and non-trivial.""" 

73 schema = ObservationInfo.model_json_schema(mode="serialization") 

74 self.assertEqual(schema.get("type"), "object") 

75 properties = schema.get("properties", {}) 

76 # Schema should describe each core property of the ObservationInfo, 

77 # not be a trivial open-ended object. 

78 self.assertIn("telescope", properties) 

79 self.assertIn("instrument", properties) 

80 self.assertIn("exposure_time", properties) 

81 self.assertIn("datetime_begin", properties) 

82 self.assertIn("location", properties) 

83 

84 def test_schema_validation_mode_generatable(self) -> None: 

85 """Validation-mode schema must also be generatable.""" 

86 schema = ObservationInfo.model_json_schema(mode="validation") 

87 self.assertEqual(schema.get("type"), "object") 

88 

89 def test_schema_excludes_internal_fields(self) -> None: 

90 """Internal runtime state must not appear in the schema.""" 

91 for mode in ("serialization", "validation"): 

92 with self.subTest(mode=mode): 

93 schema = ObservationInfo.model_json_schema(mode=mode) 

94 properties = set(schema.get("properties", {}).keys()) 

95 leaked = _INTERNAL_FIELDS & properties 

96 self.assertEqual(leaked, set(), f"Internal fields leaked into {mode} schema: {leaked}") 

97 

98 def test_schema_translator_key(self) -> None: 

99 """The schema should describe the _translator metadata key.""" 

100 schema = ObservationInfo.model_json_schema(mode="serialization") 

101 self.assertIn("_translator", schema.get("properties", {})) 

102 

103 def test_schema_allows_ext_properties(self) -> None: 

104 """Schema should allow ext_* keys via patternProperties.""" 

105 schema = ObservationInfo.model_json_schema(mode="serialization") 

106 pattern_props = schema.get("patternProperties", {}) 

107 # Some pattern keyed by ext_ prefix should accept arbitrary content. 

108 self.assertTrue( 

109 any("ext_" in key for key in pattern_props), 

110 f"Expected an ext_-prefixed patternProperties entry, got {pattern_props}", 

111 ) 

112 

113 def test_schema_astropy_field_simple_form(self) -> None: 

114 """Astropy-typed fields should use their simple form in the schema.""" 

115 schema = ObservationInfo.model_json_schema(mode="serialization") 

116 properties = schema["properties"] 

117 

118 # exposure_time: Quantity -> float (seconds) 

119 exposure_time = properties["exposure_time"] 

120 self.assertNotIn("is-instance", json.dumps(exposure_time)) 

121 self._assert_accepts_type(exposure_time, "number") 

122 

123 # datetime_begin: Time -> array of 2 numbers (TAI JD) 

124 datetime_begin = properties["datetime_begin"] 

125 self._assert_accepts_type(datetime_begin, "array") 

126 

127 # location: EarthLocation -> array of 3 numbers (geocentric m) 

128 location = properties["location"] 

129 self._assert_accepts_type(location, "array") 

130 

131 # boresight_rotation_angle: Angle -> float (degrees) 

132 rot_angle = properties["boresight_rotation_angle"] 

133 self._assert_accepts_type(rot_angle, "number") 

134 

135 # tracking_radec: SkyCoord -> array of 2 numbers 

136 tracking_radec = properties["tracking_radec"] 

137 self._assert_accepts_type(tracking_radec, "array") 

138 

139 # observing_day_offset: TimeDelta -> int (seconds) 

140 offset = properties["observing_day_offset"] 

141 self._assert_accepts_type(offset, "integer") 

142 

143 @unittest.skipUnless(HAS_JSONSCHEMA, "jsonschema package is not installed") 

144 def test_schema_validates_real_obsinfo(self) -> None: 

145 """Serialized ObservationInfo must validate against the schema.""" 

146 reference = dict( 

147 boresight_airmass=1.5, 

148 focus_z=1.0 * u.mm, 

149 temperature=15 * u.deg_C, 

150 observation_type="bias", 

151 exposure_time=5 * u.ks, 

152 detector_num=32, 

153 datetime_begin=astropy.time.Time("2021-02-15T12:00:00", format="isot", scale="utc"), 

154 ) 

155 obsinfo = makeObservationInfo(**reference) 

156 data = json.loads(obsinfo.to_json()) 

157 

158 schema = ObservationInfo.model_json_schema(mode="serialization") 

159 # This will raise if invalid. 

160 jsonschema.validate(instance=data, schema=schema) 

161 

162 @unittest.skipUnless(HAS_JSONSCHEMA, "jsonschema package is not installed") 

163 def test_schema_validates_with_extensions(self) -> None: 

164 """An ObservationInfo with ext_* properties must validate.""" 

165 obsinfo = makeObservationInfo( 

166 translator_class=_SchemaTranslator, 

167 ext_foo="bar", 

168 ext_number=42, 

169 instrument="schema_dummy", 

170 detector_name="detA", 

171 ) 

172 data = json.loads(obsinfo.to_json()) 

173 self.assertIn("ext_foo", data) 

174 self.assertIn("ext_number", data) 

175 

176 schema = ObservationInfo.model_json_schema(mode="serialization") 

177 jsonschema.validate(instance=data, schema=schema) 

178 

179 @unittest.skipUnless(HAS_JSONSCHEMA, "jsonschema package is not installed") 

180 def test_schema_validates_reference_files(self) -> None: 

181 """Reference JSON files captured before this ticket must still 

182 validate against the generated schema. 

183 

184 Guards against accidental wire-format regressions. 

185 """ 

186 schema = ObservationInfo.model_json_schema(mode="serialization") 

187 for filename in ("obsinfo-full.json", "obsinfo-full-extensions.json"): 

188 with self.subTest(filename=filename): 

189 path = os.path.join(TESTDIR, "data", filename) 

190 with open(path) as fh: 

191 data = json.load(fh) 

192 jsonschema.validate(instance=data, schema=schema) 

193 

194 def _assert_accepts_type(self, prop_schema: dict, expected_type: str) -> None: 

195 """Assert that a property schema accepts the given JSON type. 

196 

197 All ObservationInfo properties are nullable, so Pydantic emits 

198 them as ``anyOf: [<actual type>, {"type": "null"}]``. 

199 

200 Parameters 

201 ---------- 

202 prop_schema : `dict` 

203 The JSON schema entry for a single property. 

204 expected_type : `str` 

205 The JSON Schema type name that the property must accept. 

206 """ 

207 for branch in prop_schema.get("anyOf", []): 

208 if branch.get("type") == expected_type: 

209 return 

210 self.fail( 

211 f"Property schema does not accept type {expected_type!r}: {json.dumps(prop_schema, default=str)}" 

212 ) 

213 

214 

215if __name__ == "__main__": 

216 unittest.main()