Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'prgs/master' into feat/issue-334-review-merge-workflow-extract

Browse Source
This commit is contained in:
sysadmin
2026-07-07 04:38:09 -04:00
parent 25c9d20870 eb6f006a31
commit 72c20ee1e1
3 changed files with 185 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_health.py
+93 -5
View File
@@ -1,9 +1,12 @@
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
import capability_stop_terminal
import role_session_router
from role_session_router import python_bytes_have_conflict_markers
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
@@ -117,7 +120,7 @@ class TestMCPHealth(unittest.TestCase):
stderr_text,
)
@patch("role_session_router.check_mid_merge", return_value=True)
@patch("role_session_router.assess_infra_stop")
@patch(
"mcp_server.get_profile",
return_value={
@@ -125,12 +128,23 @@ class TestMCPHealth(unittest.TestCase):
"allowed_operations": ["gitea.pr.review"],
},
)
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_assess):
mock_assess.return_value = {
"infra_stop": True,
"project_root": "/repo",
"conflict_file": None,
"mid_merge": True,
"infra_stop_reasons": [
"mid-merge markers detected in /repo/.git/MERGE_HEAD (project_root=/repo)"
],
}
res = gitea_route_task_session("review_pr")
self.assertEqual(res["route_result"], "infra_stop")
self.assertIn("infra_stop", res["message"])
self.assertTrue(res.get("infra_stop_assessment", {}).get("infra_stop"))
mock_assess.assert_called()
@patch("role_session_router.check_mid_merge", return_value=True)
@patch("role_session_router.assess_infra_stop")
@patch(
"mcp_server.get_profile",
return_value={
@@ -139,9 +153,83 @@ class TestMCPHealth(unittest.TestCase):
},
)
def test_resolve_task_capability_blocks_during_merge(
self, mock_profile, mock_check
self, mock_profile, mock_assess
):
mock_assess.return_value = {
"infra_stop": True,
"project_root": "/repo",
"conflict_file": "/repo/gitea_mcp_server.py",
"mid_merge": False,
"infra_stop_reasons": [
"conflict markers detected in /repo/gitea_mcp_server.py (project_root=/repo)"
],
}
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"])
self.assertIn("infra_stop", res["exact_safe_next_action"])
self.assertEqual(
res["infra_stop_assessment"]["conflict_file"],
"/repo/gitea_mcp_server.py",
)
@patch("role_session_router.assess_infra_stop")
@patch("mcp_server._authenticated_username", return_value="reviewer-user")
@patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.pr.review", "gitea.pr.approve"],
},
)
def test_resolve_task_capability_clears_stale_terminal_when_infra_clean(
self, mock_profile, _username, mock_assess
):
mock_assess.return_value = {
"infra_stop": False,
"project_root": "/repo",
"conflict_file": None,
"mid_merge": False,
"infra_stop_reasons": [],
}
capability_stop_terminal.enter_from_capability_result({
"requested_task": "review_pr",
"required_role_kind": "reviewer",
"stop_required": True,
"active_profile": "prgs-reviewer",
"active_identity": "reviewer-user",
"exact_safe_next_action": "blocked",
})
self.assertTrue(capability_stop_terminal.is_active())
res = gitea_resolve_task_capability("review_pr", remote="prgs")
self.assertFalse(capability_stop_terminal.is_active())
self.assertTrue(res["allowed_in_current_session"])
class TestInfraStopLiveAssessment(unittest.TestCase):
def setUp(self):
self.tempdir = tempfile.mkdtemp()
def tearDown(self):
shutil.rmtree(self.tempdir, ignore_errors=True)
def test_clean_project_root_is_not_infra_stop(self):
infra = role_session_router.assess_infra_stop(self.tempdir)
self.assertFalse(infra["infra_stop"])
self.assertIsNone(infra["conflict_file"])
def test_conflict_present_then_resolved_recomputes_allowed(self):
conflict_path = os.path.join(self.tempdir, "conflicted.py")
with open(conflict_path, "wb") as handle:
handle.write(b"<<<<<<< HEAD\n")
blocked = role_session_router.assess_infra_stop(self.tempdir)
self.assertTrue(blocked["infra_stop"])
self.assertEqual(
os.path.realpath(blocked["conflict_file"]),
os.path.realpath(conflict_path),
)
os.remove(conflict_path)
cleared = role_session_router.assess_infra_stop(self.tempdir)
self.assertFalse(cleared["infra_stop"])
self.assertIsNone(cleared["conflict_file"])