test(preflight): add preflight order verification tests
This commit is contained in:
@@ -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))
|
||||
|
||||
|
||||
Reference in New Issue
Block a user