sysadminandClaude Opus 4.8
a002864a06
The gitea-workflow documentation named a gitea_edit_issue tool that no
namespace had ever registered. There was no sanctioned MCP path to change an
issue title or body at all: the only edit tool, gitea_edit_pr, PATCHes the
pull-request endpoint. An authorized body correction on issue #780 therefore
had to be recorded as a discussion comment instead.
Add edit_issue.py as the authoritative rule and gitea_edit_issue as the tool
built on it:
- Only the fields the caller names are sent, so labels, state, assignee, and
milestone cannot be overwritten from a stale read.
- A pull-request number is refused. Gitea serves pull requests from the same
/issues/{n} collection, so without that check the issue path would quietly
become a second, ungated PR edit path. gitea_edit_pr stays PR-only.
- Structurally invalid requests raise before any credential or network work;
a request that would change nothing is reported as an explicit no-op with a
next action rather than a silent success.
- Read-after-write proves the applied title/body and proves that state,
labels, assignees, and milestone did not move. Transport failures on the
pre-read, the PATCH, and the read-back are each reported, redacted, with
the correct performed/verified state.
Gates match every other issue mutation: profile permission via the shared
capability map (resolver task edit_issue, gitea.issue.comment), preflight
purity, branches worktree validation, anti-stomp inventory membership, and
audited mutation.
Fix the drift that hid this. docs/mcp-tool-inventory.md is now the canonical
registered-tool list, and mcp_tool_inventory.py compares it to the live
registry in both directions, plus checks that every tool named under skills/
is registered. The new guard immediately found a second instance of the same
defect: gitea_record_pre_review_command had lost its @mcp.tool() decorator
while the canonical review workflow still instructed reviewers to call it, so
that registration is restored.
Validation:
PASSED: venv/bin/python -m pytest tests/test_issue_781_edit_issue_tool.py -s -q
— 50 passed
FAILED: venv/bin/python -m pytest -s -q — 4095 passed, 11 failed, 6 skipped.
The identical 11 tests fail on clean master 8e149e6 with no changes applied
(254 passed, 11 failed across those five files), so they are pre-existing
and proven by a baseline run rather than asserted.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
636 lines
25 KiB
Python
636 lines
25 KiB
Python
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from mutation_profile_fixture import install_deterministic_remote_urls # noqa: E402
|
|
|
|
install_deterministic_remote_urls()
|
|
"""#781: sanctioned issue title/body editing, and the documentation drift guard.
|
|
|
|
Two defects are covered here. The first is that no MCP path could edit an issue
|
|
title or body at all, so an authorized correction had to be recorded as a
|
|
comment. The second is why nobody noticed: documentation named a tool that was
|
|
never registered, and nothing compared the two lists.
|
|
"""
|
|
import glob
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import anti_stomp_preflight # noqa: E402
|
|
import edit_issue # noqa: E402
|
|
import mcp_server # noqa: E402
|
|
import mcp_tool_inventory # noqa: E402
|
|
import task_capability_map # noqa: E402
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
CONFIG = {
|
|
"version": 2,
|
|
"contexts": {
|
|
"ctx": {
|
|
"enabled": True,
|
|
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
|
}
|
|
},
|
|
"profiles": {
|
|
"edit-author": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "author",
|
|
"username": "author-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
|
|
"forbidden_operations": [],
|
|
"allowed_repositories": [
|
|
"Scaled-Tech-Consulting/Gitea-Tools",
|
|
"Example-Org/Example-Repo",
|
|
"913443/eAgenda",
|
|
],
|
|
"execution_profile": "edit-author",
|
|
},
|
|
"read-only-author": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "author",
|
|
"username": "reader-user",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": ["gitea.read"],
|
|
"forbidden_operations": ["gitea.issue.comment"],
|
|
"allowed_repositories": [
|
|
"Scaled-Tech-Consulting/Gitea-Tools",
|
|
"Example-Org/Example-Repo",
|
|
"913443/eAgenda",
|
|
],
|
|
"execution_profile": "read-only-author",
|
|
},
|
|
},
|
|
"rules": {"allow_runtime_switching": False},
|
|
}
|
|
|
|
ISSUE_NUMBER = 9
|
|
ORIGINAL_TITLE = "fix(mcp): original title"
|
|
ORIGINAL_BODY = "Original body.\n"
|
|
NEW_TITLE = "fix(mcp): corrected title"
|
|
NEW_BODY = "Corrected body.\n"
|
|
|
|
|
|
def _registered_tool_names() -> set[str]:
|
|
manager = mcp_server.mcp._tool_manager
|
|
return set((getattr(manager, "_tools", None) or {}).keys())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule: request validation
|
|
# ---------------------------------------------------------------------------
|
|
class TestValidateEditRequest(unittest.TestCase):
|
|
def test_no_field_is_rejected(self):
|
|
with self.assertRaises(ValueError) as ctx:
|
|
edit_issue.validate_edit_request()
|
|
self.assertIn("At least one field", str(ctx.exception))
|
|
|
|
def test_blank_title_is_rejected(self):
|
|
with self.assertRaises(ValueError) as ctx:
|
|
edit_issue.validate_edit_request(title=" ")
|
|
self.assertIn("cannot be blank", str(ctx.exception))
|
|
|
|
def test_non_string_title_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
edit_issue.validate_edit_request(title=42)
|
|
|
|
def test_non_string_body_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
edit_issue.validate_edit_request(body=["not", "a", "string"])
|
|
|
|
def test_empty_body_is_a_legitimate_edit(self):
|
|
self.assertEqual(edit_issue.validate_edit_request(body=""), {"body": ""})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule: planning against the pre-image
|
|
# ---------------------------------------------------------------------------
|
|
class TestPlanIssueEdit(unittest.TestCase):
|
|
def _current(self, **overrides):
|
|
issue = {
|
|
"number": ISSUE_NUMBER,
|
|
"title": ORIGINAL_TITLE,
|
|
"body": ORIGINAL_BODY,
|
|
"state": "open",
|
|
"labels": [{"name": "type:bug"}, {"name": "mcp"}],
|
|
"assignees": [{"login": "author-user"}],
|
|
"milestone": {"title": "v1.2.0"},
|
|
}
|
|
issue.update(overrides)
|
|
return issue
|
|
|
|
def test_title_only_sends_only_the_title(self):
|
|
plan = edit_issue.plan_issue_edit(self._current(), title=NEW_TITLE)
|
|
self.assertEqual(plan["payload"], {"title": NEW_TITLE})
|
|
self.assertEqual(plan["requested_fields"], ["title"])
|
|
self.assertFalse(plan["no_op"])
|
|
|
|
def test_body_only_sends_only_the_body(self):
|
|
plan = edit_issue.plan_issue_edit(self._current(), body=NEW_BODY)
|
|
self.assertEqual(plan["payload"], {"body": NEW_BODY})
|
|
|
|
def test_combined_edit_sends_both(self):
|
|
plan = edit_issue.plan_issue_edit(
|
|
self._current(), title=NEW_TITLE, body=NEW_BODY
|
|
)
|
|
self.assertEqual(plan["payload"], {"title": NEW_TITLE, "body": NEW_BODY})
|
|
self.assertEqual(plan["requested_fields"], ["body", "title"])
|
|
|
|
def test_identical_content_is_an_explicit_no_op(self):
|
|
plan = edit_issue.plan_issue_edit(
|
|
self._current(), title=ORIGINAL_TITLE, body=ORIGINAL_BODY
|
|
)
|
|
self.assertTrue(plan["no_op"])
|
|
self.assertEqual(plan["payload"], {})
|
|
self.assertTrue(plan["reasons"])
|
|
self.assertTrue(plan["safe_next_action"])
|
|
|
|
def test_partially_unchanged_request_sends_only_the_difference(self):
|
|
plan = edit_issue.plan_issue_edit(
|
|
self._current(), title=ORIGINAL_TITLE, body=NEW_BODY
|
|
)
|
|
self.assertFalse(plan["no_op"])
|
|
self.assertEqual(plan["payload"], {"body": NEW_BODY})
|
|
self.assertEqual(plan["unchanged_fields"], ["title"])
|
|
|
|
def test_missing_body_is_compared_as_empty(self):
|
|
current = self._current()
|
|
current.pop("body")
|
|
plan = edit_issue.plan_issue_edit(current, body="")
|
|
self.assertTrue(plan["no_op"])
|
|
|
|
def test_preserved_snapshot_captures_untouched_fields(self):
|
|
plan = edit_issue.plan_issue_edit(self._current(), title=NEW_TITLE)
|
|
before = plan["preserved_before"]
|
|
self.assertEqual(before["state"], "open")
|
|
self.assertEqual(before["labels"], ["type:bug", "mcp"])
|
|
self.assertEqual(before["assignees"], ["author-user"])
|
|
self.assertEqual(before["milestone"], "v1.2.0")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule: pull requests are refused
|
|
# ---------------------------------------------------------------------------
|
|
class TestAssessIssueTarget(unittest.TestCase):
|
|
def test_issue_is_accepted(self):
|
|
target = edit_issue.assess_issue_target(
|
|
{"number": 9, "title": "t"}, issue_number=9
|
|
)
|
|
self.assertTrue(target["is_issue"])
|
|
self.assertEqual(target["reasons"], [])
|
|
|
|
def test_pull_request_is_refused_with_a_next_action(self):
|
|
target = edit_issue.assess_issue_target(
|
|
{"number": 9, "pull_request": {"merged": False}}, issue_number=9
|
|
)
|
|
self.assertFalse(target["is_issue"])
|
|
self.assertTrue(target["is_pull_request"])
|
|
self.assertIn("gitea_edit_pr", target["safe_next_action"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rule: read-after-write verification
|
|
# ---------------------------------------------------------------------------
|
|
class TestVerifyIssueEdit(unittest.TestCase):
|
|
def _plan(self, **kwargs):
|
|
current = {
|
|
"number": ISSUE_NUMBER,
|
|
"title": ORIGINAL_TITLE,
|
|
"body": ORIGINAL_BODY,
|
|
"state": "open",
|
|
"labels": [{"name": "type:bug"}],
|
|
"assignees": [],
|
|
"milestone": None,
|
|
}
|
|
return edit_issue.plan_issue_edit(current, **kwargs)
|
|
|
|
def test_applied_content_verifies(self):
|
|
plan = self._plan(title=NEW_TITLE)
|
|
observed = {
|
|
"title": NEW_TITLE,
|
|
"body": ORIGINAL_BODY,
|
|
"state": "open",
|
|
"labels": [{"name": "type:bug"}],
|
|
"assignees": [],
|
|
"milestone": None,
|
|
}
|
|
result = edit_issue.verify_issue_edit(observed, plan=plan)
|
|
self.assertTrue(result["verified"])
|
|
self.assertTrue(result["preserved_intact"])
|
|
self.assertEqual(result["applied"], {"title": NEW_TITLE})
|
|
|
|
def test_unapplied_content_fails_closed(self):
|
|
plan = self._plan(title=NEW_TITLE)
|
|
observed = {
|
|
"title": ORIGINAL_TITLE,
|
|
"state": "open",
|
|
"labels": [{"name": "type:bug"}],
|
|
}
|
|
result = edit_issue.verify_issue_edit(observed, plan=plan)
|
|
self.assertFalse(result["verified"])
|
|
self.assertEqual(result["mismatches"][0]["field"], "title")
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_dropped_label_fails_closed(self):
|
|
plan = self._plan(title=NEW_TITLE)
|
|
observed = {"title": NEW_TITLE, "state": "open", "labels": []}
|
|
result = edit_issue.verify_issue_edit(observed, plan=plan)
|
|
self.assertFalse(result["verified"])
|
|
self.assertFalse(result["preserved_intact"])
|
|
self.assertEqual(result["preserved_changed"][0]["field"], "labels")
|
|
|
|
def test_changed_state_fails_closed(self):
|
|
plan = self._plan(body=NEW_BODY)
|
|
observed = {
|
|
"body": NEW_BODY,
|
|
"state": "closed",
|
|
"labels": [{"name": "type:bug"}],
|
|
}
|
|
result = edit_issue.verify_issue_edit(observed, plan=plan)
|
|
self.assertFalse(result["verified"])
|
|
self.assertEqual(result["preserved_changed"][0]["field"], "state")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool: gitea_edit_issue against a fake Gitea
|
|
# ---------------------------------------------------------------------------
|
|
class _EditIssueToolHarness(unittest.TestCase):
|
|
def setUp(self):
|
|
self._remotes = patch.dict(
|
|
mcp_server.REMOTES,
|
|
{
|
|
"prgs": {
|
|
"host": "gitea.example.com",
|
|
"org": "Example-Org",
|
|
"repo": "Example-Repo",
|
|
}
|
|
},
|
|
)
|
|
self._remotes.start()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
self._dir = tempfile.TemporaryDirectory()
|
|
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
|
with open(self.config_path, "w", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(CONFIG))
|
|
|
|
self.issue = {
|
|
"number": ISSUE_NUMBER,
|
|
"title": ORIGINAL_TITLE,
|
|
"body": ORIGINAL_BODY,
|
|
"state": "open",
|
|
"labels": [{"name": "type:bug"}, {"name": "mcp"}],
|
|
"assignees": [{"login": "author-user"}],
|
|
"milestone": {"title": "v1.2.0"},
|
|
"html_url": "https://gitea.example.com/Example-Org/Example-Repo/issues/9",
|
|
}
|
|
self.calls: list[tuple[str, str]] = []
|
|
self.patched_payloads: list[dict] = []
|
|
|
|
patch("gitea_audit.audit_enabled", return_value=False).start()
|
|
patch("mcp_server.get_auth_header", return_value="token author-pass").start()
|
|
patch("mcp_server.api_request", side_effect=self._api).start()
|
|
self.addCleanup(patch.stopall)
|
|
|
|
def tearDown(self):
|
|
self._remotes.stop()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
self._dir.cleanup()
|
|
|
|
def _api(self, method, url, auth, payload=None):
|
|
self.calls.append((method, url))
|
|
if url.endswith("/user"):
|
|
return {"login": "author-user"}
|
|
if "/issues/" in url:
|
|
if method == "GET":
|
|
return dict(self.issue)
|
|
if method == "PATCH":
|
|
self.patched_payloads.append(dict(payload or {}))
|
|
self.issue.update(payload or {})
|
|
return dict(self.issue)
|
|
raise AssertionError(f"unexpected API call: {method} {url}")
|
|
|
|
def _env(self, profile: str = "edit-author") -> dict:
|
|
return {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": profile,
|
|
"GITEA_TOKEN_AUTHOR": "author-pass",
|
|
"PYTEST_CURRENT_TEST": os.environ.get(
|
|
"PYTEST_CURRENT_TEST", "issue_781_edit_issue"
|
|
),
|
|
}
|
|
|
|
def _edit(self, profile: str = "edit-author", **kwargs):
|
|
with patch.dict(os.environ, self._env(profile), clear=True):
|
|
return mcp_server.gitea_edit_issue(
|
|
issue_number=ISSUE_NUMBER, remote="prgs", **kwargs
|
|
)
|
|
|
|
def _patch_methods(self) -> list[str]:
|
|
return [method for method, _url in self.calls if method == "PATCH"]
|
|
|
|
|
|
class TestEditIssueSucceeds(_EditIssueToolHarness):
|
|
def test_title_only_edit(self):
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["verified"])
|
|
self.assertEqual(result["changed_fields"], ["title"])
|
|
self.assertEqual(self.patched_payloads, [{"title": NEW_TITLE}])
|
|
self.assertEqual(self.issue["title"], NEW_TITLE)
|
|
self.assertEqual(self.issue["body"], ORIGINAL_BODY)
|
|
|
|
def test_body_only_edit(self):
|
|
result = self._edit(body=NEW_BODY)
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(self.patched_payloads, [{"body": NEW_BODY}])
|
|
self.assertEqual(self.issue["title"], ORIGINAL_TITLE)
|
|
self.assertEqual(self.issue["body"], NEW_BODY)
|
|
|
|
def test_combined_edit(self):
|
|
result = self._edit(title=NEW_TITLE, body=NEW_BODY)
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(
|
|
self.patched_payloads, [{"title": NEW_TITLE, "body": NEW_BODY}]
|
|
)
|
|
self.assertEqual(result["applied"], {"title": NEW_TITLE, "body": NEW_BODY})
|
|
|
|
def test_body_can_be_cleared(self):
|
|
result = self._edit(body="")
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(self.issue["body"], "")
|
|
|
|
def test_targets_the_issue_endpoint_never_the_pull_endpoint(self):
|
|
self._edit(title=NEW_TITLE)
|
|
patched = [url for method, url in self.calls if method == "PATCH"]
|
|
self.assertTrue(patched)
|
|
for url in patched:
|
|
self.assertIn("/issues/", url)
|
|
self.assertNotIn("/pulls/", url)
|
|
|
|
def test_labels_state_assignee_and_milestone_are_provably_unchanged(self):
|
|
result = self._edit(title=NEW_TITLE)
|
|
proof = result["read_after_write"]
|
|
self.assertTrue(proof["preserved_intact"])
|
|
self.assertEqual(proof["preserved_before"], proof["preserved_after"])
|
|
self.assertEqual(proof["preserved_after"]["labels"], ["type:bug", "mcp"])
|
|
self.assertEqual(proof["preserved_after"]["state"], "open")
|
|
self.assertEqual(proof["preserved_after"]["assignees"], ["author-user"])
|
|
self.assertEqual(proof["preserved_after"]["milestone"], "v1.2.0")
|
|
|
|
def test_read_after_write_re_reads_the_issue(self):
|
|
self._edit(title=NEW_TITLE)
|
|
issue_calls = [method for method, url in self.calls if "/issues/" in url]
|
|
self.assertEqual(issue_calls, ["GET", "PATCH", "GET"])
|
|
|
|
|
|
class TestEditIssueFailsClosed(_EditIssueToolHarness):
|
|
def test_no_op_request_is_rejected_without_a_patch(self):
|
|
result = self._edit(title=ORIGINAL_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertTrue(result["no_op"])
|
|
self.assertEqual(self._patch_methods(), [])
|
|
self.assertTrue(result["reasons"])
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_invalid_request_raises_before_any_api_call(self):
|
|
with self.assertRaises(ValueError):
|
|
self._edit()
|
|
self.assertEqual(self.calls, [])
|
|
|
|
def test_blank_title_raises_before_any_api_call(self):
|
|
with self.assertRaises(ValueError):
|
|
self._edit(title=" ")
|
|
self.assertEqual(self.calls, [])
|
|
|
|
def test_authorization_failure_blocks_before_any_api_call(self):
|
|
result = self._edit(profile="read-only-author", title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertIn("permission_report", result)
|
|
self.assertEqual(
|
|
result["permission_report"]["missing_permission"],
|
|
task_capability_map.required_permission("edit_issue"),
|
|
)
|
|
self.assertEqual(self.calls, [])
|
|
|
|
def test_pull_request_target_is_refused_without_a_patch(self):
|
|
self.issue["pull_request"] = {"merged": False}
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertEqual(self._patch_methods(), [])
|
|
self.assertIn("gitea_edit_pr", result["safe_next_action"])
|
|
|
|
def test_pre_read_transport_error_is_reported(self):
|
|
def boom(method, url, auth, payload=None):
|
|
raise RuntimeError("connection reset by peer")
|
|
|
|
with patch("mcp_server.api_request", side_effect=boom):
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertTrue(result["reasons"])
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_patch_transport_error_is_reported_not_swallowed(self):
|
|
def flaky(method, url, auth, payload=None):
|
|
if method == "PATCH":
|
|
raise RuntimeError("gitea exploded")
|
|
return self._api(method, url, auth, payload)
|
|
|
|
with patch("mcp_server.api_request", side_effect=flaky):
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertIn("issue edit failed", result["reasons"][0])
|
|
self.assertEqual(self.issue["title"], ORIGINAL_TITLE)
|
|
|
|
def test_read_back_transport_error_reports_an_unverified_edit(self):
|
|
state = {"gets": 0}
|
|
|
|
def flaky(method, url, auth, payload=None):
|
|
if method == "GET" and "/issues/" in url:
|
|
state["gets"] += 1
|
|
if state["gets"] > 1:
|
|
raise RuntimeError("read timed out")
|
|
return self._api(method, url, auth, payload)
|
|
|
|
with patch("mcp_server.api_request", side_effect=flaky):
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(result["performed"])
|
|
self.assertFalse(result["verified"])
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_unapplied_edit_fails_verification(self):
|
|
def sticky(method, url, auth, payload=None):
|
|
if method == "PATCH":
|
|
self.calls.append((method, url))
|
|
# Report success but store nothing.
|
|
return dict(self.issue)
|
|
return self._api(method, url, auth, payload)
|
|
|
|
with patch("mcp_server.api_request", side_effect=sticky):
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(result["performed"])
|
|
self.assertFalse(result["verified"])
|
|
self.assertEqual(
|
|
result["read_after_write"]["mismatches"][0]["field"], "title"
|
|
)
|
|
|
|
def test_edit_that_drops_a_label_fails_verification(self):
|
|
def label_eating(method, url, auth, payload=None):
|
|
if method == "PATCH":
|
|
self.calls.append((method, url))
|
|
self.issue.update(payload or {})
|
|
self.issue["labels"] = []
|
|
return dict(self.issue)
|
|
return self._api(method, url, auth, payload)
|
|
|
|
with patch("mcp_server.api_request", side_effect=label_eating):
|
|
result = self._edit(title=NEW_TITLE)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["read_after_write"]["preserved_intact"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registration and gate wiring
|
|
# ---------------------------------------------------------------------------
|
|
class TestEditIssueRegistrationAndGates(unittest.TestCase):
|
|
def test_tool_is_registered(self):
|
|
self.assertIn("gitea_edit_issue", _registered_tool_names())
|
|
|
|
def test_resolver_task_exists_with_author_role(self):
|
|
self.assertEqual(
|
|
task_capability_map.required_permission("edit_issue"),
|
|
"gitea.issue.comment",
|
|
)
|
|
self.assertEqual(task_capability_map.required_role("edit_issue"), "author")
|
|
|
|
def test_tool_gate_matches_the_resolver_task(self):
|
|
self.assertEqual(
|
|
task_capability_map.ISSUE_MUTATION_TOOL_TASKS["gitea_edit_issue"],
|
|
"edit_issue",
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.tool_required_permission("gitea_edit_issue"),
|
|
task_capability_map.required_permission("edit_issue"),
|
|
)
|
|
|
|
def test_declared_as_an_anti_stomp_mutation_task(self):
|
|
self.assertIn("edit_issue", anti_stomp_preflight.MUTATION_TASKS)
|
|
|
|
def test_edit_pr_remains_pull_request_only(self):
|
|
import inspect
|
|
|
|
params = inspect.signature(mcp_server.gitea_edit_pr).parameters
|
|
self.assertIn("pr_number", params)
|
|
self.assertNotIn("issue_number", params)
|
|
|
|
def test_edit_issue_cannot_change_state_or_labels(self):
|
|
import inspect
|
|
|
|
params = inspect.signature(mcp_server.gitea_edit_issue).parameters
|
|
self.assertEqual(
|
|
[name for name in params if name in ("title", "body")],
|
|
["title", "body"],
|
|
)
|
|
for forbidden in ("state", "labels", "assignee", "assignees", "milestone"):
|
|
self.assertNotIn(forbidden, params)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The drift guard itself
|
|
# ---------------------------------------------------------------------------
|
|
class TestInventoryDriftRule(unittest.TestCase):
|
|
def test_missing_markers_fail_closed(self):
|
|
with self.assertRaises(ValueError):
|
|
mcp_tool_inventory.parse_documented_inventory("no markers here")
|
|
|
|
def test_documented_but_unregistered_is_drift(self):
|
|
result = mcp_tool_inventory.assess_inventory_drift(
|
|
["gitea_edit_issue", "gitea_view_issue"], ["gitea_view_issue"]
|
|
)
|
|
self.assertFalse(result["in_sync"])
|
|
self.assertEqual(result["documented_not_registered"], ["gitea_edit_issue"])
|
|
self.assertTrue(result["safe_next_action"])
|
|
|
|
def test_registered_but_undocumented_is_drift(self):
|
|
result = mcp_tool_inventory.assess_inventory_drift(
|
|
["gitea_view_issue"], ["gitea_view_issue", "gitea_edit_issue"]
|
|
)
|
|
self.assertFalse(result["in_sync"])
|
|
self.assertEqual(result["registered_not_documented"], ["gitea_edit_issue"])
|
|
|
|
def test_unsorted_inventory_is_drift(self):
|
|
result = mcp_tool_inventory.assess_inventory_drift(
|
|
["gitea_view_issue", "gitea_edit_issue"],
|
|
["gitea_view_issue", "gitea_edit_issue"],
|
|
)
|
|
self.assertFalse(result["in_sync"])
|
|
self.assertFalse(result["sorted"])
|
|
|
|
def test_module_names_are_not_treated_as_tools(self):
|
|
self.assertFalse(mcp_tool_inventory.looks_like_tool_name("gitea_auth"))
|
|
self.assertTrue(mcp_tool_inventory.looks_like_tool_name("gitea_view_issue"))
|
|
|
|
def test_unregistered_doc_reference_is_reported(self):
|
|
result = mcp_tool_inventory.assess_doc_references(
|
|
{"skills/example.md": {"gitea_edit_issue"}}, ["gitea_view_issue"]
|
|
)
|
|
self.assertFalse(result["clean"])
|
|
self.assertEqual(result["unregistered"][0]["tool"], "gitea_edit_issue")
|
|
|
|
def test_rendered_block_round_trips(self):
|
|
block = mcp_tool_inventory.render_inventory_block(
|
|
["gitea_view_issue", "gitea_edit_issue"]
|
|
)
|
|
self.assertEqual(
|
|
mcp_tool_inventory.parse_documented_inventory(block),
|
|
["gitea_edit_issue", "gitea_view_issue"],
|
|
)
|
|
|
|
|
|
class TestDocumentationMatchesRegistry(unittest.TestCase):
|
|
"""The live guard: docs and the registry must not drift apart."""
|
|
|
|
def test_documented_inventory_equals_registered_tools(self):
|
|
doc = REPO_ROOT / mcp_tool_inventory.INVENTORY_DOC_PATH
|
|
self.assertTrue(doc.exists(), f"{doc} is missing")
|
|
documented = mcp_tool_inventory.parse_documented_inventory(
|
|
doc.read_text(encoding="utf-8")
|
|
)
|
|
result = mcp_tool_inventory.assess_inventory_drift(
|
|
documented, _registered_tool_names()
|
|
)
|
|
self.assertTrue(result["in_sync"], "; ".join(result["reasons"]))
|
|
|
|
def test_every_tool_named_in_the_skills_is_registered(self):
|
|
references: dict[str, set[str]] = {}
|
|
pattern = str(REPO_ROOT / "skills" / "**" / "*.md")
|
|
paths = glob.glob(pattern, recursive=True)
|
|
self.assertTrue(paths, "no skill documents found to check")
|
|
for path in paths:
|
|
text = Path(path).read_text(encoding="utf-8")
|
|
names = mcp_tool_inventory.extract_tool_references(text)
|
|
if names:
|
|
references[str(Path(path).relative_to(REPO_ROOT))] = names
|
|
result = mcp_tool_inventory.assess_doc_references(
|
|
references, _registered_tool_names()
|
|
)
|
|
self.assertTrue(result["clean"], "; ".join(result["reasons"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|