FabledGrimoire/tests/test_snippet_drift_check.py

"""Tests for the drift check — does a snippet still match its source?

The check itself runs agent-side (Grimoire has no checkout), so what's testable
here is everything around it: the verdict's shape, the rule that a verdict
expires when the code it blessed is edited away, and the two dialects of the
filter predicate — SQL for the browse/keyword arms, Python for the semantic
arm's already-fetched candidates.
"""
from types import SimpleNamespace

import pytest

from grimoire.services import knowledge as knowledge_svc
from grimoire.services import snippets as snippets_svc
from grimoire.services.snippets import (
    code_sha,
    compose_data,
    compose_verification,
    verification_view,
)


def _note(data=None, body=""):
    return SimpleNamespace(data=data, body=body, title="n — w", tags=["snippet"])


# --- the verdict record ---------------------------------------------------


def test_unknown_status_is_rejected_not_stored():
    """A typo'd status would be stored happily and then never match any filter —
    invisible rot in the thing built to make rot visible."""
    with pytest.raises(ValueError):
        compose_verification(status="broekn", checked_code_sha="abc")


@pytest.mark.parametrize("status", ["ok", "missing", "moved", "changed"])
def test_every_documented_status_is_accepted(status):
    out = compose_verification(status=status, checked_code_sha="abc")
    assert out["status"] == status
    assert out["code_sha"] == "abc"
    assert out["checked_at"]


def test_blank_detail_and_path_are_omitted_not_stored_empty():
    """Keeps `data` sparse, and keeps a containment match from tripping on ""."""
    out = compose_verification(status="ok", checked_code_sha="abc")
    assert "detail" not in out and "path" not in out


# --- expiry by edit -------------------------------------------------------


def test_code_sha_ignores_reformatting_but_not_meaning():
    """A verdict must not expire because an editor stripped trailing whitespace,
    and must expire when the code actually changes."""
    assert code_sha("def f():\n    return 1") == code_sha("def f():   \n    return 1\n")
    assert code_sha("def f():\n    return 1") != code_sha("def f():\n    return 2")


def test_verdict_stays_current_while_code_is_unchanged():
    code = "def f():\n    return 1"
    fields = {"code": code, "verification": compose_verification(
        status="ok", checked_code_sha=code_sha(code)
    )}
    view = verification_view(_note(), fields)
    assert view["status"] == "ok"
    assert view["current"] is True
    assert view["needs_attention"] is False


def test_an_ok_verdict_expires_once_the_code_is_edited():
    """The whole reason the verdict carries a hash: an OK verdict must never go
    on vouching for code nobody checked."""
    fields = {"code": "def f():\n    return 2", "verification": compose_verification(
        status="ok", checked_code_sha=code_sha("def f():\n    return 1")
    )}
    view = verification_view(_note(), fields)
    assert view["current"] is False
    assert view["needs_attention"] is True


def test_never_checked_reads_as_unverified():
    view = verification_view(_note(), {"code": "x = 1"})
    assert view["status"] == "unverified"
    assert view["current"] is False


def test_a_failing_verdict_needs_attention_even_while_current():
    code = "x = 1"
    fields = {"code": code, "verification": compose_verification(
        status="missing", checked_code_sha=code_sha(code)
    )}
    assert verification_view(_note(), fields)["needs_attention"] is True


# --- the data mirror ------------------------------------------------------


def test_compose_data_mirrors_the_current_code_hash():
    """`data.code_sha` is what makes "this verdict expired" an SQL-expressible
    comparison rather than a post-filter that would break pagination totals."""
    out = compose_data(name="n", code="def f():\n    return 1")
    assert out["code_sha"] == code_sha("def f():\n    return 1")


def test_compose_data_omits_the_hash_when_there_is_no_code():
    assert "code_sha" not in compose_data(name="n")


def test_the_code_itself_is_never_copied_into_data():
    """data indexes AROUND the code; duplicating the blob would be pure weight."""
    out = compose_data(name="n", code="secret_looking_blob")
    assert "code" not in out
    assert "secret_looking_blob" not in str(out)


# --- the filter, Python dialect -------------------------------------------


def _data(status, checked_sha, current_sha):
    return {
        "verification": {"status": status, "code_sha": checked_sha},
        "code_sha": current_sha,
    }


@pytest.mark.parametrize(
    "value, expected",
    [
        ("", True),            # no filter
        ("ok", True),
        ("attention", False),
        ("drifted", False),
        ("unverified", False),
    ],
)
def test_python_dialect_on_a_current_ok_verdict(value, expected):
    data = _data("ok", "aaa", "aaa")
    assert knowledge_svc.verification_matches(data, value) is expected


@pytest.mark.parametrize(
    "value, expected",
    [("ok", False), ("attention", True), ("drifted", False), ("unverified", False)],
)
def test_python_dialect_on_an_expired_ok_verdict(value, expected):
    """Expired-but-ok is the case that motivated `attention` existing at all: it
    is not `drifted` (nothing was found wrong) and not `unverified` (a check did
    happen), yet it plainly needs looking at."""
    data = _data("ok", "aaa", "bbb")
    assert knowledge_svc.verification_matches(data, value) is expected


@pytest.mark.parametrize(
    "value, expected",
    [("ok", False), ("attention", True), ("drifted", True), ("changed", True),
     ("missing", False)],
)
def test_python_dialect_on_a_drifted_verdict(value, expected):
    data = _data("changed", "aaa", "aaa")
    assert knowledge_svc.verification_matches(data, value) is expected


