diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index d9f1029..47e24b7 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -5,6 +5,7 @@ the MCP protocol) with mocked API responses. """ import json import os +os.environ["GITEA_TEST_ENVIRONMENT"] = "1" import sys import unittest from unittest.mock import patch, MagicMock @@ -2475,3 +2476,88 @@ class TestIssueLocking(unittest.TestCase): 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)) + + +class TestPreflightVerification(unittest.TestCase): + """Test workspace edits and pre-flight ordering verification.""" + + def setUp(self): + self.preflight_path = "/tmp/gitea_preflight_check.json" + if os.path.exists(self.preflight_path): + os.remove(self.preflight_path) + os.environ["GITEA_TEST_FORCE_DIRTY"] = "1" + + def tearDown(self): + if os.path.exists(self.preflight_path): + os.remove(self.preflight_path) + os.environ.pop("GITEA_TEST_FORCE_DIRTY", None) + + @patch("subprocess.run") + def test_record_preflight_detects_violation_whoami(self, mock_run): + mock_run.return_value = MagicMock(stdout="M mcp_server.py\n") + from mcp_server import record_preflight_check, verify_preflight_purity + record_preflight_check("whoami") + with open(self.preflight_path, "r") as f: + data = json.load(f) + self.assertTrue(data.get("whoami_preflight_violation")) + self.assertTrue(data.get("whoami_called")) + + data["capability_called"] = True + with open(self.preflight_path, "w") as f: + json.dump(data, f) + + with self.assertRaises(RuntimeError) as ctx: + verify_preflight_purity("prgs") + self.assertIn("Workspace file edits occurred before gitea_whoami", str(ctx.exception)) + + @patch("subprocess.run") + def test_record_preflight_detects_violation_capability(self, mock_run): + mock_run.return_value = MagicMock(stdout="M mcp_server.py\n") + from mcp_server import record_preflight_check, verify_preflight_purity + record_preflight_check("capability", resolved_role="author") + with open(self.preflight_path, "r") as f: + data = json.load(f) + self.assertTrue(data.get("capability_preflight_violation")) + self.assertTrue(data.get("capability_called")) + self.assertEqual(data.get("role"), "author") + + data["whoami_called"] = True + with open(self.preflight_path, "w") as f: + json.dump(data, f) + + with self.assertRaises(RuntimeError) as ctx: + verify_preflight_purity("prgs") + self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability", str(ctx.exception)) + + @patch("subprocess.run") + def test_verify_preflight_reviewer_edits_blocked(self, mock_run): + with open(self.preflight_path, "w") as f: + json.dump({ + "whoami_called": True, + "capability_called": True, + "role": "reviewer" + }, f) + mock_run.return_value = MagicMock(stdout="M review_proofs.py\n") + from mcp_server import verify_preflight_purity + with self.assertRaises(RuntimeError) as ctx: + verify_preflight_purity("prgs") + self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception)) + + @patch("subprocess.run") + def test_verify_preflight_clean_reviewer_allowed(self, mock_run): + with open(self.preflight_path, "w") as f: + json.dump({ + "whoami_called": True, + "capability_called": True, + "role": "reviewer" + }, f) + mock_run.return_value = MagicMock(stdout="") + from mcp_server import verify_preflight_purity + verify_preflight_purity("prgs") + + def test_verify_preflight_skipped_fails(self): + from mcp_server import verify_preflight_purity + with self.assertRaises(RuntimeError) as ctx: + verify_preflight_purity("prgs") + self.assertIn("verification were skipped", str(ctx.exception)) +