Coverage for tests/test_frozen_schemas.py: 98%

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 importlib.metadata 

15import json 

16from pathlib import Path 

17from typing import Any, ClassVar 

18 

19import pydantic 

20import pytest 

21 

22from lsst.images.serialization import ( 

23 ArchiveTree, 

24 FrozenSchemaError, 

25 InputArchive, 

26 available_schema_classes, 

27 check_frozen_schemas, 

28 dump_schema, 

29 frozen_schema_filename, 

30 frozen_schema_path, 

31 is_development_version, 

32 write_frozen_schemas, 

33) 

34from lsst.images.serialization._frozen_schemas import _summary_description 

35 

36REPO_SCHEMA_DIR = Path(__file__).parent.parent / "schemas" 

37 

38 

39class _ForeignArchiveTree(ArchiveTree): 

40 """Concrete ArchiveTree defined outside lsst.images; registers itself on 

41 class creation, like the test doubles in other test modules. 

42 """ 

43 

44 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_foreign" 

45 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

46 MIN_READ_VERSION: ClassVar[int] = 1 

47 PUBLIC_TYPE: ClassVar[type] = object 

48 

49 def deserialize( 

50 self, archive: InputArchive[Any], **kwargs: Any 

51 ) -> Any: # pragma: no cover - never invoked 

52 raise NotImplementedError() 

53 

54 

55class _CustomBaseArchiveTree(ArchiveTree): 

56 """Concrete ArchiveTree with a third-party schema URL base.""" 

57 

58 SCHEMA_URL_BASE: ClassVar[str] = "https://example.org/schemas" 

59 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_custom_base" 

60 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

61 MIN_READ_VERSION: ClassVar[int] = 1 

62 PUBLIC_TYPE: ClassVar[type] = object 

63 

64 def deserialize( 

65 self, archive: InputArchive[Any], **kwargs: Any 

66 ) -> Any: # pragma: no cover - never invoked 

67 raise NotImplementedError() 

68 

69 

70class _DevDoubleArchiveTree(ArchiveTree): 

71 """Development-version double, defined in the test package.""" 

72 

73 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_development" 

74 SCHEMA_VERSION: ClassVar[str] = "1.0.0.dev0" 

75 MIN_READ_VERSION: ClassVar[int] = 1 

76 PUBLIC_TYPE: ClassVar[type] = object 

77 

78 def deserialize( 

79 self, archive: InputArchive[Any], **kwargs: Any 

80 ) -> Any: # pragma: no cover - never invoked 

81 raise NotImplementedError() 

82 

83 

84def test_custom_schema_url_base() -> None: 

85 """Verify a subclass can override SCHEMA_URL_BASE so its schema URL and 

86 $id are minted under its own documentation site. 

87 """ 

88 expected = "https://example.org/schemas/frozen_schemas_test_custom_base-1.0.0" 

89 assert _CustomBaseArchiveTree().schema_url == expected 

90 schema = dump_schema(_CustomBaseArchiveTree) 

91 assert schema["$id"] == expected 

92 assert schema["title"] == "frozen_schemas_test_custom_base" 

93 

94 

95def test_available_schema_classes() -> None: 

96 """Verify enumeration finds the package's schemas, sorted by name.""" 

97 classes = available_schema_classes() 

98 names = [cls.SCHEMA_NAME for cls in classes] 

99 assert names == sorted(names) 

100 assert "image" in names 

101 assert "cell_coadd" in names # lazily-loaded built-in provider 

102 

103 

104def test_available_schema_classes_excludes_foreign_modules() -> None: 

105 """Verify classes registered from outside lsst.images (e.g. test 

106 doubles) are not treated as package-owned schemas. 

107 """ 

108 names = [cls.SCHEMA_NAME for cls in available_schema_classes()] 

109 assert "frozen_schemas_test_foreign" not in names 

110 

111 

112def test_available_schema_classes_package_filter() -> None: 

113 """Verify the package argument selects schemas by defining module, so an 

114 external package can freeze its own schemas. 

115 """ 

116 classes = available_schema_classes(package=_ForeignArchiveTree.__module__) 

117 names = [cls.SCHEMA_NAME for cls in classes] 

118 assert "frozen_schemas_test_foreign" in names 

119 assert "image" not in names 

120 

121 

122def test_available_schema_classes_discovers_entry_points(monkeypatch: pytest.MonkeyPatch) -> None: 

123 """Verify schemas provided only through the ``lsst.images.schemas`` 

124 entry point group are discovered without prior registration. 

125 """ 

126 state: dict[str, type[ArchiveTree]] = {} 

127 

128 def _load() -> type[ArchiveTree]: 

129 if "cls" not in state: 129 ↛ 143line 129 didn't jump to line 143 because the condition on line 129 was always true

130 

131 class _EntryPointArchiveTree(ArchiveTree): 

