Viewing files at commit 523859ca17adCommit detailsView latest

Browse files
scripts/oakbench/fixture_registry.py 122 lines · 4.6 KB · python
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