FabledGrimoire/tests/test_rule_moment_doors.py

"""Every door that writes a rule can mount it on moments.

THE SHAPE, NOT ONE FUNCTION. test_system_tagging_door_parity records how a
capability goes missing: whichever door nobody exercised for a kind never
grows the parameter. So this file asserts the whole table — every rule and
preference write, on both doors, takes `moments` — and that each one checks
the names BEFORE its create or update. A refusal after the write would leave
a rule created without the mounts it asked for, and the caller reading the
error would reasonably believe nothing happened.
"""
from __future__ import annotations

import ast
import pathlib
from unittest.mock import AsyncMock, patch

import pytest

from tests.helpers import fake_rule
from tests.helpers import plain_rule_detail as _plain_detail

ROOT = pathlib.Path(__file__).resolve().parents[1] / "src" / "grimoire"

# (tool, the service write it must validate before)
MCP_DOORS = [
    ("create_rule", "create_rule"),
    ("create_project_rule", "create_project_rule"),
    ("update_rule", "update_rule"),
    ("create_preference", "create_rule"),
    ("update_preference", "update_rule"),
]
# (route handler, the service write it must validate before)
REST_DOORS = [
    ("create_rule", "create_rule"),
    ("update_rule", "update_rule"),
    ("create_project_rule", "create_project_rule"),
]


def _fn(path: pathlib.Path, name: str):
    for node in ast.parse(path.read_text()).body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name:
            return node
    raise AssertionError(f"{path.name} has no {name}")


def _first_line_calling(fn, attr: str) -> int | None:
    lines = [
        n.lineno for n in ast.walk(fn)
        if isinstance(n, ast.Call) and (
            (isinstance(n.func, ast.Attribute) and n.func.attr == attr)
            or (isinstance(n.func, ast.Name) and n.func.id == attr)
        )
    ]
    return min(lines) if lines else None


@pytest.mark.parametrize("tool,write", MCP_DOORS)
def test_every_mcp_rule_door_takes_moments_and_checks_them_first(tool, write):
    fn = _fn(ROOT / "mcp" / "tools" / "rulebooks.py", tool)
    params = {a.arg for a in fn.args.args + fn.args.kwonlyargs}
    assert "moments" in params, f"{tool} cannot mount a rule"
    check = _first_line_calling(fn, "require_moments")
    wrote = _first_line_calling(fn, write)
    assert check is not None and wrote is not None, (tool, check, wrote)
    assert check < wrote, f"{tool} validates moments after it has already written"
    detail = [n for n in ast.walk(fn) if isinstance(n, ast.Call)
              and isinstance(n.func, ast.Attribute) and n.func.attr == "rule_detail"]
    assert any(any(isinstance(a, ast.Name) and a.id == "moments" for a in c.args)
               for c in detail), f"{tool} never hands its moments to rule_detail"


@pytest.mark.parametrize("handler,write", REST_DOORS)
def test_every_rest_rule_door_takes_moments_and_checks_them_first(handler, write):
    fn = _fn(ROOT / "routes" / "rulebooks.py", handler)
    check = _first_line_calling(fn, "_moments_or_refusal")
    wrote = _first_line_calling(fn, write)
    assert check is not None and wrote is not None, (handler, check, wrote)
    assert check < wrote, f"{handler} validates moments after it has already written"


def test_the_guard_can_fail():
    """A door with no check at all must be reported, not skipped."""
    fn = ast.parse("async def f():\n    await svc.create_rule()\n").body[0]
    assert _first_line_calling(fn, "require_moments") is None


async def test_an_unknown_moment_is_refused_before_anything_is_written():
    create = AsyncMock(return_value=fake_rule(id=5))
    with patch("grimoire.mcp.tools.rulebooks.rulebooks_svc.create_rule", create), \
         patch("grimoire.mcp.tools.rulebooks.dedup_svc.find_duplicate_rule", AsyncMock()), \
         _plain_detail(), patch("grimoire.mcp.tools.rulebooks.current_user_id", lambda: 7):
        from grimoire.mcp.tools.rulebooks import create_rule
        with pytest.raises(ValueError, match="unknown moment"):
            await create_rule(
                topic_id=10, title="t", statement="s",
                when_to_apply="when the moment this fixture stands in for arises",
                moments=["work.finished"],
            )
    create.assert_not_awaited()


async def test_the_door_hands_rule_detail_the_normalised_moments():
    seen = {}

    async def detail(_uid, rule, _system_ids=None, moments=None):
        seen["moments"] = moments
        return rule.to_dict()

    with patch("grimoire.mcp.tools.rulebooks.rulebooks_svc.update_rule",
               AsyncMock(return_value=fake_rule(id=5))), \
         patch("grimoire.mcp.tools.rulebooks.rulebooks_svc.rule_detail", detail), \
         patch("grimoire.mcp.tools.rulebooks.current_user_id", lambda: 7):
        from grimoire.mcp.tools.rulebooks import update_rule
        await update_rule(rule_id=5, moments=[" Work.Finish", "reply.report", "work.finish"])
    assert seen["moments"] == ["work.finish", "reply.report"]


async def test_omitting_moments_leaves_the_mounts_alone():
    seen = {}

    async def detail(_uid, rule, _system_ids=None, moments="unset"):
        seen["moments"] = moments
        return rule.to_dict()

    with patch("grimoire.mcp.tools.rulebooks.rulebooks_svc.update_rule",
               AsyncMock(return_value=fake_rule(id=5))), \
         patch("grimoire.mcp.tools.rulebooks.rulebooks_svc.rule_detail", detail), \
         patch("grimoire.mcp.tools.rulebooks.current_user_id", lambda: 7):
        from grimoire.mcp.tools.rulebooks import update_rule
        await update_rule(rule_id=5, title="renamed")
    assert seen["moments"] is None