132 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_entry_point" 

133 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

134 MIN_READ_VERSION: ClassVar[int] = 1 

135 PUBLIC_TYPE: ClassVar[type] = object 

136 

137 def deserialize( 

138 self, archive: InputArchive[Any], **kwargs: Any 

139 ) -> Any: # pragma: no cover - never invoked 

140 raise NotImplementedError() 

141 

142 state["cls"] = _EntryPointArchiveTree 

143 return state["cls"] 

144 

145 class _FakeEntryPoint: 

146 name = "frozen_schemas_test_entry_point" 

147 value = "fake.module:FakeClass" 

148 

149 def load(self) -> type[ArchiveTree]: 

150 return _load() 

151 

152 real_entry_points = importlib.metadata.entry_points 

153 

154 def _fake_entry_points(**kwargs: Any) -> Any: 

155 if kwargs.get("group") == "lsst.images.schemas": 155 ↛ 160line 155 didn't jump to line 160 because the condition on line 155 was always true

156 eps = [_FakeEntryPoint()] 

157 if (name := kwargs.get("name")) is not None: 

158 eps = [ep for ep in eps if ep.name == name] 

159 return eps 

160 return real_entry_points(**kwargs) 

161 

162 monkeypatch.setattr(importlib.metadata, "entry_points", _fake_entry_points) 

163 classes = available_schema_classes(package=_ForeignArchiveTree.__module__) 

164 assert "frozen_schemas_test_entry_point" in [cls.SCHEMA_NAME for cls in classes] 

165 

166 

167def test_write_skips_development_schemas(tmp_path: Path) -> None: 

168 """Verify a development schema is never frozen.""" 

169 write_frozen_schemas(tmp_path, package=_DevDoubleArchiveTree.__module__) 

170 assert not frozen_schema_path(tmp_path, _DevDoubleArchiveTree).exists() 

171 

172 

173def test_check_skips_development_schemas(tmp_path: Path) -> None: 

174 """Verify check does not report a missing file for a development schema.""" 

175 problems = check_frozen_schemas(tmp_path, package=_DevDoubleArchiveTree.__module__) 

176 assert all("development" not in p for p in problems) 

177 

178 

179def test_write_refuses_to_change_finalized(tmp_path: Path) -> None: 

180 """Verify a finalized frozen file cannot be overwritten.""" 

181 write_frozen_schemas(tmp_path) 

