FabledGrimoire/tests/test_project_fence.py

"""The project fence, end to end: plugin/hooks/grimoire_project_fence.sh run
through a real shell against real git repos and a stand-in server.

A session started in one repo edits, commits and pushes in another, and makes
Grimoire writes. Each act aimed elsewhere is asked about in the server's
words; acts in the session's own repo never reach the server.
"""
from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path

from tests.helpers import http_sink, need_tools

HOOK = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "grimoire_project_fence.sh"
PATH = "/api/plugin/write-scope"
HOME_URL = "https://forge.example/team/home.git"
OTHER_URL = "https://forge.example/team/other.git"
ASK = json.dumps({"allowed": False, "targets": [{"id": 9, "title": "Other"}], "issue": False,
                  "reason": "This session belongs to Home. Editing a file in Other, a different project."}).encode()
OK = b'{"allowed": true, "targets": [], "issue": false, "reason": ""}'


def _git(*args, cwd):
    subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True,
                   env={**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@example.com",
                        "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@example.com"})


def _repo(tmp_path: Path, name: str, remote: str) -> Path:
    path = tmp_path / name
    path.mkdir()
    _git("init", "-q", cwd=path)
    _git("remote", "add", "origin", remote, cwd=path)
    return path


def _run(tmp_path, port, event, project_dir):
    need_tools("bash", "curl", "awk", "git")
    env = {"PATH": os.environ["PATH"], "GRIMOIRE_URL": f"http://127.0.0.1:{port}",
           "GRIMOIRE_TOKEN": "t", "TMPDIR": str(tmp_path / "tmp"), "HOME": str(tmp_path),
           "CLAUDE_PROJECT_DIR": str(project_dir)}
    (tmp_path / "tmp").mkdir(exist_ok=True)
    event = {"session_id": "s-fence", "hook_event_name": "PreToolUse", **event}
    out = subprocess.run(["bash", str(HOOK)], input=json.dumps(event), capture_output=True,
                         text=True, env=env, timeout=30)
    assert out.returncode == 0, out.stderr
    return out.stdout.strip()


def _asked(out: str) -> str:
    decision = json.loads(out)["hookSpecificOutput"]
    assert decision["permissionDecision"] == "ask"
    return decision["permissionDecisionReason"]


def _edit(path, cwd):
    return {"tool_name": "Edit", "cwd": str(cwd),
            "tool_input": {"file_path": str(path), "old_string": "a", "new_string": "b"}}


def _bash(command, cwd):
    return {"tool_name": "Bash", "cwd": str(cwd), "tool_input": {"command": command}}


def test_an_edit_in_the_sessions_own_repo_never_asks_the_server(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, _edit(home / "src" / "a.py", home), home)
    assert out == "" and seen == []


def test_an_edit_in_another_projects_repo_is_asked_about_in_the_servers_words(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    other = _repo(tmp_path, "other", OTHER_URL)
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, _edit(other / "new_dir" / "b.py", home), home)
    assert "a different project" in _asked(out)
    q = seen[0]
    assert q["_path"] == PATH
    assert q["repo"] == [HOME_URL] and q["target_repo"] == [OTHER_URL]
    assert q["act"] == ["Editing a file in"]


