59 lines
2.5 KiB
Python
59 lines
2.5 KiB
Python
import os
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import role_session_router
|
|
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
|
|
|
|
class TestMCPHealth(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
self.temp_file = os.path.join(self.project_root, "test_temp_conflict.py")
|
|
if os.path.exists(self.temp_file):
|
|
os.remove(self.temp_file)
|
|
|
|
def tearDown(self):
|
|
if os.path.exists(self.temp_file):
|
|
os.remove(self.temp_file)
|
|
|
|
def test_startup_conflict_detection(self):
|
|
# Create a Python file with conflict markers constructed dynamically
|
|
with open(self.temp_file, "w") as f:
|
|
f.write("<" * 7 + " HEAD\n")
|
|
f.write("print('hello')\n")
|
|
f.write("=" * 7 + "\n")
|
|
f.write("print('world')\n")
|
|
f.write(">" * 7 + " main\n")
|
|
|
|
# Run mcp_server.py
|
|
venv_python = os.path.join(self.project_root, "venv", "bin", "python")
|
|
script_path = os.path.join(self.project_root, "mcp_server.py")
|
|
res = subprocess.run(
|
|
[venv_python, script_path],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
cwd=self.project_root
|
|
)
|
|
|
|
self.assertEqual(res.returncode, 1)
|
|
stderr = res.stderr.decode()
|
|
self.assertIn("infra_stop", stderr)
|
|
self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
|
|
|
|
@patch("role_session_router.check_mid_merge", return_value=True)
|
|
@patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
|
|
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
|
|
res = gitea_route_task_session("review_pr")
|
|
self.assertEqual(res["route_result"], "infra_stop")
|
|
self.assertIn("infra_stop", res["message"])
|
|
|
|
@patch("role_session_router.check_mid_merge", return_value=True)
|
|
@patch("mcp_server.get_profile", return_value={"profile_name": "prgs-reviewer", "allowed_operations": ["gitea.pr.review"]})
|
|
def test_resolve_task_capability_blocks_during_merge(self, mock_profile, mock_check):
|
|
res = gitea_resolve_task_capability("review_pr")
|
|
self.assertTrue(res["infra_stop"])
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertIn("infra_stop", res["exact_safe_next_action"])
|