C.E.L_Slide_test2/tests/phase_z2_ai_fallback/test_cache.py

"""IMP-46 u2 — Persistent JSON cache backend tests.

Scope (Stage 2 plan, u2):

* Replaced ``NotImplementedError`` marker with a real persistent backend
  at ``data/frame_cache/{frame_id}/{signature_hash}.json``.
* Preserved IMP-33 u6 dual write gate: ``visual_check_passed`` AND
  ``user_approved`` BOTH required (loud :class:`AiFallbackCacheGateError`
  before any filesystem touch).
* Round-trip every :class:`ProposalKind`; round-trip ``slide_css`` None
  *and* set; missing or corrupt files miss silently.
* Fingerprint *comparison* is u3; here we only check that the field is
  persisted.

All filesystem writes are scoped to ``tmp_path`` via
``monkeypatch.setattr`` on the module-level :data:`CACHE_ROOT`, so the
production directory is never touched by these tests.
"""
from __future__ import annotations

import json
import pathlib

import pytest

from src.phase_z2_ai_fallback import cache as cache_mod
from src.phase_z2_ai_fallback.cache import (
    AiFallbackCacheGateError,
    KEY_DELIMITER,
    SCHEMA_VERSION,
    read_proposal,
    save_proposal,
)
from src.phase_z2_ai_fallback.schema import AiFallbackProposal, ProposalKind


_FRAME_ID = "1171281190"
_SIG_HASH = "a" * 64  # SHA256-shaped placeholder; cache is shape-agnostic.
_KEY = f"{_FRAME_ID}{KEY_DELIMITER}{_SIG_HASH}"


def _proposal(
    kind: ProposalKind = ProposalKind.BUILDER_OPTIONS_PATCH,
    payload: dict | None = None,
) -> AiFallbackProposal:
    return AiFallbackProposal(
        proposal_kind=kind,
        payload=payload if payload is not None else {"item_parser": "bullet_v2"},
        rationale="u2-test",
    )


@pytest.fixture(autouse=True)
def _isolated_cache_root(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch):
    """Redirect the cache root to an isolated tmp directory for every test."""
    monkeypatch.setattr(cache_mod, "CACHE_ROOT", tmp_path / "frame_cache")
    yield tmp_path / "frame_cache"


# -- read_proposal --------------------------------------------------------


def test_read_proposal_returns_none_for_missing_file():
    assert read_proposal(_KEY) is None


def test_read_proposal_rejects_empty_key():
    with pytest.raises(ValueError):
        read_proposal("")


def test_read_proposal_rejects_non_string_key():
    with pytest.raises(ValueError):
        read_proposal(None)  # type: ignore[arg-type]


def test_read_proposal_returns_none_for_legacy_key_format():
    """Router back-compat: pre-u4 cache_key (no '::') misses silently."""
    assert read_proposal("frame:1171281190:cardinality:many") is None


def test_read_proposal_returns_none_for_corrupt_json(_isolated_cache_root: pathlib.Path):
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("{not valid json", encoding="utf-8")
    assert read_proposal(_KEY) is None


def test_read_proposal_returns_none_for_non_dict_root(_isolated_cache_root: pathlib.Path):
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("[]", encoding="utf-8")
    assert read_proposal(_KEY) is None


def test_read_proposal_returns_none_when_payload_proposal_missing(
    _isolated_cache_root: pathlib.Path,
):
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps({"schema_version": 1}), encoding="utf-8")
    assert read_proposal(_KEY) is None


def test_read_proposal_returns_none_for_forbidden_proposal_kind(
    _isolated_cache_root: pathlib.Path,
):
    path = _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "proposal": {"proposal_kind": "mdx_text", "payload": {}, "rationale": ""},
                "slide_css": None,
                "fingerprints": {},
            }
        ),
        encoding="utf-8",
    )
    assert read_proposal(_KEY) is None


# -- save_proposal: write gates -------------------------------------------


def test_save_rejects_when_visual_check_failed():
    with pytest.raises(AiFallbackCacheGateError) as exc:
        save_proposal(
            _KEY, _proposal(), visual_check_passed=False, user_approved=True
        )
    assert "visual_check_passed" in str(exc.value)


