Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions plane_mcp/tools/workitem.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,9 @@

PRIORITIES = get_args(PriorityEnum)

# Plane's floor for an assignable project member; guest sits below it.
_MEMBER_ROLE = 15

WRITE_FIELDS = (
"name",
"assignees",
Expand Down Expand Up @@ -177,6 +180,37 @@ def _scoped_pql(pql: str, project_id: str) -> str:
return f"({pql}) AND {scope}" if pql else scope


def _unassignable(client: Any, workspace_slug: str, project_id: str, wanted: list[str]) -> str | None:
"""Name the assignee ids Plane would drop, because dropping them is invisible.

Validation filters an assignee it will not accept out of the payload, and an
update deletes the work item's existing assignees before writing what is left.
So one unassignable id both fails to apply and clears the field, under a 200.
"""
try:
members = client.projects.get_members(workspace_slug=workspace_slug, project_id=project_id)
except HttpError as exc:
logger.warning("could not read members of project %s (%s); skipping the assignee check", project_id, exc)
return None
# Community Edition omits role and is_active, so an absent one cannot disqualify.
assignable = {
member.id
for member in members
if member.id and member.is_active is not False and (member.role is None or member.role >= _MEMBER_ROLE)
}
if not assignable: # nothing readable to judge against; never block a write that works today
return None
rejected = [user_id for user_id in wanted if user_id not in assignable]
if not rejected:
return None
return (
f"Error: not assignable in this project: {', '.join(rejected)}. Plane drops an assignee it "
"will not accept without reporting it, and clears the work item's existing assignees in the "
"process. Only active project members at member role or above can be assigned -- "
"`member list_project` lists them."
)


def _description_html(description_html: str, description_stripped: str) -> str | None:
"""Plane recomputes the stripped form server-side, so plain text must be wrapped."""
if description_html:
Expand Down Expand Up @@ -352,6 +386,10 @@ def write_payload() -> dict[str, Any]:
if not project_id:
return missing(action, "project_id")

if action in ("create", "update") and (wanted := coerce_list(assignees)):
if error := _unassignable(client, workspace_slug, project_id, wanted):
return error

if action == "create":
if not name:
return missing(action, "name")
Expand Down Expand Up @@ -399,6 +437,9 @@ def write_payload() -> dict[str, Any]:
return missing(action, f"add_{field[:-1]}_id or remove_{field[:-1]}_id")
# Either side takes one id or several, so adding three assignees is one call.
adding, removing = coerce_list(add) or [], coerce_list(remove) or []
if field == "assignees" and adding:
if error := _unassignable(client, workspace_slug, project_id, adding):
return error
current = client.work_items.retrieve(
workspace_slug=workspace_slug, project_id=project_id, work_item_id=workitem_id
)
Expand Down
8 changes: 7 additions & 1 deletion tests/test_coercion.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from __future__ import annotations

import pytest
from plane.models.projects import ProjectMember

from plane_mcp.coercion import accepted_types, coerce_arguments

Expand Down Expand Up @@ -152,10 +153,15 @@ def test_booleans_are_not_treated_as_numbers():


class _Recorder:
"""Stands in for the Plane client; records the first SDK call and stops."""
"""Stands in for the Plane client; records the first SDK call and stops.

A write carrying assignees reads the project's members first, so that lookup
has to answer -- and answer with the assignee -- rather than be the recorded call.
"""

def __init__(self):
self.kwargs: dict = {}
self.get_members = lambda **_: [ProjectMember(id=ASSIGNEE)]

def __getattr__(self, _name):
return self
Expand Down
96 changes: 96 additions & 0 deletions tests/tools/test_dispatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,11 @@
import inspect

import pytest
from plane.errors.errors import HttpError
from plane.models.projects import ProjectMember

from plane_mcp.toolkit.spec import action_names
from plane_mcp.tools.workitem import _MEMBER_ROLE

# Values that satisfy a parameter well enough to reach the SDK call. Enum-valued
# parameters need a member of their own vocabulary; the rest take a plausible id.
Expand Down Expand Up @@ -228,6 +231,99 @@ def test_a_valid_priority_still_reaches_the_sdk(registered, spy):
assert spy.recorder.only().kwargs["data"].priority == "urgent"


# The same defect one layer out: Plane filters an assignee it will not accept out of
# the payload, and the update deletes the existing assignees before writing what is
# left. So naming one both fails to assign and clears the field, under a 200.

ALICE = ("alice", 20, True)


def _project_members(*rows):
return [
ProjectMember(id=id_, email=f"{id_}@example.com", role=role, is_active=active) for id_, role, active in rows
]


@pytest.mark.parametrize(
("members", "why"),
[
([ALICE], "not a member of this project"),
([ALICE, ("bob", 5, True)], "a guest, below the assignable role"),
([ALICE, ("bob", 20, False)], "an inactive membership"),
],
ids=["non-member", "guest", "inactive"],
)
def test_an_unassignable_assignee_is_refused_before_the_write(members, why, registered, spy):
spy.returns["projects.get_members"] = _project_members(*members)

result = registered["workitem"].fn(action="update", project_id="p", workitem_id="w", assignees=["bob"])

assert isinstance(result, str) and result.startswith("Error:"), f"{why}: {result}"
assert "bob" in result, f"{why}: refused without naming who was rejected"
assert "work_items.update" not in spy.recorder.methods, f"{why}: wrote anyway, clearing the assignees"


@pytest.mark.parametrize("role", [20, _MEMBER_ROLE], ids=["above the floor", "exactly the floor"])
def test_an_assignable_member_reaches_the_sdk(role, registered, spy):
spy.returns["projects.get_members"] = _project_members(("bob", role, True))

registered["workitem"].fn(action="create", project_id="p", name="x", assignees=["bob"])

create = next(c for c in spy.recorder.calls if c.method == "work_items.create")
assert create.kwargs["data"].assignees == ["bob"]


def test_membership_alone_decides_when_the_edition_omits_role(registered, spy):
"""Community Edition answers with identity fields only; an absent role cannot disqualify."""
spy.returns["projects.get_members"] = [ProjectMember(id="bob")]

registered["workitem"].fn(action="create", project_id="p", name="x", assignees=["bob"])

assert "work_items.create" in spy.recorder.methods


@pytest.mark.parametrize(
"members",
[HttpError("nope", 404), []],
ids=["lookup fails", "nothing to judge against"],
)
def test_an_unusable_member_list_does_not_block_the_write(members, registered, spy):
"""The check guards a write that works today; it must not become a new way to fail."""
spy.returns["projects.get_members"] = members

registered["workitem"].fn(action="create", project_id="p", name="x", assignees=["bob"])

assert "work_items.create" in spy.recorder.methods


def test_a_write_carrying_no_assignees_costs_no_extra_request(registered, spy):
registered["workitem"].fn(action="update", project_id="p", workitem_id="w", name="renamed")

assert spy.recorder.methods == ["work_items.update"]


def test_adding_an_unassignable_assignee_is_refused(registered, spy):
spy.returns["projects.get_members"] = _project_members(ALICE)

result = registered["workitem"].fn(action="manage_assignee", project_id="p", workitem_id="w", add_user_id="bob")

assert isinstance(result, str) and result.startswith("Error:"), result
assert "work_items.update" not in spy.recorder.methods


def test_removing_an_assignee_is_never_blocked(registered, spy):
"""Whoever is on the item may no longer be assignable; that must not trap them there."""
from types import SimpleNamespace

spy.returns["projects.get_members"] = _project_members(ALICE)
spy.returns["work_items.retrieve"] = SimpleNamespace(assignees=["bob"], labels=[])

registered["workitem"].fn(action="manage_assignee", project_id="p", workitem_id="w", remove_user_id="bob")

update = next(c for c in spy.recorder.calls if c.method == "work_items.update")
assert update.kwargs["data"].assignees == []


@pytest.mark.parametrize(
("end_date", "edits"),
[
Expand Down