scripts.test_green_ledger

Reference › scripts › test_green_ledger

scripts/test_green_ledger.py is the RUL-032 proven-content ledger, the first of three enforcement mechanisms named by that ruling: it records which test modules passed on exactly which content, so run_tests can reuse a pass instead of re-running a module whose inputs did not change. run_tests imports it, it imports affected_tests, and tests/scripts/test_test_green_ledger.py covers it. The ledger lives in reports/test-green-ledger.json.

The key of a module is a sha256 over the bytes of every file it depends on: the test module itself, its import closure (closure_for, built from the dependency graph and test import map by _build_closure_maps and cached), and the conftest.py files from the repository root down (_conftest_chain). key_inputs lists those paths in hash order and module_key hashes them. load_ledger reads the ledger, record_green records a passing module under its key, partition splits a list of modules into reused and to-run, and _write_atomic writes the file atomically. The command line takes exactly one of --status PATH... (report reused or to-run for each given module) and --seed-from REPORT (seed the ledger from a run_tests.py --json report such as reports/run-tests-latest.json).

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/test_green_ledger.py

#!/usr/bin/env python3
"""RUL-032 proven-content ledger — enforcement mechanism 1 of 3 named by
``rulings/RUL-032-tests-run-once-per-code-state.md`` (Mohamed, 2026-09-22,
verbatim): *"with the rule enforced that test only run once unless the code
in the test or code is changed, and watch for escalation to full suite when
the tests have already been run, this should never happen."*

A test module whose CONTENT KEY — itself, its ``src/`` import closure, its
``conftest.py`` chain, and ``pyproject.toml`` — exactly matches the key it
was last recorded GREEN at does not need to run again, whatever the git
identity ``scripts/run_tests.py``'s report-level reuse check
(``_try_reuse_report``) is doing: that check is HEAD-commit + dirty-tree
scoped (report-level, all-or-nothing); this one is module-level and
content-keyed, so a subset run still reuses everything it proves it can.

Key definition (the ruling, verbatim, its four inputs): sha256 over, in
order, the bytes of the test module; every ``src/`` file in its transitive
import closure (sorted by path — the same closure
``scripts/affected_tests.py`` walks); every ``conftest.py`` from the repo
root down to the module's own directory; and ``pyproject.toml``. Nothing
else — in particular an ``env/*.env`` edit never changes a module's key
(mechanism 3: ``scripts/affected_tests.py`` no longer escalates on one
either). A missing input (deleted file, moved module) is skipped, never
fatal — hashing simply omits it, which still changes the digest relative to
a run where it existed, so a genuinely stale key does not silently pass.

Ledger: ``reports/test-green-ledger.json`` (git-ignored, like the duration
ledgers — per-run/per-tree bookkeeping, not source) —
``{module: {"key", "passed", "run_id", "head_commit", "ts"}}``, written
atomically (tmp file + ``os.replace``).

CLI:
    python scripts/test_green_ledger.py --status tests/gateway/test_x.py ...
    python scripts/test_green_ledger.py --seed-from reports/run-tests-latest.json
"""

from __future__ import annotations

import argparse
import functools
import hashlib
import json
import os
import subprocess
import sys
import time
from collections.abc import Callable
from pathlib import Path
from typing import Any

_REPO_ROOT_FOR_IMPORT = Path(__file__).resolve().parent.parent
# Invoked directly (``python scripts/test_green_ledger.py``), Python puts
# THIS FILE'S OWN directory (REPO_ROOT/scripts) on sys.path[0] -- not
# REPO_ROOT -- so ``from scripts.affected_tests import ...`` below fails
# with "No module named 'scripts'" even though ``scripts/__init__.py``
# makes it a real package (same issue, same fix, as scripts/run_tests.py's
# module docstring documents). Harmless when already present (pytest's
# rootdir insertion, ``-m``, or a prior interpreter run already did this).
if str(_REPO_ROOT_FOR_IMPORT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT_FOR_IMPORT))

from scripts.affected_tests import (  # noqa: E402
    MAX_DEPTH,
    SRC_IMPORT_PREFIX,
    _load_ast,
    build_module_file_map,
    build_test_import_map,
    transitive_dependents,
)

REPO_ROOT = Path(__file__).resolve().parent.parent

#: The canonical ledger location. Every read/write function below takes an
#: explicit ``path``/override so a test run never has to mutate the real
#: ledger a concurrent ``run_tests.py`` invocation depends on.
LEDGER_PATH = REPO_ROOT / "reports" / "test-green-ledger.json"


