Coverage for tests/test_schema

1# This file is part of lsst-images.

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.

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"""The fixture coverage reporter, exercised through purpose-built schemas.

13The mechanics are pinned with doubles rather than the package's own schemas,

14so a new model or a new fixture variant cannot turn these into failures. The

15checks against the committed tree assert structure that stays true as the tree

16grows, never counts.

17"""

19from __future__ import annotations

21from pathlib import Path

22from typing import Any, ClassVar

24import pydantic

25import pytest

27from lsst.images.serialization import ArchiveTree, InputArchive

28from lsst.images.tests import (

29 canonical_fixture_text,

30 format_coverage_report,

31 schema_coverage,

32)

34FIXTURE_DIR = Path(__file__).parent / "data" / "schemas"

37class _Leaf(ArchiveTree):

38 """A stamped sub-schema a container can hold."""

40 SCHEMA_NAME: ClassVar[str] = "coverage_leaf"

41 SCHEMA_VERSION: ClassVar[str] = "1.0.0"

42 MIN_READ_VERSION: ClassVar[int] = 1

43 PUBLIC_TYPE: ClassVar[type] = dict

45 tag: str = pydantic.Field(default="", description="A value to round-trip.")

47 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:

48 """Return nothing; these doubles are never deserialized."""

49 raise NotImplementedError()

52class _OtherLeaf(ArchiveTree):

53 """A second stamped sub-schema, so a union has more than one candidate."""

55 SCHEMA_NAME: ClassVar[str] = "coverage_other_leaf"

56 SCHEMA_VERSION: ClassVar[str] = "1.0.0"

57 MIN_READ_VERSION: ClassVar[int] = 1

58 PUBLIC_TYPE: ClassVar[type] = dict

60 other: int = pydantic.Field(default=0, description="A value to round-trip.")

62 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:

63 """Return nothing; these doubles are never deserialized."""

64 raise NotImplementedError()

67class _Inner(pydantic.BaseModel):

68 """An unstamped nested model, whose paths belong to its container."""

70 depth: int = pydantic.Field(default=0, description="A nested value.")

71 note: str = pydantic.Field(default="", description="Another nested value.")

74class _Container(ArchiveTree):

75 """A composite schema holding leaves in several kinds of position."""

77 SCHEMA_NAME: ClassVar[str] = "coverage_container"

78 SCHEMA_VERSION: ClassVar[str] = "1.0.0"

79 MIN_READ_VERSION: ClassVar[int] = 1

80 PUBLIC_TYPE: ClassVar[type] = dict

82 chosen: _Leaf | _OtherLeaf = pydantic.Field(

83 default_factory=_Leaf, description="A union position with two candidates."

84 )

85 entries: list[_Leaf] = pydantic.Field(default_factory=list, description="A list position.")

86 table: dict[str, _Leaf] = pydantic.Field(default_factory=dict, description="A mapping position.")

87 inner: _Inner = pydantic.Field(default_factory=_Inner, description="An unstamped nested model.")

89 def deserialize(self, archive: InputArchive[Any], **kwargs: Any) -> Any:

90 """Return nothing; these doubles are never deserialized."""

91 raise NotImplementedError()

94_PACKAGE = _Container.__module__

95_CONTAINER_KEY = ("coverage_container", "1.0.0")

96_LEAF_KEY = ("coverage_leaf", "1.0.0")

97_OTHER_KEY = ("coverage_other_leaf", "1.0.0")

100def _write(directory: Path, tree: ArchiveTree, *, retired: bool = False) -> Path:

101 """Write a tree as a canonical fixture and return its path.

102

103 Parameters

104 ----------

105 directory

106 Root of the fixture tree to write into.

107 tree

108 Instance to serialize.

109 retired

110 Whether to place the file in the schema's ``retired`` subdirectory.

111 """

112 cls = type(tree)

113 parent = directory / cls.SCHEMA_NAME / ("retired" if retired else "")

114 parent.mkdir(parents=True, exist_ok=True)

115 path = parent / f"{cls.SCHEMA_NAME}-{cls.SCHEMA_VERSION}.json"

116 path.write_text(canonical_fixture_text(tree))

117 return path

118

119

120def test_container_fixture_credits_its_embedded_sub_schema(tmp_path: Path) -> None:

121 """Verify a sub-schema is credited by the container that embeds it.

122

123 This is the property that makes per-container attribution unnecessary: a

124 leaf reached only from a container still has that container as a source.

125 """

126 _write(tmp_path, _Container(chosen=_Leaf(tag="x")))

127 report = schema_coverage(tmp_path, package=_PACKAGE)

128 assert report.schemas[_LEAF_KEY].sources == {"coverage_container-1.0.0.json"}

129 assert ".tag" in report.schemas[_LEAF_KEY].expressed

130

131

132def test_coverage_is_keyed_by_schema_never_by_container(tmp_path: Path) -> None:

133 """Verify the report's keys are schemas, so a gap cannot name a container.

134

135 A key that mentioned the containing fixture is what would let the report

136 claim one container "fails to cover" a schema another one exercises.

137 """

138 _write(tmp_path, _Container(chosen=_Leaf()))

139 report = schema_coverage(tmp_path, package=_PACKAGE)

140 assert set(report.schemas) == {_CONTAINER_KEY, _LEAF_KEY, _OTHER_KEY}

141

