bvandeusenandClaude Opus 5.5
55f0f96016
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Failing after 11s
CI & Build / TypeScript typecheck (push) Successful in 32s
CI & Build / Python tests (push) Failing after 1m23s
CI & Build / integration (push) Successful in 1m53s
CI & Build / Build & push image (push) Skipped
GET /api/plugin/write-scope answers the plugin's fence. The session's project comes from repo / project_id, as every plugin endpoint reads it. The act is either: - a directory (target_repo / target_project_id), for a file edit or a git command; or - an MCP call (tool, plus input: the call's short arguments). It returns allowed, the other projects the act would change, and the reason the operator is asked. It is a GET so a read-scoped key still fences file edits and git. services/write_scope.WRITE_TARGETS lists every write tool with the arguments that locate its write: a project id, or a task, note, milestone, System, rule, relation, trash batch, repo or family decision, each traced to its project. A write that lands in no project (install settings, user-level records, the judgments a session is asked for anywhere) has an empty entry. The one cross-project write allowed is create_task(kind="issue") as a new, unplanned, open issue. A project the caller cannot read is never named. tests/test_write_scope.py: - requires an entry for every write tool; - requires every declared argument to exist on the tool; - refuses a locating argument (project_id, task_id, milestone_id and the rest) that a write tool takes without declaring it. The integration test traces each locator from real rows. Milestone 583, step 2. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
3.8 KiB
Python
87 lines
3.8 KiB
Python
"""Real-Postgres proof that a write is traced to the project it lands in.
|
|
|
|
Two projects of one user: the session's own and another. Each locator kind is
|
|
resolved from a real row, and check_act answers for a session standing in
|
|
the first.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from grimoire.models import async_session
|
|
from grimoire.models.milestone import Milestone
|
|
from grimoire.models.note import Note
|
|
from grimoire.models.project import Project
|
|
from grimoire.models.system import System
|
|
from grimoire.services import write_scope
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
|
|
|
|
OWNER = "write_scope_owner"
|
|
BATCH = "write-scope-test-batch"
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def two_projects():
|
|
async with async_session() as s:
|
|
user = await ensure_user(s, OWNER)
|
|
for model in (Note, Milestone, System, Project):
|
|
for row in (await s.execute(select(model).where(model.user_id == user.id))).scalars():
|
|
await s.delete(row)
|
|
await s.flush()
|
|
home = Project(user_id=user.id, title="Home")
|
|
other = Project(user_id=user.id, title="Elsewhere")
|
|
s.add_all([home, other])
|
|
await s.flush()
|
|
plan = Milestone(user_id=user.id, project_id=other.id, title="Their plan")
|
|
area = System(user_id=user.id, project_id=other.id, name="Their area")
|
|
s.add_all([plan, area])
|
|
await s.flush()
|
|
theirs = Note(user_id=user.id, project_id=other.id, milestone_id=plan.id,
|
|
title="Their task", body="", note_type="note", task_kind="work",
|
|
status="todo")
|
|
mine = Note(user_id=user.id, project_id=home.id, title="My task", body="",
|
|
note_type="note", task_kind="work", status="todo")
|
|
trashed = Note(user_id=user.id, project_id=other.id, title="Trashed", body="",
|
|
note_type="note", deleted_batch_id=BATCH)
|
|
s.add_all([theirs, mine, trashed])
|
|
await s.commit()
|
|
return {"user": user.id, "home": home.id, "other": other.id, "plan": plan.id,
|
|
"area": area.id, "theirs": theirs.id, "mine": mine.id}
|
|
|
|
|
|
async def test_each_locator_finds_the_project_of_the_row_it_names(two_projects):
|
|
ids = two_projects
|
|
other, home, uid = ids["other"], ids["home"], ids["user"]
|
|
|
|
async def lands(tool, **args):
|
|
return await write_scope.tool_targets(uid, tool, args)
|
|
|
|
assert await lands("add_task_log", task_id=ids["theirs"]) == {other}
|
|
assert await lands("update_task", task_id=ids["mine"]) == {home}
|
|
assert await lands("update_task", task_id=ids["mine"], milestone_id=ids["plan"]) == {home, other}
|
|
assert await lands("delete_milestone", milestone_id=ids["plan"]) == {other}
|
|
assert await lands("update_system", system_id=ids["area"]) == {other}
|
|
assert await lands("restore", deleted_batch_id=BATCH) == {other}
|
|
assert await lands("create_rule", topic_id=1) == set()
|
|
|
|
|
|
async def test_a_session_in_one_project_is_asked_before_writing_in_the_other(two_projects):
|
|
ids = two_projects
|
|
uid, home, other = ids["user"], ids["home"], ids["other"]
|
|
|
|
own = await write_scope.check_act(uid, home, tool="add_task_log",
|
|
args={"task_id": ids["mine"]})
|
|
assert own["allowed"] is True
|
|
|
|
theirs = await write_scope.check_act(uid, home, tool="add_task_log",
|
|
args={"task_id": ids["theirs"]})
|
|
assert theirs["allowed"] is False
|
|
assert theirs["targets"] == [{"id": other, "title": "Elsewhere"}]
|
|
assert "Home" in theirs["reason"] and "Elsewhere" in theirs["reason"]
|
|
|
|
issue = await write_scope.check_act(uid, home, tool="create_task",
|
|
args={"project_id": other, "kind": "issue"})
|
|
assert issue["allowed"] is True and issue["issue"] is True
|