diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 19d168d..8e233d4 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -257,6 +257,7 @@ import gitea_config # noqa: E402 import capability_stop_terminal # noqa: E402 import issue_duplicate_gate # noqa: E402 import role_session_router # noqa: E402 +import task_capability_map # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -555,6 +556,12 @@ def gitea_create_issue( "number": None, "reasons": block_reasons, } + blocked = _profile_permission_block( + task_capability_map.required_permission("create_issue"), + number=None, + ) + if blocked: + return blocked verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -2494,6 +2501,10 @@ def gitea_close_issue( Returns: dict with 'success' boolean and 'message'. """ + blocked = _profile_permission_block( + task_capability_map.required_permission("close_issue")) + if blocked: + return blocked verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -2698,6 +2709,25 @@ def _profile_operation_gate(op: str) -> list[str]: return [f"profile is not allowed to {op}"] +def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None: + """Structured permission denial for gated tools (#69, #142). + + Returns a block dict when the active profile forbids *required_operation*, + or ``None`` when the gate passes. Never performs network I/O. + """ + reasons = _profile_operation_gate(required_operation) + if not reasons: + return None + blocked = { + "success": False, + "performed": False, + "reasons": reasons, + "permission_report": _permission_block_report(required_operation), + } + blocked.update(extra_fields) + return blocked + + @mcp.tool() def gitea_list_issue_comments( issue_number: int, @@ -3882,6 +3912,10 @@ def gitea_mark_issue( if action not in ("start", "done"): raise ValueError(f"action must be 'start' or 'done', got '{action}'") + blocked = _profile_permission_block( + task_capability_map.required_permission("mark_issue")) + if blocked: + return blocked verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -4005,6 +4039,10 @@ def gitea_set_issue_labels( Returns: list of all labels currently applied to the issue. """ + blocked = _profile_permission_block( + task_capability_map.required_permission("set_issue_labels")) + if blocked: + return blocked verify_preflight_purity(remote) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) @@ -4199,86 +4237,13 @@ def gitea_resolve_task_capability( remote: Known remote instance name. host: Optional override for the Gitea host. """ - TASK_MAP = { - "create_issue": { - "permission": "gitea.issue.create", - "role": "author", - }, - "comment_issue": { - "permission": "gitea.issue.comment", - "role": "author", - }, - "close_issue": { - "permission": "gitea.issue.close", - "role": "author", - }, - "claim_issue": { - "permission": "gitea.issue.comment", - "role": "author", - }, - "mark_issue": { - "permission": "gitea.issue.comment", - "role": "author", - }, - "create_branch": { - "permission": "gitea.branch.create", - "role": "author", - }, - "push_branch": { - "permission": "gitea.branch.push", - "role": "author", - }, - "create_pr": { - "permission": "gitea.pr.create", - "role": "author", - }, - "comment_pr": { - "permission": "gitea.pr.comment", - "role": "author", - }, - # PR closure is a first-class capability (#216): distinct from - # comment_pr (gitea.pr.comment) and from ordinary PR edits, which - # need no dedicated capability. gitea_edit_pr(state='closed') is - # gated on the same operation, so no edit-path fallback exists. - "close_pr": { - "permission": "gitea.pr.close", - "role": "author", - }, - "address_pr_change_requests": { - "permission": "gitea.branch.push", - "role": "author", - }, - "review_pr": { - "permission": "gitea.pr.review", - "role": "reviewer", - }, - "merge_pr": { - "permission": "gitea.pr.merge", - "role": "reviewer", - }, - "blind_pr_queue_review": { - "permission": "gitea.pr.review", - "role": "reviewer", - }, - "request_changes_pr": { - "permission": "gitea.pr.request_changes", - "role": "reviewer", - }, - "approve_pr": { - "permission": "gitea.pr.approve", - "role": "reviewer", - }, - "delete_branch": { - "permission": "gitea.branch.delete", - "role": "author", - }, - } + TASK_MAP = task_capability_map.TASK_CAPABILITY_MAP if task not in TASK_MAP: raise ValueError(f"Unknown task/action: '{task}' (fail closed)") - required_permission = TASK_MAP[task]["permission"] - required_role = TASK_MAP[task]["role"] + required_permission = task_capability_map.required_permission(task) + required_role = task_capability_map.required_role(task) if required_role == "reviewer" and role_session_router.check_mid_merge(): profile = get_profile() diff --git a/task_capability_map.py b/task_capability_map.py new file mode 100644 index 0000000..fd3c344 --- /dev/null +++ b/task_capability_map.py @@ -0,0 +1,113 @@ +"""Shared task→permission map for resolver and tool gates (#69). + +``gitea_resolve_task_capability`` and issue-mutating MCP tools must agree on +which profile operation each task requires. This module is the single source +of truth; regression tests assert tool gates cannot drift from it. +""" + +from __future__ import annotations + +TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { + "create_issue": { + "permission": "gitea.issue.create", + "role": "author", + }, + "comment_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "close_issue": { + "permission": "gitea.issue.close", + "role": "author", + }, + "claim_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "mark_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "set_issue_labels": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "create_branch": { + "permission": "gitea.branch.create", + "role": "author", + }, + "push_branch": { + "permission": "gitea.branch.push", + "role": "author", + }, + "create_pr": { + "permission": "gitea.pr.create", + "role": "author", + }, + "comment_pr": { + "permission": "gitea.pr.comment", + "role": "author", + }, + "close_pr": { + "permission": "gitea.pr.close", + "role": "author", + }, + "address_pr_change_requests": { + "permission": "gitea.branch.push", + "role": "author", + }, + "review_pr": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, + "merge_pr": { + "permission": "gitea.pr.merge", + "role": "reviewer", + }, + "blind_pr_queue_review": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, + "request_changes_pr": { + "permission": "gitea.pr.request_changes", + "role": "reviewer", + }, + "approve_pr": { + "permission": "gitea.pr.approve", + "role": "reviewer", + }, + "delete_branch": { + "permission": "gitea.branch.delete", + "role": "author", + }, +} + +# Issue-mutating MCP tools and their resolver task keys. +ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = { + "gitea_create_issue": "create_issue", + "gitea_close_issue": "close_issue", + "gitea_create_issue_comment": "comment_issue", + "gitea_mark_issue": "mark_issue", + "gitea_set_issue_labels": "set_issue_labels", +} + + +def required_permission(task: str) -> str: + """Return the canonical operation a *task* requires (fail closed).""" + try: + return TASK_CAPABILITY_MAP[task]["permission"] + except KeyError as exc: + raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc + + +def required_role(task: str) -> str: + """Return author/reviewer role kind for *task* (fail closed).""" + try: + return TASK_CAPABILITY_MAP[task]["role"] + except KeyError as exc: + raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc + + +def tool_required_permission(tool_name: str) -> str: + """Return the operation an issue-mutating tool must gate on.""" + return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name]) \ No newline at end of file diff --git a/tests/test_audit.py b/tests/test_audit.py index 6594739..4efbe65 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -162,7 +162,8 @@ class _AuditWiringBase(unittest.TestCase): def _env(self, **extra): env = {"GITEA_AUDIT_LOG": self.audit_path, "GITEA_PROFILE_NAME": "gitea-author", - "GITEA_ALLOWED_OPERATIONS": "read,merge"} + "GITEA_ALLOWED_OPERATIONS": ( + "read,merge,gitea.issue.create,gitea.issue.close")} env.update(extra) return env @@ -237,7 +238,10 @@ class TestSimpleToolAudit(_AuditWiringBase): def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role): # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file. mock_api.return_value = {"number": 1, "html_url": "http://x/1"} - with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True): + with patch.dict(os.environ, { + "GITEA_PROFILE_NAME": "gitea-author", + "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", + }, clear=True): gitea_create_issue(title="x", remote="prgs") issue_posts = [ c for c in mock_api.call_args_list if c.args[0] == "POST" diff --git a/tests/test_issue_duplicate_gate.py b/tests/test_issue_duplicate_gate.py index 1ddb39c..fe13303 100644 --- a/tests/test_issue_duplicate_gate.py +++ b/tests/test_issue_duplicate_gate.py @@ -19,6 +19,9 @@ from mcp_server import gitea_create_issue # noqa: E402 from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402 FAKE_AUTH = "Basic dGVzdDp0ZXN0" +ISSUE_WRITE_ENV = { + "GITEA_ALLOWED_OPERATIONS": "gitea.issue.create", +} CANONICAL_TITLE = ( "Add hard queue-target resolution wall before PR inventory " "or empty-queue claims" @@ -118,7 +121,8 @@ class TestCreateIssueMCPGate(unittest.TestCase): mock_get_all.return_value = [ {"number": 201, "title": CANONICAL_TITLE, "state": "open"}, ] - result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs") + with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs") self.assertFalse(result["performed"]) self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE) mock_api.assert_not_called() @@ -135,13 +139,14 @@ class TestCreateIssueMCPGate(unittest.TestCase): {"number": 201, "title": CANONICAL_TITLE, "state": "open"}, ] mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"} - result = gitea_create_issue( - title=CANONICAL_TITLE, - body="Follow-up scope", - remote="prgs", - allow_duplicate_override=True, - split_from_issue=201, - ) + with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_create_issue( + title=CANONICAL_TITLE, + body="Follow-up scope", + remote="prgs", + allow_duplicate_override=True, + split_from_issue=201, + ) self.assertEqual(result["number"], 210) payload = mock_api.call_args[0][3] self.assertIn("Operator-approved split from #201.", payload["body"]) @@ -156,7 +161,8 @@ class TestCreateIssueMCPGate(unittest.TestCase): ): mock_role_check.return_value = (True, []) mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} - result = gitea_create_issue(title="Unique new issue", body="body text") + with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_create_issue(title="Unique new issue", body="body text") self.assertEqual(result["number"], 1) diff --git a/tests/test_issue_write_tool_gates.py b/tests/test_issue_write_tool_gates.py new file mode 100644 index 0000000..afe865d --- /dev/null +++ b/tests/test_issue_write_tool_gates.py @@ -0,0 +1,210 @@ +"""Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69).""" +import json +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import mcp_server +import task_capability_map + +CONFIG = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "comment-only": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "commenter", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read", "gitea.issue.comment"], + "forbidden_operations": [ + "gitea.issue.create", "gitea.issue.close"], + "execution_profile": "comment-only", + }, + "full-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": [ + "gitea.read", "gitea.issue.create", "gitea.issue.close", + "gitea.issue.comment"], + "forbidden_operations": [], + "execution_profile": "full-author", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + +ISSUE_WRITE_TASKS = ( + "create_issue", "close_issue", "comment_issue", "mark_issue", + "set_issue_labels", +) + +TOOL_CALLS = { + "create_issue": lambda: mcp_server.gitea_create_issue( + title="New issue", remote="prgs"), + "close_issue": lambda: mcp_server.gitea_close_issue( + issue_number=9, remote="prgs"), + "comment_issue": lambda: mcp_server.gitea_create_issue_comment( + issue_number=9, body="hello", remote="prgs"), + "mark_issue": lambda: mcp_server.gitea_mark_issue( + issue_number=9, action="start", remote="prgs"), + "set_issue_labels": lambda: mcp_server.gitea_set_issue_labels( + issue_number=9, labels=["bug"], remote="prgs"), +} + + +class IssueWriteGateBase(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)) + + def tearDown(self): + self._remotes.stop() + mcp_server._IDENTITY_CACHE.clear() + self._dir.cleanup() + + def _env(self, profile: str) -> dict: + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + } + + +class TestResolverToolAlignment(unittest.TestCase): + def test_issue_mutation_tools_match_resolver_permissions(self): + for tool_name, task_key in ( + task_capability_map.ISSUE_MUTATION_TOOL_TASKS.items() + ): + self.assertEqual( + task_capability_map.tool_required_permission(tool_name), + task_capability_map.required_permission(task_key), + msg=f"{tool_name} gate drift from resolver task {task_key}", + ) + + +class TestDeniedIssueWritesFailClosed(IssueWriteGateBase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolver_denied_tasks_block_raw_tools( + self, _auth, mock_api, _get_all, _role + ): + mock_api.return_value = {"number": 1} + with patch.dict(os.environ, self._env("comment-only"), clear=True): + for task in ISSUE_WRITE_TASKS: + with self.subTest(task=task): + resolve = mcp_server.gitea_resolve_task_capability( + task=task, remote="prgs") + if resolve["allowed_in_current_session"]: + continue + result = TOOL_CALLS[task]() + self.assertFalse(result.get("success", True), task) + self.assertFalse(result.get("performed", True), task) + self.assertTrue(result.get("reasons"), task) + self.assertIn("permission_report", result, task) + self.assertEqual( + result["permission_report"]["missing_permission"], + resolve["required_operation_permission"], + ) + + +class TestAllowedIssueWritesProceed(IssueWriteGateBase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolver_allowed_create_issue_proceeds( + self, _auth, mock_api, _get_all, _role + ): + mock_api.return_value = {"number": 42, "html_url": "https://x/42"} + with patch.dict(os.environ, self._env("full-author"), clear=True): + resolve = mcp_server.gitea_resolve_task_capability( + task="create_issue", remote="prgs") + self.assertTrue(resolve["allowed_in_current_session"]) + result = mcp_server.gitea_create_issue(title="Allowed", remote="prgs") + self.assertEqual(result.get("number"), 42) + post_calls = [ + c for c in mock_api.call_args_list if c.args[0] == "POST" + ] + self.assertTrue(post_calls) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolver_allowed_close_issue_proceeds(self, _auth, mock_api): + mock_api.return_value = {"state": "closed"} + with patch.dict(os.environ, self._env("full-author"), clear=True): + resolve = mcp_server.gitea_resolve_task_capability( + task="close_issue", remote="prgs") + self.assertTrue(resolve["allowed_in_current_session"]) + result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs") + self.assertTrue(result.get("success")) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api): + def api_side_effect(method, url, auth, payload=None): + if method == "GET" and url.endswith("/labels?limit=100"): + return [{"id": 10, "name": "status:in-progress"}] + if method == "POST" and "/issues/9/labels" in url: + return [{"name": "status:in-progress"}] + if method == "GET" and url.endswith("/user"): + return {"login": "author-user"} + return {} + mock_api.side_effect = api_side_effect + with patch.dict(os.environ, self._env("full-author"), clear=True): + resolve = mcp_server.gitea_resolve_task_capability( + task="mark_issue", remote="prgs") + self.assertTrue(resolve["allowed_in_current_session"]) + result = mcp_server.gitea_mark_issue( + issue_number=9, action="start", remote="prgs") + self.assertTrue(result.get("success")) + + +class TestCreateIssueMissingPermissionReport(IssueWriteGateBase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_create_issue_without_create_permission_blocked( + self, _auth, mock_api, _get_all, _role + ): + with patch.dict(os.environ, self._env("comment-only"), clear=True): + result = mcp_server.gitea_create_issue(title="Blocked", remote="prgs") + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertIsNone(result["number"]) + self.assertEqual( + result["permission_report"]["missing_permission"], + "gitea.issue.create", + ) + mock_api.assert_not_called() + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 8aeae6b..e2b5e30 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -47,6 +47,13 @@ import mcp_server FAKE_AUTH = "Basic dGVzdDp0ZXN0" +# Issue-write tools are profile-gated (#69). +ISSUE_WRITE_ENV = { + "GITEA_ALLOWED_OPERATIONS": ( + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" + ), +} + # --------------------------------------------------------------------------- # Create Issue @@ -60,7 +67,7 @@ class TestCreateIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_creates_issue(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} - with patch.dict(os.environ, {}, clear=True): + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): result = gitea_create_issue(title="Test issue", body="body text") self.assertEqual(result["number"], 1) self.assertNotIn("url", result) @@ -77,7 +84,8 @@ class TestCreateIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} - with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True): + env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} + with patch.dict(os.environ, env, clear=True): result = gitea_create_issue(title="Test issue", body="body text") self.assertIn("issues/1", result["url"]) @@ -88,7 +96,8 @@ class TestCreateIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"} - result = gitea_create_issue(title="Test", remote="prgs") + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_create_issue(title="Test", remote="prgs") self.assertEqual(result["number"], 5) url = mock_api.call_args[0][1] self.assertIn("gitea.prgs.cc", url) @@ -136,7 +145,8 @@ class TestCloseIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_closes_issue(self, _auth, mock_api): mock_api.return_value = {"state": "closed"} - result = gitea_close_issue(issue_number=42) + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_close_issue(issue_number=42) self.assertTrue(result["success"]) self.assertIn("42", result["message"]) patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH") @@ -235,7 +245,8 @@ class TestMarkIssue(unittest.TestCase): [{"id": 10, "name": "status:in-progress"}], [{"name": "status:in-progress"}], ] - result = gitea_mark_issue(issue_number=5, action="start") + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_mark_issue(issue_number=5, action="start") self.assertTrue(result["success"]) self.assertIn("claimed", result["message"]) @@ -246,7 +257,8 @@ class TestMarkIssue(unittest.TestCase): [{"id": 10, "name": "status:in-progress"}], None, ] - result = gitea_mark_issue(issue_number=5, action="done") + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + result = gitea_mark_issue(issue_number=5, action="done") self.assertTrue(result["success"]) self.assertIn("released", result["message"]) @@ -258,8 +270,9 @@ class TestMarkIssue(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_missing_label_raises(self, _auth, mock_api): mock_api.return_value = [] # no labels exist - with self.assertRaises(RuntimeError): - gitea_mark_issue(issue_number=5, action="start") + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with self.assertRaises(RuntimeError): + gitea_mark_issue(issue_number=5, action="start") # --------------------------------------------------------------------------- @@ -269,12 +282,14 @@ class TestAuthErrors(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=None) def test_no_credentials_raises(self, _auth): - with self.assertRaises(RuntimeError): - gitea_create_issue(title="test") + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with self.assertRaises(RuntimeError): + gitea_create_issue(title="test") def test_unknown_remote_raises(self): - with self.assertRaises(ValueError): - gitea_create_issue(title="test", remote="nonexistent") + with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): + with self.assertRaises(ValueError): + gitea_create_issue(title="test", remote="nonexistent") # --------------------------------------------------------------------------- @@ -1939,7 +1954,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase): patch("gitea_audit.audit_enabled", return_value=True).start() self.mock_audit = patch("gitea_audit.write_event").start() # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216). - patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start() + patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start() def tearDown(self): patch.stopall()