diff --git a/tests/test_credentials.py b/tests/test_credentials.py index 5f93178..b054917 100644 --- a/tests/test_credentials.py +++ b/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") diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index e2b5e30..aa90d88 100644 --- a/tests/test_mcp_server.py +++ b/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,37 +2812,49 @@ 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 self.assertRaises(RuntimeError) as ctx: - gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs") + 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 self.assertRaises(ValueError) as ctx: - gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs") + 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) - 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)) + 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 self.assertRaises(ValueError) as ctx: - gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs") + 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))