Gitea-Tools/tests/test_workspace_guard_alignment.py

"""Tests for runtime_context / mutation-guard workspace alignment (#460)."""

from __future__ import annotations

import os
import sys
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import author_mutation_worktree as amw  # noqa: E402
import gitea_mcp_server as srv  # noqa: E402

current_file_path = Path(__file__).resolve()
if "branches" in current_file_path.parts:
    CONTROL_ROOT = str(current_file_path.parents[3])
    BRANCHES_WORKTREE = str(current_file_path.parents[1])
else:
    CONTROL_ROOT = str(current_file_path.parents[1])
    BRANCHES_WORKTREE = str(current_file_path.parents[1] / "branches" / "mock-worktree")
MCP_PROCESS_ROOT = BRANCHES_WORKTREE


class TestCanonicalRepoRoot(unittest.TestCase):
    @mock.patch("subprocess.run")
    def test_resolves_main_repo_from_branches_worktree(self, mock_run):
        mock_run.return_value = MagicMock(
            returncode=0,
            stdout=f"{CONTROL_ROOT}/.git\n",
        )
        root = amw.resolve_canonical_repo_root(BRANCHES_WORKTREE, MCP_PROCESS_ROOT)
        self.assertEqual(root, CONTROL_ROOT)

    def test_falls_back_when_git_unavailable(self):
        root = amw.resolve_canonical_repo_root("/missing/path", MCP_PROCESS_ROOT)
        self.assertEqual(root, os.path.realpath(MCP_PROCESS_ROOT))


class TestWorkspaceRepoMembership(unittest.TestCase):
    @mock.patch("os.path.isdir", return_value=True)
    @mock.patch("os.path.exists", return_value=True)
    @mock.patch("subprocess.run")
    def test_valid_branches_worktree_accepted(self, mock_run, *_exists):
        mock_run.return_value = MagicMock(
            returncode=0,
            stdout=f"{CONTROL_ROOT}/.git\n",
        )
        result = amw.assess_workspace_repo_membership(
            workspace_path=BRANCHES_WORKTREE,
            canonical_repo_root=CONTROL_ROOT,
        )
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])

    @mock.patch("os.path.isdir", return_value=True)
    @mock.patch("os.path.exists", return_value=True)
    @mock.patch("subprocess.run")
    def test_wrong_repo_rejected(self, mock_run, *_exists):
        mock_run.return_value = MagicMock(
            returncode=0,
            stdout="/other/repo/.git\n",
        )
        result = amw.assess_workspace_repo_membership(
            workspace_path=BRANCHES_WORKTREE,
            canonical_repo_root=CONTROL_ROOT,
        )
        self.assertTrue(result["block"])
        self.assertIn("does not belong", result["reasons"][0])

    @mock.patch("os.path.exists", return_value=False)
    def test_missing_worktree_rejected(self, *_exists):
        result = amw.assess_workspace_repo_membership(
            workspace_path=f"{CONTROL_ROOT}/branches/missing-worktree",
            canonical_repo_root=CONTROL_ROOT,
        )
        self.assertTrue(result["block"])
        self.assertIn("does not exist", result["reasons"][0])


class TestRuntimeContextGuardAlignment(unittest.TestCase):
    def setUp(self):
        srv._preflight_whoami_called = True
        srv._preflight_capability_called = True
        srv._preflight_resolved_role = "author"
        self._orig_in_test = srv._preflight_in_test_mode
        srv._preflight_in_test_mode = lambda: False
        self._env_patch = mock.patch.dict(
            os.environ,
            {},
            clear=False,
        )
        self._env_patch.start()
        os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
        os.environ.pop("GITEA_AUTHOR_WORKTREE", None)

    def tearDown(self):
        srv._preflight_in_test_mode = self._orig_in_test
        self._env_patch.stop()

    @mock.patch("subprocess.run")
    @mock.patch("os.path.isdir", return_value=True)
    @mock.patch("os.path.exists", return_value=True)
    def test_runtime_context_and_guard_share_resolved_workspace(
        self, _exists, _isdir, mock_run
    ):
        def run_side_effect(cmd, *args, **kwargs):
            res = MagicMock(returncode=0)
            if "--git-common-dir" in cmd:
                res.stdout = f"{CONTROL_ROOT}/.git\n"
            elif "--show-toplevel" in cmd:
                cwd = cmd[cmd.index("-C") + 1] if "-C" in cmd else ""
                res.stdout = f"{cwd}\n"
            else:
                res.stdout = f"{CONTROL_ROOT}\n"
            return res
        mock_run.side_effect = run_side_effect
        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
            ctx = srv._resolve_author_mutation_context(BRANCHES_WORKTREE)
            status = srv.assess_preflight_status(worktree_path=BRANCHES_WORKTREE)
        self.assertEqual(ctx["workspace_path"], os.path.realpath(BRANCHES_WORKTREE))
        self.assertEqual(ctx["canonical_repo_root"], CONTROL_ROOT)
        self.assertFalse(ctx["roots_aligned"])
        self.assertEqual(
            status["preflight_workspace"]["active_task_workspace_root"],
            os.path.realpath(BRANCHES_WORKTREE),
        )
        self.assertEqual(
            status["preflight_workspace"]["canonical_repository_root"],
            CONTROL_ROOT,
        )
        self.assertIn("workspace_root_mismatch", status["preflight_workspace"])

    @mock.patch("subprocess.run")
    @mock.patch("os.path.isdir", return_value=True)
    @mock.patch("os.path.exists", return_value=True)
    def test_declared_branches_worktree_passes_when_mcp_root_differs(
        self, _exists, _isdir, mock_run
    ):
        mock_run.return_value = MagicMock(
            returncode=0,
            stdout=f"{CONTROL_ROOT}/.git\n",
        )
        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
            with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE)

    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
    @mock.patch(
        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value={
            "current_branch": "master",
            "head_sha": "a" * 40,
            "porcelain_status": "",
        },
    )
    def test_stable_checkout_still_rejected(self, _git, _remote_sha):
        with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
            with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                with self.assertRaises(RuntimeError) as ctx:
                    srv.verify_preflight_purity()
        self.assertIn("stable control checkout", str(ctx.exception))

    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
    @mock.patch(
        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
        return_value={
            "current_branch": "master",
            "head_sha": "a" * 40,
            "porcelain_status": "",
        },
    )
    @mock.patch("os.path.isdir", return_value=True)
    @mock.patch("os.path.exists", return_value=True)
    @mock.patch("subprocess.run")
    def test_non_branches_worktree_rejected(
        self, mock_run, mock_exists, mock_isdir, _git, _remote_sha,
    ):
        outside = "/tmp/outside-repo-checkout"
        mock_run.return_value = MagicMock(
            returncode=0,
            stdout=f"{CONTROL_ROOT}/.git\n",
        )
        with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
            with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                with self.assertRaises(RuntimeError) as ctx:
                    srv.verify_preflight_purity(worktree_path=outside)
        self.assertIn("not under", str(ctx.exception))


if __name__ == "__main__":
    unittest.main()