def closure_for(
    test_path: str,
    *,
    repo_root: Path,
    ast_src_path: Path | None = None,
    ast_tests_path: Path | None = None,
    max_depth: int = MAX_DEPTH,
    load_ast: Callable[..., dict[str, Any]] = _load_ast,
) -> set[str]:
    """Repo-relative ``src/`` file paths in ``test_path``'s transitive import
    closure.

    ``scripts/affected_tests.py`` walks this same closure BACKWARDS: from a
    changed src module, BFS forward through ``reverse_dep`` (who imports the
    thing that imports the thing that changed) to find every module that
    could break, then looks up which tests import any of those. This walks
    it FORWARDS instead, from one test's own direct ``orchestrata_ultimate.*``
    imports, through the FORWARD dependency graph (a module's own
    dependencies) — the set of src files whose change would, by that same
    selector logic, cause it to select THIS test. Built entirely from the
    selector's own maps (``build_module_file_map``, ``build_test_import_map``,
    ``transitive_dependents`` — direction-agnostic BFS over an adjacency
    dict, handed the forward graph instead of the selector's reverse one) —
    no second import walker.

    A repo_root with no serialized AST to load and no way to regenerate one
    (a synthetic/sandboxed test tree that is not a full checkout) yields an
    EMPTY closure rather than raising -- ``module_key`` still hashes the
    test module itself, its conftest chain, and pyproject.toml, so a key
    computed this way is degraded (misses src-side changes) but never
    absent, and callers (``partition``, ``record_green``) never crash on a
    tree that legitimately has no closure to see.

    ``ast-ultimate.json``/``ast-tests.json`` are tens of MB on a real repo
    (measured 2026-09-22: 89MB/60MB) -- a bulk caller (``--seed-from`` over
    197 modules; ``partition``/``record_green`` over a whole batch) calling
    this once per module must not re-read and re-parse them from disk every
    time. When ``load_ast`` is left at its default, the loaded/derived maps
    are cached per ``(src path, tests path, repo_root)`` for the life of
    this process (``_cached_closure_maps``); a caller-supplied ``load_ast``
    (test injection) always re-reads, since it did not opt into caching.
    """
    src_path = ast_src_path or repo_root / "plans" / "ast-ultimate.json"
    tests_path = ast_tests_path or repo_root / "plans" / "ast-tests.json"
    try:
        if load_ast is _load_ast:
            dependency_graph, test_import_map, file_of = _cached_closure_maps(
                src_path, tests_path, repo_root
            )
        else:
            ast_src = load_ast(
                src_path,
                "src/orchestrata_ultimate",
                "orchestrata-ultimate",
                repo_root=repo_root,
            )
            ast_tests = load_ast(
                tests_path,
                "tests",
                "orchestrata-ultimate-tests",
                repo_root=repo_root,
            )
            dependency_graph, test_import_map, file_of = _build_closure_maps(
                ast_src, ast_tests, repo_root=repo_root
            )
    except OSError, subprocess.CalledProcessError, json.JSONDecodeError:
        return set()
    # Invert test_import_map for this ONE test: the dotted modules it
    # directly imports are exactly the keys whose value set contains it.
    direct = {dotted for dotted, tests in test_import_map.items() if test_path in tests}
    forward_graph = {mod: set(deps) for mod, deps in dependency_graph.items()}
    closure_dotted: set[str] = set()
    for start in direct:
        closure_dotted |= transitive_dependents(start, forward_graph, max_depth)
    return {file_of[d] for d in closure_dotted if d in file_of}


def _build_closure_maps(
    ast_src: dict[str, Any], ast_tests: dict[str, Any], *, repo_root: Path
) -> tuple[dict[str, list[str]], dict[str, set[str]], dict[str, str]]:
    """``(dependency_graph, test_import_map, {dotted module: file path})``
    from already-loaded ``ast_src``/``ast_tests`` dicts -- the part of
    ``closure_for`` that is pure computation, shared by the cached and the
    test-injected-loader paths so there is exactly one place this map
    construction happens."""
    module_file_map = build_module_file_map(ast_src, repo_root=repo_root)
    dependency_graph: dict[str, list[str]] = ast_src.get("dependency_graph", {})
    internal_modules = frozenset(dependency_graph)
    test_import_map = build_test_import_map(
        ast_tests, src_prefix=SRC_IMPORT_PREFIX, internal_modules=internal_modules
    )
    file_of = {dotted: path for path, dotted in module_file_map.items()}
    return dependency_graph, test_import_map, file_of


