"""Publication of an unpublished local commit (#812 AC20).

Entry point B of #812: a registered worktree, clean, on its issue branch,
holding a local commit that has never been published. Exact-owner lease renewal
refuses such a claim for want of an observable remote head, and every existing
publication path is lock-derived, so the two predicates close a cycle around
work that is otherwise complete.

These tests exercise the disposition through its *evidence*, never through any
particular issue number: every case uses an arbitrary issue number against a
synthetic repository, and the same assertions hold for any other. Nothing here
reads, writes, or references the live protected worktree named in #812 AC17 —
that content is preserved evidence for the duration of this work, so the
fixtures below build their own repositories from scratch.

The remote is a local bare repository, so publication and read-after-write
verification are genuinely executed rather than mocked.
"""

from __future__ import annotations

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

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

import branch_publish  # noqa: E402
import issue_lock_provenance  # noqa: E402
import issue_lock_renewal  # noqa: E402
import issue_lock_store  # noqa: E402
import mcp_server  # noqa: E402
from mutation_profile_fixture import shared_mutation_env  # noqa: E402

ISSUE = 9812
BRANCH = f"feat/issue-{ISSUE}-publish-fixture"
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
GIT_REMOTE = "prgs"


def _ts(hours: int) -> str:
    return (
        (datetime.now(timezone.utc) + timedelta(hours=hours))
        .isoformat()
        .replace("+00:00", "Z")
    )


