Coverage for tests/test_frozen_schema_dependencies.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. 

11"""Tests for the freeze-time check that a finalized schema depends only on 

12other finalized schemas. 

13 

14The doubles live in their own module so the ``package`` filter used by 

15``write_frozen_schemas`` selects exactly them, leaving the doubles in 

16``test_frozen_schemas`` unaffected. 

17""" 

18 

19from __future__ import annotations 

20 

21from pathlib import Path 

22from typing import Any, ClassVar 

23 

24import pydantic 

25import pytest 

26 

27from lsst.images.serialization import ( 

28 ArchiveTree, 

29 FrozenSchemaError, 

30 InputArchive, 

31 available_schema_classes, 

32 dump_schema, 

33 frozen_schema_path, 

34 is_development_version, 

35 schema_dependencies, 

36 write_frozen_schemas, 

37) 

38from lsst.images.serialization._frozen_schemas import _canonical_text 

39 

40PACKAGE = __name__ 

41 

42 

43class _DevLeaf(ArchiveTree): 

44 """Development schema that others depend on.""" 

45 

46 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_dev_leaf" 

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

48 MIN_READ_VERSION: ClassVar[int] = 1 

49 PUBLIC_TYPE: ClassVar[type] = object 

50 

51 note: str = pydantic.Field(default="", description="A field.") 

52 

