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()