def test_save_rejects_when_user_not_approved():
    with pytest.raises(AiFallbackCacheGateError) as exc:
        save_proposal(
            _KEY, _proposal(), visual_check_passed=True, user_approved=False
        )
    assert "user_approved" in str(exc.value)


def test_save_rejects_when_both_gates_false():
    with pytest.raises(AiFallbackCacheGateError):
        save_proposal(
            _KEY, _proposal(), visual_check_passed=False, user_approved=False
        )


def test_save_gate_violation_does_not_touch_filesystem(
    _isolated_cache_root: pathlib.Path,
):
    with pytest.raises(AiFallbackCacheGateError):
        save_proposal(
            _KEY, _proposal(), visual_check_passed=False, user_approved=True
        )
    # Cache root may or may not exist depending on fixture order, but the
    # frame_id directory must NOT exist when the gate rejects the write.
    assert not (_isolated_cache_root / _FRAME_ID).exists()


def test_save_rejects_empty_key():
    with pytest.raises(ValueError):
        save_proposal(
            "", _proposal(), visual_check_passed=True, user_approved=True
        )


def test_save_rejects_non_proposal_object():
    with pytest.raises(TypeError):
        save_proposal(
            _KEY,
            {"proposal_kind": "builder_options_patch"},  # type: ignore[arg-type]
            visual_check_passed=True,
            user_approved=True,
        )


def test_save_rejects_legacy_key_format():
    """Writes must use the structural ``frame_id::signature_hash`` form."""
    with pytest.raises(ValueError):
        save_proposal(
            "frame:1171281190:cardinality:many",
            _proposal(),
            visual_check_passed=True,
            user_approved=True,
        )


def test_save_rejects_slide_css_non_string():
    with pytest.raises(TypeError):
        save_proposal(
            _KEY,
            _proposal(),
            visual_check_passed=True,
            user_approved=True,
            slide_css=123,  # type: ignore[arg-type]
        )


def test_save_rejects_fingerprints_non_dict():
    with pytest.raises(TypeError):
        save_proposal(
            _KEY,
            _proposal(),
            visual_check_passed=True,
            user_approved=True,
            fingerprints=["contract_sha", "abc"],  # type: ignore[arg-type]
        )


def test_gate_error_is_not_notimplementederror():
    """The persistent backend no longer raises ``NotImplementedError`` —
    callers must distinguish gate violation from absent persistence."""
    assert not issubclass(AiFallbackCacheGateError, NotImplementedError)


# -- save_proposal: persistence + round-trip ------------------------------


def test_save_creates_parent_directories(_isolated_cache_root: pathlib.Path):
    assert not (_isolated_cache_root / _FRAME_ID).exists()
    save_proposal(
        _KEY, _proposal(), visual_check_passed=True, user_approved=True
    )
    assert (_isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json").is_file()


def test_save_returns_resolved_path(_isolated_cache_root: pathlib.Path):
    path = save_proposal(
        _KEY, _proposal(), visual_check_passed=True, user_approved=True
    )
    assert path == _isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json"


def test_save_payload_includes_schema_version(_isolated_cache_root: pathlib.Path):
    path = save_proposal(
        _KEY, _proposal(), visual_check_passed=True, user_approved=True
    )
    data = json.loads(path.read_text(encoding="utf-8"))
    assert data["schema_version"] == SCHEMA_VERSION


def test_save_payload_includes_proposal_dump(_isolated_cache_root: pathlib.Path):
    proposal = _proposal(payload={"item_parser": "pillar_item"})
    path = save_proposal(
        _KEY, proposal, visual_check_passed=True, user_approved=True
    )
    data = json.loads(path.read_text(encoding="utf-8"))
    assert data["proposal"] == proposal.model_dump(mode="json")


def test_round_trip_default_slide_css_is_none(_isolated_cache_root: pathlib.Path):
    path = save_proposal(
        _KEY, _proposal(), visual_check_passed=True, user_approved=True
    )
    data = json.loads(path.read_text(encoding="utf-8"))
    assert data["slide_css"] is None
    assert data["fingerprints"] == {}


def test_round_trip_with_slide_css_set(_isolated_cache_root: pathlib.Path):
    css = ".slide { padding: 40px; }"
    path = save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        slide_css=css,
    )
    data = json.loads(path.read_text(encoding="utf-8"))
    assert data["slide_css"] == css


