547 lines
26 KiB
Python
547 lines
26 KiB
Python
"""Regression tests for Issue #973: validated cross-repository canonical roots."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import gitea_config
|
|
import namespace_workspace_binding as nwb
|
|
import canonical_repository_root as crr
|
|
import stable_control_runtime
|
|
import gitea_mcp_server as mcp_server
|
|
|
|
|
|
class TestIssue973RecognizedEnvKeys(unittest.TestCase):
|
|
"""Test recognized environment variable keys under #973."""
|
|
|
|
def test_recognized_gitea_env_keys(self):
|
|
for key in (
|
|
"GITEA_CANONICAL_REPOSITORY_ROOT",
|
|
"GITEA_REVIEWER_WORKTREE",
|
|
"GITEA_MERGER_WORKTREE",
|
|
"GITEA_MCP_SESSION_STATE_TTL_HOURS",
|
|
):
|
|
self.assertIn(key, gitea_config.RECOGNIZED_GITEA_ENV_KEYS)
|
|
|
|
def test_get_unconsumed_gitea_env_overrides_ignores_recognized(self):
|
|
env = {
|
|
"GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path",
|
|
"GITEA_REVIEWER_WORKTREE": "/some/reviewer/path",
|
|
"GITEA_MERGER_WORKTREE": "/some/merger/path",
|
|
"GITEA_MCP_SESSION_STATE_TTL_HOURS": "24",
|
|
"GITEA_UNRECOGNIZED_FOO_VAR": "bar",
|
|
}
|
|
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
|
|
self.assertNotIn("GITEA_CANONICAL_REPOSITORY_ROOT", unconsumed)
|
|
self.assertNotIn("GITEA_REVIEWER_WORKTREE", unconsumed)
|
|
self.assertNotIn("GITEA_MERGER_WORKTREE", unconsumed)
|
|
self.assertNotIn("GITEA_MCP_SESSION_STATE_TTL_HOURS", unconsumed)
|
|
self.assertIn("GITEA_UNRECOGNIZED_FOO_VAR", unconsumed)
|
|
|
|
|
|
class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase):
|
|
"""Test workspace binding and canonical root validation for cross-repo namespaces."""
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.tmp_dir = os.path.realpath(self._tmp.name)
|
|
|
|
# Create simulated installation root
|
|
self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools")
|
|
os.makedirs(self.install_root)
|
|
subprocess.run(["git", "init", "-b", "master"], cwd=self.install_root, check=True)
|
|
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.install_root, check=True)
|
|
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.install_root, check=True)
|
|
with open(os.path.join(self.install_root, "README.md"), "w") as f:
|
|
f.write("install\n")
|
|
subprocess.run(["git", "add", "README.md"], cwd=self.install_root, check=True)
|
|
subprocess.run(["git", "commit", "-m", "initial"], cwd=self.install_root, check=True)
|
|
|
|
# Create simulated target repository root
|
|
self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane")
|
|
os.makedirs(self.target_root)
|
|
subprocess.run(["git", "init", "-b", "master"], cwd=self.target_root, check=True)
|
|
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.target_root, check=True)
|
|
subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.target_root, check=True)
|
|
with open(os.path.join(self.target_root, "README.md"), "w") as f:
|
|
f.write("target\n")
|
|
subprocess.run(["git", "add", "README.md"], cwd=self.target_root, check=True)
|
|
subprocess.run(["git", "commit", "-m", "initial"], cwd=self.target_root, check=True)
|
|
|
|
# Add remotes to simulate real git repositories with identities
|
|
subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"], cwd=self.install_root, check=True)
|
|
subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"], cwd=self.target_root, check=True)
|
|
|
|
# Create simulated foreign repository root
|
|
self.evil_root = os.path.join(self.tmp_dir, "Evil-Repo")
|
|
os.makedirs(self.evil_root)
|
|
subprocess.run(["git", "init", "-b", "master"], cwd=self.evil_root, check=True)
|
|
subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.evil_root, check=True)
|
|
subprocess.run(["git", "config", "user.name", "Evil User"], cwd=self.evil_root, check=True)
|
|
with open(os.path.join(self.evil_root, "README.md"), "w") as f:
|
|
f.write("evil\n")
|
|
subprocess.run(["git", "add", "README.md"], cwd=self.evil_root, check=True)
|
|
subprocess.run(["git", "commit", "-m", "initial"], cwd=self.evil_root, check=True)
|
|
subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Someone-Else/Evil-Repo.git"], cwd=self.evil_root, check=True)
|
|
|
|
# Create branches/ directory and a valid registered worktree in target repository
|
|
self.target_branches = os.path.join(self.target_root, "branches")
|
|
self.target_worktree = os.path.join(self.target_branches, "rev-pr-99")
|
|
subprocess.run(["git", "worktree", "add", "-b", "rev-pr-99", self.target_worktree], cwd=self.target_root, check=True)
|
|
|
|
def tearDown(self):
|
|
self._tmp.cleanup()
|
|
|
|
def test_valid_same_repository_configuration(self):
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=None,
|
|
)
|
|
self.assertEqual(ctx["canonical_repo_root"], self.install_root)
|
|
self.assertTrue(ctx["roots_aligned"])
|
|
self.assertTrue(ctx["canonical_root_assessment"]["proven"])
|
|
|
|
def test_valid_cross_repo_canonical_root(self):
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertEqual(ctx["canonical_repo_root"], self.target_root)
|
|
self.assertTrue(ctx["roots_aligned"])
|
|
self.assertTrue(ctx["canonical_root_assessment"]["proven"])
|
|
|
|
def test_expected_repository_identity_match(self):
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="test",
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
)
|
|
self.assertTrue(assessment["proven"])
|
|
self.assertFalse(assessment["block"])
|
|
|
|
def test_foreign_repository_identity_mismatch(self):
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.evil_root,
|
|
source="test",
|
|
expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(assessment["proven"])
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
|
|
|
|
def test_native_repository_binding_mismatch(self):
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.evil_root,
|
|
expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
|
|
remote="prgs",
|
|
)
|
|
self.assertFalse(ctx["roots_aligned"])
|
|
self.assertFalse(ctx["canonical_root_assessment"]["proven"])
|
|
self.assertTrue(any("identity mismatch" in r for r in ctx["canonical_root_assessment"]["reasons"]))
|
|
|
|
def test_unpatched_foreign_configured_root_derives_expected_from_process_root_and_blocks(self):
|
|
"""B8: Production path test where foreign configured root cannot self-authorize."""
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server, "_configured_canonical_root", return_value=(self.evil_root, "env")):
|
|
expected_slug = mcp_server._resolve_expected_repository_slug("prgs")
|
|
self.assertEqual(expected_slug, "Scaled-Tech-Consulting/Gitea-Tools")
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.evil_root,
|
|
source="env",
|
|
expected_slug=expected_slug,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
)
|
|
self.assertFalse(assessment["proven"])
|
|
self.assertTrue(assessment["block"])
|
|
self.assertEqual(assessment["resolved_slug"], "Someone-Else/Evil-Repo")
|
|
self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
|
|
|
|
def test_unpatched_valid_cross_repo_matching_session_context(self):
|
|
"""B8: Valid cross-repo namespace matches when session context is bound to target repo."""
|
|
bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx):
|
|
expected_slug = mcp_server._resolve_expected_repository_slug("prgs")
|
|
self.assertEqual(expected_slug, "Scaled-Tech-Consulting/mcp-control-plane")
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=expected_slug,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
)
|
|
self.assertTrue(assessment["proven"])
|
|
self.assertFalse(assessment["block"])
|
|
self.assertEqual(assessment["resolved_slug"], "Scaled-Tech-Consulting/mcp-control-plane")
|
|
|
|
def test_unprovable_expected_identity_fails_closed(self):
|
|
"""B8: If expected repository identity is unprovable for a configured root, fail closed."""
|
|
no_remote_root = os.path.join(self.tmp_dir, "no-remote-process-root")
|
|
os.makedirs(no_remote_root)
|
|
subprocess.run(["git", "init", "-b", "master"], cwd=no_remote_root, check=True)
|
|
with patch.object(mcp_server, "PROJECT_ROOT", no_remote_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context", return_value=None):
|
|
expected_slug = mcp_server._resolve_expected_repository_slug("prgs")
|
|
self.assertIsNone(expected_slug)
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=expected_slug,
|
|
process_project_root=no_remote_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
)
|
|
self.assertFalse(assessment["proven"])
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"]))
|
|
|
|
def test_missing_canonical_root(self):
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="author",
|
|
worktree_path=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root="",
|
|
)
|
|
self.assertEqual(ctx["canonical_repo_root"], self.install_root)
|
|
self.assertTrue(ctx["roots_aligned"])
|
|
|
|
def test_nonexistent_configured_canonical_root(self):
|
|
nonexistent = os.path.join(self.tmp_dir, "nonexistent-repo")
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=nonexistent,
|
|
)
|
|
self.assertFalse(ctx["roots_aligned"])
|
|
self.assertFalse(ctx["canonical_root_assessment"]["proven"])
|
|
self.assertTrue(any("does not exist" in r for r in ctx["canonical_root_assessment"]["reasons"]))
|
|
|
|
def test_non_git_configured_canonical_root(self):
|
|
non_git = os.path.join(self.tmp_dir, "non-git-dir")
|
|
os.makedirs(non_git)
|
|
ctx = nwb.resolve_namespace_mutation_context(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=non_git,
|
|
)
|
|
self.assertFalse(ctx["roots_aligned"])
|
|
self.assertFalse(ctx["canonical_root_assessment"]["proven"])
|
|
self.assertTrue(any("not a git repository" in r for r in ctx["canonical_root_assessment"]["reasons"]))
|
|
|
|
def test_git_common_directory_membership_matching(self):
|
|
valid, err = nwb.verify_git_common_directory_membership(
|
|
self.target_worktree, self.target_root
|
|
)
|
|
self.assertTrue(valid, err)
|
|
self.assertIsNone(err)
|
|
|
|
def test_foreign_git_common_directory(self):
|
|
# Foreign worktree created under install_root
|
|
install_branches = os.path.join(self.install_root, "branches")
|
|
foreign_wt = os.path.join(install_branches, "foreign-wt")
|
|
subprocess.run(["git", "worktree", "add", "-b", "foreign-wt", foreign_wt], cwd=self.install_root, check=True)
|
|
|
|
valid, err = nwb.verify_git_common_directory_membership(
|
|
foreign_wt, self.target_root
|
|
)
|
|
self.assertFalse(valid)
|
|
self.assertIn("does not match", err)
|
|
|
|
def test_normalized_path_aliases(self):
|
|
alias_path = self.target_worktree + "/../rev-pr-99/./"
|
|
valid, err = nwb.verify_git_common_directory_membership(
|
|
alias_path, self.target_root
|
|
)
|
|
self.assertTrue(valid, err)
|
|
|
|
def test_safe_symlink_identity(self):
|
|
link_path = os.path.join(self.target_branches, "symlink-rev-99")
|
|
try:
|
|
os.symlink(self.target_worktree, link_path)
|
|
valid, err = nwb.verify_git_common_directory_membership(
|
|
link_path, self.target_root
|
|
)
|
|
self.assertTrue(valid, err)
|
|
finally:
|
|
if os.path.exists(link_path):
|
|
os.unlink(link_path)
|
|
|
|
def test_symlink_escape_or_foreign_alias(self):
|
|
outside_dir = os.path.join(self.tmp_dir, "outside-target")
|
|
os.makedirs(outside_dir)
|
|
link_escape = os.path.join(self.target_branches, "escape-link")
|
|
try:
|
|
os.symlink(outside_dir, link_escape)
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind="reviewer",
|
|
worktree_path=link_escape,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertTrue(assessment["block"])
|
|
finally:
|
|
if os.path.exists(link_escape):
|
|
os.unlink(link_escape)
|
|
|
|
def test_reviewer_worktree_registered_and_valid(self):
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind="reviewer",
|
|
worktree_path=self.target_worktree,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertFalse(assessment["block"])
|
|
|
|
def test_reviewer_worktree_unregistered_blocks(self):
|
|
unreg_wt = os.path.join(self.target_branches, "unregistered-reviewer")
|
|
os.makedirs(unreg_wt)
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind="reviewer",
|
|
worktree_path=unreg_wt,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"]))
|
|
|
|
def test_merger_worktree_registered_and_valid(self):
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind="merger",
|
|
worktree_path=self.target_worktree,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertFalse(assessment["block"])
|
|
|
|
def test_merger_worktree_unregistered_blocks(self):
|
|
unreg_wt = os.path.join(self.target_branches, "unregistered-merger")
|
|
os.makedirs(unreg_wt)
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind="merger",
|
|
worktree_path=unreg_wt,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"]))
|
|
|
|
def test_reviewer_or_merger_worktree_outside_branches_blocks(self):
|
|
outside_wt = os.path.join(self.target_root, "outside_branches_wt")
|
|
os.makedirs(outside_wt)
|
|
for r_kind in ("reviewer", "merger"):
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind=r_kind,
|
|
worktree_path=outside_wt,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("is not under" in r for r in assessment["reasons"]))
|
|
|
|
def test_nonexistent_reviewer_or_merger_worktree_blocks(self):
|
|
nonexistent_wt = os.path.join(self.target_branches, "nonexistent-wt")
|
|
for r_kind in ("reviewer", "merger"):
|
|
assessment = nwb.assess_namespace_mutation_workspace(
|
|
role_kind=r_kind,
|
|
worktree_path=nonexistent_wt,
|
|
worktree=None,
|
|
process_project_root=self.install_root,
|
|
env={},
|
|
configured_canonical_root=self.target_root,
|
|
expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
|
|
)
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("does not exist" in r for r in assessment["reasons"]))
|
|
|
|
def test_safe_and_unsafe_mutation_alignment_outcomes(self):
|
|
# Safe alignment (same repo)
|
|
report_safe = stable_control_runtime.build_runtime_report(
|
|
process_root=self.install_root,
|
|
checkout_branch="master",
|
|
runtime_head="abcdef123456",
|
|
active_task_workspace=self.install_root,
|
|
canonical_repository_root=self.install_root,
|
|
workspace_roots_aligned=True,
|
|
)
|
|
gate_safe = stable_control_runtime.assess_runtime_mutation_gate(report_safe)
|
|
self.assertFalse(gate_safe["block"])
|
|
|
|
# Unsafe alignment
|
|
report_unsafe = stable_control_runtime.build_runtime_report(
|
|
process_root=self.install_root,
|
|
checkout_branch="master",
|
|
runtime_head="abcdef123456",
|
|
active_task_workspace=self.target_worktree,
|
|
canonical_repository_root=self.target_root,
|
|
workspace_roots_aligned=False,
|
|
)
|
|
gate_unsafe = stable_control_runtime.assess_runtime_mutation_gate(report_unsafe)
|
|
self.assertTrue(gate_unsafe["block"])
|
|
|
|
def test_reviewer_lease_lifecycle_production_path(self):
|
|
"""B5: Automated regression for reviewer lease acquire and release through production path."""
|
|
mock_whoami = {
|
|
"authenticated": True,
|
|
"username": "sysadmin",
|
|
"remote": "prgs",
|
|
"profile": {
|
|
"profile_name": "prgs-reviewer",
|
|
"role": "reviewer",
|
|
"role_kind": "reviewer",
|
|
"allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.approve", "gitea.pr.request_changes"],
|
|
"forbidden_operations": [],
|
|
},
|
|
}
|
|
|
|
def mock_api_request(method, url, auth=None, json_data=None):
|
|
if method == "GET":
|
|
return {"number": 99, "head": {"sha": "abc1234"}, "state": "open", "merged": False, "merged_at": None}
|
|
elif method == "POST":
|
|
return {"id": 9999, "body": (json_data or {}).get("body", "")}
|
|
return {}
|
|
|
|
with patch.object(mcp_server, "gitea_whoami", return_value=mock_whoami), \
|
|
patch.object(mcp_server, "get_profile", return_value=mock_whoami["profile"]), \
|
|
patch.object(mcp_server, "_effective_workspace_role", return_value="reviewer"), \
|
|
patch.object(mcp_server, "_configured_canonical_root", return_value=(self.target_root, "env")), \
|
|
patch.object(mcp_server, "_reviewer_session_worktree", return_value=self.target_worktree), \
|
|
patch.object(mcp_server, "_auth", return_value="token mock-token"), \
|
|
patch.object(mcp_server, "_fetch_pr_comments", return_value=[]), \
|
|
patch.object(mcp_server, "api_request", side_effect=mock_api_request), \
|
|
patch("reviewer_pr_lease.assess_acquire_lease", return_value={"acquire_allowed": True, "reasons": [], "lease_body": "<!-- LEASE -->"}), \
|
|
patch("reviewer_pr_lease.find_active_reviewer_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \
|
|
patch("reviewer_pr_lease.get_session_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \
|
|
patch("reviewer_pr_lease.clear_session_lease") as mock_clear:
|
|
|
|
acq_res = mcp_server.gitea_acquire_reviewer_pr_lease(
|
|
pr_number=99,
|
|
remote="prgs",
|
|
worktree=self.target_worktree,
|
|
org="Scaled-Tech-Consulting",
|
|
repo="mcp-control-plane",
|
|
)
|
|
self.assertTrue(acq_res.get("success"), acq_res)
|
|
|
|
rel_res = mcp_server.gitea_release_reviewer_pr_lease(
|
|
pr_number=99,
|
|
worktree=self.target_worktree,
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="mcp-control-plane",
|
|
)
|
|
self.assertTrue(rel_res.get("success"), rel_res)
|
|
mock_clear.assert_called_once()
|
|
|
|
def test_unbound_session_derives_and_retains_configured_target_repository(self):
|
|
"""B9: Unbound session with a legitimate configured target root derives and retains target repository."""
|
|
prof = {
|
|
"profile_name": "prgs-author",
|
|
"allowed_repositories": ["Scaled-Tech-Consulting/mcp-control-plane"],
|
|
"canonical_repository_root": self.target_root,
|
|
}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context", return_value=None):
|
|
slug, reasons = mcp_server._canonical_repository_slug(prof, "prgs")
|
|
self.assertEqual(slug, "Scaled-Tech-Consulting/mcp-control-plane", f"reasons: {reasons}")
|
|
self.assertEqual(reasons, [])
|
|
res = mcp_server._trusted_session_repository(prof, "prgs")
|
|
self.assertEqual(res["org"], "Scaled-Tech-Consulting")
|
|
self.assertEqual(res["repository"], "mcp-control-plane")
|
|
|
|
def test_process_root_a_configured_target_b_retains_b(self):
|
|
"""B9: Process root A (Gitea-Tools) and configured target B (mcp-control-plane) keep B as expected identity when bound."""
|
|
bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx):
|
|
expected = mcp_server._resolve_expected_repository_slug("prgs")
|
|
self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane")
|
|
|
|
def test_request_supplied_repository_c_cannot_replace_derived_or_bound_repository_b(self):
|
|
"""B9: Request-supplied repository C (Timesheet) cannot replace bound repository B (mcp-control-plane)."""
|
|
bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"}
|
|
with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
|
|
patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx):
|
|
expected = mcp_server._resolve_expected_repository_slug("prgs", org="Scaled-Tech-Consulting", repo="Timesheet")
|
|
self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane")
|
|
|
|
def test_known_expected_repo_a_plus_candidate_root_b_fails_closed(self):
|
|
"""B9: Known expected repository A plus candidate root B fails closed in validation mode."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="validation",
|
|
)
|
|
self.assertFalse(assessment["proven"])
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
|
|
|
|
def test_validation_mode_with_unprovable_expected_identity_fails_closed(self):
|
|
"""B9: Validation mode with expected_slug=None and require_binding=True fails closed."""
|
|
assessment = crr.assess_canonical_repository_root(
|
|
configured_value=self.target_root,
|
|
source="env",
|
|
expected_slug=None,
|
|
process_project_root=self.install_root,
|
|
remote="prgs",
|
|
require_binding=True,
|
|
mode="validation",
|
|
)
|
|
self.assertFalse(assessment["proven"])
|
|
self.assertTrue(assessment["block"])
|
|
self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|