142

143def test_union_position_reports_candidates_and_what_was_reached(tmp_path: Path) -> None:

144 """Verify a union position names every candidate and those reached."""

145 _write(tmp_path, _Container(chosen=_Leaf()))

146 report = schema_coverage(tmp_path, package=_PACKAGE)

147 (position,) = [p for p in report.schemas[_CONTAINER_KEY].positions if p.path == ".chosen"]

148 assert position.candidates == {"coverage_leaf", "coverage_other_leaf"}

149 assert position.reached == {"coverage_leaf"}

150 assert position.missing == {"coverage_other_leaf"}

151

152

153def test_a_second_fixture_closes_a_union_gap(tmp_path: Path) -> None:

154 """Verify reaching the other branch empties the position's missing set.

155

156 A variant added to widen coverage should show up here as the gap closing,

157 which is the whole point of reporting the position.

158 """

159 _write(tmp_path, _Container(chosen=_Leaf()))

160 other = tmp_path / "coverage_container" / "coverage_container-1.0.0-other.json"

161 other.write_text(canonical_fixture_text(_Container(chosen=_OtherLeaf(other=3))))

162 report = schema_coverage(tmp_path, package=_PACKAGE)

163 (position,) = [p for p in report.schemas[_CONTAINER_KEY].positions if p.path == ".chosen"]

164 assert position.reached == {"coverage_leaf", "coverage_other_leaf"}

165 assert not position.missing

166

167

168def test_list_and_mapping_positions_normalize_away_data_keys(tmp_path: Path) -> None:

169 """Verify list indices and mapping keys do not appear as schema paths.

170

171 An index and a mapping key are chosen by whoever wrote the data, so a

172 fixture with two list entries must not read as two distinct positions.

173 """

174 _write(

175 tmp_path,

176 _Container(entries=[_Leaf(tag="a"), _Leaf(tag="b")], table={"first": _Leaf(tag="c")}),

177 )

178 report = schema_coverage(tmp_path, package=_PACKAGE)

179 paths = {p.path for p in report.schemas[_CONTAINER_KEY].positions}

180 assert ".entries[]" in paths

181 assert ".table" in paths

182 assert not any(path.startswith(".entries[0") or path.startswith(".table.") for path in paths)

183

184

185def test_absent_roots_collapse_an_unset_subtree(tmp_path: Path) -> None:

186 """Verify an unset field reports one gap, not one per descendant.

187

188 ``butler_info`` is `None` unless a butler wrote the file, and its

189 ``exclude_if`` drops it from the dump entirely, taking its whole declared

190 subtree with it. Only the outermost path is actionable: populate that

191 field and everything beneath it becomes reachable.

192 """

193 _write(tmp_path, _Container())

194 coverage = schema_coverage(tmp_path, package=_PACKAGE).schemas[_CONTAINER_KEY]

195 assert ".butler_info.dataset" in coverage.absent

196 assert ".butler_info.dataset" not in coverage.absent_roots

197 assert ".butler_info" in coverage.absent_roots

198

199

200def test_retired_fixtures_do_not_credit_coverage(tmp_path: Path) -> None:

201 """Verify a retired fixture contributes nothing.

202

203 A retired fixture is expected not to validate, so counting it would credit

204 coverage to a shape current code rejects.

205 """

206 _write(tmp_path, _Leaf(tag="x"), retired=True)

207 coverage = schema_coverage(tmp_path, package=_PACKAGE).schemas[_LEAF_KEY]

208 assert not coverage.sources

209 assert ".tag" in coverage.absent

210

211

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

213 """Verify a schema nothing reaches is reported rather than omitted."""

214 report = schema_coverage(tmp_path, package=_PACKAGE)

215 coverage = report.schemas[_OTHER_KEY]

216 assert not coverage.sources

217 assert not coverage.expressed

218 assert "no fixture reaches this schema" in format_coverage_report(report, schema=coverage.name)

219

220

221def test_format_marks_gaps_and_full_positions_differently(tmp_path: Path) -> None:

222 """Verify the text output distinguishes a reached position from a gap."""

223 _write(tmp_path, _Container(chosen=_Leaf()))

224 report = schema_coverage(tmp_path, package=_PACKAGE)

225 text = format_coverage_report(report, schema="coverage_container")

226 assert "gap .chosen [coverage_leaf] of {coverage_leaf, coverage_other_leaf}" in text

227 assert "holds" not in text.split(".chosen")[0].splitlines()[-1]

228

229

230def test_committed_tree_credits_a_sub_schema_from_a_container() -> None:

231 """Verify the real fixture tree exercises cross-container attribution.

232

233 Asserted structurally rather than by count: some schema in the committed

234 tree is credited by a fixture that is not its own, which is what the

235 aggregation exists to capture.

236 """

237 report = schema_coverage(FIXTURE_DIR)

238 borrowed = {

239 coverage.name: sorted(coverage.sources)

240 for coverage in report.schemas.values()

241 if any(not source.startswith(f"{coverage.name}-") for source in coverage.sources)

242 }

243 assert borrowed, "no schema is credited by a container's fixture"

244

245

246@pytest.mark.parametrize("schema", ["cell_coadd", "visit_image"])

247def test_committed_composites_report_their_sub_schema_positions(schema: str) -> None:

Coverage for tests/test_schema_coverage.py: 100%

110 statements