"""Regression tests for durable author worktree resolution (#618)."""

from __future__ import annotations

import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from unittest.mock import MagicMock, patch

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
import namespace_workspace_binding as nwb  # noqa: E402

FAKE_AUTH = {"Authorization": "token test-token"}
current_file_path = Path(__file__).resolve()
if "branches" in current_file_path.parts:
    CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3])
else:
    CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1])


class TestDurableAuthorWorktreeResolution(unittest.TestCase):
    def test_missing_author_env_fails_closed_no_control_fallback(self):
        missing = "/nonexistent/branches/mcp-author-clean-ns"
        result = amw.resolve_durable_author_worktree(
            process_project_root=CONTROL_CHECKOUT_ROOT,
            author_worktree_env=missing,
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
        )
        self.assertTrue(result["block"])
        self.assertTrue(result["bound_worktree_missing"])
        self.assertFalse(result["silent_control_fallback"])
        self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, result["reasons"][0])
        self.assertNotEqual(
            os.path.realpath(result["workspace_path"]),
            os.path.realpath(CONTROL_CHECKOUT_ROOT),
        )

    def test_missing_active_env_fails_closed(self):
        missing = "/nonexistent/branches/deleted-active"
        result = amw.resolve_durable_author_worktree(
            process_project_root=CONTROL_CHECKOUT_ROOT,
            active_worktree_env=missing,
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
        )
        self.assertTrue(result["block"])
        self.assertTrue(result["bound_worktree_missing"])
        self.assertIn(amw.ACTIVE_WORKTREE_ENV, result["workspace_binding_source"])

    def test_derives_from_active_author_issue_lock(self):
        lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock")
        result = amw.resolve_durable_author_worktree(
            process_project_root=CONTROL_CHECKOUT_ROOT,
            session_lock_worktree=lock_wt,
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
            validate=False,
        )
        self.assertEqual(
            result["workspace_path"], os.path.realpath(os.path.abspath(lock_wt))
        )
        self.assertIn("issue lock", result["workspace_binding_source"])

    def test_explicit_worktree_path_wins_over_lock(self):
        explicit = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-explicit")
        lock_wt = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "issue-618-lock")
        result = amw.resolve_durable_author_worktree(
            worktree_path=explicit,
            process_project_root=CONTROL_CHECKOUT_ROOT,
            session_lock_worktree=lock_wt,
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
            validate=False,
        )
        self.assertEqual(
            result["workspace_path"], os.path.realpath(os.path.abspath(explicit))
        )
        self.assertEqual(result["workspace_binding_source"], "worktree_path argument")

    def test_no_binding_does_not_silently_use_control_checkout(self):
        result = amw.resolve_durable_author_worktree(
            process_project_root=CONTROL_CHECKOUT_ROOT,
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
        )
        self.assertTrue(result["block"])
        self.assertFalse(result["silent_control_fallback"])
        blob = " ".join(result["reasons"])
        self.assertIn("control checkout", blob)
        self.assertIn("forbidden", blob)

    def test_process_root_under_branches_is_allowed(self):
        branches_root = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "session-wt")
        result = amw.resolve_durable_author_worktree(
            process_project_root=branches_root,
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
            validate=False,
        )
        self.assertFalse(result["block"])
        self.assertEqual(
            result["workspace_path"], os.path.realpath(branches_root)
        )
        self.assertIn("branches/", result["workspace_binding_source"])

    def test_lock_ownership_mismatch_fails_closed(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = tmp
            branches = os.path.join(root, "branches")
            os.makedirs(os.path.join(branches, "a"))
            os.makedirs(os.path.join(branches, "b"))
            # Seed a fake .git so membership/list may soft-fail without hard error
            os.makedirs(os.path.join(root, ".git"))
            result = amw.resolve_durable_author_worktree(
                worktree_path=os.path.join(branches, "a"),
                process_project_root=root,
                session_lock_worktree=os.path.join(branches, "b"),
                canonical_repo_root=root,
                validate=True,
            )
            self.assertTrue(result["block"])
            self.assertTrue(
                any("lock" in r.lower() and "match" in r.lower() for r in result["reasons"])
            )

    def test_traversal_safety_blocks_escape(self):
        assessment = amw.assess_path_traversal_safety(
            path="/tmp/other-repo/branches/evil",
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
        )
        self.assertTrue(assessment["block"])
        self.assertTrue(any("escapes" in r for r in assessment["reasons"]))

    def test_bound_worktree_existence_reports_null_git_root(self):
        assessment = amw.assess_bound_worktree_existence(
            configured_path="/nonexistent/branches/gone",
            binding_source=f"{amw.AUTHOR_WORKTREE_ENV} environment variable",
            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
            profile_name="prgs-author",
        )
        self.assertTrue(assessment["block"])
        self.assertIsNone(assessment["inspected_git_root"])
        self.assertFalse(assessment["path_exists"])
        msg = amw.format_bound_worktree_missing_error(assessment)
        self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, msg)
        self.assertIn("prgs-author", msg)
        self.assertIn("recreate or repoint", msg.lower())