182 (cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"] 

183 path = frozen_schema_path(tmp_path, cls) 

184 stale = json.loads(path.read_text()) 

185 stale["description"] = "changed by hand" 

186 path.write_text(json.dumps(stale, indent=2, sort_keys=True) + "\n") 

187 with pytest.raises(FrozenSchemaError, match="bump"): 

188 write_frozen_schemas(tmp_path) 

189 

190 

191def test_dump_schema_has_id_and_title() -> None: 

192 """Verify the dumped schema carries the canonical $id and title.""" 

193 (cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"] 

194 schema = dump_schema(cls) 

195 assert schema["$id"] == f"https://images.lsst.io/schemas/image-{cls.SCHEMA_VERSION}" 

196 assert schema["title"] == "image" 

197 assert frozen_schema_filename(cls) == f"image-{cls.SCHEMA_VERSION}.json" 

198 

199 

200def test_dumped_schema_ids_match_declaring_class() -> None: 

201 """Verify every dumped schema's $id and title come from the declaring 

202 class, not an inherited concrete schema (e.g. visit_image, which 

203 subclasses masked_image). 

204 """ 

205 for cls in available_schema_classes(): 

206 schema = dump_schema(cls) 

207 expected = f"https://images.lsst.io/schemas/{cls.SCHEMA_NAME}-{cls.SCHEMA_VERSION}" 

208 assert schema["$id"] == expected, cls.SCHEMA_NAME 

209 assert schema["title"] == cls.SCHEMA_NAME 

210 

211 

212def test_write_and_check_round_trip(tmp_path: Path) -> None: 

213 """Verify write produces files that check accepts, and rewrite is a 

214 no-op. 

215 """ 

216 changed = write_frozen_schemas(tmp_path) 

217 assert changed 

218 assert check_frozen_schemas(tmp_path) == [] 

219 assert write_frozen_schemas(tmp_path) == [] 

220 

221 

222def test_frozen_layout_per_schema_subdirectory(tmp_path: Path) -> None: 

223 """Verify frozen files are laid out as ``{name}/{name}-{version}.json`` 

224 so the directory scales as versions accumulate. 

225 """ 

226 write_frozen_schemas(tmp_path) 

227 (image_cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"] 

228 path = frozen_schema_path(tmp_path, image_cls) 

229 assert path == tmp_path / "image" / frozen_schema_filename(image_cls) 

230 assert path.exists() 

231 assert not list(tmp_path.glob("*.json")) 

232 

233 

234def test_check_reports_missing_and_stale(tmp_path: Path) -> None: 

235 """Verify check flags a missing file and a stale file.""" 

236 write_frozen_schemas(tmp_path) 

237 (cls,) = [c for c in available_schema_classes() if c.SCHEMA_NAME == "image"] 

238 path = frozen_schema_path(tmp_path, cls) 

239 stale = json.loads(path.read_text()) 

240 stale["description"] = "something else" 

241 path.write_text(json.dumps(stale, indent=2, sort_keys=True) + "\n") 

242 problems = check_frozen_schemas(tmp_path) 

243 assert any("bump SCHEMA_VERSION" in p for p in problems) 

244 path.unlink() 

245 problems = check_frozen_schemas(tmp_path) 

246 assert any("missing" in p for p in problems) 

247 

248 

249def test_write_preserves_superseded_versions(tmp_path: Path) -> None: 

250 """Verify write never deletes a frozen file for an old schema version.""" 

251 old = tmp_path / "image" / "image-0.9.9.json" 

252 old.parent.mkdir() 

253 old.write_text("{}\n") 

254 write_frozen_schemas(tmp_path) 

255 assert old.exists() 

256 

257 

258_DEVELOPMENT_SCHEMAS = { 

259 "color_image", 

260} 

261 

262 

263def test_development_schemas_are_unfrozen() -> None: 

264 """Verify exactly the intended schemas are development (unfrozen) and the 

265 rest are finalized (frozen). 

266 """ 

267 for cls in available_schema_classes(): 

268 frozen = frozen_schema_path(REPO_SCHEMA_DIR, cls) 

269 if cls.SCHEMA_NAME in _DEVELOPMENT_SCHEMAS: 

270 assert is_development_version(cls.SCHEMA_VERSION), cls.SCHEMA_NAME 

271 assert not frozen.exists(), f"{cls.SCHEMA_NAME} should be unfrozen" 

272 else: 

273 assert not is_development_version(cls.SCHEMA_VERSION), cls.SCHEMA_NAME 

274 assert frozen.exists(), f"{cls.SCHEMA_NAME} should be frozen" 

275 

276 

277def test_committed_frozen_schemas_are_current() -> None: 

278 """Verify the git-committed frozen schema files match the models. 

279 

280 A failure here means a serialization model changed; run 

281 ``lsst-images-admin schemas write`` and commit the result. 

282 """ 

283 problems = check_frozen_schemas(REPO_SCHEMA_DIR) 

284 assert not problems, ( 

285 "Frozen schema files are stale; run 'lsst-images-admin schemas write' " 

286 "and commit the result: " + ", ".join(problems) 

287 ) 

288 

289 

290class _NotesArchiveTree(ArchiveTree): 

291 """Summary line for the notes tree. 

292 

293 Extended summary paragraph that should be kept. 

294 

295 Notes 

296 ----- 

297 Implementation detail that must not appear in the schema description. 

298 """ 

299 

300 SCHEMA_NAME: ClassVar[str] = "frozen_schemas_test_notes" 

301 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

302 MIN_READ_VERSION: ClassVar[int] = 1 

303 PUBLIC_TYPE: ClassVar[type] = object 

304 

305 label: str = pydantic.Field(default="", description="A one-line field description with no sections.") 

306 

307 def deserialize( 

308 self, archive: InputArchive[Any], **kwargs: Any 

309 ) -> Any: # pragma: no cover - never invoked 

310 raise NotImplementedError() 

311 

312 

313@pytest.mark.parametrize( 

314 ("text", "expected"), 

315 [ 

316 ("Just a summary.", "Just a summary."), 

317 ("Summary.\n\nExtended paragraph.", "Summary.\n\nExtended paragraph."), 

318 ("Summary.\n\nNotes\n-----\nDetail.", "Summary."), 

319 ("Summary.\n\nMore.\n\nParameters\n----------\nx\n A thing.", "Summary.\n\nMore."), 

320 ("A dash - not a section header.", "A dash - not a section header."), 

321 ], 

322) 

323def test_summary_description(text: str, expected: str) -> None: 

324 """Verify the summary extends up to (not into) the first numpydoc section 

325 header, and text without a section header is returned unchanged. 

326 """ 

327 assert _summary_description(text) == expected 

328 

329 

330def test_dump_schema_description_drops_numpydoc_sections() -> None: 

331 """Verify a class docstring's numpydoc sections are dropped from the 

332 schema description, while a one-line field description is untouched. 

333 """ 

334 schema = dump_schema(_NotesArchiveTree) 

335 assert schema["description"] == ( 

336 "Summary line for the notes tree.\n\nExtended summary paragraph that should be kept." 

337 ) 

338 assert "Notes" not in schema["description"] 

339 assert schema["properties"]["label"]["description"] == "A one-line field description with no sections."