content-find-agent-new/tests/snapshot.py

"""Tiny golden-snapshot helper for case-replay tests (V2-M0C).

Zero third-party deps (no syrupy). A snapshot stores an approved JSON projection
of replay output; later runs compare against it and fail on drift. Regenerate
intentionally with UPDATE_SNAPSHOTS=1.

    from tests.snapshot import assert_matches
    assert_matches("real_id45/decision_summary", summary, subset_keys=["rejected_content_count"])

Golden files live under tests/fixtures/snapshots/{name}.json.
"""

from __future__ import annotations

import json
import os
from pathlib import Path
from typing import Any

SNAPSHOT_DIR = Path("tests/fixtures/snapshots")


def _updating() -> bool:
    return os.environ.get("UPDATE_SNAPSHOTS", "") not in {"", "0", "false", "False"}


def _canonical(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, indent=2, sort_keys=True)


def _project(actual: Any, subset_keys: list[str] | None) -> Any:
    """Keep only the given dot-path keys (avoids snapshotting volatile fields)."""
    if subset_keys is None:
        return actual
    out: dict[str, Any] = {}
    for dotted in subset_keys:
        cur: Any = actual
        for part in dotted.split("."):
            cur = cur.get(part) if isinstance(cur, dict) else None
        out[dotted] = cur
    return out


def _first_diff(expected: str, actual: str) -> str:
    exp_lines = expected.splitlines()
    act_lines = actual.splitlines()
    for i in range(max(len(exp_lines), len(act_lines))):
        e = exp_lines[i] if i < len(exp_lines) else "<missing>"
        a = act_lines[i] if i < len(act_lines) else "<missing>"
        if e != a:
            return f"line {i + 1}:\n  expected: {e}\n  actual:   {a}"
    return "<no line diff>"


def assert_matches(name: str, actual: Any, *, subset_keys: list[str] | None = None) -> None:
    golden = SNAPSHOT_DIR / f"{name}.json"
    projected = _canonical(_project(actual, subset_keys))

    if not golden.exists():
        if _updating():
            golden.parent.mkdir(parents=True, exist_ok=True)
            golden.write_text(projected + "\n", encoding="utf-8")
            return
        raise AssertionError(
            f"snapshot missing: {golden}. Run with UPDATE_SNAPSHOTS=1 to create it."
        )

    expected = golden.read_text(encoding="utf-8").rstrip("\n")
    if projected == expected:
        return
    if _updating():
        golden.write_text(projected + "\n", encoding="utf-8")
        return
    raise AssertionError(
        f"snapshot mismatch for {name}:\n{_first_diff(expected, projected)}\n"
        f"(run UPDATE_SNAPSHOTS=1 to accept the new output)"
    )