| 1 | """Fixture registry loading and manifest verification.
|
| 2 |
|
| 3 | The registry is metadata, not result data: fixture ids, seeds, generator
|
| 4 | versions, hosting, and expected manifests are committed so rows can later
|
| 5 | point to reproducible fixture identity without committing generated bytes.
|
| 6 | """
|
| 7 |
|
| 8 | from __future__ import annotations
|
| 9 |
|
| 10 | import hashlib
|
| 11 | import json
|
| 12 | from dataclasses import dataclass
|
| 13 | from pathlib import Path
|
| 14 | from typing import Any
|
| 15 |
|
| 16 | ROOT = Path(__file__).resolve().parents[2]
|
| 17 | DEFAULT_FIXTURE_REGISTRY = ROOT / "config" / "fixtures.json"
|
| 18 |
|
| 19 |
|
| 20 | @dataclass(frozen=True)
|
| 21 | class FixtureSpec:
|
| 22 | fixture_id: str
|
| 23 | generator: str
|
| 24 | version: str
|
| 25 | seed: str
|
| 26 | description: str
|
| 27 | manifest: str | None
|
| 28 | license: str
|
| 29 | hosting: tuple[str, ...]
|
| 30 | status: str = "active"
|
| 31 |
|
| 32 | def row_fields(self) -> dict[str, Any]:
|
| 33 | return {
|
| 34 | "fixture_id": self.fixture_id,
|
| 35 | "fixture_generator": self.generator,
|
| 36 | "fixture_version": self.version,
|
| 37 | "fixture_seed": self.seed,
|
| 38 | "fixture_manifest_sha256": self.manifest,
|
| 39 | }
|
| 40 |
|
| 41 |
|
| 42 | def load_fixture_registry(path: Path = DEFAULT_FIXTURE_REGISTRY) -> dict[str, FixtureSpec]:
|
| 43 | raw = json.loads(path.read_text())
|
| 44 | if raw.get("schema_version") != 1:
|
| 45 | raise ValueError(f"unsupported fixture registry schema_version: {raw.get('schema_version')!r}")
|
| 46 | fixtures = raw.get("fixtures")
|
| 47 | if not isinstance(fixtures, dict):
|
| 48 | raise ValueError("fixture registry requires an object-valued 'fixtures' key")
|
| 49 | return {fixture_id: _parse_fixture(fixture_id, data) for fixture_id, data in fixtures.items()}
|
| 50 |
|
| 51 |
|
| 52 | def _parse_fixture(fixture_id: str, data: dict[str, Any]) -> FixtureSpec:
|
| 53 | missing = [
|
| 54 | key
|
| 55 | for key in ("generator", "version", "seed", "description", "license", "hosting")
|
| 56 | if key not in data
|
| 57 | ]
|
| 58 | if missing:
|
| 59 | raise ValueError(f"fixture {fixture_id!r} missing required keys: {', '.join(missing)}")
|
| 60 | hosting = data["hosting"]
|
| 61 | if not isinstance(hosting, list) or not all(isinstance(item, str) for item in hosting):
|
| 62 | raise ValueError(f"fixture {fixture_id!r} hosting must be a list of strings")
|
| 63 | manifest = data.get("manifest")
|
| 64 | if manifest is not None and (
|
| 65 | not isinstance(manifest, str) or len(manifest) != 64 or any(ch not in "0123456789abcdef" for ch in manifest)
|
| 66 | ):
|
| 67 | raise ValueError(f"fixture {fixture_id!r} manifest must be a lowercase sha256 hex string or null")
|
| 68 | return FixtureSpec(
|
| 69 | fixture_id=fixture_id,
|
| 70 | generator=str(data["generator"]),
|
| 71 | version=str(data["version"]),
|
| 72 | seed=str(data["seed"]),
|
| 73 | description=str(data["description"]),
|
| 74 | manifest=manifest,
|
| 75 | license=str(data["license"]),
|
| 76 | hosting=tuple(hosting),
|
| 77 | status=str(data.get("status", "active")),
|
| 78 | )
|
| 79 |
|
| 80 |
|
| 81 | def tree_manifest_sha256(root: Path) -> str:
|
| 82 | """Hash paths, file modes, and file contents under ``root`` deterministically."""
|
| 83 | digest = hashlib.sha256()
|
| 84 | for path in sorted(item for item in root.rglob("*") if item.is_file()):
|
| 85 | rel = path.relative_to(root).as_posix()
|
| 86 | mode = path.stat().st_mode & 0o777
|
| 87 | digest.update(f"{rel}\0{mode:o}\0".encode("utf-8"))
|
| 88 | with path.open("rb") as fh:
|
| 89 | for chunk in iter(lambda: fh.read(1024 * 1024), b""):
|
| 90 | digest.update(chunk)
|
| 91 | digest.update(b"\0")
|
| 92 | return digest.hexdigest()
|
| 93 |
|
| 94 |
|
| 95 | def verify_fixture(spec: FixtureSpec, root: Path) -> tuple[str, str | None]:
|
| 96 | """Verify a fixture tree against the spec's pinned manifest.
|
| 97 |
|
| 98 | Returns ``(status, actual_manifest)`` where status is one of:
|
| 99 |
|
| 100 | - ``"verified"``: manifest is pinned and the tree hash matches it.
|
| 101 | - ``"unverifiable:manifest_null"``: the tree exists and is non-empty but
|
| 102 | the registry pins no manifest; nothing can be verified (null means
|
| 103 | unmeasured, never a silent pass).
|
| 104 | - ``"failed:<reason>"``: the tree is missing (``path_missing``), not a
|
| 105 | directory (``not_a_directory``), empty (``tree_empty``), or its hash
|
| 106 | does not match the pinned manifest (``manifest_mismatch``).
|
| 107 |
|
| 108 | ``actual_manifest`` is ``None`` whenever no meaningful tree hash exists
|
| 109 | (missing, non-directory, or empty roots).
|
| 110 | """
|
| 111 | if not root.exists():
|
| 112 | return "failed:path_missing", None
|
| 113 | if not root.is_dir():
|
| 114 | return "failed:not_a_directory", None
|
| 115 | if not any(item.is_file() for item in root.rglob("*")):
|
| 116 | return "failed:tree_empty", None
|
| 117 | actual = tree_manifest_sha256(root)
|
| 118 | if spec.manifest is None:
|
| 119 | return "unverifiable:manifest_null", actual
|
| 120 | if actual == spec.manifest:
|
| 121 | return "verified", actual
|
| 122 | return "failed:manifest_mismatch", actual
|