def test_round_trip_with_fingerprints(_isolated_cache_root: pathlib.Path):
    fingerprints = {
        "contract_sha": "c" * 64,
        "partial_sha": "p" * 64,
        "catalog_sha": "x" * 64,
    }
    path = save_proposal(
        _KEY,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
        fingerprints=fingerprints,
    )
    data = json.loads(path.read_text(encoding="utf-8"))
    assert data["fingerprints"] == fingerprints


def test_read_returns_proposal_after_save(_isolated_cache_root: pathlib.Path):
    original = _proposal(payload={"key": "value"})
    save_proposal(
        _KEY, original, visual_check_passed=True, user_approved=True
    )
    loaded = read_proposal(_KEY)
    assert loaded is not None
    assert loaded.proposal_kind == original.proposal_kind
    assert loaded.payload == original.payload
    assert loaded.rationale == original.rationale


@pytest.mark.parametrize("kind", list(ProposalKind))
def test_round_trip_all_proposal_kinds(
    kind: ProposalKind, _isolated_cache_root: pathlib.Path
):
    """Every whitelisted ProposalKind survives save → read unchanged."""
    if kind is ProposalKind.PARTIAL_OVERRIDES:
        payload = {"slots": {"pillar_1": "alpha"}}
    elif kind is ProposalKind.SLOT_MAPPING_PROPOSAL:
        payload = {"mapping": [{"from": "a", "to": "b"}]}
    else:
        payload = {"item_parser": "bullet_v2"}
    save_proposal(
        _KEY,
        _proposal(kind=kind, payload=payload),
        visual_check_passed=True,
        user_approved=True,
    )
    loaded = read_proposal(_KEY)
    assert loaded is not None
    assert loaded.proposal_kind is kind
    assert loaded.payload == payload


def test_save_overwrites_existing_entry(_isolated_cache_root: pathlib.Path):
    save_proposal(
        _KEY,
        _proposal(payload={"v": 1}),
        visual_check_passed=True,
        user_approved=True,
    )
    save_proposal(
        _KEY,
        _proposal(payload={"v": 2}),
        visual_check_passed=True,
        user_approved=True,
    )
    loaded = read_proposal(_KEY)
    assert loaded is not None
    assert loaded.payload == {"v": 2}


def test_file_layout_uses_frame_id_directory(_isolated_cache_root: pathlib.Path):
    """Storage layout = ``frame_id/`` directory, ``signature_hash.json`` file."""
    other_frame_key = f"{_FRAME_ID}_other{KEY_DELIMITER}{_SIG_HASH}"
    save_proposal(
        _KEY, _proposal(), visual_check_passed=True, user_approved=True
    )
    save_proposal(
        other_frame_key,
        _proposal(),
        visual_check_passed=True,
        user_approved=True,
    )
    assert (_isolated_cache_root / _FRAME_ID / f"{_SIG_HASH}.json").is_file()
    assert (
        _isolated_cache_root / f"{_FRAME_ID}_other" / f"{_SIG_HASH}.json"
    ).is_file()


def test_different_signature_hashes_isolated(_isolated_cache_root: pathlib.Path):
    """Two distinct signature hashes under the same frame_id never collide."""
    key_a = f"{_FRAME_ID}{KEY_DELIMITER}{'a' * 64}"
    key_b = f"{_FRAME_ID}{KEY_DELIMITER}{'b' * 64}"
    save_proposal(
        key_a,
        _proposal(payload={"sig": "a"}),
        visual_check_passed=True,
        user_approved=True,
    )
    save_proposal(
        key_b,
        _proposal(payload={"sig": "b"}),
        visual_check_passed=True,
        user_approved=True,
    )
    loaded_a = read_proposal(key_a)
    loaded_b = read_proposal(key_b)
    assert loaded_a is not None and loaded_a.payload == {"sig": "a"}
    assert loaded_b is not None and loaded_b.payload == {"sig": "b"}


