FabledGrimoire/tests/test_write_scope.py

"""Where a write lands, so the plugin can hold a session to its own project.

The guards here keep services/write_scope.WRITE_TARGETS whole: every write
tool says which of its arguments locate the write, and an argument that would
locate one can't be left out. The rest covers the one cross-project write
allowed (filing an issue) and the answer check_act gives.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest
from quart import g

from grimoire.services import write_scope
from grimoire.services.write_scope import (
    LOCATOR_NAMES, LOCATORS, REFERENCES, WRITE_TARGETS, check_act, is_issue_filing,
)


def _write_tool_params() -> dict[str, set[str]]:
    from grimoire.mcp.server import _WRITE_TOOLS, build_mcp_server

    manager = build_mcp_server()._tool_manager
    return {
        name: set(manager.get_tool(name).parameters.get("properties", {}))
        for name in _WRITE_TOOLS
    }


def test_every_write_tool_says_where_it_writes():
    from grimoire.mcp.server import _WRITE_TOOLS

    missing = sorted(_WRITE_TOOLS - set(WRITE_TARGETS))
    assert not missing, (
        f"write tools with no entry in WRITE_TARGETS: {missing}. Say which "
        "arguments locate each one's write, or {} when it lands in no project."
    )
    extra = sorted(set(WRITE_TARGETS) - _WRITE_TOOLS)
    assert not extra, f"WRITE_TARGETS names tools that aren't write tools: {extra}"


def test_every_declared_argument_is_one_the_tool_takes_and_has_a_known_kind():
    params = _write_tool_params()
    for tool, locators in WRITE_TARGETS.items():
        for arg, kind in locators.items():
            assert arg in params[tool], f"{tool} has no argument {arg!r}"
            assert kind in LOCATORS, f"{tool}.{arg}: unknown locator {kind!r}"


def test_an_argument_that_would_locate_a_write_cannot_be_left_out():
    params = _write_tool_params()
    undeclared = sorted(
        f"{tool}.{arg}"
        for tool, names in params.items()
        for arg in names & LOCATOR_NAMES
        if arg not in WRITE_TARGETS[tool] and arg not in REFERENCES.get(tool, {})
    )
    assert not undeclared, (
        f"write tools take these locating arguments without declaring them in "
        f"WRITE_TARGETS, or saying in REFERENCES why they only point: {undeclared}"
    )


def test_a_reference_is_an_argument_the_tool_takes_and_not_also_a_locator():
    params = _write_tool_params()
    for tool, refs in REFERENCES.items():
        for arg in refs:
            assert arg in params[tool], f"REFERENCES names {tool}.{arg}, which it doesn't take"
            assert arg not in WRITE_TARGETS[tool], f"{tool}.{arg} is both a locator and a reference"


def test_no_write_tool_carries_a_read_prefix():
    # The fence hook skips the server for these prefixes, so a write named
    # like a read would pass unchecked.
    from grimoire.mcp.server import _WRITE_TOOLS

    read_named = sorted(t for t in _WRITE_TOOLS if t.startswith(write_scope.READ_PREFIXES))
    assert not read_named, f"write tools named like reads: {read_named}"


def test_the_locator_scan_sees_the_arguments_it_guards():
    # The guard above passes vacuously if the scan finds no parameters at all.
    params = _write_tool_params()
    assert "task_id" in params["update_task"] & LOCATOR_NAMES
    assert "project_id" in params["create_task"] & LOCATOR_NAMES


@pytest.mark.parametrize("args, expected", [
    ({"project_id": 7, "kind": "issue"}, True),
    ({"project_id": 7, "kind": "issue", "status": "todo"}, True),
    ({"project_id": 7, "kind": "work"}, False),
    ({"project_id": 7}, False),
    ({"project_id": 7, "kind": "issue", "status": "in_progress"}, False),
    ({"project_id": 7, "kind": "issue", "milestone_id": 3}, False),
    ({"project_id": 7, "kind": "issue", "parent_id": 9}, False),
    ({"project_id": 7, "kind": "issue", "milestone_id": 0}, True),
])
def test_filing_an_issue_is_a_new_open_unplanned_issue(args, expected):
    assert is_issue_filing("create_task", args) is expected


def test_only_create_task_files_an_issue():
    assert is_issue_filing("update_task", {"task_id": 4, "kind": "issue"}) is False
    assert is_issue_filing("create_records", {"project_id": 7, "kind": "issue"}) is False


async def test_ids_of_zero_or_below_locate_nothing():
    # update_task's project_id=0 leaves it unchanged and -1 clears it.
    assert await write_scope.tool_targets(1, "update_task", {"project_id": -1}) == set()
    assert await write_scope.tool_targets(1, "create_task", {"project_id": 0}) == set()
    assert await write_scope.tool_targets(1, "create_note", {"project_id": 12}) == {12}


async def test_reads_and_unknown_tools_land_nowhere():
    assert await write_scope.tool_targets(1, "get_task", {"task_id": 5}) == set()
    assert await write_scope.tool_targets(1, "no_such_tool", {"project_id": 5}) == set()


def _patched(landing: set[int], readable: bool = True):
    return (
        patch.object(write_scope, "tool_targets", AsyncMock(return_value=landing)),
        patch("grimoire.services.access.can_read_project", AsyncMock(return_value=readable)),
        patch.object(write_scope, "_titles", AsyncMock(return_value={2: "Home", 9: "Elsewhere"})),
    )


async def test_a_write_in_the_sessions_own_project_goes_ahead():
    a, b, c = _patched({2})
    with a, b, c:
        out = await check_act(1, 2, tool="update_task", args={"task_id": 5})
    assert out == {"allowed": True, "targets": [], "issue": False, "reason": ""}


async def test_a_write_in_another_project_asks_and_says_how_to_file_an_issue_instead():
    a, b, c = _patched({9})
    with a, b, c:
        out = await check_act(1, 2, tool="update_task", args={"task_id": 5})
    assert out["allowed"] is False
    assert out["targets"] == [{"id": 9, "title": "Elsewhere"}]
    assert "Home (project 2)" in out["reason"]
    assert "Elsewhere (project 9)" in out["reason"]
    assert 'create_task(project_id=9, kind="issue")' in out["reason"]


async def test_filing_an_issue_in_another_project_goes_ahead():
    a, b, c = _patched({9})
    with a, b, c:
        out = await check_act(1, 2, tool="create_task", args={"project_id": 9, "kind": "issue"})
    assert out["allowed"] is True and out["issue"] is True


async def test_a_session_with_no_project_is_held_to_nothing():
    a, b, c = _patched({9})
    with a, b, c:
        out = await check_act(1, 0, tool="update_task", args={"task_id": 5})
    assert out["allowed"] is True


async def test_a_project_the_caller_cannot_read_is_not_named():
    a, b, c = _patched({9}, readable=False)
    with a, b, c:
        out = await check_act(1, 2, tool="update_task", args={"task_id": 5})
    assert out == {"allowed": True, "targets": [], "issue": False, "reason": ""}


async def test_a_directory_in_another_project_asks_in_the_words_of_the_act():
    with patch("grimoire.services.access.can_read_project", AsyncMock(return_value=True)), \
         patch.object(write_scope, "_titles", AsyncMock(return_value={2: "Home", 9: "Elsewhere"})):
        out = await check_act(1, 2, target_project=9, act="git push in")
        same = await check_act(1, 2, target_project=2, act="git push in")
    assert out["allowed"] is False
    assert "git push in Elsewhere (project 9)" in out["reason"]
    assert same["allowed"] is True


async def test_the_route_reads_the_session_the_tool_and_its_input():
    from grimoire.app import create_app
    from grimoire.routes import plugin

    app = create_app()
    answer = {"allowed": False, "targets": [{"id": 9, "title": "x"}], "issue": False, "reason": "r"}
    check = AsyncMock(return_value=dict(answer))
    url = ("/api/plugin/write-scope?project_id=2"
           "&tool=mcp__plugin_grimoire_grimoire__update_task"
           "&input=%7B%22task_id%22%3A5%7D&act=x")
    async with app.test_request_context(url):
        g.user = SimpleNamespace(id=1)
        with patch.object(plugin.write_scope_svc, "check_act", check), \
             patch.object(plugin, "can_read_project", AsyncMock(return_value=True)):
            resp = await plugin.write_scope.__wrapped__()
    body = await resp.get_json()
    assert body["allowed"] is False and body["session_project"] == 2
    _, kwargs = check.call_args
    assert check.call_args.args == (1, 2)
    assert kwargs["tool"] == "update_task" and kwargs["args"] == {"task_id": 5}


async def test_the_route_resolves_a_target_repo_to_its_project():
    from grimoire.app import create_app
    from grimoire.routes import plugin

    app = create_app()
    check = AsyncMock(return_value={"allowed": True, "targets": [], "issue": False, "reason": ""})
    url = "/api/plugin/write-scope?project_id=2&target_repo=https://forge.example/o/r.git&act=git%20push%20in"
    async with app.test_request_context(url):
        g.user = SimpleNamespace(id=1)
        with patch.object(plugin.write_scope_svc, "check_act", check), \
             patch.object(plugin, "can_read_project", AsyncMock(return_value=True)), \
             patch.object(plugin.repo_bindings_svc, "resolve_project", AsyncMock(return_value=9)):
            await plugin.write_scope.__wrapped__()
    assert check.call_args.kwargs["target_project"] == 9
    assert check.call_args.kwargs["act"] == "git push in"


async def test_the_route_refuses_input_that_is_not_an_object():
    from grimoire.app import create_app
    from grimoire.routes import plugin

    app = create_app()
    async with app.test_request_context("/api/plugin/write-scope?tool=update_task&input=%5B1%5D"):
        g.user = SimpleNamespace(id=1)
        with patch.object(plugin, "can_read_project", AsyncMock(return_value=True)):
            resp, status = await plugin.write_scope.__wrapped__()
    assert status == 400