tests/test_credentials.py

@@ -107,6 +107,24 @@ class TestGetCredentials(unittest.TestCase):
 class TestGetAuthHeader(unittest.TestCase):
     """Test the get_auth_header function."""
 
+    def setUp(self):
+        self._saved_env = os.environ.copy()
+        self._saved_configs = gitea_auth.DYNAMIC_CONFIGS.copy()
+        for key in (
+            "GITEA_MCP_CONFIG",
+            "GITEA_MCP_PROFILE",
+            "GITEA_TOKEN",
+            "GITEA_TOKEN_PRGS",
+        ):
+            os.environ.pop(key, None)
+        gitea_auth.DYNAMIC_CONFIGS.clear()
+
+    def tearDown(self):
+        os.environ.clear()
+        os.environ.update(self._saved_env)
+        gitea_auth.DYNAMIC_CONFIGS.clear()
+        gitea_auth.DYNAMIC_CONFIGS.update(self._saved_configs)
+
     @patch("gitea_auth.get_credentials", return_value=("user", "pass"))
     def test_returns_basic_header(self, _cred):
         header = gitea_auth.get_auth_header("example.com")

tests/test_mcp_server.py

@@ -54,6 +54,17 @@ ISSUE_WRITE_ENV = {
     ),
 }
 
+# create_pr is gated on author profile + namespace (#69, #209).
+CREATE_PR_ENV = {
+    "GITEA_PROFILE_NAME": "author-test",
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.read,gitea.pr.create,gitea.branch.push"
+    ),
+    "GITEA_FORBIDDEN_OPERATIONS": (
+        "gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
+    ),
+}
+
 
 # ---------------------------------------------------------------------------
 # Create Issue
@@ -109,14 +120,16 @@ class TestCreateIssue(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCreatePR(unittest.TestCase):
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
         mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        with patch.dict(os.environ, {}, clear=True):
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertEqual(result["number"], 3)
         self.assertNotIn("url", result)
@@ -124,14 +137,17 @@ class TestCreatePR(unittest.TestCase):
         self.assertEqual(payload["head"], "feat/x")
         self.assertEqual(payload["base"], "main")
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
         mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
-        with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
+        env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
+        with patch.dict(os.environ, env, clear=True):
             result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
         self.assertIn("pulls/3", result["url"])
 
@@ -2796,35 +2812,47 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_missing_lock_fails(self, _auth):
+    def test_create_pr_missing_lock_fails(self, _auth, _role):
         if os.path.exists("/tmp/gitea_issue_lock.json"):
             os.remove("/tmp/gitea_issue_lock.json")
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(RuntimeError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("Issue lock is missing", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_branch_mismatch_fails(self, _auth):
+    def test_create_pr_branch_mismatch_fails(self, _auth, _role):
         with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
             json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
         self.assertIn("does not match locked branch", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_forbidden_terms_fails(self, _auth):
+    def test_create_pr_forbidden_terms_fails(self, _auth, _role):
         with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
             json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             for term in ("equivalent to #196", "related to #196", "same as #196"):
                 with self.assertRaises(ValueError) as ctx:
                     gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
                 self.assertIn("contains forbidden term", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_create_pr_missing_closes_ref_fails(self, _auth):
+    def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
         with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
             json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
             with self.assertRaises(ValueError) as ctx:
                 gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))