class TestNamespaceAuthorNoDemotion(unittest.TestCase):
    def test_author_missing_env_not_demoted_to_process_root(self):
        missing = "/nonexistent/branches/mcp-author-clean-ns"
        demotions: list[str] = []
        path, source = nwb.resolve_namespace_workspace(
            role_kind="author",
            process_project_root=CONTROL_CHECKOUT_ROOT,
            env={amw.AUTHOR_WORKTREE_ENV: missing},
            demotions=demotions,
            verify_paths=True,
        )
        self.assertIn("AUTHOR", source)
        self.assertNotEqual(os.path.realpath(path), os.path.realpath(CONTROL_CHECKOUT_ROOT))
        self.assertTrue(any("not demoted" in d for d in demotions))

    def test_reviewer_still_demotes_missing_env(self):
        """#702 demotion retained for non-author roles."""
        demotions: list[str] = []
        path, source = nwb.resolve_namespace_workspace(
            role_kind="reviewer",
            process_project_root=CONTROL_CHECKOUT_ROOT,
            env={"GITEA_ACTIVE_WORKTREE": "/nonexistent/branches/review-gone"},
            demotions=demotions,
            verify_paths=True,
        )
        self.assertEqual(source, "MCP server process root (default)")
        self.assertEqual(path, os.path.realpath(CONTROL_CHECKOUT_ROOT))
        self.assertTrue(demotions)

    def test_mutation_context_surfaces_missing_binding_health(self):
        ctx = nwb.resolve_namespace_mutation_context(
            role_kind="author",
            worktree_path=None,
            process_project_root=CONTROL_CHECKOUT_ROOT,
            env={amw.AUTHOR_WORKTREE_ENV: "/nonexistent/branches/mcp-author-clean-ns"},
            profile_name="prgs-author",
        )
        self.assertTrue(ctx.get("bound_worktree_missing"))
        self.assertTrue(ctx.get("author_worktree_block"))
        self.assertIsNone(ctx.get("inspected_git_root"))
        self.assertFalse(ctx.get("path_exists"))