@functools.lru_cache(maxsize=4)
def _cached_closure_maps(
    ast_src_path: Path, ast_tests_path: Path, repo_root: Path
) -> tuple[dict[str, list[str]], dict[str, set[str]], dict[str, str]]:
    ast_src = _load_ast(
        ast_src_path,
        "src/orchestrata_ultimate",
        "orchestrata-ultimate",
        repo_root=repo_root,
    )
    ast_tests = _load_ast(
        ast_tests_path,
        "tests",
        "orchestrata-ultimate-tests",
        repo_root=repo_root,
    )
    return _build_closure_maps(ast_src, ast_tests, repo_root=repo_root)


def _conftest_chain(test_path: str) -> list[str]:
    """Repo-relative ``conftest.py`` candidate paths from the repo root down
    to ``test_path``'s own directory, root-first. Candidates, not existence
    checks — a directory with no conftest simply contributes nothing when
    hashed (missing inputs are skipped, never fatal)."""
    chain = ["conftest.py"]
    cur = Path()
    for part in Path(test_path).parent.parts:
        cur = cur / part
        chain.append(str(cur / "conftest.py"))
    return chain


def key_inputs(test_path: str, *, closure: set[str]) -> list[str]:
    """Every repo-relative path ``module_key`` hashes, in the exact order it
    hashes them. Exposed separately so ``--seed-from`` can ask "did any of
    THIS module's inputs change since the recorded run" without
    recomputing (and needing file bytes for) the digest itself."""
    return [test_path, *sorted(closure), *_conftest_chain(test_path), "pyproject.toml"]


def module_key(test_path: str, *, repo_root: Path, closure: set[str]) -> str:
    """sha256 over the concatenated bytes of ``key_inputs`` (test module;
    closure src files sorted by path; conftest chain; pyproject.toml), in
    that order. A missing input is skipped, never fatal."""
    h = hashlib.sha256()
    for rel in key_inputs(test_path, closure=closure):
        try:
            h.update((repo_root / rel).read_bytes())
        except OSError:
            continue
    return h.hexdigest()