def test_parse_key_rejects_triple_delimiter():
    """Two ``::`` markers (extra delimiter inside signature) is rejected."""
    assert (
        read_proposal(
            f"{_FRAME_ID}{KEY_DELIMITER}{_SIG_HASH}{KEY_DELIMITER}extra"
        )
        is None
    )


# -- IMP-46 u5: auto_cache gate (2^3 truth table) -------------------------
#
# Three booleans: visual_check_passed (V), user_approved (U), auto_cache (A).
# Contract: V=True AND (U=True OR A=True) -> persist; else gate-raise.
# V is never bypassable; A=True only relaxes U=False.

_GATE_TRUTH_TABLE = [
    # (V,     U,     A,     expect_persist)
    (False, False, False, False),
    (False, False, True,  False),
    (False, True,  False, False),
    (False, True,  True,  False),
    (True,  False, False, False),
    (True,  False, True,  True),
    (True,  True,  False, True),
    (True,  True,  True,  True),
]


@pytest.mark.parametrize("v,u,a,expect_persist", _GATE_TRUTH_TABLE)
def test_save_gate_truth_table(
    v: bool,
    u: bool,
    a: bool,
    expect_persist: bool,
    _isolated_cache_root: pathlib.Path,
) -> None:
    """IMP-46 u5 — exhaustive 2^3 enumeration of (V, U, A) -> {persist, raise}."""
    if expect_persist:
        path = save_proposal(
            _KEY,
            _proposal(payload={"v": int(v), "u": int(u), "a": int(a)}),
            visual_check_passed=v,
            user_approved=u,
            auto_cache=a,
        )
        assert path.is_file(), f"truth row (V={v}, U={u}, A={a}) must persist"
    else:
        with pytest.raises(AiFallbackCacheGateError):
            save_proposal(
                _KEY,
                _proposal(),
                visual_check_passed=v,
                user_approved=u,
                auto_cache=a,
            )
        # Gate violations must never touch the filesystem (parent dir absent).
        assert not (_isolated_cache_root / _FRAME_ID).exists(), (
            f"truth row (V={v}, U={u}, A={a}) leaked a directory"
        )


def test_auto_cache_default_off_preserves_dual_gate_semantics(
    _isolated_cache_root: pathlib.Path,
) -> None:
    """Calling save_proposal without ``auto_cache`` keeps the IMP-46 u2 behaviour."""
    with pytest.raises(AiFallbackCacheGateError) as exc:
        save_proposal(
            _KEY, _proposal(), visual_check_passed=True, user_approved=False
        )
    assert "user_approved" in str(exc.value)
    assert not (_isolated_cache_root / _FRAME_ID).exists()


def test_auto_cache_cannot_bypass_visual_check() -> None:
    """``visual_check_passed=False`` raises even with ``auto_cache=True``."""
    with pytest.raises(AiFallbackCacheGateError) as exc:
        save_proposal(
            _KEY,
            _proposal(),
            visual_check_passed=False,
            user_approved=True,
            auto_cache=True,
        )
    assert "visual_check_passed" in str(exc.value)


def test_auto_cache_bypass_user_approved_persists(
    _isolated_cache_root: pathlib.Path,
) -> None:
    """``auto_cache=True`` with ``user_approved=False`` persists the proposal."""
    path = save_proposal(
        _KEY,
        _proposal(payload={"bypass": "user"}),
        visual_check_passed=True,
        user_approved=False,
        auto_cache=True,
    )
    assert path.is_file()
    loaded = read_proposal(_KEY)
    assert loaded is not None
    assert loaded.payload == {"bypass": "user"}


def test_auto_cache_rejects_non_bool() -> None:
    """``auto_cache`` must be a bool (loud TypeError, symmetric with other kwargs)."""
    with pytest.raises(TypeError):
        save_proposal(
            _KEY,
            _proposal(),
            visual_check_passed=True,
            user_approved=True,
            auto_cache="yes",  # type: ignore[arg-type]
        )


def test_auto_cache_is_keyword_only() -> None:
    """``auto_cache`` must be passed by keyword (positional rejected)."""
    import inspect

    sig = inspect.signature(save_proposal)
    param = sig.parameters["auto_cache"]
    assert param.kind is inspect.Parameter.KEYWORD_ONLY
    assert param.default is False