class _PublishBase(unittest.TestCase):
    """Real git repo + real bare remote + durable lock naming the caller.

    The recorded owner pid is deliberately **this live process**. That mirrors
    the production shape #812 documents, where the pid belongs to a long-running
    MCP daemon rather than to a dead author client, and it proves publication
    never depends on a dead process (#812 AC24).
    """

    def setUp(self):
        self.lock_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.lock_dir.cleanup)
        self.origin = tempfile.mkdtemp(prefix="issue812-origin-")
        self.repo = tempfile.mkdtemp(prefix="issue812-work-")
        for path in (self.origin, self.repo):
            self.addCleanup(
                lambda p=path: subprocess.run(["rm", "-rf", p], check=False)
            )
        self._init_repos()
        self.remotes = patch.dict(
            mcp_server.REMOTES,
            {"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
        )
        self.remotes.start()
        self.addCleanup(patch.stopall)
        mcp_server._IDENTITY_CACHE.clear()

    # ── fixture construction ─────────────────────────────────────────────
    def _git(self, *args, cwd=None):
        return subprocess.run(
            ["git", "-C", cwd or self.repo, *args],
            capture_output=True,
            text=True,
            check=True,
        )

    def _init_repos(self):
        subprocess.run(
            ["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True
        )
        self._git("init", "-q", "-b", "master")
        self._git("config", "user.email", "test@example.com")
        self._git("config", "user.name", "Test")
        self._git("remote", "add", GIT_REMOTE, self.origin)

        with open(os.path.join(self.repo, "seed.txt"), "w") as fh:
            fh.write("seed\n")
        self._git("add", "seed.txt")
        self._git("commit", "-q", "-m", "seed")
        self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
        self._git("push", "-q", GIT_REMOTE, "master")

        self._git("checkout", "-q", "-b", BRANCH)
        with open(os.path.join(self.repo, "work.txt"), "w") as fh:
            fh.write("unpublished implementation\n")
        self._git("add", "work.txt")
        self._git("commit", "-q", "-m", "unpublished implementation")
        self.head_sha = self._git("rev-parse", "HEAD").stdout.strip()
        self.worktree = os.path.realpath(self.repo)

    def lock_path(self):
        return issue_lock_store.lock_file_path(
            remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
            lock_dir=self.lock_dir.name,
        )

    def write_lock(self, **overrides):
        path = self.lock_path()
        claimant = overrides.pop(
            "claimant", {"username": IDENTITY, "profile": PROFILE}
        )
        pid = overrides.pop("session_pid", os.getpid())
        lease = {
            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
            "issue_number": ISSUE,
            "pr_number": None,
            "branch": overrides.get("branch_name", BRANCH),
            "worktree_path": overrides.get("worktree_path", self.worktree),
            "claimant": claimant,
            "created_at": _ts(-2),
            "last_heartbeat_at": _ts(-2),
            # Expired: entry point B's lease has lapsed, which is precisely why
            # renewal — and therefore a published head — is needed.
            "expires_at": _ts(-1),
        }
        lease.update(overrides.pop("work_lease", {}))
        data = {
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "remote": "prgs",
            "org": ORG,
            "repo": REPO,
            "worktree_path": self.worktree,
            "session_pid": pid,
            "pid": pid,
            "lock_generation": 1,
            "work_lease": lease,
            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
                tool="gitea_lock_issue", claimant=claimant
            ),
        }
        data.update(overrides)
        data["lock_file_path"] = path
        issue_lock_store.save_lock_file(path, data)
        return path

    def _tool_env(self):
        env = shared_mutation_env(
            PROFILE, include_example_repo=True,
            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
        )
        env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
        # These tests repoint PROJECT_ROOT at a synthetic repository so the
        # registered-worktree proof runs for real. Pin the parity gate to the
        # server's own startup head so that repointing does not read as a stale
        # daemon; the gate itself stays live and enforced.
        startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or ""
        env["GITEA_TEST_CURRENT_HEAD"] = startup_head
        env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head
        return env

    # ── tool driver ──────────────────────────────────────────────────────
    def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
        """Drive the public publication tool against the synthetic fixture."""
        env = self._tool_env()
        with patch(
            "mcp_server._list_open_pulls", return_value=list(open_prs or [])
        ), patch(
            "mcp_server._auth", return_value="token x"
        ), patch(
            "mcp_server.get_auth_header", return_value="token x"
        ), patch(
            "mcp_server._work_lease_claimant",
            return_value={"username": IDENTITY, "profile": PROFILE},
        ), patch.object(
            mcp_server, "PROJECT_ROOT", self.repo
        ), patch.dict(os.environ, env, clear=True):
            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
            return mcp_server.gitea_publish_unpublished_issue_branch(
                issue_number=kwargs.pop("issue_number", ISSUE),
                branch_name=kwargs.pop("branch_name", BRANCH),
                worktree_path=kwargs.pop("worktree_path", self.worktree),
                expected_head=expected_head or self.head_sha,
                remote="prgs",
                git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE),
                **kwargs,
            )

    def remote_head(self, branch=BRANCH):
        res = subprocess.run(
            ["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch],
            capture_output=True, text=True, check=False,
        )
        return (res.stdout or "").strip() or None


class TestSuccessfulPublication(_PublishBase):
    """AC20 — the branch becomes observable and is verified after the write."""

    def test_publishes_clean_unpublished_commit(self):
        self.write_lock()
        self.assertIsNone(self.remote_head(), "fixture must start unpublished")

        result = self.run_publish()

        self.assertTrue(result["success"], result.get("reasons"))
        self.assertTrue(result["performed"])
        self.assertTrue(result["published"])
        self.assertTrue(result["verified"], "read-after-write must be proven")
        self.assertEqual(result["remote_head_sha"], self.head_sha)
        self.assertEqual(self.remote_head(), self.head_sha)

    def test_publication_does_not_rewrite_the_commit(self):
        self.write_lock()
        self.run_publish()
        # The published object is the same commit, not a copy or a rewrite.
        self.assertEqual(self.remote_head(), self.head_sha)
        self.assertEqual(
            self._git("rev-parse", "HEAD").stdout.strip(), self.head_sha
        )

    def test_exact_next_action_names_the_lock_call(self):
        self.write_lock()
        result = self.run_publish()
        self.assertIn("gitea_lock_issue", result["exact_next_action"])


class TestFailsClosed(_PublishBase):
    """AC20/AC9 — each refusal reason, exercised independently."""

    def test_changed_local_head_refuses(self):
        self.write_lock()
        stale = self.base_sha  # a real commit, but not the declared head
        result = self.run_publish(expected_head=stale)
        self.assertFalse(result["success"])
        self.assertTrue(
            any("local commit changed" in r for r in result["reasons"]),
            result["reasons"],
        )
        self.assertIsNone(self.remote_head(), "refusal must not publish")

    def test_abbreviated_sha_refuses(self):
        self.write_lock()
        result = self.run_publish(expected_head=self.head_sha[:8])
        self.assertFalse(result["success"])
        self.assertTrue(
            any("40-character" in r for r in result["reasons"]), result["reasons"]
        )

    def test_dirty_tracked_worktree_refuses(self):
        self.write_lock()
        with open(os.path.join(self.repo, "work.txt"), "a") as fh:
            fh.write("uncommitted edit\n")

        result = self.run_publish()

        self.assertFalse(result["success"])
        self.assertTrue(
            any("dirty tracked files" in r for r in result["reasons"]),
            result["reasons"],
        )
        self.assertIn("work.txt", result["evidence"]["dirty_tracked_files"])
        self.assertIsNone(self.remote_head())

    def test_untracked_file_refuses(self):
        self.write_lock()
        with open(os.path.join(self.repo, "stray.txt"), "w") as fh:
            fh.write("not committed\n")

        result = self.run_publish()

        self.assertFalse(result["success"])
        self.assertTrue(
            any("untracked files" in r for r in result["reasons"]), result["reasons"]
        )
        self.assertIn("stray.txt", result["evidence"]["untracked_files"])
        self.assertIsNone(self.remote_head())

    def test_unexpected_remote_head_refuses(self):
        """A remote head that is not an ancestor must never be overwritten."""
        self.write_lock()
        # Publish a divergent commit to the branch from a separate line.
        self._git("checkout", "-q", "-b", "divergent", self.base_sha)
        with open(os.path.join(self.repo, "other.txt"), "w") as fh:
            fh.write("someone else's work\n")
        self._git("add", "other.txt")
        self._git("commit", "-q", "-m", "divergent")
        divergent = self._git("rev-parse", "HEAD").stdout.strip()
        self._git("push", "-q", GIT_REMOTE, f"{divergent}:refs/heads/{BRANCH}")
        self._git("checkout", "-q", BRANCH)

        result = self.run_publish()

        self.assertFalse(result["success"])
        self.assertTrue(
            any("not an ancestor" in r for r in result["reasons"]), result["reasons"]
        )
        self.assertEqual(
            self.remote_head(), divergent, "the other head must survive intact"
        )

    def test_fast_forward_remote_head_is_allowed(self):
        """An ancestor head is an honest fast-forward, not a conflict."""
        self.write_lock()
        self._git("push", "-q", GIT_REMOTE, f"{self.base_sha}:refs/heads/{BRANCH}")

        result = self.run_publish()

        self.assertTrue(result["success"], result.get("reasons"))
        self.assertTrue(result["evidence"]["fast_forward_from_remote"])
        self.assertEqual(self.remote_head(), self.head_sha)

    def test_content_hash_mismatch_refuses(self):
        self.write_lock()
        wrong = {"work.txt": "0" * 64}

        result = self.run_publish(expected_file_hashes=wrong)

        self.assertFalse(result["success"])
        self.assertTrue(
            any("declared content hashes" in r for r in result["reasons"]),
            result["reasons"],
        )
        self.assertFalse(result["evidence"]["file_hashes_verified"])
        self.assertIsNone(self.remote_head())

    def test_matching_content_hashes_publish(self):
        self.write_lock()
        digests = branch_publish.hash_worktree_files(self.worktree, ["work.txt"])

        result = self.run_publish(expected_file_hashes=digests)

        self.assertTrue(result["success"], result.get("reasons"))
        self.assertTrue(result["evidence"]["file_hashes_verified"])

    def test_missing_declared_file_refuses(self):
        self.write_lock()
        result = self.run_publish(expected_file_hashes={"absent.txt": "0" * 64})
        self.assertFalse(result["success"])
        self.assertTrue(
            any("missing or unreadable" in r for r in result["reasons"]),
            result["reasons"],
        )

    def test_foreign_claimant_refuses(self):
        """Ownership comes from the durable record, not from the caller."""
        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})

        result = self.run_publish()

        self.assertFalse(result["success"])
        self.assertTrue(
            any("foreign claim" in r for r in result["reasons"]), result["reasons"]
        )
        self.assertIsNone(self.remote_head())

    def test_foreign_profile_refuses(self):
        self.write_lock(
            claimant={"username": IDENTITY, "profile": "test-reviewer-prgs"}
        )
        result = self.run_publish()
        self.assertFalse(result["success"])
        self.assertTrue(
            any("claimant profile" in r for r in result["reasons"]), result["reasons"]
        )

    def test_absent_lock_record_refuses(self):
        """No recorded claim means this cannot be used to bypass the lock."""
        result = self.run_publish()  # no write_lock()

        self.assertFalse(result["success"])
        self.assertTrue(
            any("no durable issue-lock record" in r for r in result["reasons"]),
            result["reasons"],
        )
        self.assertIsNone(self.remote_head())

    def test_branch_mismatch_against_lock_refuses(self):
        self.write_lock(branch_name=f"feat/issue-{ISSUE}-different")
        result = self.run_publish()
        self.assertFalse(result["success"])
        self.assertTrue(
            any("records branch" in r for r in result["reasons"]), result["reasons"]
        )

    def test_worktree_mismatch_against_lock_refuses(self):
        self.write_lock(worktree_path="/tmp/some/other/worktree")
        result = self.run_publish()
        self.assertFalse(result["success"])
        self.assertTrue(
            any("records worktree" in r for r in result["reasons"]), result["reasons"]
        )

    def test_competing_open_pr_on_another_branch_refuses(self):
        self.write_lock()
        competing = [{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]

        result = self.run_publish(open_prs=competing)

        self.assertFalse(result["success"])
        self.assertTrue(
            any("already claim issue" in r for r in result["reasons"]),
            result["reasons"],
        )
        self.assertIsNone(self.remote_head())

    def test_open_pr_on_the_same_branch_is_not_competing(self):
        """This branch's own PR is not a rival claim against itself."""
        self.write_lock()
        own = [{"number": 77, "head": {"ref": BRANCH}}]

        result = self.run_publish(open_prs=own)

        self.assertTrue(result["success"], result.get("reasons"))


class TestGuardStrictnessPreserved(_PublishBase):
    """AC15 — publication is an operation, never a weakening of the guards."""

    def test_non_issue_branch_refuses(self):
        self._git("checkout", "-q", "-b", "scratch/not-issue-linked")
        self.write_lock(branch_name="scratch/not-issue-linked")

        result = self.run_publish(branch_name="scratch/not-issue-linked")

        self.assertFalse(result["success"])
        self.assertTrue(
            any("issue-linked" in r for r in result["reasons"]), result["reasons"]
        )

    def test_stable_branch_refuses(self):
        self.write_lock(branch_name="master")
        result = self.run_publish(branch_name="master")
        self.assertFalse(result["success"])
        self.assertTrue(
            any("issue-linked" in r or "stable branch" in r for r in result["reasons"]),
            result["reasons"],
        )

    def test_branch_number_must_match_the_issue(self):
        other = "feat/issue-7777-mismatched"
        self._git("checkout", "-q", "-b", other)
        self.write_lock(branch_name=other)
        result = self.run_publish(branch_name=other)
        self.assertFalse(result["success"])
        self.assertTrue(
            any("does not carry issue number" in r for r in result["reasons"]),
            result["reasons"],
        )

    def test_unregistered_worktree_refuses(self):
        """#713 — an improvised directory is not a registered worktree."""
        path = self.write_lock()
        assessment = branch_publish.assess_unpublished_commit_publication(
            issue_lock_store.read_lock_file(path),
            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
            expected_head=self.head_sha, remote="prgs", org=ORG, repo=REPO,
            identity=IDENTITY, profile=PROFILE,
            worktree_state={
                "current_branch": BRANCH, "porcelain_status": "",
                "head_sha": self.head_sha,
            },
            worktree_registered=False,
            remote_probe={"probe_ok": True, "remote_branch_exists": False},
        )
        self.assertEqual(assessment["outcome"], branch_publish.REFUSED)
        self.assertTrue(
            any("not listed in git worktree list" in r
                for r in assessment["reasons"]),
            assessment["reasons"],
        )

    def test_unobservable_remote_refuses(self):
        """An unknown remote state must not be mistaken for an absent branch."""
        self.write_lock()
        result = self.run_publish(git_remote_name="no-such-remote")
        self.assertFalse(result["success"])
        self.assertTrue(
            any("could not be observed" in r for r in result["reasons"]),
            result["reasons"],
        )


class TestRecordSeparation(_PublishBase):
    """AC23 — the durable issue lock and the workflow lease are distinct."""

    def test_publication_leaves_the_issue_lock_byte_identical(self):
        path = self.write_lock()
        with open(path, "rb") as fh:
            before = fh.read()

        result = self.run_publish()

        self.assertTrue(result["success"], result.get("reasons"))
        with open(path, "rb") as fh:
            after = fh.read()
        self.assertEqual(before, after, "publication must not mutate the lock record")
        self.assertFalse(result["issue_lock_record_mutated"])
        self.assertFalse(result["workflow_lease_touched"])

    def test_refusal_also_reports_untouched_records(self):
        result = self.run_publish()  # refuses: no lock record
        self.assertFalse(result["issue_lock_record_mutated"])
        self.assertFalse(result["workflow_lease_touched"])

    def test_lock_generation_is_not_advanced(self):
        path = self.write_lock()
        self.run_publish()
        lock = issue_lock_store.read_lock_file(path)
        self.assertEqual(lock["lock_generation"], 1)


class TestTruthfulProcessEvidence(_PublishBase):
    """AC24 — a live daemon pid is never represented as a dead process."""

    def test_live_recorded_pid_does_not_block_publication(self):
        # The recorded pid is this live process, standing in for the live MCP
        # daemon. Reclaim would refuse here; publication legitimately does not.
        path = self.write_lock(session_pid=os.getpid())
        lock = issue_lock_store.read_lock_file(path)
        self.assertEqual(lock["pid"], os.getpid())

        result = self.run_publish()

        self.assertTrue(result["success"], result.get("reasons"))
        self.assertEqual(self.remote_head(), self.head_sha)

    def test_liveness_is_not_consulted_as_evidence(self):
        self.write_lock(session_pid=os.getpid())
        result = self.run_publish()
        self.assertFalse(result["evidence"]["owner_pid_liveness_consulted"])

    def test_reclaim_still_refuses_for_the_same_live_pid(self):
        """Publication does not soften the reclaim predicate it routes around."""
        path = self.write_lock(session_pid=os.getpid())
        lock = issue_lock_store.read_lock_file(path)
        reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
        self.assertFalse(reclaim["reclaim_allowed"])


class TestIdempotentRetry(_PublishBase):
    """AC20 — retry is safe and read-after-write is proven every time."""

    def test_second_publication_reports_already_published(self):
        self.write_lock()
        first = self.run_publish()
        self.assertTrue(first["performed"])

        second = self.run_publish()

        self.assertTrue(second["success"], second.get("reasons"))
        self.assertFalse(second["performed"], "no second push is needed")
        self.assertTrue(second["published"])
        self.assertTrue(second["verified"])
        self.assertEqual(second["outcome"], branch_publish.ALREADY_PUBLISHED)
        self.assertEqual(self.remote_head(), self.head_sha)


class TestDryRun(_PublishBase):
    """AC12 — dry run reports the decision and mutates nothing."""

    def test_dry_run_reports_intent_without_publishing(self):
        self.write_lock()

        result = self.run_publish(dry_run=True)

        self.assertTrue(result["success"])
        self.assertTrue(result["dry_run"])
        self.assertTrue(result["would_publish"])
        self.assertFalse(result["performed"])
        self.assertIsNone(self.remote_head(), "dry run must not publish")

    def test_dry_run_and_apply_agree_on_a_refusal(self):
        """AC11 — the reported decision does not depend on which mode ran."""
        self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})

        dry = self.run_publish(dry_run=True)
        applied = self.run_publish()

        self.assertFalse(dry["success"])
        self.assertFalse(applied["success"])
        self.assertEqual(dry["reasons"], applied["reasons"])