53 def deserialize( 

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

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

56 raise NotImplementedError() 

57 

58 

59class _FinalLeaf(ArchiveTree): 

60 """Finalized schema with no dependencies of its own.""" 

61 

62 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_final_leaf" 

63 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

64 MIN_READ_VERSION: ClassVar[int] = 1 

65 PUBLIC_TYPE: ClassVar[type] = object 

66 

67 note: str = pydantic.Field(default="", description="A field.") 

68 

69 def deserialize( 

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

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

72 raise NotImplementedError() 

73 

74 

75class _FinalInheritsDev(_DevLeaf): 

76 """Finalized schema that inherits from a development schema. 

77 

78 The ``cell_coadd`` / ``masked_image`` shape. 

79 """ 

80 

81 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_inherits_dev" 

82 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

83 

84 

85class _FinalEmbedsDev(ArchiveTree): 

86 """Finalized schema with a field referencing a development schema.""" 

87 

88 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_embeds_dev" 

89 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

90 MIN_READ_VERSION: ClassVar[int] = 1 

91 PUBLIC_TYPE: ClassVar[type] = object 

92 

93 child: _DevLeaf = pydantic.Field(description="An embedded development schema.") 

94 

95 def deserialize( 

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

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

98 raise NotImplementedError() 

99 

100 

101class _FinalEmbedsDevIndirectly(ArchiveTree): 

102 """Finalized schema whose dependency on a development schema is indirect. 

103 

104 Reaches ``_DevLeaf`` only through a container and an optional union, and 

105 only via another finalized schema, so a non-transitive check that looked 

106 at bare annotations would miss it. 

107 """ 

108 

109 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_embeds_dev_indirect" 

110 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

111 MIN_READ_VERSION: ClassVar[int] = 1 

112 PUBLIC_TYPE: ClassVar[type] = object 

113 

114 children: dict[str, _FinalEmbedsDev | None] = pydantic.Field( 

115 default_factory=dict, description="Indirectly reaches the development schema." 

116 ) 

117 

118 def deserialize( 

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

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

121 raise NotImplementedError() 

122 

123 

124class _FinalClean(ArchiveTree): 

125 """Finalized schema depending only on another finalized schema.""" 

126 

127 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_clean" 

128 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

129 MIN_READ_VERSION: ClassVar[int] = 1 

130 PUBLIC_TYPE: ClassVar[type] = object 

131 

132 children: list[_FinalLeaf] = pydantic.Field( 

133 default_factory=list, description="Embedded finalized schemas." 

134 ) 

135 

136 def deserialize( 

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

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

139 raise NotImplementedError() 

140 

141 

142class _SelfRecursive(ArchiveTree): 

143 """Finalized schema that references itself.""" 

144 

145 SCHEMA_NAME: ClassVar[str] = "frozen_schema_deps_test_recursive" 

146 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

147 MIN_READ_VERSION: ClassVar[int] = 1 

148 PUBLIC_TYPE: ClassVar[type] = object 

149 

150 child: _SelfRecursive | None = pydantic.Field(default=None, description="Itself.") 

151 

152 def deserialize( 

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

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

155 raise NotImplementedError() 

156 

157 

158def test_dependencies_follow_inheritance() -> None: 

159 """Verify a schema-declaring base class counts as a dependency.""" 

160 assert set(schema_dependencies(_FinalInheritsDev)) == {_DevLeaf.SCHEMA_NAME} 

161 

162 

163def test_dependencies_follow_fields() -> None: 

164 """Verify an embedded model counts as a dependency.""" 

165 assert set(schema_dependencies(_FinalEmbedsDev)) == {_DevLeaf.SCHEMA_NAME} 

166 

167 

168def test_dependencies_are_transitive() -> None: 

169 """Verify dependencies are followed through containers and unions.""" 

170 assert set(schema_dependencies(_FinalEmbedsDevIndirectly)) == { 

171 _FinalEmbedsDev.SCHEMA_NAME, 

172 _DevLeaf.SCHEMA_NAME, 

173 } 

174 

175 

176def test_dependencies_exclude_self() -> None: 

177 """Verify a self-referential schema is not its own dependency.""" 

178 assert schema_dependencies(_SelfRecursive) == {} 

179 

180 

181def test_dependencies_of_clean_schema() -> None: 

182 """Verify a clean finalized schema reports only finalized dependencies.""" 

183 assert set(schema_dependencies(_FinalClean)) == {_FinalLeaf.SCHEMA_NAME} 

184 

185 

186def test_freeze_refuses_schema_inheriting_development(tmp_path: Path) -> None: 

187 """Verify freezing fails when a finalized schema depends on a dev one.""" 

188 with pytest.raises(FrozenSchemaError, match=_DevLeaf.SCHEMA_NAME): 

189 write_frozen_schemas(tmp_path, package=PACKAGE) 

190 

191 

192def test_freeze_error_names_the_dependent_schema(tmp_path: Path) -> None: 

193 """Verify the error identifies which schema may not be frozen.""" 

194 with pytest.raises(FrozenSchemaError) as caught: 

195 write_frozen_schemas(tmp_path, package=PACKAGE) 

196 message = str(caught.value) 

197 assert _DevLeaf.SCHEMA_VERSION in message 

198 assert any( 

199 name in message 

200 for name in ( 

201 _FinalInheritsDev.SCHEMA_NAME, 

202 _FinalEmbedsDev.SCHEMA_NAME, 

203 _FinalEmbedsDevIndirectly.SCHEMA_NAME, 

204 ) 

205 ) 

206 

207 

208def test_shipped_schemas_have_no_development_dependencies() -> None: 

209 """Verify no finalized schema in the package depends on a development one. 

210 

211 This is the invariant the freeze-time check exists to protect, asserted 

212 over the real schema set rather than the doubles above. 

213 """ 

214 offenders = { 

215 cls.SCHEMA_NAME: sorted( 

216 f"{name}-{dependency.SCHEMA_VERSION}" 

217 for name, dependency in schema_dependencies(cls).items() 

218 if is_development_version(dependency.SCHEMA_VERSION) 

219 ) 

220 for cls in available_schema_classes() 

221 if not is_development_version(cls.SCHEMA_VERSION) 

222 } 

223 assert not {name: dev for name, dev in offenders.items() if dev} 

224 

225 

226def test_already_frozen_file_is_not_rechecked(tmp_path: Path) -> None: 

227 """Verify the check applies only at freeze time. 

228 

229 An already-committed frozen file is left alone even when its dependencies 

230 are still in development, so a pre-existing violation does not become a 

231 permanent failure. 

232 """ 

233 for cls in (_FinalInheritsDev, _FinalEmbedsDev, _FinalEmbedsDevIndirectly): 

234 path = frozen_schema_path(tmp_path, cls) 

235 path.parent.mkdir(parents=True, exist_ok=True) 

236 path.write_text(_canonical_text(dump_schema(cls))) 

237 write_frozen_schemas(tmp_path, package=PACKAGE) 

238 assert frozen_schema_path(tmp_path, _FinalClean).exists() 

239 assert not frozen_schema_path(tmp_path, _DevLeaf).exists()