def test_an_edit_the_server_allows_goes_ahead_silently(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    other = _repo(tmp_path, "other", OTHER_URL)
    with http_sink(by_path={PATH: OK}) as (port, seen):
        out = _run(tmp_path, port, _edit(other / "b.py", home), home)
    assert out == "" and len(seen) == 1


def test_an_edit_outside_every_repo_is_no_projects_business(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    loose = tmp_path / "loose"
    loose.mkdir()
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, _edit(loose / "notes.txt", home), home)
    assert out == "" and seen == []


def test_a_commit_through_git_dash_c_is_checked_against_that_repo(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    other = _repo(tmp_path, "other", OTHER_URL)
    command = f'git -C {other} commit -m "tidy && push later"'
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, _bash(command, home), home)
    _asked(out)
    assert seen[0]["target_repo"] == [OTHER_URL]
    assert seen[0]["act"] == ["git commit in"]


def test_a_push_after_cd_is_checked_against_the_repo_it_moved_to(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    _repo(tmp_path, "other", OTHER_URL)
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, _bash("cd ../other && git push origin dev", home), home)
    _asked(out)
    assert seen[0]["target_repo"] == [OTHER_URL] and seen[0]["act"] == ["git push in"]


def test_git_commands_that_change_nothing_and_writes_at_home_stay_local(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    other = _repo(tmp_path, "other", OTHER_URL)
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        assert _run(tmp_path, port, _bash(f"git -C {other} status && git -C {other} log -1", home), home) == ""
        assert _run(tmp_path, port, _bash("git commit -m x && git push", home), home) == ""
        assert _run(tmp_path, port, _bash("echo hello", home), home) == ""
    assert seen == []


def test_a_worktree_of_another_repo_belongs_to_that_repo(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    other = _repo(tmp_path, "other", OTHER_URL)
    _git("commit", "-q", "--allow-empty", "-m", "start", cwd=other)
    tree = tmp_path / "other-tree"
    _git("worktree", "add", "-q", str(tree), cwd=other)
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, _edit(tree / "c.py", home), home)
    _asked(out)
    assert seen[0]["target_repo"] == [OTHER_URL]


def test_a_grimoire_write_sends_its_short_arguments_and_is_asked_about(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    event = {"tool_name": "mcp__plugin_grimoire_grimoire__update_task", "cwd": str(home),
             "tool_input": {"task_id": 5, "status": "done", "body": "x" * 500,
                            "system_ids": [3, 4]}}
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        out = _run(tmp_path, port, event, home)
    _asked(out)
    q = seen[0]
    assert q["tool"] == ["update_task"]
    assert json.loads(q["input"][0]) == {"task_id": "5", "status": "done", "system_ids": ["3", "4"]}


def test_a_grimoire_read_never_asks_the_server(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    event = {"tool_name": "mcp__plugin_grimoire_grimoire__get_task", "cwd": str(home),
             "tool_input": {"task_id": 5}}
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        assert _run(tmp_path, port, event, home) == ""
    assert seen == []


def test_a_session_started_outside_any_project_takes_the_first_one_it_enters(tmp_path):
    loose = tmp_path / "loose"
    loose.mkdir()
    write = {"tool_name": "mcp__plugin_grimoire_grimoire__add_task_log", "cwd": str(loose),
             "tool_input": {"task_id": 5, "content": "c"}}
    with http_sink(by_path={PATH: ASK}) as (port, seen):
        # Nothing to hold the session to yet.
        assert _run(tmp_path, port, write, loose) == ""
        assert seen == []
        for pid in (7, 8):
            _run(tmp_path, port, {"hook_event_name": "PostToolUse", "cwd": str(loose),
                                  "tool_name": "mcp__plugin_grimoire_grimoire__enter_project",
                                  "tool_input": {"project_id": pid}}, loose)
        _asked(_run(tmp_path, port, write, loose))
    assert seen[0]["fallback_project_id"] == ["7"]
    assert "repo" not in seen[0]


def test_an_unanswered_check_lets_the_act_through_and_says_so_once(tmp_path):
    home = _repo(tmp_path, "home", HOME_URL)
    other = _repo(tmp_path, "other", OTHER_URL)
    first = _run(tmp_path, 1, _edit(other / "b.py", home), home)
    ctx = json.loads(first)["hookSpecificOutput"]["additionalContext"]
    assert ctx.startswith("> Grimoire did not answer the project fence check")
    assert _run(tmp_path, 1, _edit(other / "b.py", home), home) == ""


def test_the_entered_project_outlives_a_compaction_and_ends_with_the_session(tmp_path):
    # A compaction loses the context, not the session: the project it entered
    # stays its project. Session end clears it.
    loose = tmp_path / "loose"
    loose.mkdir()
    _run(tmp_path, 1, {"hook_event_name": "PostToolUse", "cwd": str(loose),
                       "tool_name": "mcp__plugin_grimoire_grimoire__enter_project",
                       "tool_input": {"project_id": 7}}, loose)
    entered = tmp_path / "tmp" / "grimoire-fence" / "s-fence.entered"
    assert entered.read_text() == "7"

    env = {"PATH": os.environ["PATH"], "TMPDIR": str(tmp_path / "tmp"), "HOME": str(tmp_path)}
    hooks = HOOK.parent
    for hook, event in (("grimoire_session_context.sh", {"source": "compact"}),
                        ("grimoire_session_end.sh", {"reason": "other"})):
        subprocess.run(["bash", str(hooks / hook)], input=json.dumps({"session_id": "s-fence", **event}),
                       capture_output=True, text=True, env=env, timeout=60, check=True)
        if hook == "grimoire_session_context.sh":
            assert entered.exists(), "a compaction cleared the project the session entered"
    assert not entered.exists()