sysadmin
0801e2455b
Resolve gitea_mcp_server.py conflicts after master advanced with #724 (side-effect-free resolver), #725 (merger lease), #728 (PR-sync), and #702 stale-binding recovery: - Keep #714 fail-closed session immutability: no auto profile switch (_ensure_matching_profile / _try_auto_switch_for_operation). - Retain master #709 _authenticated_actor identity helper. - Preserve resolve-path auto_recover=False (#685) and report-only stale_binding_recovery coexistence with runtime_reconnect_required. - Cross-host / cross-repository substitution still fails closed without mutating the pinned session context. - Regression coverage for merge coexistence, dual-module identity-cache isolation in conftest, and structured unknown_task fail-closed. Closes #714 conflict remediation path for PR #715.
746 lines
30 KiB
Python
746 lines
30 KiB
Python
"""#714: fail closed on cross-host MCP profile drift and capability substitution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
import gitea_config # noqa: E402
|
|
import mcp_server # noqa: E402
|
|
import session_context_binding as session_ctx # noqa: E402
|
|
|
|
|
|
CONFIG_714 = {
|
|
"version": 2,
|
|
"allow_runtime_switching": True,
|
|
"contexts": {
|
|
"prgs": {
|
|
"enabled": True,
|
|
"gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
|
|
},
|
|
"mdcps": {
|
|
"enabled": True,
|
|
"gitea": {"enabled": True, "base_url": "https://gitea.dadeschools.net"},
|
|
},
|
|
},
|
|
"profiles": {
|
|
"prgs-author": {
|
|
"enabled": True,
|
|
"context": "prgs",
|
|
"role": "author",
|
|
"username": "jcwalker3",
|
|
"base_url": "https://gitea.prgs.cc",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"},
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.issue.create",
|
|
"gitea.issue.comment",
|
|
"gitea.issue.close",
|
|
"gitea.pr.create",
|
|
"gitea.pr.comment",
|
|
"gitea.branch.create",
|
|
"gitea.branch.push",
|
|
"gitea.repo.commit",
|
|
],
|
|
"forbidden_operations": [
|
|
"gitea.pr.approve",
|
|
"gitea.pr.merge",
|
|
"gitea.pr.request_changes",
|
|
],
|
|
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools"],
|
|
"execution_profile": "prgs-author",
|
|
},
|
|
"mdcps-reviewer": {
|
|
"enabled": True,
|
|
"context": "mdcps",
|
|
"role": "reviewer",
|
|
"username": "913443",
|
|
"base_url": "https://gitea.dadeschools.net",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_REVIEWER"},
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.pr.review",
|
|
"gitea.pr.comment",
|
|
"gitea.pr.approve",
|
|
"gitea.pr.request_changes",
|
|
],
|
|
"forbidden_operations": [
|
|
"gitea.branch.create",
|
|
"gitea.branch.push",
|
|
"gitea.pr.create",
|
|
"gitea.pr.merge",
|
|
"gitea.repo.commit",
|
|
],
|
|
"allowed_repositories": ["913443/eAgenda"],
|
|
"execution_profile": "mdcps-reviewer",
|
|
},
|
|
"mdcps-author": {
|
|
"enabled": True,
|
|
"context": "mdcps",
|
|
"role": "author",
|
|
"username": "913443",
|
|
"base_url": "https://gitea.dadeschools.net",
|
|
"auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_AUTHOR"},
|
|
"allowed_operations": [
|
|
"gitea.read",
|
|
"gitea.issue.create",
|
|
"gitea.issue.comment",
|
|
"gitea.pr.create",
|
|
"gitea.pr.comment",
|
|
"gitea.branch.create",
|
|
"gitea.branch.push",
|
|
"gitea.repo.commit",
|
|
],
|
|
"forbidden_operations": [
|
|
"gitea.pr.approve",
|
|
"gitea.pr.merge",
|
|
"gitea.pr.request_changes",
|
|
],
|
|
"allowed_repositories": ["913443/eAgenda"],
|
|
"execution_profile": "mdcps-author",
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
class TestIssue714SessionContextImmutability(unittest.TestCase):
|
|
def setUp(self):
|
|
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_714))
|
|
self._remotes = patch.dict(
|
|
mcp_server.REMOTES,
|
|
{
|
|
"dadeschools": {
|
|
"host": "gitea.dadeschools.net",
|
|
"org": "913443",
|
|
"repo": "eAgenda",
|
|
},
|
|
"prgs": {
|
|
"host": "gitea.prgs.cc",
|
|
"org": "Scaled-Tech-Consulting",
|
|
"repo": "Gitea-Tools",
|
|
},
|
|
},
|
|
clear=False,
|
|
)
|
|
self._remotes.start()
|
|
def _url(name):
|
|
if name == "dadeschools":
|
|
return "https://gitea.dadeschools.net/913443/eAgenda.git"
|
|
if name == "prgs":
|
|
return "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
|
return None
|
|
self._url_patch = patch.object(mcp_server, "_local_git_remote_url", side_effect=_url)
|
|
self._url_patch.start()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
mcp_server._MUTATION_AUTHORITY = None
|
|
self._env = {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": "mdcps-reviewer",
|
|
"GITEA_TOKEN_MDCPS_REVIEWER": "mdcps-reviewer-token",
|
|
"GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
|
|
"GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
|
|
}
|
|
|
|
def tearDown(self):
|
|
self._url_patch.stop()
|
|
self._remotes.stop()
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
gitea_config._active_profile_override = None
|
|
mcp_server._MUTATION_AUTHORITY = None
|
|
self._dir.cleanup()
|
|
|
|
def _api_side_effect(self, method, url, header):
|
|
# Token identity mapping for whoami endpoint
|
|
auth = str(header)
|
|
if "mdcps-reviewer-token" in auth or "mdcps-author-token" in auth:
|
|
login = "913443"
|
|
else:
|
|
login = "jcwalker3"
|
|
return {"login": login, "full_name": "Test", "id": 1, "email": "[email protected]"}
|
|
|
|
def test_pin_mdcps_reviewer_never_drifts_to_prgs_author(self):
|
|
"""Reproduce the incident: pin mdcps-reviewer then resolve comment_issue."""
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
act = mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
self.assertTrue(act["success"])
|
|
self.assertEqual(act["after_profile"], "mdcps-reviewer")
|
|
|
|
who1 = mcp_server.gitea_whoami(remote="dadeschools")
|
|
self.assertEqual(who1["profile"]["profile_name"], "mdcps-reviewer")
|
|
self.assertEqual(who1["username"], "913443")
|
|
|
|
# Capability that mdcps-reviewer lacks — must NOT switch to prgs-author
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="comment_issue", remote="dadeschools"
|
|
)
|
|
self.assertEqual(res["active_profile"], "mdcps-reviewer")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertFalse(res.get("auto_profile_substitution", True))
|
|
self.assertNotIn("prgs-author", res["matching_configured_profile"])
|
|
self.assertIn("mdcps-author", res["matching_configured_profile"])
|
|
|
|
who2 = mcp_server.gitea_whoami(remote="dadeschools")
|
|
rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
|
|
self.assertEqual(who2["profile"]["profile_name"], "mdcps-reviewer")
|
|
self.assertEqual(rt["active_profile"], "mdcps-reviewer")
|
|
# Still pinned — never prgs-author
|
|
self.assertEqual(
|
|
gitea_config.selected_profile_name(), "mdcps-reviewer"
|
|
)
|
|
|
|
def test_repeated_calls_remain_on_mdcps_reviewer(self):
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
for _ in range(5):
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="review_pr", remote="dadeschools"
|
|
)
|
|
self.assertEqual(res["active_profile"], "mdcps-reviewer")
|
|
who = mcp_server.gitea_whoami(remote="dadeschools")
|
|
self.assertEqual(who["profile"]["profile_name"], "mdcps-reviewer")
|
|
rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
|
|
self.assertEqual(rt["active_profile"], "mdcps-reviewer")
|
|
|
|
def test_dadeschools_request_cannot_resolve_through_prgs_author(self):
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="comment_issue", remote="dadeschools"
|
|
)
|
|
self.assertNotEqual(res["active_profile"], "prgs-author")
|
|
self.assertNotIn("prgs-author", res["matching_configured_profile"])
|
|
# Gate must not silently switch either
|
|
blocked = mcp_server._profile_permission_block(
|
|
"gitea.issue.comment", remote="dadeschools"
|
|
)
|
|
self.assertIsNotNone(blocked)
|
|
self.assertFalse(blocked.get("success", True))
|
|
self.assertEqual(
|
|
gitea_config.selected_profile_name(), "mdcps-reviewer"
|
|
)
|
|
|
|
def test_prgs_request_cannot_resolve_through_mdcps_profile(self):
|
|
"""Reverse of the incident: a pinned MDCPS profile must never serve a
|
|
prgs request, nor be silently swapped for the prgs-side profile."""
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-author", remote="dadeschools"
|
|
)
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="create_issue", remote="prgs"
|
|
)
|
|
self.assertNotEqual(res["active_profile"], "prgs-author")
|
|
self.assertEqual(res["active_profile"], "mdcps-author")
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
blocked = mcp_server._session_context_mutation_block(remote="prgs")
|
|
self.assertIsNotNone(blocked)
|
|
self.assertTrue(any("cross-host" in r for r in blocked["reasons"]))
|
|
self.assertEqual(gitea_config.selected_profile_name(), "mdcps-author")
|
|
|
|
def test_unsupported_capability_fails_closed_structured(self):
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="comment_issue", remote="dadeschools"
|
|
)
|
|
self.assertFalse(res["allowed_in_current_session"])
|
|
self.assertTrue(res["stop_required"])
|
|
self.assertIn("reason", res)
|
|
self.assertIn("exact_safe_next_action", res)
|
|
self.assertIn("activate_profile", res["exact_safe_next_action"])
|
|
# Unknown task still fail closed (structured denial after #723;
|
|
# never raises into internal_error and never substitutes profile).
|
|
unknown = mcp_server.gitea_resolve_task_capability(
|
|
task="reopen_issue", remote="dadeschools"
|
|
)
|
|
self.assertFalse(unknown.get("allowed_in_current_session"))
|
|
self.assertTrue(unknown.get("stop_required"))
|
|
self.assertEqual(unknown.get("reason_code"), "unknown_task")
|
|
self.assertEqual(unknown.get("mutation_performed"), False)
|
|
self.assertEqual(unknown.get("active_profile"), "mdcps-reviewer")
|
|
self.assertNotEqual(unknown.get("active_profile"), "prgs-author")
|
|
|
|
def test_identity_mismatch_blocks_mutation(self):
|
|
"""Profile expects 913443 but authenticated as jcwalker3."""
|
|
|
|
def wrong_identity(method, url, header):
|
|
return {
|
|
"login": "jcwalker3",
|
|
"full_name": "Wrong",
|
|
"id": 9,
|
|
"email": "[email protected]",
|
|
}
|
|
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=wrong_identity):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
blocked = mcp_server._session_context_mutation_block(
|
|
remote="dadeschools"
|
|
)
|
|
self.assertIsNotNone(blocked)
|
|
self.assertTrue(
|
|
any("identity mismatch" in r for r in blocked["reasons"])
|
|
)
|
|
|
|
def test_repository_host_mismatch_blocks_mutation(self):
|
|
"""mdcps-reviewer cannot serve prgs remote."""
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
blocked = mcp_server._session_context_mutation_block(remote="prgs")
|
|
self.assertIsNotNone(blocked)
|
|
self.assertTrue(
|
|
any("cross-host" in r for r in blocked["reasons"])
|
|
)
|
|
|
|
def test_drift_cannot_reach_write(self):
|
|
"""Simulated mid-session profile override cannot pass permission gate."""
|
|
with patch.dict(os.environ, self._env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
mcp_server.gitea_activate_profile(
|
|
profile_name="mdcps-reviewer", remote="dadeschools"
|
|
)
|
|
# Bind session as mdcps-reviewer
|
|
self.assertEqual(
|
|
session_ctx.get_session_context()["profile_name"],
|
|
"mdcps-reviewer",
|
|
)
|
|
# Hostile override without activate_profile
|
|
gitea_config._active_profile_override = "prgs-author"
|
|
blocked = mcp_server._session_context_mutation_block(
|
|
remote="dadeschools"
|
|
)
|
|
self.assertIsNotNone(blocked)
|
|
self.assertTrue(any("drift" in r or "cross-host" in r for r in blocked["reasons"]))
|
|
|
|
def test_static_prgs_author_still_works(self):
|
|
env = {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": "prgs-author",
|
|
"GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
|
|
}
|
|
with patch.dict(os.environ, env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
gitea_config._active_profile_override = None
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="create_issue", remote="prgs"
|
|
)
|
|
self.assertEqual(res["active_profile"], "prgs-author")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertEqual(res["active_identity"], "jcwalker3")
|
|
|
|
def test_static_mdcps_author_still_works(self):
|
|
env = {
|
|
"GITEA_MCP_CONFIG": self.config_path,
|
|
"GITEA_MCP_PROFILE": "mdcps-author",
|
|
"GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
|
|
}
|
|
with patch.dict(os.environ, env, clear=False):
|
|
with patch("mcp_server.api_request", side_effect=self._api_side_effect):
|
|
gitea_config._active_profile_override = None
|
|
res = mcp_server.gitea_resolve_task_capability(
|
|
task="comment_issue", remote="dadeschools"
|
|
)
|
|
self.assertEqual(res["active_profile"], "mdcps-author")
|
|
self.assertTrue(res["allowed_in_current_session"])
|
|
self.assertNotIn("prgs-author", res["matching_configured_profile"])
|
|
|
|
|
|
class TestSessionContextBindingUnit(unittest.TestCase):
|
|
def test_profile_matches_remote_by_base_url(self):
|
|
remotes = {
|
|
"dadeschools": {"host": "gitea.dadeschools.net"},
|
|
"prgs": {"host": "gitea.prgs.cc"},
|
|
}
|
|
mdcps = {"base_url": "https://gitea.dadeschools.net", "context": "mdcps"}
|
|
prgs = {"base_url": "https://gitea.prgs.cc", "context": "prgs"}
|
|
self.assertTrue(
|
|
session_ctx.profile_matches_remote(mdcps, "dadeschools", remotes)
|
|
)
|
|
self.assertFalse(session_ctx.profile_matches_remote(mdcps, "prgs", remotes))
|
|
self.assertTrue(session_ctx.profile_matches_remote(prgs, "prgs", remotes))
|
|
|
|
def test_bind_and_detect_drift(self):
|
|
session_ctx.bind_session_context(
|
|
profile_name="mdcps-reviewer",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="913443",
|
|
source="test",
|
|
)
|
|
ok = session_ctx.assess_session_context(
|
|
profile_name="mdcps-reviewer",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="913443",
|
|
)
|
|
self.assertTrue(ok["proven"])
|
|
bad = session_ctx.assess_session_context(
|
|
profile_name="prgs-author",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="jcwalker3",
|
|
)
|
|
self.assertTrue(bad["block"])
|
|
self.assertTrue(any("drift" in r for r in bad["reasons"]))
|
|
|
|
def test_bound_context_survives_multiple_calls_and_exceptions(self):
|
|
original = session_ctx.bind_session_context(
|
|
profile_name="mdcps-reviewer",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="913443",
|
|
source="test",
|
|
)
|
|
|
|
try:
|
|
for _ in range(3):
|
|
observed = session_ctx.seed_session_context_if_unbound(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
source="interleaved-test-call",
|
|
)
|
|
self.assertEqual(observed, original)
|
|
raise RuntimeError("simulated caller failure")
|
|
except RuntimeError:
|
|
pass
|
|
|
|
self.assertEqual(session_ctx.get_session_context(), original)
|
|
drift = session_ctx.assess_session_context(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
require_bound=True,
|
|
)
|
|
self.assertTrue(drift["block"])
|
|
|
|
def test_parallel_calls_cannot_overwrite_established_binding(self):
|
|
original = session_ctx.bind_session_context(
|
|
profile_name="mdcps-reviewer",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="913443",
|
|
source="test",
|
|
)
|
|
barrier = threading.Barrier(9)
|
|
results = []
|
|
results_lock = threading.Lock()
|
|
|
|
def competing_seed(index: int) -> None:
|
|
barrier.wait()
|
|
result = session_ctx.seed_session_context_if_unbound(
|
|
profile_name=f"prgs-author-{index}",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity=f"other-{index}",
|
|
source="parallel-test-call",
|
|
)
|
|
with results_lock:
|
|
results.append(result)
|
|
|
|
threads = [threading.Thread(target=competing_seed, args=(i,)) for i in range(8)]
|
|
for thread in threads:
|
|
thread.start()
|
|
barrier.wait()
|
|
for thread in threads:
|
|
thread.join(timeout=5)
|
|
|
|
self.assertFalse(any(thread.is_alive() for thread in threads))
|
|
self.assertEqual(results, [original] * 8)
|
|
self.assertEqual(session_ctx.get_session_context(), original)
|
|
|
|
def test_concurrent_initialization_has_single_winning_binding(self):
|
|
"""Racing first-binds from an unbound context: exactly one winner, and
|
|
every racer observes that same complete binding (never a partial one)."""
|
|
self.assertIsNone(session_ctx.get_session_context())
|
|
thread_count = 8
|
|
barrier = threading.Barrier(thread_count)
|
|
results = []
|
|
results_lock = threading.Lock()
|
|
|
|
def racing_seed(index: int) -> None:
|
|
barrier.wait()
|
|
result = session_ctx.seed_session_context_if_unbound(
|
|
profile_name=f"prgs-author-{index}",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity=f"user-{index}",
|
|
source="concurrent-init-test",
|
|
)
|
|
with results_lock:
|
|
results.append(result)
|
|
|
|
threads = [
|
|
threading.Thread(target=racing_seed, args=(i,))
|
|
for i in range(thread_count)
|
|
]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join(timeout=5)
|
|
|
|
self.assertFalse(any(thread.is_alive() for thread in threads))
|
|
winner = session_ctx.get_session_context()
|
|
self.assertIsNotNone(winner)
|
|
self.assertEqual(len(results), thread_count)
|
|
self.assertEqual(results, [winner] * thread_count)
|
|
# A losing racer's identity must never be spliced into the winner.
|
|
self.assertEqual(
|
|
winner["identity"], winner["profile_name"].replace("prgs-author-", "user-")
|
|
)
|
|
|
|
def test_repository_mismatch_blocks_before_mutation(self):
|
|
"""Same host and profile, different repository, must still fail closed."""
|
|
session_ctx.bind_session_context(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
repository="Gitea-Tools",
|
|
org="Scaled-Tech-Consulting",
|
|
source="test",
|
|
)
|
|
same = session_ctx.assess_session_context(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
repository="Gitea-Tools",
|
|
org="Scaled-Tech-Consulting",
|
|
require_bound=True,
|
|
)
|
|
self.assertTrue(same["proven"])
|
|
other_repo = session_ctx.assess_session_context(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
repository="eAgenda",
|
|
org="Scaled-Tech-Consulting",
|
|
require_bound=True,
|
|
)
|
|
self.assertTrue(other_repo["block"])
|
|
self.assertTrue(
|
|
any("repository drift" in r for r in other_repo["reasons"])
|
|
)
|
|
other_org = session_ctx.assess_session_context(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
repository="Gitea-Tools",
|
|
org="913443",
|
|
require_bound=True,
|
|
)
|
|
self.assertTrue(other_org["block"])
|
|
self.assertTrue(any("org drift" in r for r in other_org["reasons"]))
|
|
|
|
def test_returned_snapshot_cannot_mutate_binding(self):
|
|
session_ctx.bind_session_context(
|
|
profile_name="mdcps-reviewer",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="913443",
|
|
source="test",
|
|
)
|
|
snapshot = session_ctx.get_session_context()
|
|
snapshot["profile_name"] = "prgs-author"
|
|
self.assertEqual(
|
|
session_ctx.get_session_context()["profile_name"], "mdcps-reviewer"
|
|
)
|
|
|
|
|
|
class TestSessionContextTestBoundaryIsolation(unittest.TestCase):
|
|
def test_mdcps_binding_starts_clean_and_does_not_escape_test(self):
|
|
self.assertIsNone(session_ctx.get_session_context())
|
|
session_ctx.bind_session_context(
|
|
profile_name="mdcps-reviewer",
|
|
remote="dadeschools",
|
|
host="gitea.dadeschools.net",
|
|
identity="913443",
|
|
source="test-boundary",
|
|
)
|
|
|
|
def test_prgs_binding_starts_clean_and_does_not_escape_test(self):
|
|
self.assertIsNone(session_ctx.get_session_context())
|
|
session_ctx.bind_session_context(
|
|
profile_name="prgs-author",
|
|
remote="prgs",
|
|
host="gitea.prgs.cc",
|
|
identity="jcwalker3",
|
|
source="test-boundary",
|
|
)
|
|
|
|
def test_reset_helper_is_rejected_outside_pytest_boundary(self):
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with self.assertRaises(RuntimeError):
|
|
session_ctx._reset_session_context_for_testing()
|
|
|
|
|
|
class TestIssue714MergeCoexistenceWithMaster(unittest.TestCase):
|
|
"""Conflict-resolution regressions after merging advanced master into #715.
|
|
|
|
Preserves:
|
|
- #714 immutable session / no auto profile substitution
|
|
- #685 side-effect-free resolve (auto_recover=False)
|
|
- #709 actor identity helper from master
|
|
"""
|
|
|
|
def test_auto_switch_helpers_remain_fail_closed(self):
|
|
self.assertFalse(mcp_server._try_auto_switch_for_operation("gitea.pr.review"))
|
|
self.assertFalse(
|
|
mcp_server._try_auto_switch_for_operation(
|
|
"gitea.issue.comment", host="gitea.prgs.cc"
|
|
)
|
|
)
|
|
# Active profile is prgs-author; cannot match mdcps review permission
|
|
# and must not switch.
|
|
with patch.object(
|
|
mcp_server,
|
|
"get_profile",
|
|
return_value={
|
|
"profile_name": "prgs-author",
|
|
"allowed_operations": ["gitea.read", "gitea.issue.create"],
|
|
"forbidden_operations": ["gitea.pr.approve"],
|
|
"base_url": "https://gitea.prgs.cc",
|
|
"context": "prgs",
|
|
},
|
|
):
|
|
matched = mcp_server._ensure_matching_profile(
|
|
"gitea.pr.approve", "reviewer", remote="prgs"
|
|
)
|
|
self.assertIsNone(matched)
|
|
self.assertNotEqual(
|
|
gitea_config.selected_profile_name() or "prgs-author",
|
|
"mdcps-reviewer",
|
|
)
|
|
|
|
def test_authenticated_actor_helper_survives_merge(self):
|
|
"""Master #709 F7 actor identity coexists with #714 immutability."""
|
|
self.assertTrue(hasattr(mcp_server, "_authenticated_actor"))
|
|
with patch(
|
|
"mcp_server.get_auth_header", return_value={"Authorization": "token x"}
|
|
), patch(
|
|
"mcp_server.api_request",
|
|
return_value={"id": 42, "login": "sysadmin"},
|
|
):
|
|
mcp_server._ACTOR_IDENTITY_CACHE.clear()
|
|
actor = mcp_server._authenticated_actor("gitea.prgs.cc")
|
|
self.assertEqual(actor.get("user_id"), 42)
|
|
self.assertEqual(actor.get("login"), "sysadmin")
|
|
# Cached; second call does not re-request
|
|
with patch("mcp_server.api_request") as mock_api:
|
|
actor2 = mcp_server._authenticated_actor("gitea.prgs.cc")
|
|
mock_api.assert_not_called()
|
|
self.assertEqual(actor2.get("user_id"), 42)
|
|
|
|
def test_resolve_still_report_only_after_master_merge(self):
|
|
"""#685: resolve path must not pass auto_recover=True after conflict merge."""
|
|
allowed = [
|
|
"gitea.read",
|
|
"gitea.issue.create",
|
|
"gitea.issue.comment",
|
|
"gitea.pr.create",
|
|
"gitea.pr.comment",
|
|
"gitea.branch.create",
|
|
"gitea.branch.push",
|
|
"gitea.repo.commit",
|
|
]
|
|
profile = {
|
|
"profile_name": "prgs-author",
|
|
"role": "author",
|
|
"allowed_operations": allowed,
|
|
"forbidden_operations": [],
|
|
"base_url": "https://gitea.prgs.cc",
|
|
"context": "prgs",
|
|
}
|
|
config = {
|
|
"contexts": {
|
|
"prgs": {
|
|
"enabled": True,
|
|
"gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
|
|
}
|
|
},
|
|
"profiles": {
|
|
"prgs-author": {
|
|
"role": "author",
|
|
"allowed_operations": allowed,
|
|
"forbidden_operations": [],
|
|
"base_url": "https://gitea.prgs.cc",
|
|
"context": "prgs",
|
|
}
|
|
},
|
|
}
|
|
fake_binding = {
|
|
"classification": "unbound",
|
|
"active_worktree": None,
|
|
"reasons": [],
|
|
}
|
|
with patch.dict(
|
|
os.environ, {"GITEA_MCP_PROFILE": "prgs-author"}, clear=False
|
|
), patch.object(
|
|
mcp_server, "get_profile", return_value=profile
|
|
), patch.object(
|
|
mcp_server.gitea_config, "load_config", return_value=config
|
|
), patch.object(
|
|
mcp_server, "_authenticated_username", return_value="jcwalker3"
|
|
), patch.object(
|
|
mcp_server,
|
|
"_assess_stale_active_binding",
|
|
return_value=fake_binding,
|
|
) as mock_assess, patch.object(
|
|
mcp_server, "record_preflight_check", return_value=None
|
|
), patch.object(
|
|
mcp_server, "record_mutation_authority", return_value=None
|
|
), patch.object(
|
|
mcp_server, "init_review_decision_lock", return_value=None
|
|
):
|
|
result = mcp_server.gitea_resolve_task_capability(
|
|
task="create_issue", remote="prgs"
|
|
)
|
|
mock_assess.assert_called()
|
|
for call in mock_assess.call_args_list:
|
|
self.assertFalse(call.kwargs.get("auto_recover", True))
|
|
self.assertEqual(result.get("mutation_performed"), False)
|
|
# Session context must not have been rewritten by resolve
|
|
# (report-only; any prior binding stays; unbound stays unbound unless
|
|
# seed-on-read path is intentional — resolve must not activate peers).
|
|
self.assertNotEqual(result.get("active_profile"), "mdcps-reviewer")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|