sysadminandClaude Opus 4.8
2b04108f93
test_skip_scan_walk_root_skips_sibling_worktrees_only derived its fixture paths from __file__, so the branches/ sibling-worktree skip assertion only held when the suite ran from a branches/<slug> worktree whose orchestration checkout has a real branches/ directory — skip_python_scan_walk_root skips a branches/ walk root only when <project_root>/branches exists on disk. From the main checkout or any plain clone the test failed with 'AssertionError: False is not true'. Build the <main>/branches/<worktree> layout in a TemporaryDirectory and assert against those paths instead. All four behavioural assertions are preserved and no production code changes. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
259 lines
10 KiB
Python
259 lines
10 KiB
Python
"""Tests for pre-task role/session router (#206)."""
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_config
|
|
import mcp_server
|
|
import role_session_router
|
|
from review_proofs import assess_role_route_handoff
|
|
|
|
CONFIG = {
|
|
"version": 2,
|
|
"contexts": {
|
|
"ctx": {
|
|
"enabled": True,
|
|
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
|
}
|
|
},
|
|
"profiles": {
|
|
"prgs-author": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "author",
|
|
"username": "jcwalker3",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.issue.create", "gitea.pr.create",
|
|
"gitea.branch.push", "gitea.issue.comment",
|
|
],
|
|
"forbidden_operations": [
|
|
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
|
|
],
|
|
"execution_profile": "prgs-author",
|
|
},
|
|
"prgs-reviewer": {
|
|
"enabled": True,
|
|
"context": "ctx",
|
|
"role": "reviewer",
|
|
"username": "sysadmin",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
|
"allowed_operations": [
|
|
"gitea.read", "gitea.pr.review", "gitea.pr.approve",
|
|
"gitea.pr.merge", "gitea.issue.comment",
|
|
],
|
|
"forbidden_operations": [
|
|
"gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
|
|
],
|
|
"execution_profile": "prgs-reviewer",
|
|
},
|
|
},
|
|
"rules": {"allow_runtime_switching": False},
|
|
}
|
|
|
|
|
|
class TestRoleSessionRouter(unittest.TestCase):
|
|
def setUp(self):
|
|
role_session_router.clear_route_state()
|
|
self._remotes = patch.dict(mcp_server.REMOTES, {
|
|
"prgs": {
|
|
"host": "gitea.example.com",
|
|
"org": "Scaled-Tech-Consulting",
|
|
"repo": "Gitea-Tools",
|
|
},
|
|
})
|
|
self._remotes.start()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
self._dir = tempfile.TemporaryDirectory()
|
|
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
|
with open(self.config_path, "w", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(CONFIG))
|
|
|
|
def tearDown(self):
|
|
self._remotes.stop()
|
|
role_session_router.clear_route_state()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
self._dir.cleanup()
|
|
|
|
def _env(self, profile):
|
|
return {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": profile,
|
|
"GITEA_TOKEN_AUTHOR": "author-pass",
|
|
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
|
}
|
|
|
|
def test_reviewer_task_under_author_profile_wrong_role_stop(self):
|
|
with patch.dict(os.environ, self._env("prgs-author")):
|
|
route = mcp_server.gitea_route_task_session(
|
|
task_type="review_pr", remote="prgs"
|
|
)
|
|
self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
|
|
self.assertFalse(route["downstream_allowed"])
|
|
self.assertIn("Wrong role/session for reviewer task", route["message"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_issue_creation_blocked_after_reviewer_wrong_role_stop(
|
|
self, _auth, mock_api
|
|
):
|
|
with patch.dict(os.environ, self._env("prgs-author")):
|
|
mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
|
|
result = mcp_server.gitea_create_issue(
|
|
title="process issue fallback",
|
|
body="should not post",
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(result.get("success", True))
|
|
self.assertFalse(result.get("performed", True))
|
|
issue_posts = [
|
|
c for c in mock_api.call_args_list
|
|
if c.args[0] == "POST" and str(c.args[1]).endswith("/issues")
|
|
]
|
|
self.assertEqual(issue_posts, [])
|
|
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.api_request", return_value={"number": 888})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_author_task_allowed_after_capability_resolve(
|
|
self, _auth, _api, _get_all
|
|
):
|
|
"""#228: explicit create_issue capability clears reviewer wrong_role_stop."""
|
|
with patch.dict(os.environ, self._env("prgs-author")):
|
|
mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
|
|
cap = mcp_server.gitea_resolve_task_capability(
|
|
task="create_issue", remote="prgs"
|
|
)
|
|
self.assertTrue(cap["allowed_in_current_session"])
|
|
result = mcp_server.gitea_create_issue(
|
|
title="operation-scoped author recovery",
|
|
remote="prgs",
|
|
)
|
|
self.assertEqual(result.get("number"), 888)
|
|
|
|
@patch("mcp_server.api_get_all", return_value=[])
|
|
@patch("mcp_server.api_request", return_value={"number": 999})
|
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
|
def test_author_task_allowed_after_explicit_author_route(
|
|
self, _auth, _api, _get_all
|
|
):
|
|
with patch.dict(os.environ, self._env("prgs-author")):
|
|
route = mcp_server.gitea_route_task_session(
|
|
task_type="create_issue", remote="prgs"
|
|
)
|
|
self.assertEqual(
|
|
route["route_result"], role_session_router.ROUTE_ALLOWED
|
|
)
|
|
result = mcp_server.gitea_create_issue(
|
|
title="legit author task",
|
|
remote="prgs",
|
|
)
|
|
self.assertEqual(result.get("number"), 999)
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("prgs-reviewer")):
|
|
route = mcp_server.gitea_route_task_session(
|
|
task_type="review_pr", remote="prgs"
|
|
)
|
|
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
|
|
self.assertTrue(route["downstream_allowed"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
|
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
|
def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
|
|
with patch.dict(os.environ, self._env("prgs-reviewer")):
|
|
route = mcp_server.gitea_route_task_session(
|
|
task_type="create_issue", remote="prgs"
|
|
)
|
|
self.assertEqual(
|
|
route["route_result"], role_session_router.ROUTE_TO_AUTHOR
|
|
)
|
|
self.assertFalse(route["downstream_allowed"])
|
|
|
|
def test_activate_profile_blocked_in_static_mode(self):
|
|
with patch.dict(os.environ, self._env("prgs-author")):
|
|
result = mcp_server.gitea_activate_profile(
|
|
profile_name="prgs-reviewer", remote="prgs"
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertIn("switching is disabled", result["message"])
|
|
|
|
def test_handoff_requires_route_fields(self):
|
|
incomplete = assess_role_route_handoff("Task done.")
|
|
self.assertFalse(incomplete["complete"])
|
|
self.assertIn("Task type", incomplete["missing_fields"])
|
|
|
|
complete = assess_role_route_handoff(
|
|
"\n".join([
|
|
"Task type: review_pr",
|
|
"Required role: reviewer",
|
|
"Active role: author",
|
|
"Route result: wrong_role_stop",
|
|
]),
|
|
route_result={"route_result": "wrong_role_stop"},
|
|
)
|
|
self.assertTrue(complete["complete"])
|
|
|
|
|
|
class TestCheckMidMerge(unittest.TestCase):
|
|
def test_skip_scan_walk_root_skips_sibling_worktrees_only(self):
|
|
# Hermetic <main>/branches/<worktree> layout (#283): deriving these
|
|
# paths from __file__ made the test pass only when the suite ran from
|
|
# a branches/ worktree, because skip_python_scan_walk_root skips a
|
|
# branches/ walk root only when <project_root>/branches exists.
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
main_root = os.path.join(tmp, "main")
|
|
worktree_root = os.path.join(main_root, "branches", "fix-issue-1")
|
|
os.makedirs(os.path.join(worktree_root, "tests"))
|
|
|
|
self.assertFalse(
|
|
role_session_router.skip_python_scan_walk_root(
|
|
main_root, main_root
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
role_session_router.skip_python_scan_walk_root(
|
|
main_root, os.path.join(main_root, "branches", "fix-issue-1")
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
role_session_router.skip_python_scan_walk_root(
|
|
worktree_root, worktree_root
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
role_session_router.skip_python_scan_walk_root(
|
|
worktree_root, os.path.join(worktree_root, "tests")
|
|
)
|
|
)
|
|
|
|
def test_decorative_equals_banner_is_not_mid_merge(self):
|
|
self.assertFalse(role_session_router.check_mid_merge())
|
|
|
|
def test_python_bytes_have_conflict_markers_rejects_decorative_equals(self):
|
|
banner = b"===========================================\n"
|
|
self.assertFalse(role_session_router.python_bytes_have_conflict_markers(banner))
|
|
|
|
def test_python_bytes_have_conflict_markers_detects_real_markers(self):
|
|
self.assertTrue(
|
|
role_session_router.python_bytes_have_conflict_markers(b"<<<<<<< HEAD\n")
|
|
)
|
|
self.assertTrue(
|
|
role_session_router.python_bytes_have_conflict_markers(b"=======\n")
|
|
)
|
|
self.assertTrue(
|
|
role_session_router.python_bytes_have_conflict_markers(b">>>>>>> topic\n")
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |