"""#727 / PR #728: author ownership when tracking issue number != PR number.

Regression for REQUEST_CHANGES: gitea_update_pr_branch_by_merge must not require
issue_number == pr_number. Ownership is proven via linked issue + lock/branch
context and fails closed for unrelated issues.
"""

from __future__ import annotations

import os
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch

import issue_lock_store
import gitea_mcp_server as mcp


def _live_lock(
    *,
    issue_number: int,
    branch_name: str = "feat/issue-727-pr-sync-status",
    worktree_path: str = "/tmp/branches/issue-727-pr-sync-status",
    remote: str = "prgs",
    org: str = "Scaled-Tech-Consulting",
    repo: str = "Gitea-Tools",
) -> dict:
    now = datetime.now(timezone.utc)
    return {
        "issue_number": issue_number,
        "branch_name": branch_name,
        "worktree_path": worktree_path,
        "remote": remote,
        "org": org,
        "repo": repo,
        "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
        "acquired_at": now.isoformat(),
        "expires_at": (now + timedelta(hours=2)).isoformat(),
        "owner_pid": os.getpid(),
        "status": "active",
    }


class TestAuthorOwnershipIssuePrMismatch(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.TemporaryDirectory(prefix="gitea-issue-locks-")
        self.lock_dir = self._tmp.name
        os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir

    def tearDown(self):
        # Clear session pointer for this process.
        try:
            ptr = issue_lock_store.session_pointer_path(self.lock_dir)
            if os.path.isfile(ptr):
                os.unlink(ptr)
        except Exception:
            pass
        os.environ.pop("GITEA_ISSUE_LOCK_DIR", None)
        self._tmp.cleanup()

    def test_issue_727_pr_728_session_lock_accepted(self):
        """Tracking issue #727 lock is valid ownership for PR #728."""
        lock = _live_lock(issue_number=727)
        issue_lock_store.bind_session_lock(lock)
        result = mcp._prove_author_ownership_for_pr(
            pr_number=728,
            pr_title="feat: pr sync",
            pr_body="Closes #727\n\nNative PR sync lifecycle.",
            source_branch="feat/issue-727-pr-sync-status",
            remote="prgs",
            host=None,
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path=lock["worktree_path"],
        )
        self.assertTrue(result["proven"], result)
        self.assertTrue(result["has_author_lock"])
        self.assertEqual(result["matched_issue"], 727)
        self.assertEqual(result["matched_via"], "session_lock")
        self.assertIn(727, result["linked_issues"])

    def test_issue_727_pr_728_durable_lock_accepted(self):
        lock = _live_lock(issue_number=727)
        path = issue_lock_store.lock_file_path(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            issue_number=727,
            lock_dir=self.lock_dir,
        )
        issue_lock_store.save_lock_file(path, lock)
        result = mcp._prove_author_ownership_for_pr(
            pr_number=728,
            pr_title="feat: pr sync",
            pr_body="Fixes #727",
            source_branch="feat/issue-727-pr-sync-status",
            remote="prgs",
            host=None,
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertTrue(result["proven"], result)
        self.assertEqual(result["matched_issue"], 727)
        self.assertEqual(result["matched_via"], "durable_lock")

    def test_legacy_same_number_still_accepted(self):
        lock = _live_lock(
            issue_number=100,
            branch_name="fix/issue-100",
            worktree_path="/tmp/branches/issue-100",
        )
        issue_lock_store.bind_session_lock(lock)
        result = mcp._prove_author_ownership_for_pr(
            pr_number=100,
            pr_title="fix something",
            pr_body="Closes #100",
            source_branch="fix/issue-100",
            remote="prgs",
            host=None,
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path=lock["worktree_path"],
        )
        self.assertTrue(result["proven"], result)
        self.assertEqual(result["matched_issue"], 100)

    def test_unrelated_issue_lock_rejected(self):
        """Lock for issue #999 must not authorize PR #728 linked only to #727."""
        lock = _live_lock(
            issue_number=999,
            branch_name="feat/issue-999",
            worktree_path="/tmp/branches/issue-999",
        )
        issue_lock_store.bind_session_lock(lock)
        path = issue_lock_store.lock_file_path(
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            issue_number=999,
            lock_dir=self.lock_dir,
        )
        issue_lock_store.save_lock_file(path, lock)
        result = mcp._prove_author_ownership_for_pr(
            pr_number=728,
            pr_title="feat: pr sync",
            pr_body="Closes #727",
            source_branch="feat/issue-727-pr-sync-status",
            remote="prgs",
            host=None,
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path=lock["worktree_path"],
        )
        self.assertFalse(result["proven"], result)
        self.assertFalse(result["has_author_lock"])
        self.assertIsNone(result["matched_issue"])
        self.assertTrue(result["reasons"])

    def test_branch_mismatch_on_session_lock_fail_closed(self):
        lock = _live_lock(
            issue_number=727,
            branch_name="feat/other-branch",
        )
        issue_lock_store.bind_session_lock(lock)
        result = mcp._prove_author_ownership_for_pr(
            pr_number=728,
            pr_title="feat: pr sync",
            pr_body="Closes #727",
            source_branch="feat/issue-727-pr-sync-status",
            remote="prgs",
            host=None,
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            worktree_path=lock["worktree_path"],
        )
        self.assertFalse(result["proven"], result)
        self.assertTrue(any("branch" in r for r in result["reasons"]))

    def test_no_lock_fail_closed(self):
        result = mcp._prove_author_ownership_for_pr(
            pr_number=728,
            pr_title="feat: pr sync",
            pr_body="Closes #727",
            source_branch="feat/issue-727-pr-sync-status",
            remote="prgs",
            host=None,
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
        )
        self.assertFalse(result["proven"], result)
        self.assertTrue(any("no live author issue lock" in r for r in result["reasons"]))


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