class TestCreateIssueAndCommentAgreeOnMissingWorktree(unittest.TestCase):
    """AC3/AC4: create_issue and create_issue_comment enforce the same rule."""

    def setUp(self):
        srv._preflight_whoami_called = True
        srv._preflight_capability_called = True
        srv._preflight_resolved_role = "author"
        srv._preflight_resolved_task = None
        srv._preflight_whoami_violation = False
        srv._preflight_capability_violation = False
        self._orig_in_test = srv._preflight_in_test_mode
        srv._preflight_in_test_mode = lambda: False
        self._lock_patch = patch(
            "gitea_mcp_server._session_author_lock_worktree", return_value=None
        )
        self._lock_patch.start()
        self.addCleanup(self._restore)

    def _restore(self):
        srv._preflight_in_test_mode = self._orig_in_test
        srv._preflight_resolved_task = None
        self._lock_patch.stop()
        os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None)
        os.environ.pop(amw.ACTIVE_WORKTREE_ENV, None)

    def _assert_blocked_missing(self, result_or_exc):
        if isinstance(result_or_exc, BaseException):
            blob = str(result_or_exc)
        else:
            blob = " ".join(
                str(x)
                for x in (
                    result_or_exc.get("reasons") or [],
                    result_or_exc.get("message"),
                    result_or_exc.get("blocker_kind"),
                )
                if x
            )
            if not blob:
                blob = str(result_or_exc)
        self.assertTrue(
            amw.BOUND_WORKTREE_MISSING_MESSAGE in blob
            or "does not exist" in blob
            or "bound worktree" in blob.lower(),
            msg=blob,
        )

    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
    @patch(
        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
        return_value=(True, []),
    )
    @patch("gitea_mcp_server.api_request")
    @patch("gitea_mcp_server.api_get_all", return_value=[])
    def test_create_issue_blocked_when_author_env_missing(
        self, _get_all, mock_api, _role, _ns, _prof, _auth
    ):
        missing = os.path.join(
            CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env"
        )
        os.environ[amw.AUTHOR_WORKTREE_ENV] = missing
        srv._preflight_resolved_task = "create_issue"
        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
            try:
                res = srv.gitea_create_issue(title="Test issue", body="body text here")
            except RuntimeError as exc:
                self._assert_blocked_missing(exc)
            else:
                self.assertFalse(res.get("success", True) and res.get("number"))
                self._assert_blocked_missing(res)
        mock_api.assert_not_called()

    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
    @patch("gitea_mcp_server.api_request")
    def test_create_issue_comment_blocked_when_author_env_missing(self, mock_api, _auth):
        missing = os.path.join(
            CONTROL_CHECKOUT_ROOT, "branches", "nonexistent-618-author-env"
        )
        os.environ[amw.AUTHOR_WORKTREE_ENV] = missing
        srv._preflight_resolved_task = "comment_issue"
        author_env = {
            "GITEA_PROFILE_NAME": "gitea-author",
            "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
            amw.AUTHOR_WORKTREE_ENV: missing,
        }
        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
            with patch.dict(os.environ, author_env, clear=False):
                try:
                    res = srv.gitea_create_issue_comment(
                        issue_number=618,
                        body="evidence comment",
                        remote="prgs",
                    )
                except RuntimeError as exc:
                    self._assert_blocked_missing(exc)
                else:
                    self.assertFalse(res.get("success", True))
                    self._assert_blocked_missing(res)
        mock_api.assert_not_called()


class TestRuntimeContextUnhealthyMissingWorktree(unittest.TestCase):
    def setUp(self):
        srv._preflight_whoami_called = True
        srv._preflight_capability_called = True
        srv._preflight_resolved_role = "author"
        srv._preflight_whoami_violation = False
        srv._preflight_capability_violation = False
        self._lock_patch = patch(
            "gitea_mcp_server._session_author_lock_worktree", return_value=None
        )
        self._lock_patch.start()

    def tearDown(self):
        self._lock_patch.stop()
        os.environ.pop(amw.AUTHOR_WORKTREE_ENV, None)

    def test_assess_preflight_reports_null_git_root_and_missing(self):
        missing = "/nonexistent/branches/mcp-author-clean-ns"
        os.environ[amw.AUTHOR_WORKTREE_ENV] = missing
        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
            with patch("gitea_mcp_server.get_profile", return_value={
                "profile_name": "prgs-author",
                "allowed_operations": ["gitea.pr.create"],
                "forbidden_operations": [],
            }):
                status = srv.assess_preflight_status()
        self.assertFalse(status["preflight_ready"])
        blob = " ".join(status["preflight_block_reasons"])
        self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, blob)
        details = status["preflight_workspace"]
        self.assertIsNotNone(details)
        self.assertTrue(details.get("bound_worktree_missing"))
        self.assertIsNone(details.get("inspected_git_root"))
        self.assertFalse(details.get("path_exists"))
        self.assertFalse(details.get("workspace_healthy"))


class TestThreadLedgerExample(unittest.TestCase):
    def test_bound_worktree_missing_ledger_example_exists(self):
        import thread_state_ledger_examples as examples

        names = [name for name, _h, _l in examples.EXAMPLES]
        self.assertIn("bound_worktree_missing_blocker", names)
        for name, _handoff, ledger in examples.EXAMPLES:
            if name == "bound_worktree_missing_blocker":
                self.assertIn(amw.BOUND_WORKTREE_MISSING_MESSAGE, ledger)
                self.assertIn("inspected_git_root", ledger)
                self.assertIn("operator", ledger.lower())
                break


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