class TestRenewalUnblocked(_PublishBase):
    """AC20/AC21 — renewal is permitted only after verified publication."""

    def _renewal(self, remote_head):
        return issue_lock_renewal.assess_exact_owner_lease_renewal(
            issue_lock_store.read_lock_file(self.lock_path()),
            issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
            remote="prgs", org=ORG, repo=REPO,
            identity=IDENTITY, profile=PROFILE,
            current_branch=BRANCH, porcelain_status="", worktree_exists=True,
            head_sha=self.head_sha, remote_head_sha=remote_head,
        )

    def test_renewal_refuses_before_publication(self):
        self.write_lock()
        decision = self._renewal(None)
        self.assertFalse(decision["renewal_sanctioned"])
        self.assertTrue(
            any("unpublished branch" in r for r in decision["reasons"]),
            decision["reasons"],
        )

    def test_renewal_is_sanctioned_after_publication(self):
        self.write_lock()
        result = self.run_publish()
        self.assertTrue(result["verified"], result.get("reasons"))

        decision = self._renewal(self.remote_head())

        self.assertTrue(decision["renewal_sanctioned"], decision["reasons"])


class TestProtectedAssetUntouched(unittest.TestCase):
    """AC17 — no test or fixture may reference the protected worktree."""

    def test_no_reference_to_the_protected_worktree(self):
        here = os.path.dirname(os.path.abspath(__file__))
        root = os.path.dirname(here)
        needle = "issue-635-project-registry" + "-api"
        for path in (
            os.path.join(here, "test_issue_812_publish_unpublished_commit.py"),
            os.path.join(root, "branch_publish.py"),
        ):
            with open(path, "r", encoding="utf-8") as fh:
                body = fh.read()
            self.assertNotIn(needle, body)


if __name__ == "__main__":  # pragma: no cover
    unittest.main()