"""Issue #706 F1 integration regression (review 457). The unit tests in ``test_issue_706_canonical_repository_root.py`` exercise ``crr.assess_canonical_repository_root`` / ``session_ctx.seed_session_context_if_unbound`` with a *preselected* slug, so they never drive the real end-to-end path that review 457 found broken: _seed_session_context -> _trusted_session_repository -> _workspace_repository_slug -> _local_git_remote_url (cwd=PROJECT_ROOT, always Gitea-Tools) Before the fix, the session repository identity was pinned from the *install* checkout remote even when a cross-repository ``canonical_repository_root`` was configured to an external repository (e.g. mcp-control-plane). The mutation preflight then derived ``expected_slug`` from that install-derived pin and ``_enforce_canonical_repository_root`` failed closed on a self-inflicted identity mismatch, so the stated cross-repo namespaces stayed blocked. These tests use two *real* temporary git repositories and drive the live ``mcp_server._seed_session_context`` and ``mcp_server._enforce_canonical_repository_root`` functions, asserting the session pins the *configured target* identity and that enforcement accepts the target worktree — while every fail-closed property is preserved. """ 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 mcp_server # noqa: E402 # loads gitea_mcp_server.py into this namespace import canonical_repository_root as crr # noqa: E402 import session_context_binding as session_ctx # noqa: E402 INSTALL_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane" TARGET_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git" OTHER_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/other-repo.git" def _git(cwd: str, *args: str) -> str: return subprocess.run( ["git", "-C", cwd, *args], capture_output=True, text=True, check=True ).stdout.strip() def _init_repo(path: Path, remote_url: str) -> str: 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", "prgs", 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, wt: Path, branch: str) -> str: _git(repo_root, "worktree", "add", "-q", "-b", branch, str(wt)) return os.path.realpath(str(wt)) def _profile(role: str, *, canonical: str | None = None, allowed=(TARGET_SLUG,)) -> dict: p = { "profile_name": f"mcp-control-plane-{role}", "role": role, "username": "svc", "allowed_operations": ["gitea.read"], "forbidden_operations": [], "allowed_repositories": list(allowed), } if canonical is not None: p["canonical_repository_root"] = canonical return p class _Base(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() tmp = Path(self._tmp.name) self.install = _init_repo(tmp / "Gitea-Tools", INSTALL_URL) self.target = _init_repo(tmp / "mcp-control-plane", TARGET_URL) (Path(self.target) / "branches").mkdir() self.target_wt = _add_worktree( self.target, Path(self.target) / "branches" / "author-issue-1", "feat/issue-1", ) # PROJECT_ROOT and the install git remote are the Gitea-Tools install # checkout — exactly the source that (mis)seeded the session before. self._p_root = mock.patch.object(mcp_server, "PROJECT_ROOT", self.install) self._p_url = mock.patch.object( mcp_server, "_local_git_remote_url", return_value=INSTALL_URL ) self._p_root.start() self._p_url.start() session_ctx._reset_session_context_for_testing() def tearDown(self): mock.patch.stopall() session_ctx._reset_session_context_for_testing() def _seed(self, profile, env=None): session_ctx._reset_session_context_for_testing() with mock.patch.object(mcp_server, "get_profile", return_value=profile), \ mock.patch.dict(os.environ, env or {}, clear=False): return mcp_server._seed_session_context( profile=profile, remote="prgs", host="gitea.prgs.cc", identity="svc", ) def _enforce(self, profile, env=None): with mock.patch.object(mcp_server, "get_profile", return_value=profile), \ mock.patch.dict(os.environ, env or {}, clear=False): mcp_server._enforce_canonical_repository_root( self.target_wt, remote="prgs" ) class TestF1SeedsConfiguredTargetIdentity(_Base): def test_env_config_seeds_target_not_install(self): ctx = self._seed( _profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: self.target}, ) self.assertEqual(ctx["org"], "Scaled-Tech-Consulting") self.assertEqual(ctx["repository"], "mcp-control-plane") # Regression guard: must NOT be the install repo. self.assertNotEqual(ctx["repository"], "Gitea-Tools") def test_profile_field_seeds_target_not_install(self): ctx = self._seed(_profile("merger", canonical=self.target)) self.assertEqual(ctx["org"], "Scaled-Tech-Consulting") self.assertEqual(ctx["repository"], "mcp-control-plane") self.assertNotEqual(ctx["repository"], "Gitea-Tools") def test_enforce_accepts_target_after_seed_reviewer(self): prof = _profile("reviewer", canonical=self.target) self._seed(prof) # Would raise RuntimeError on the self-inflicted identity mismatch # before the fix. self._enforce(prof) def test_enforce_accepts_target_after_seed_merger(self): prof = _profile("merger", canonical=self.target) self._seed(prof) self._enforce(prof) def test_env_overrides_profile_for_seed(self): # profile points at install; env points at the real target → env wins. prof = _profile("reviewer", canonical=self.install, allowed=(TARGET_SLUG,)) ctx = self._seed(prof, env={crr.CANONICAL_ROOT_ENV: self.target}) self.assertEqual(ctx["repository"], "mcp-control-plane") class TestUnconfiguredDefaultUnchanged(_Base): def test_unconfigured_keeps_install_identity(self): prof = _profile("author", allowed=("Scaled-Tech-Consulting/Gitea-Tools",)) ctx = self._seed(prof) # no env, no profile canonical field self.assertEqual(ctx["repository"], "Gitea-Tools") self.assertEqual(ctx["org"], "Scaled-Tech-Consulting") class TestFailClosed(_Base): def _trusted(self, profile, env=None, *, for_mutation=True): with mock.patch.object(mcp_server, "get_profile", return_value=profile), \ mock.patch.dict(os.environ, env or {}, clear=False): return mcp_server._trusted_session_repository( profile, "prgs", for_mutation=for_mutation ) def test_nonexistent_configured_root_fails_closed(self): res = self._trusted( _profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: self.target + "-missing"}, ) self.assertIsNone(res["repository"]) self.assertTrue(res["reasons"]) def test_non_git_configured_root_fails_closed(self): plain = Path(self._tmp.name) / "plain" plain.mkdir() res = self._trusted( _profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: str(plain)} ) self.assertIsNone(res["repository"]) self.assertTrue(any("git" in r for r in res["reasons"])) def test_unallowlisted_target_identity_fails_closed(self): # Configured target is valid, but the profile does not authorize it. res = self._trusted( _profile("reviewer", canonical=self.target, allowed=("Scaled-Tech-Consulting/Gitea-Tools",)), ) self.assertIsNone(res["repository"]) self.assertTrue(any("scope" in r.lower() for r in res["reasons"])) def test_forged_identity_conflict_fails_closed_at_enforce(self): # Seed the target, then present a *different* configured root at # enforcement time: the session pin no longer matches → fail closed. other = _init_repo(Path(self._tmp.name) / "other", OTHER_URL) prof_seed = _profile("reviewer", canonical=self.target) self._seed(prof_seed) prof_drift = _profile( "reviewer", canonical=other, allowed=(TARGET_SLUG, "Scaled-Tech-Consulting/other-repo"), ) with self.assertRaises(RuntimeError): self._enforce(prof_drift) if __name__ == "__main__": unittest.main()