def test_python_dialect_on_a_row_with_no_data_at_all():
    """Rows predating migration 0070 have no `data`. They are unverified, which
    is true — nobody has checked them."""
    assert knowledge_svc.verification_matches(None, "unverified") is True
    assert knowledge_svc.verification_matches(None, "attention") is False
    assert knowledge_svc.verification_matches(None, "ok") is False


@pytest.mark.parametrize(
    "value, expected",
    [("ok", False), ("attention", True), ("drifted", False), ("unverified", False)],
)
def test_python_dialect_on_a_push_invalidated_ok_verdict(value, expected):
    """A push touched the recorded location since the verdict. Like the
    expired case, it is neither drifted (nothing found wrong) nor unverified (a
    check happened) — but the repo moved under the blessing, so `attention`
    must include it and `ok` must not."""
    data = _data("ok", "aaa", "aaa")
    data["verification"]["invalidated_by"] = {"commit_sha": "d" * 40, "at": "t"}
    assert knowledge_svc.verification_matches(data, value) is expected


def test_push_invalidation_reads_as_attention_and_clears_on_reverify():
    """The flag rides the verdict dict, so recording ANY fresh verdict clears
    it by construction — compose_verification builds a new dict. No clearing
    branch exists to forget."""
    sha = code_sha("def f(): pass")
    verdict = compose_verification(status="ok", checked_code_sha=sha)
    verdict["invalidated_by"] = {"commit_sha": "d" * 40, "at": "t"}
    fields = {"code": "def f(): pass", "verification": verdict}
    view = verification_view(_note(), fields)
    assert view["needs_attention"] is True
    assert view["invalidated_by"]["commit_sha"] == "d" * 40

    fresh = compose_verification(status="ok", checked_code_sha=sha)
    assert "invalidated_by" not in fresh
    view2 = verification_view(_note(), {"code": "def f(): pass", "verification": fresh})
    assert view2["needs_attention"] is False


# --- the filter, SQL dialect ----------------------------------------------


def test_no_filter_produces_no_clause():
    assert knowledge_svc._verification_clause("") is None


@pytest.mark.parametrize(
    "value", ["ok", "drifted", "unverified", "attention", "missing", "moved", "changed"]
)
def test_every_filter_value_compiles_to_sql(value):
    """The two dialects have to agree, and the SQL one has to be expressible at
    all — the expired comparison relies on jsonpath reading two fields of the
    same row, which is easy to get wrong silently."""
    clause = knowledge_svc._verification_clause(value)
    assert clause is not None
    assert str(clause)


def test_a_status_value_is_json_quoted_into_the_jsonpath():
    """Statuses are validated on WRITE, but the filter takes caller input
    directly — a quote must not be able to break out of the jsonpath and turn
    a narrowing filter into a widening one."""
    hostile = 'x" || @.status == "ok'
    clause = knowledge_svc._verification_clause(hostile)
    params = list(clause.compile().params.values())
    jsonpath = next(p for p in params if isinstance(p, str) and "status" in p)
    # The injected quote is escaped, so the whole hostile value stays ONE string
    # literal rather than closing it and appending a second condition. Asserting
    # on the exact json.dumps form rather than counting "@.status" — the hostile
    # value legitimately contains that text inside the literal.
    import json as _json

    assert jsonpath == f"$.verification ? (@.status == {_json.dumps(hostile)})"
    assert '\\"' in jsonpath


# --- write path -----------------------------------------------------------


async def test_recording_a_verdict_does_not_touch_the_body():
    """A verdict is metadata about the snippet, not part of it. Writing it into
    the body would put it into the embedding and change what the record recalls
    against."""
    from unittest.mock import AsyncMock, patch

    note = SimpleNamespace(
        id=5, user_id=1, note_type="snippet", deleted_at=None,
        title="n — w", body="```python\nx = 1\n```", tags=["snippet"], data=None,
    )
    with (
        patch.object(snippets_svc, "get_snippet", AsyncMock(return_value=note)),
        patch("grimoire.services.access.can_write_note", AsyncMock(return_value=True)),
        patch.object(
            snippets_svc.notes_svc, "update_note", AsyncMock(return_value=note)
        ) as upd,
    ):
        await snippets_svc.record_verification(1, 5, status="moved", detail="renamed")

    assert set(upd.call_args.kwargs) == {"data"}
    stored = upd.call_args.kwargs["data"]["verification"]
    assert stored["status"] == "moved"
    assert stored["detail"] == "renamed"


async def test_recording_a_verdict_requires_write_access():
    """Being able to read a snippet someone shared must not let you mark it
    broken — a verdict changes how the record is presented to them."""
    from unittest.mock import AsyncMock, patch

    note = SimpleNamespace(id=5, user_id=2, note_type="snippet", deleted_at=None,
                           title="n — w", body="", tags=[], data=None)
    with (
        patch.object(snippets_svc, "get_snippet", AsyncMock(return_value=note)),
        patch("grimoire.services.access.can_write_note", AsyncMock(return_value=False)),
        patch.object(snippets_svc.notes_svc, "update_note", AsyncMock()) as upd,
    ):
        assert await snippets_svc.record_verification(1, 5, status="ok") is None
    upd.assert_not_called()