| 1 |
"""Repo-level index of checked-in determinism goldens. |
| 2 |
|
| 3 |
Separate from the per-store `dlm.lock`: this file tracks which |
| 4 |
runtime tuples have an approved golden under `tests/golden/determinism/`. |
| 5 |
The canonical path is `.determinism/lock.json`. |
| 6 |
""" |
| 7 |
|
| 8 |
from __future__ import annotations |
| 9 |
|
| 10 |
import json |
| 11 |
from collections.abc import Mapping |
| 12 |
from datetime import UTC, datetime |
| 13 |
from pathlib import Path |
| 14 |
from typing import Final |
| 15 |
|
| 16 |
from pydantic import BaseModel, ConfigDict, Field |
| 17 |
|
| 18 |
from dlm.io.atomic import write_text |
| 19 |
from dlm.lock.errors import GoldenIndexSchemaError, GoldenIndexWriteError |
| 20 |
|
| 21 |
GOLDEN_INDEX_RELATIVE_PATH: Final[str] = ".determinism/lock.json" |
| 22 |
CURRENT_GOLDEN_INDEX_VERSION: Final[int] = 1 |
| 23 |
|
| 24 |
|
| 25 |
class DeterminismGoldenEntry(BaseModel): |
| 26 |
"""One approved tuple golden tracked at repo scope.""" |
| 27 |
|
| 28 |
model_config = ConfigDict(extra="forbid", frozen=True) |
| 29 |
|
| 30 |
golden_relpath: str = Field( |
| 31 |
..., |
| 32 |
pattern=r"^tests/golden/determinism/tuple-[0-9a-f]{16}\.json$", |
| 33 |
) |
| 34 |
adapter_sha256: str = Field(..., pattern=r"^[0-9a-f]{64}$") |
| 35 |
platform: str = Field(..., min_length=1) |
| 36 |
pinned_versions: dict[str, str] = Field(default_factory=dict) |
| 37 |
|
| 38 |
|
| 39 |
class DeterminismGoldenIndex(BaseModel): |
| 40 |
"""Checked-in set of approved determinism goldens.""" |
| 41 |
|
| 42 |
model_config = ConfigDict(extra="forbid", frozen=True) |
| 43 |
|
| 44 |
lock_version: int = Field(CURRENT_GOLDEN_INDEX_VERSION, ge=1) |
| 45 |
updated_at: datetime |
| 46 |
goldens: tuple[DeterminismGoldenEntry, ...] = () |
| 47 |
|
| 48 |
|
| 49 |
def golden_index_path(repo_root: Path) -> Path: |
| 50 |
"""Return `<repo_root>/.determinism/lock.json`.""" |
| 51 |
|
| 52 |
return repo_root / GOLDEN_INDEX_RELATIVE_PATH |
| 53 |
|
| 54 |
|
| 55 |
def write_golden_index(repo_root: Path, index: DeterminismGoldenIndex) -> Path: |
| 56 |
"""Atomically persist the repo-level determinism-golden index.""" |
| 57 |
|
| 58 |
target = golden_index_path(repo_root) |
| 59 |
if index.lock_version != CURRENT_GOLDEN_INDEX_VERSION: |
| 60 |
raise GoldenIndexWriteError( |
| 61 |
path=target, |
| 62 |
reason=( |
| 63 |
f"lock_version={index.lock_version!r} != writer's " |
| 64 |
f"CURRENT_GOLDEN_INDEX_VERSION={CURRENT_GOLDEN_INDEX_VERSION}" |
| 65 |
), |
| 66 |
) |
| 67 |
target.parent.mkdir(parents=True, exist_ok=True) |
| 68 |
payload = index.model_dump(mode="json") |
| 69 |
text = json.dumps(payload, indent=2, sort_keys=True) + "\n" |
| 70 |
write_text(target, text) |
| 71 |
return target |
| 72 |
|
| 73 |
|
| 74 |
def load_golden_index(repo_root: Path) -> DeterminismGoldenIndex | None: |
| 75 |
"""Read `.determinism/lock.json`, returning `None` when absent.""" |
| 76 |
|
| 77 |
path = golden_index_path(repo_root) |
| 78 |
if not path.is_file(): |
| 79 |
return None |
| 80 |
|
| 81 |
try: |
| 82 |
raw = path.read_text(encoding="utf-8") |
| 83 |
except OSError as exc: |
| 84 |
raise GoldenIndexSchemaError(path, f"unreadable: {exc}") from exc |
| 85 |
|
| 86 |
try: |
| 87 |
payload = json.loads(raw) |
| 88 |
except json.JSONDecodeError as exc: |
| 89 |
raise GoldenIndexSchemaError(path, f"invalid JSON: {exc}") from exc |
| 90 |
|
| 91 |
if not isinstance(payload, dict): |
| 92 |
raise GoldenIndexSchemaError( |
| 93 |
path, |
| 94 |
f"top-level JSON must be an object, got {type(payload).__name__}", |
| 95 |
) |
| 96 |
|
| 97 |
version = payload.get("lock_version") |
| 98 |
if version != CURRENT_GOLDEN_INDEX_VERSION: |
| 99 |
raise GoldenIndexSchemaError( |
| 100 |
path, |
| 101 |
f"unsupported lock_version {version!r} (reader expects {CURRENT_GOLDEN_INDEX_VERSION})", |
| 102 |
) |
| 103 |
|
| 104 |
try: |
| 105 |
return DeterminismGoldenIndex.model_validate(payload) |
| 106 |
except Exception as exc: |
| 107 |
raise GoldenIndexSchemaError(path, f"schema validation: {exc}") from exc |
| 108 |
|
| 109 |
|
| 110 |
def upsert_golden_index( |
| 111 |
repo_root: Path, |
| 112 |
*, |
| 113 |
golden_relpath: str, |
| 114 |
adapter_sha256: str, |
| 115 |
platform: str, |
| 116 |
pinned_versions: Mapping[str, str], |
| 117 |
) -> Path: |
| 118 |
"""Insert or replace one tuple golden in `.determinism/lock.json`.""" |
| 119 |
|
| 120 |
current = load_golden_index(repo_root) |
| 121 |
entries = {} if current is None else {entry.golden_relpath: entry for entry in current.goldens} |
| 122 |
entries[golden_relpath] = DeterminismGoldenEntry( |
| 123 |
golden_relpath=golden_relpath, |
| 124 |
adapter_sha256=adapter_sha256, |
| 125 |
platform=platform, |
| 126 |
pinned_versions=dict(sorted(pinned_versions.items())), |
| 127 |
) |
| 128 |
updated = DeterminismGoldenIndex( |
| 129 |
updated_at=_utcnow(), |
| 130 |
goldens=tuple(sorted(entries.values(), key=lambda entry: entry.golden_relpath)), |
| 131 |
) |
| 132 |
return write_golden_index(repo_root, updated) |
| 133 |
|
| 134 |
|
| 135 |
def _utcnow() -> datetime: |
| 136 |
return datetime.now(UTC).replace(tzinfo=None, microsecond=0) |