"""Issue #706: immutable startup/profile canonical_repository_root capability. Cross-repository MCP namespaces (e.g. ``eagenda-author``) must bind their canonical repository root to the *configured target repository*, not to the Gitea-Tools install checkout the server script lives in. These tests prove: * the configured binding is resolved from profile/env with env precedence, * the target repository identity and git common-directory membership are validated (AC3), * the branches-only guard (#274) is enforced *inside the target repo* (AC4), * missing / conflicting / forged bindings fail closed (AC5), * prgs and mdcps namespaces stay simultaneously isolated (AC6), * the session binding pins the canonical root immutably, * no weakening of the install-root single-repo default (AC8). Uses two distinct real git repositories/worktrees (AC7). """ from __future__ import annotations import os import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import canonical_repository_root as crr # noqa: E402 import gitea_config # noqa: E402 import namespace_workspace_binding as nwb # noqa: E402 import session_context_binding as session_ctx # noqa: E402 def _git(cwd: str, *args: str) -> str: res = subprocess.run( ["git", "-C", cwd, *args], capture_output=True, text=True, check=True, ) return res.stdout.strip() def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str: """Create a real git repo with one commit and a remote; return realpath root.""" path.mkdir(parents=True, exist_ok=True) _git(str(path), "init", "-q") _git(str(path), "config", "user.email", "t@example.com") _git(str(path), "config", "user.name", "Test") _git(str(path), "remote", "add", remote_name, remote_url) (path / "README.md").write_text("seed\n") _git(str(path), "add", "README.md") _git(str(path), "commit", "-q", "-m", "seed") return os.path.realpath(str(path)) def _add_worktree(repo_root: str, worktree_path: Path, branch: str) -> str: _git(repo_root, "worktree", "add", "-q", "-b", branch, str(worktree_path)) return os.path.realpath(str(worktree_path)) # --------------------------------------------------------------------------- # configured_canonical_root: resolution + precedence # --------------------------------------------------------------------------- class TestConfiguredCanonicalRoot(unittest.TestCase): def test_unconfigured_returns_none(self): value, source = crr.configured_canonical_root({}, {}) self.assertIsNone(value) self.assertIsNone(source) def test_profile_field_used(self): value, source = crr.configured_canonical_root( {"canonical_repository_root": "/repo/eAgenda"}, {} ) self.assertEqual(value, "/repo/eAgenda") self.assertIn("profile", source) def test_env_overrides_profile(self): value, source = crr.configured_canonical_root( {"canonical_repository_root": "/repo/eAgenda"}, {crr.CANONICAL_ROOT_ENV: "/repo/other"}, ) self.assertEqual(value, "/repo/other") self.assertIn(crr.CANONICAL_ROOT_ENV, source) def test_blank_values_ignored(self): value, source = crr.configured_canonical_root( {"canonical_repository_root": " "}, {crr.CANONICAL_ROOT_ENV: ""}, ) self.assertIsNone(value) self.assertIsNone(source) # --------------------------------------------------------------------------- # assess_canonical_repository_root: default (single-repo) path # --------------------------------------------------------------------------- class TestUnconfiguredFallback(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.tmp = self._tmp.name def tearDown(self): self._tmp.cleanup() def test_unconfigured_falls_back_to_process_root(self): root = _init_repo( Path(self.tmp) / "install", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git", ) got = crr.assess_canonical_repository_root( configured_value=None, source=None, expected_slug=None, process_project_root=root, ) self.assertTrue(got["proven"]) self.assertFalse(got["block"]) self.assertFalse(got["configured"]) self.assertEqual(got["canonical_repo_root"], root) # --------------------------------------------------------------------------- # assess_canonical_repository_root: configured cross-repo path # --------------------------------------------------------------------------- class TestConfiguredCrossRepo(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.tmp = Path(self._tmp.name) self.install = _init_repo( self.tmp / "Gitea-Tools", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git", ) self.target = _init_repo( self.tmp / "mcp-control-plane", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git", ) def tearDown(self): self._tmp.cleanup() def test_valid_configured_root_binds_to_target(self): got = crr.assess_canonical_repository_root( configured_value=self.target, source="profile canonical_repository_root", expected_slug="Scaled-Tech-Consulting/mcp-control-plane", process_project_root=self.install, ) self.assertTrue(got["proven"], got.get("reasons")) self.assertFalse(got["block"]) self.assertTrue(got["configured"]) self.assertEqual(got["canonical_repo_root"], self.target) self.assertNotEqual(got["canonical_repo_root"], self.install) def test_missing_path_fails_closed(self): got = crr.assess_canonical_repository_root( configured_value=str(self.tmp / "does-not-exist"), source="profile canonical_repository_root", expected_slug=None, process_project_root=self.install, ) self.assertTrue(got["block"]) self.assertFalse(got["proven"]) self.assertTrue(any("exist" in r for r in got["reasons"])) def test_non_git_path_fails_closed(self): plain = self.tmp / "plain" plain.mkdir() got = crr.assess_canonical_repository_root( configured_value=str(plain), source="profile canonical_repository_root", expected_slug=None, process_project_root=self.install, ) self.assertTrue(got["block"]) self.assertTrue(any("git" in r for r in got["reasons"])) def test_identity_mismatch_fails_closed(self): # Configured root is the install repo, but session expects the target # repo identity: a forged/conflicting binding. got = crr.assess_canonical_repository_root( configured_value=self.install, source="profile canonical_repository_root", expected_slug="Scaled-Tech-Consulting/mcp-control-plane", process_project_root=self.install, ) self.assertTrue(got["block"]) self.assertTrue(any("identity" in r for r in got["reasons"])) def test_unprovable_identity_blocks_when_required(self): noremote = _init_repo(self.tmp / "noremote-src", "x", remote_name="origin") _git(noremote, "remote", "remove", "origin") got = crr.assess_canonical_repository_root( configured_value=noremote, source="profile canonical_repository_root", expected_slug="Scaled-Tech-Consulting/mcp-control-plane", process_project_root=self.install, require_binding=True, ) self.assertTrue(got["block"]) def test_repository_identity_slug_reads_remote(self): self.assertEqual( crr.repository_identity_slug(self.target), "Scaled-Tech-Consulting/mcp-control-plane", ) # --------------------------------------------------------------------------- # Workspace membership + branches-only enforced inside the TARGET repo (AC4) # --------------------------------------------------------------------------- class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.tmp = Path(self._tmp.name) self.install = _init_repo( self.tmp / "Gitea-Tools", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git", ) self.target = _init_repo( self.tmp / "mcp-control-plane", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git", ) (Path(self.target) / "branches").mkdir() self.target_wt = _add_worktree( self.target, Path(self.target) / "branches" / "author-issue-1", "feat/issue-1", ) def tearDown(self): self._tmp.cleanup() def test_context_canonical_root_is_configured_target(self): ctx = nwb.resolve_namespace_mutation_context( role_kind="author", worktree_path=self.target_wt, process_project_root=self.install, env={}, configured_canonical_root=self.target, ) self.assertEqual(ctx["canonical_repo_root"], self.target) self.assertFalse(ctx["roots_aligned"]) def test_target_worktree_is_member_of_target_root(self): got = nwb.amw.assess_workspace_repo_membership( workspace_path=self.target_wt, canonical_repo_root=self.target, ) self.assertTrue(got["proven"], got.get("reasons")) def test_target_worktree_not_member_of_install_root(self): got = nwb.amw.assess_workspace_repo_membership( workspace_path=self.target_wt, canonical_repo_root=self.install, ) self.assertTrue(got["block"]) def test_branches_guard_passes_inside_target(self): assessment = nwb.assess_namespace_mutation_workspace( role_kind="author", worktree_path=self.target_wt, worktree=None, process_project_root=self.install, env={}, current_branch="feat/issue-1", configured_canonical_root=self.target, ) self.assertFalse(assessment["block"], assessment.get("reasons")) self.assertEqual(assessment["canonical_repo_root"], self.target) def test_target_control_checkout_blocks_branches_guard(self): assessment = nwb.assess_namespace_mutation_workspace( role_kind="author", worktree_path=self.target, # stable target checkout, not branches/ worktree=None, process_project_root=self.install, env={}, current_branch="master", configured_canonical_root=self.target, ) self.assertTrue(assessment["block"]) # --------------------------------------------------------------------------- # Immutable session pin (AC1/AC5) # --------------------------------------------------------------------------- class TestSessionCanonicalRootPin(unittest.TestCase): def setUp(self): os.environ["PYTEST_CURRENT_TEST"] = "t" session_ctx._reset_session_context_for_testing() def tearDown(self): session_ctx._reset_session_context_for_testing() def test_seed_stores_canonical_root(self): ctx = session_ctx.seed_session_context_if_unbound( profile_name="mcp-control-plane-author", remote="prgs", host="gitea.prgs.cc", identity="svc", repository="mcp-control-plane", org="Scaled-Tech-Consulting", canonical_repository_root="/repo/mcp-control-plane", ) self.assertEqual(ctx["canonical_repository_root"], "/repo/mcp-control-plane") def test_canonical_root_drift_fails_closed(self): session_ctx.seed_session_context_if_unbound( profile_name="mcp-control-plane-author", remote="prgs", host="gitea.prgs.cc", identity="svc", repository="mcp-control-plane", org="Scaled-Tech-Consulting", canonical_repository_root="/repo/mcp-control-plane", ) got = session_ctx.assess_session_context( profile_name="mcp-control-plane-author", remote="prgs", canonical_repository_root="/repo/forged-elsewhere", ) self.assertTrue(got["block"]) self.assertTrue(any("canonical" in r for r in got["reasons"])) def test_matching_canonical_root_passes(self): session_ctx.seed_session_context_if_unbound( profile_name="mcp-control-plane-author", remote="prgs", host="gitea.prgs.cc", identity="svc", repository="mcp-control-plane", org="Scaled-Tech-Consulting", canonical_repository_root="/repo/mcp-control-plane", ) got = session_ctx.assess_session_context( profile_name="mcp-control-plane-author", remote="prgs", canonical_repository_root="/repo/mcp-control-plane", ) self.assertFalse(got["block"], got["reasons"]) # --------------------------------------------------------------------------- # Simultaneous prgs / mdcps isolation (AC6) # --------------------------------------------------------------------------- class TestSimultaneousIsolation(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.tmp = Path(self._tmp.name) self.prgs = _init_repo( self.tmp / "prgs-repo", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git", ) self.mdcps = _init_repo( self.tmp / "mdcps-repo", "https://gitea.dadeschools.net/dadeschools/eAgenda.git", ) def tearDown(self): self._tmp.cleanup() def test_each_namespace_binds_its_own_target(self): a = crr.assess_canonical_repository_root( configured_value=self.prgs, source="env", expected_slug="Scaled-Tech-Consulting/mcp-control-plane", process_project_root=self.tmp.as_posix(), ) b = crr.assess_canonical_repository_root( configured_value=self.mdcps, source="env", expected_slug="dadeschools/eAgenda", process_project_root=self.tmp.as_posix(), ) self.assertEqual(a["canonical_repo_root"], self.prgs) self.assertEqual(b["canonical_repo_root"], self.mdcps) self.assertNotEqual(a["canonical_repo_root"], b["canonical_repo_root"]) def test_cross_wired_identity_blocks(self): # prgs path claimed under the mdcps identity → forged binding. got = crr.assess_canonical_repository_root( configured_value=self.prgs, source="env", expected_slug="dadeschools/eAgenda", process_project_root=self.tmp.as_posix(), ) self.assertTrue(got["block"]) # --------------------------------------------------------------------------- # Config validation of the profile field (AC5: malformed bindings fail closed) # --------------------------------------------------------------------------- class TestConfigValidation(unittest.TestCase): def test_absent_is_allowed(self): # single-repo default: no field configured gitea_config._validate_canonical_repository_root("p", None) def test_valid_absolute_path_ok(self): gitea_config._validate_canonical_repository_root( "p", "/Users/x/Development/mcp-control-plane" ) def test_relative_path_rejected(self): with self.assertRaises(gitea_config.ConfigError): gitea_config._validate_canonical_repository_root( "p", "relative/path" ) def test_empty_string_rejected(self): with self.assertRaises(gitea_config.ConfigError): gitea_config._validate_canonical_repository_root("p", " ") def test_non_string_rejected(self): with self.assertRaises(gitea_config.ConfigError): gitea_config._validate_canonical_repository_root("p", ["/x"]) if __name__ == "__main__": unittest.main()