def load_ledger(path: Path | None = None) -> dict[str, dict[str, Any]]:
    """The ledger dict, or ``{}`` when the file is absent/unreadable/corrupt
    — the ledger is a cache, never a source of truth strong enough to be
    worth failing a run over."""
    p = path or LEDGER_PATH
    try:
        return json.loads(p.read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        return {}


def _write_atomic(path: Path, doc: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(json.dumps(doc, indent=2) + "\n", encoding="utf-8")
    os.replace(tmp, path)


def record_green(
    modules_with_counts: dict[str, int],
    run_id: str,
    head: str,
    *,
    repo_root: Path | None = None,
    path: Path | None = None,
) -> None:
    """Stamp every module in ``modules_with_counts`` green at its CURRENT
    content key. Called by ``scripts/run_tests.py`` after a batch/solo round
    for every module that passed, and by ``--seed-from`` for every module a
    prior report proves is still green."""
    repo_root = repo_root if repo_root is not None else REPO_ROOT
    p = path or LEDGER_PATH
    ledger = load_ledger(p)
    now = time.time()
    for test_path, passed in modules_with_counts.items():
        closure = closure_for(test_path, repo_root=repo_root)
        key = module_key(test_path, repo_root=repo_root, closure=closure)
        ledger[test_path] = {
            "key": key,
            "passed": int(passed),
            "run_id": run_id,
            "head_commit": head,
            "ts": now,
        }
    _write_atomic(p, ledger)


def partition(
    modules: list[str], *, repo_root: Path | None = None, path: Path | None = None
) -> tuple[list[str], dict[str, dict[str, Any]]]:
    """Split ``modules`` into ``(to_run, reused)``. ``reused`` maps module ->
    its recorded ledger entry, for every module whose CURRENT content key
    matches the key it was last recorded green at; everything else — never
    recorded, or recorded at a stale key — goes to ``to_run``."""
    repo_root = repo_root if repo_root is not None else REPO_ROOT
    ledger = load_ledger(path)
    to_run: list[str] = []
    reused: dict[str, dict[str, Any]] = {}
    for m in modules:
        entry = ledger.get(m)
        if entry is None:
            to_run.append(m)
            continue
        closure = closure_for(m, repo_root=repo_root)
        key = module_key(m, repo_root=repo_root, closure=closure)
        if key == entry.get("key"):
            reused[m] = entry
        else:
            to_run.append(m)
    return to_run, reused


def _cmd_status(paths: list[str], *, repo_root: Path | None = None) -> int:
    repo_root = repo_root if repo_root is not None else REPO_ROOT
    to_run, reused = partition(paths, repo_root=repo_root)
    for m in paths:
        if m in reused:
            entry = reused[m]
            head = str(entry.get("head_commit") or "")[:8]
            print(
                f"[test_green_ledger] reused  {m}  "
                f"(passed={entry.get('passed')}, head={head or 'unknown'})"
            )
        else:
            print(f"[test_green_ledger] run     {m}")
    print(f"[test_green_ledger] {len(reused)} reused, {len(to_run)} to run")
    return 0


def _is_ancestor(candidate: str, ref: str, *, repo_root: Path) -> bool:
    result = subprocess.run(
        ["git", "merge-base", "--is-ancestor", candidate, ref],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        check=False,
    )
    return result.returncode == 0


def _changed_since(rev: str, *, repo_root: Path) -> set[str] | None:
    result = subprocess.run(
        ["git", "diff", "--name-only", f"{rev}..HEAD"],
        cwd=str(repo_root),
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return None
    return {line.strip() for line in result.stdout.splitlines() if line.strip()}


def _cmd_seed_from(
    report_path: Path,
    *,
    repo_root: Path | None = None,
    ledger_path: Path | None = None,
) -> int:
    """Record every module ``report_path`` (a ``run_tests.py --json`` report)
    proved green as green in the ledger — but ONLY when the report's
    ``head_commit`` is an ancestor of HEAD and nothing that has changed
    since touches that module's own key inputs. A module that fails either
    check is refused and named, not silently dropped."""
    repo_root = repo_root if repo_root is not None else REPO_ROOT
    try:
        report = json.loads(report_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        print(f"[test_green_ledger] cannot read {report_path}: {exc}", flush=True)
        return 1
    head = report.get("head_commit")
    counts: dict[str, int] = report.get("module_test_counts") or {}
    if not head:
        print(
            f"[test_green_ledger] refused ALL {len(counts)} module(s) — "
            f"{report_path} has no head_commit",
            flush=True,
        )
        return 1
    if not _is_ancestor(head, "HEAD", repo_root=repo_root):
        print(
            f"[test_green_ledger] refused ALL {len(counts)} module(s) — "
            f"head_commit {head} is not an ancestor of HEAD",
            flush=True,
        )
        return 1
    changed = _changed_since(head, repo_root=repo_root)
    if changed is None:
        print(
            f"[test_green_ledger] refused ALL {len(counts)} module(s) — "
            f"`git diff {head}..HEAD` failed",
            flush=True,
        )
        return 1
    seeded: dict[str, int] = {}
    refused: list[tuple[str, str]] = []
    for test_path, count in sorted(counts.items()):
        if not (repo_root / test_path).exists():
            refused.append((test_path, "no longer exists on disk"))
            continue
        closure = closure_for(test_path, repo_root=repo_root)
        inputs = set(key_inputs(test_path, closure=closure))
        touched = sorted(changed & inputs)
        if touched:
            refused.append(
                (
                    test_path,
                    f"key input(s) changed since {head[:8]}: {', '.join(touched)}",
                )
            )
            continue
        seeded[test_path] = int(count)
    if seeded:
        record_green(
            seeded,
            run_id=f"seed:{report_path.name}",
            head=head,
            repo_root=repo_root,
            path=ledger_path,
        )
    print(
        f"[test_green_ledger] seeded {len(seeded)}/{len(counts)} module(s) "
        f"from {report_path} at head {head[:8]}",
        flush=True,
    )
    for test_path, reason in refused:
        print(f"[test_green_ledger] refused {test_path}: {reason}", flush=True)
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="RUL-032 proven-content ledger — reports/test-green-ledger.json"
    )
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument(
        "--status",
        nargs="+",
        metavar="PATH",
        help="Report reused/to-run for each given test module path.",
    )
    group.add_argument(
        "--seed-from",
        metavar="REPORT",
        help="Seed the ledger from a run_tests.py --json report "
        "(e.g. reports/run-tests-latest.json).",
    )
    args = parser.parse_args(argv)
    if args.status:
        return _cmd_status(args.status)
    return _cmd_seed_from(Path(args.seed_from))


if __name__ == "__main__":
    sys.exit(main())