Coverage for python/lsst/images/tests/_schema_coverage.py: 97%
184 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-24 09:06 +0000
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"""What the committed schema fixtures do and do not exercise.
13Coverage is attributed per schema, not per fixture file. Every
14`~lsst.images.serialization.ArchiveTree` serializes its own
15``schema_version`` and ``schema_url``, root or embedded, so fixture JSON is
16self-describing at every depth: a walk can switch attribution at each nested
17stamp and credit what it finds to the schema that owns it. A sub-model
18reached from several containers is therefore credited by all of them, and no
19report can claim a container "fails to cover" a schema that another container
20exercises.
22Two things are reported. Property coverage is the set of paths a schema
23declares that no fixture expresses. Sub-schema positions are the places a
24schema can hold another stamped schema, with the candidates each position
25admits and the ones fixtures actually put there; a composite model is under
26no obligation to reach every candidate, but knowing which it misses is what
27tells you whether a fixture set matches what you assumed it covered.
29See :ref:`lsst.images-schema-fixtures` for the fixture tree this reads.
30"""
32from __future__ import annotations
34__all__ = (
35 "CoverageReport",
36 "SchemaCoverage",
37 "SubSchemaPosition",
38 "format_coverage_report",
39 "schema_coverage",
40)
42import dataclasses
43import json
44from collections import defaultdict
45from pathlib import Path
47from ..serialization import ArchiveTree, available_schema_classes, dump_schema
48from ._schema_fixtures import fixture_version, iter_schema_fixtures
50_SCHEMA_URL_KEY = "x-lsst-schema-url"
51"""Key `~lsst.images.serialization.dump_schema` writes the canonical URL of a
52nested published schema under.
54Draft 2020-12 makes ``$id`` start a new resolution scope, which would break
55the root-relative references pydantic generates, so nested definitions carry
56their identity under this non-reserved key instead.
57"""
60def _schema_name_from_url(url: str) -> str:
61 """Return the schema name encoded in a canonical schema URL.
63 Parameters
64 ----------
65 url
66 Canonical schema URL, as ``{base}/{name}-{version}``.
67 """
68 return url.rsplit("/", 1)[-1].rsplit("-", 1)[0]
71def _stamp_key(value: object) -> tuple[str, str] | None:
72 """Return the ``(name, version)`` a JSON value stamps itself with.
74 Parameters
75 ----------
76 value
77 Raw JSON-like value to inspect.
79 Returns
80 -------
81 `tuple` [ `str`, `str` ] or `None`
82 The schema name and fixture-filename version, or `None` if the value
83 is not a stamped tree. A malformed stamp yields `None` rather than
84 raising; `~lsst.images.tests.check_schema_fixtures` is what reports
85 such a fixture as broken.
86 """
87 if not isinstance(value, dict):
88 return None
89 url = value.get("schema_url")
90 version = value.get("schema_version")
91 if not isinstance(url, str) or not isinstance(version, str):
92 return None
93 try:
94 return _schema_name_from_url(url), fixture_version(version)
95 except Exception:
96 return None
99def _credit(
100 value: object,
101 key: tuple[str, str],
102 path: str,
103 expressed: dict[tuple[str, str], set[str]],
104 embeddings: dict[tuple[str, str], dict[str, set[str]]],
105) -> None:
106 """Credit the paths a raw fixture value expresses to their owning schema.
108 Parameters
109 ----------
110 value
111 Raw JSON-like value to walk.
112 key
113 Schema name and version of the nearest enclosing stamped tree.
114 path
115 Path of ``value`` within that tree, empty at the tree's own root.
116 expressed
117 Accumulator mapping each schema to the paths it expresses.
118 embeddings
119 Accumulator mapping each schema to the sub-schema names found at each
120 of its paths.
122 Notes
123 -----
124 List indices collapse to ``[]``, because an index is data rather than a
125 position in the schema. Attribution switches at every nested stamp, and
126 the nested path is recorded in the container as well: the container did
127 put something there, even though the contents belong to the sub-schema.
128 """
129 if path:
130 expressed[key].add(path)
131 if isinstance(value, dict):
132 for name, sub in value.items():
133 child_path = f"{path}.{name}"
134 if (child_key := _stamp_key(sub)) is not None:
135 expressed[key].add(child_path)
136 embeddings[key].setdefault(child_path, set()).add(child_key[0])
137 _credit(sub, child_key, "", expressed, embeddings)
138 else:
139 _credit(sub, key, child_path, expressed, embeddings)
140 elif isinstance(value, list):
141 for item in value:
142 item_path = f"{path}[]"
143 if (item_key := _stamp_key(item)) is not None:
144 expressed[key].add(item_path)
145 embeddings[key].setdefault(item_path, set()).add(item_key[0])
146 _credit(item, item_key, "", expressed, embeddings)
147 else:
148 _credit(item, key, item_path, expressed, embeddings)
151def _resolve(node: object, defs: dict[str, object], seen: frozenset[str]) -> tuple[object, frozenset[str]]:
152 """Follow a ``$ref`` chain to the definition it names.
154 Parameters
155 ----------
156 node
157 Schema node, which may be a ``$ref``.
158 defs
159 The document's ``$defs`` mapping.
160 seen
161 Definition names already followed on this branch, which stops a
162 recursive model (a sum of fields that may themselves be sums) from
163 looping.
165 Returns
166 -------
167 resolved : `object`
168 The resolved node, or `None` if the chain revisits a definition.
169 seen : `frozenset` [ `str` ]
170 ``seen`` extended with the names followed here.
171 """
172 while isinstance(node, dict) and "$ref" in node:
173 name = str(node["$ref"]).rsplit("/", 1)[-1]
174 if name in seen:
175 return None, seen
176 seen = seen | {name}
177 node = defs.get(name, {})
178 return node, seen
181def _candidates(node: object, defs: dict[str, object], seen: frozenset[str]) -> frozenset[str]:
182 """Return the stamped schema names a schema node may hold.
184 Parameters
185 ----------
186 node
187 Schema node describing a position.
188 defs
189 The document's ``$defs`` mapping.
190 seen
191 Definition names already followed on this branch.
193 Notes
194 -----
195 A node carrying the nested-schema URL key is itself a candidate.
196 Otherwise its ``anyOf`` and ``oneOf`` branches are unioned, which is what
197 turns a plain union of PSF models, or a discriminated union of field
198 models, into the set of schemas that position admits. A branch that is
199 ``null`` or unconstrained contributes nothing.
200 """
201 resolved, seen = _resolve(node, defs, seen)
202 if not isinstance(resolved, dict):
203 return frozenset()
204 if isinstance(url := resolved.get(_SCHEMA_URL_KEY), str):
205 return frozenset({_schema_name_from_url(url)})
206 found: set[str] = set()
207 for branch in (*resolved.get("anyOf", ()), *resolved.get("oneOf", ())):
208 found |= _candidates(branch, defs, seen)
209 return frozenset(found)
212def _declare(
213 node: object,
214 defs: dict[str, object],
215 prefix: str,
216 seen: frozenset[str],
217 paths: set[str],
218 positions: dict[str, frozenset[str]],
219 mappings: set[str],
220) -> None:
221 """Collect the paths and sub-schema positions a schema node declares.
223 Parameters
224 ----------
225 node
226 Schema node to walk.
227 defs
228 The document's ``$defs`` mapping.
229 prefix
230 Path of ``node`` within the document, empty at the root.
231 seen
232 Definition names already followed on this branch.
233 paths
234 Accumulator of declared paths.
235 positions
236 Accumulator mapping a path to the sub-schema names it admits.
237 mappings
238 Accumulator of paths whose children are data-keyed rather than
239 declared, so a credited path below one can be truncated back to it.
241 Notes
242 -----
243 A position that admits a stamped sub-schema is a boundary: it is recorded
244 and not descended into, because everything below it belongs to that
245 sub-schema and is credited there. This is the same boundary `_credit`
246 switches attribution at, which is what makes the two sides comparable.
247 """
248 resolved, seen = _resolve(node, defs, seen)
249 if not isinstance(resolved, dict):
250 return
251 for branch in (*resolved.get("anyOf", ()), *resolved.get("oneOf", ())):
252 _declare(branch, defs, prefix, seen, paths, positions, mappings)
253 for name, sub in resolved.get("properties", {}).items():
254 path = f"{prefix}.{name}"
255 paths.add(path)
256 if candidates := _candidates(sub, defs, seen):
257 positions[path] = candidates
258 continue
259 _declare(sub, defs, path, seen, paths, positions, mappings)
260 for keyword in ("items", "contains"):
261 if (items := resolved.get(keyword)) is not None:
262 path = f"{prefix}[]"
263 if candidates := _candidates(items, defs, seen):
264 positions[path] = candidates
265 else:
266 _declare(items, defs, path, seen, paths, positions, mappings)
267 for item in resolved.get("prefixItems", ()):
268 _declare(item, defs, f"{prefix}[]", seen, paths, positions, mappings)
269 if isinstance(values := resolved.get("additionalProperties"), dict):
270 mappings.add(prefix)
271 if candidates := _candidates(values, defs, seen):
272 positions[prefix] = candidates
275def _truncate(path: str, mappings: frozenset[str]) -> str:
276 """Return a credited path cut back to its nearest data-keyed ancestor.
278 Parameters
279 ----------
280 path
281 Credited path, which may descend through a mapping's data keys.
282 mappings
283 Paths whose children are data keys rather than declared properties.
285 Notes
286 -----
287 A mapping's keys are values chosen by whoever wrote the data, so they are
288 not schema positions and cannot be compared against declared paths. The
289 longest matching ancestor wins, so a mapping nested inside another is cut
290 at the inner one.
291 """
292 best = ""
293 for mapping in mappings:
294 if path.startswith(f"{mapping}.") and len(mapping) > len(best):
295 best = mapping
296 return best or path
299@dataclasses.dataclass(frozen=True)
300class SubSchemaPosition:
301 """One place a schema can hold another stamped schema."""
303 path: str
304 """Path of the position within its containing schema."""
306 candidates: frozenset[str]
307 """Schema names the position admits."""
309 reached: frozenset[str]
310 """Schema names some committed fixture actually put there."""
312 @property
313 def missing(self) -> frozenset[str]:
314 """Candidates no fixture ever put at this position."""
315 return self.candidates - self.reached
318@dataclasses.dataclass(frozen=True)
319class SchemaCoverage:
320 """What the fixture tree exercises of one schema version."""
322 name: str
323 """Schema name."""
325 version: str
326 """Fixture-filename version, e.g. ``1.0.0`` or ``1.0.0.dev``."""
328 expressed: frozenset[str]
329 """Declared paths that at least one fixture expresses."""
331 absent: frozenset[str]
332 """Declared paths no fixture expresses."""
334 positions: tuple[SubSchemaPosition, ...]
335 """Sub-schema positions, ordered by path."""
337 sources: frozenset[str]
338 """Names of the fixture files that credit this schema, whether as their
339 own top-level tree or by embedding it."""
341 @property
342 def absent_roots(self) -> frozenset[str]:
343 """The shallowest absent path of each absent subtree.
345 A field left unset takes its whole declared subtree with it, so
346 ``butler_info`` being `None` everywhere would otherwise report every
347 path beneath it as a separate gap. Only the outermost is actionable:
348 populate that field and its children become reachable.
349 """
350 return frozenset(
351 path
352 for path in self.absent
353 if not any(path.startswith(f"{other}.") or path.startswith(f"{other}[") for other in self.absent)
354 )
357@dataclasses.dataclass(frozen=True)
358class CoverageReport:
359 """Coverage of every schema in scope, keyed by name and version."""
361 schemas: dict[tuple[str, str], SchemaCoverage]
362 """Coverage per ``(name, version)``."""
364 @property
365 def positions_missing_candidates(self) -> tuple[tuple[str, SubSchemaPosition], ...]:
366 """Positions with an unreached candidate, as ``(schema, position)``."""
367 return tuple(
368 (f"{cov.name} {cov.version}", position)
369 for cov in self.schemas.values()
370 for position in cov.positions
371 if position.missing
372 )
375def _coverage_for(
376 tree_cls: type[ArchiveTree],
377 expressed: dict[tuple[str, str], set[str]],
378 embeddings: dict[tuple[str, str], dict[str, set[str]]],
379 sources: dict[tuple[str, str], set[str]],
380) -> SchemaCoverage:
381 """Build one schema's coverage from the credited walk results.
383 Parameters
384 ----------
385 tree_cls
386 Serialization model class to report on.
387 expressed
388 Paths credited to each schema.
389 embeddings
390 Sub-schema names credited at each path of each schema.
391 sources
392 Fixture filenames crediting each schema.
393 """
394 key = (tree_cls.SCHEMA_NAME, fixture_version(tree_cls.SCHEMA_VERSION))
395 document = dump_schema(tree_cls)
396 defs = document.get("$defs", {})
397 paths: set[str] = set()
398 positions: dict[str, frozenset[str]] = {}
399 mappings: set[str] = set()
400 _declare(document, defs, "", frozenset(), paths, positions, mappings)
401 frozen_mappings = frozenset(mappings)
402 seen = {_truncate(path, frozen_mappings) for path in expressed.get(key, ())}
403 reached: dict[str, set[str]] = defaultdict(set)
404 for path, names in embeddings.get(key, {}).items():
405 reached[_truncate(path, frozen_mappings)] |= names
406 declared = paths | set(positions)
407 return SchemaCoverage(
408 name=key[0],
409 version=key[1],
410 expressed=frozenset(declared & seen),
411 absent=frozenset(declared - seen),
412 positions=tuple(
413 SubSchemaPosition(
414 path=path,
415 candidates=candidates,
416 reached=frozenset(reached.get(path, ())),
417 )
418 for path, candidates in sorted(positions.items())
419 ),
420 sources=frozenset(sources.get(key, ())),
421 )
424def schema_coverage(directory: Path, *, package: str = "lsst.images") -> CoverageReport:
425 """Report what the committed fixtures exercise of each schema.
427 Parameters
428 ----------
429 directory
430 Directory holding the fixture tree.
431 package
432 Package whose schemas to report on; see
433 `~lsst.images.serialization.available_schema_classes`.
435 Returns
436 -------
437 `CoverageReport`
438 Coverage keyed by schema name and fixture-filename version.
440 Notes
441 -----
442 Every non-retired fixture in the tree contributes, whatever schema it is
443 a fixture *of*: coverage is credited to the schema that owns each stamped
444 subtree, so a sub-model embedded by several containers is credited by all
445 of them. Retired fixtures are excluded because they do not validate.
447 This reports reach, not judgement. A path counted as expressed only
448 proves some fixture put a value there, not that the value is interesting;
449 what pins payload data is described in
450 :ref:`lsst.images-schema-fixtures`.
451 """
452 expressed: dict[tuple[str, str], set[str]] = defaultdict(set)
453 embeddings: dict[tuple[str, str], dict[str, set[str]]] = defaultdict(dict)
454 sources: dict[tuple[str, str], set[str]] = defaultdict(set)
455 for fixture in iter_schema_fixtures(directory):
456 if fixture.retired:
457 continue
458 try:
459 content = json.loads(fixture.path.read_text())
460 except (OSError, UnicodeError, json.JSONDecodeError):
461 # A fixture that will not parse is check_schema_fixtures's problem
462 # to report; skipping it here keeps this function a pure reporter.
463 continue
464 if (key := _stamp_key(content)) is None: 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true
465 continue
466 # Walk into per-fixture accumulators first, so which schemas this one
467 # file credits is known exactly, then merge. A schema embedded by a
468 # container is credited by it even when it owns no fixture itself.
469 seen_paths: dict[tuple[str, str], set[str]] = defaultdict(set)
470 seen_embeddings: dict[tuple[str, str], dict[str, set[str]]] = defaultdict(dict)
471 _credit(content, key, "", seen_paths, seen_embeddings)
472 for credited, paths in seen_paths.items():
473 expressed[credited] |= paths
474 for credited, found in seen_embeddings.items():
475 for path, names in found.items():
476 embeddings[credited].setdefault(path, set()).update(names)
477 for credited in set(seen_paths) | set(seen_embeddings):
478 sources[credited].add(fixture.path.name)
479 return CoverageReport(
480 schemas={
481 (tree_cls.SCHEMA_NAME, fixture_version(tree_cls.SCHEMA_VERSION)): _coverage_for(
482 tree_cls, expressed, embeddings, sources
483 )
484 for tree_cls in available_schema_classes(package)
485 }
486 )
489def format_coverage_report(report: CoverageReport, *, schema: str | None = None) -> str:
490 """Render a coverage report as text.
492 Parameters
493 ----------
494 report
495 The report to render.
496 schema
497 Restrict the output to this schema name, or `None` for all of them.
499 Returns
500 -------
501 `str`
502 The rendered report, one block per schema.
504 Notes
505 -----
506 Absent paths are collapsed to the root of each absent subtree, and every
507 sub-schema position is listed whether or not it has a gap: a position that
508 reaches all its candidates is as much of an answer to "what does this
509 fixture set cover?" as one that does not. A gap is marked ``gap`` rather
510 than reported as an error, because a composite model is under no
511 obligation to hold every candidate its schema admits.
512 """
513 lines: list[str] = []
514 for key in sorted(report.schemas):
515 coverage = report.schemas[key]
516 if schema is not None and coverage.name != schema:
517 continue
518 declared = len(coverage.expressed) + len(coverage.absent)
519 lines.append(
520 f"{coverage.name} {coverage.version}: "
521 f"{len(coverage.expressed)}/{declared} paths, {len(coverage.sources)} fixture(s)"
522 )
523 if not coverage.sources:
524 lines.append(" no fixture reaches this schema")
525 for path in sorted(coverage.absent_roots):
526 lines.append(f" absent {path}")
527 for position in coverage.positions:
528 marker = "gap " if position.missing else "holds "
529 reached = ", ".join(sorted(position.reached)) or "nothing"
530 if position.candidates == position.reached:
531 lines.append(f" {marker}{position.path} [{reached}]")
532 else:
533 lines.append(
534 f" {marker}{position.path} [{reached}] of {{{', '.join(sorted(position.candidates))}}}"
535 )
536 return "\n".join(lines)