"""Tests for the Gitea issue↔PR linkage console (#645, Phase 3).

Covers the acceptance criteria of the issue:

* AC1 — issue↔PR linkage is visible for the selected project/repo, in both
  directions, with the evidence that produced each edge.
* AC2 — the latest canonical handoff (CTH) is summarized for a focused thread.
* AC3 — an external Gitea link appears only under the admin reveal opt-in.
* AC4 — every case is driven by mocked Gitea payloads; no network.

Plus the invariants this console must not violate: a partial or failed read is
never rendered as "no link exists", an unfetched thread is never rendered as
"no handoff", redaction happens before display, and the surface stays read-only.
"""

from __future__ import annotations

import json
import os
import sys
import unittest
from pathlib import Path
from unittest import mock

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

from tests.webui_testclient import TestClient

from canonical_thread_handoff import format_cth_body
from webui.app import create_app
from webui.linkage_loader import (
    EVIDENCE_BRANCH,
    EVIDENCE_CLOSES,
    EVIDENCE_REFERENCE,
    HANDOFF_LOADED,
    HANDOFF_NOT_LOADED,
    HANDOFF_UNAVAILABLE,
    LinkageSnapshot,
    load_linkage_snapshot,
    resolve_linkage,
    resolve_pr_links,
    snapshot_to_dict,
    summarize_handoff,
)
from webui.linkage_views import render_linkage_page
from webui.nav import nav_hrefs
from webui.queue_loader import PaginationMeta


def _pagination(*, complete: bool = True, count: int = 0) -> PaginationMeta:
    return PaginationMeta(
        page=1,
        per_page=50,
        returned_count=count,
        has_more=not complete,
        is_final_page=complete,
        inventory_complete=complete,
        pages_fetched=1,
    )


def _pr(
    number: int,
    *,
    title: str = "",
    body: str = "",
    head: str = "",
    state: str = "open",
    labels: tuple[str, ...] = (),
) -> dict:
    return {
        "number": number,
        "title": title or f"pr {number}",
        "body": body,
        "state": state,
        "head": {"ref": head},
        "labels": [{"name": name} for name in labels],
    }


def _issue(
    number: int,
    *,
    title: str = "",
    state: str = "open",
    labels: tuple[str, ...] = (),
) -> dict:
    return {
        "number": number,
        "title": title or f"issue {number}",
        "state": state,
        "labels": [{"name": name} for name in labels],
    }


def _fetcher(items: list[dict], *, complete: bool = True):
    def _fetch(*_args, **_kwargs):
        return items, _pagination(complete=complete, count=len(items))

    return _fetch


def _load(
    issues: list[dict],
    prs: list[dict],
    *,
    complete: bool = True,
    **kwargs,
) -> LinkageSnapshot:
    return load_linkage_snapshot(
        fetch_prs=_fetcher(prs, complete=complete),
        fetch_issues=_fetcher(issues, complete=complete),
        **kwargs,
    )


def _cth(comment_id: int, *, created_at: str, status: str, next_owner: str) -> dict:
    return {
        "id": comment_id,
        "created_at": created_at,
        "user": {"login": "jcwalker3"},
        "body": format_cth_body(
            cth_type="Author Handoff",
            status=status,
            next_owner=next_owner,
            current_blocker="none",
            decision="implemented",
            proof="full suite green",
            next_action="review PR",
            ready_to_paste_prompt="Review PR #902 now.",
        ),
    }


class TestLinkageEvidence(unittest.TestCase):
    """AC1 — every edge records how it was found, and keeps all candidates."""

    def test_closes_keyword_in_body_is_strongest_evidence(self):
        links = resolve_pr_links(_pr(902, body="Closes #643"))
        self.assertEqual([link.issue_number for link in links], [643])
        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))
        self.assertTrue(links[0].closes)

    def test_closes_keyword_in_title_counts(self):
        links = resolve_pr_links(_pr(902, title="feat(webui): preview (Closes #643)"))
        self.assertEqual(links[0].evidence, (EVIDENCE_CLOSES,))

    def test_canonical_branch_marker_links_without_a_keyword(self):
        links = resolve_pr_links(_pr(902, head="feat/issue-643-request-preview"))
        self.assertEqual([link.issue_number for link in links], [643])
        self.assertEqual(links[0].evidence, (EVIDENCE_BRANCH,))
        self.assertFalse(links[0].closes)

    def test_non_canonical_branch_is_not_treated_as_a_marker(self):
        self.assertEqual(resolve_pr_links(_pr(902, head="issue-643-preview")), ())

    def test_bare_mention_is_recorded_as_the_weakest_evidence(self):
        links = resolve_pr_links(_pr(902, body="context in #643"))
        self.assertEqual(links[0].evidence, (EVIDENCE_REFERENCE,))
        self.assertFalse(links[0].closes)

    def test_several_evidence_kinds_merge_onto_one_edge(self):
        links = resolve_pr_links(
            _pr(902, body="Closes #643 — see #643", head="feat/issue-643-preview")
        )
        self.assertEqual(len(links), 1)
        self.assertEqual(
            links[0].evidence,
            (EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE),
        )

    def test_stronger_evidence_sorts_first(self):
        links = resolve_pr_links(_pr(902, body="Closes #643, related #700"))
        self.assertEqual([link.issue_number for link in links], [643, 700])

    def test_self_reference_is_not_linkage(self):
        links = resolve_pr_links(_pr(902, body="supersedes #902"))
        self.assertEqual(links, ())

    def test_every_candidate_is_kept_never_collapsed_to_a_guess(self):
        links = resolve_pr_links(_pr(902, body="Closes #643\nCloses #644"))
        self.assertEqual([link.issue_number for link in links], [643, 644])


class TestLinkageIndex(unittest.TestCase):
    def test_issue_direction_ignores_mention_only_edges(self):
        index = resolve_linkage([_pr(902, body="context in #643")])
        self.assertIsNone(index.issue_prs.get(643))
        self.assertEqual(index.pr_links[902][0].evidence, (EVIDENCE_REFERENCE,))

    def test_contested_issue_is_reported_when_two_prs_claim_it(self):
        index = resolve_linkage(
            [_pr(902, body="Closes #643"), _pr(903, head="feat/issue-643-again")]
        )
        self.assertEqual(index.contested_issues(), (643,))
        self.assertEqual(index.issue_prs[643], (902, 903))

    def test_single_claim_is_not_contested(self):
        index = resolve_linkage([_pr(902, body="Closes #643")])
        self.assertEqual(index.contested_issues(), ())

    def test_ambiguous_when_two_issues_tie_at_the_strongest_evidence(self):
        index = resolve_linkage([_pr(902, body="Closes #643\nCloses #644")])
        self.assertTrue(index.ambiguous(902))

    def test_weaker_candidate_alongside_a_stronger_one_is_not_ambiguous(self):
        index = resolve_linkage([_pr(902, body="Closes #643, see #700")])
        self.assertFalse(index.ambiguous(902))
        self.assertEqual(index.primary_issue(902).issue_number, 643)

    def test_malformed_pr_row_is_skipped_not_raised_on(self):
        index = resolve_linkage([{"title": "no number"}, _pr(902, body="Closes #643")])
        self.assertEqual(sorted(index.pr_links), [902])


class TestLinkageSnapshot(unittest.TestCase):
    """AC1 — linkage is visible per project/repo, in both directions."""

    def test_both_directions_are_populated(self):
        snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
        self.assertTrue(snapshot.ok)
        self.assertEqual([node.number for node in snapshot.issues], [643])
        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
        self.assertEqual(snapshot.prs[0].links[0].issue_number, 643)

    def test_repo_scope_comes_from_the_registry_project(self):
        snapshot = _load([], [])
        self.assertIn("/", snapshot.repo_label)
        self.assertTrue(snapshot.project_id)

    def test_unknown_project_fails_closed_with_a_reason(self):
        snapshot = _load([_issue(643)], [], project_id="no-such-project")
        self.assertFalse(snapshot.ok)
        self.assertIn("not found in registry", snapshot.fetch_error)
        self.assertEqual(snapshot.issues, ())

    def test_orphan_pr_is_identifiable(self):
        snapshot = _load([], [_pr(902), _pr(903, body="Closes #643")])
        self.assertEqual([node.number for node in snapshot.orphan_prs], [902])

    def test_state_scope_defaults_to_open_and_is_reported(self):
        self.assertEqual(_load([], []).state_scope, "open")
        self.assertEqual(_load([], [], state="all").state_scope, "all")

    def test_unsupported_state_falls_back_to_open(self):
        self.assertEqual(_load([], [], state="../etc").state_scope, "open")

    def test_state_is_passed_through_to_the_fetchers(self):
        seen: list[str] = []

        def _fetch(*_args, **kwargs):
            seen.append(kwargs.get("state", ""))
            return [], _pagination()

        load_linkage_snapshot(state="all", fetch_prs=_fetch, fetch_issues=_fetch)
        self.assertEqual(seen, ["all", "all"])


class TestPartialInventoryIsNotAnAbsenceClaim(unittest.TestCase):
    """An empty edge list from a partial read must never read as 'no link'."""

    def test_incomplete_pagination_marks_links_non_authoritative(self):
        snapshot = _load([_issue(643)], [], complete=False)
        self.assertFalse(snapshot.inventory_complete)
        self.assertFalse(snapshot.issues[0].links_authoritative)

    def test_complete_pagination_marks_links_authoritative(self):
        snapshot = _load([_issue(643)], [], complete=True)
        self.assertTrue(snapshot.inventory_complete)
        self.assertTrue(snapshot.issues[0].links_authoritative)

    def test_partial_window_renders_a_qualified_empty_cell(self):
        html = render_linkage_page(_load([_issue(643)], [], complete=False))
        self.assertIn("none found (partial inventory)", html)

    def test_complete_window_renders_a_plain_none(self):
        html = render_linkage_page(_load([_issue(643)], [], complete=True))
        self.assertNotIn("partial inventory", html)
        self.assertIn(">none<", html)

    def test_missing_credentials_fail_closed_without_a_table(self):
        with mock.patch(
            "webui.linkage_loader._offline_test_mode", return_value=False
        ), mock.patch("webui.linkage_loader.get_auth_header", return_value=""):
            snapshot = load_linkage_snapshot()
        self.assertFalse(snapshot.ok)
        self.assertIn("credentials unavailable", snapshot.fetch_error)
        html = render_linkage_page(snapshot)
        self.assertIn("Linkage unavailable", html)
        self.assertNotIn("Issues → pull requests", html)

    def test_fetch_failure_is_reported_not_raised(self):
        def _boom(*_args, **_kwargs):
            raise RuntimeError("gitea 502")

        snapshot = load_linkage_snapshot(fetch_prs=_boom, fetch_issues=_boom)
        self.assertFalse(snapshot.ok)
        self.assertIn("Gitea fetch failed", snapshot.fetch_error)


class TestHandoffSummary(unittest.TestCase):
    """AC2 — the latest canonical handoff is summarized for a focused thread."""

    def test_latest_cth_wins(self):
        summary = summarize_handoff([
            _cth(1, created_at="2026-07-24T10:00:00Z", status="in progress",
                 next_owner="author"),
            _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
                 next_owner="reviewer"),
        ])
        self.assertEqual(summary.comment_id, 2)
        self.assertEqual(summary.status, "PR-open")
        self.assertEqual(summary.next_owner, "reviewer")
        self.assertTrue(summary.cth_type_known)

    def test_thread_without_a_cth_summarizes_to_none(self):
        self.assertIsNone(summarize_handoff([{"id": 1, "body": "ordinary comment"}]))

    def test_unknown_heading_is_reported_not_republished(self):
        summary = summarize_handoff([
            {
                "id": 5,
                "created_at": "2026-07-25T10:00:00Z",
                "user": {"login": "someone"},
                "body": "<!-- cth:v1 -->\n## CTH: Totally Made Up\n\nStatus: odd\n",
            }
        ])
        self.assertFalse(summary.cth_type_known)
        self.assertEqual(summary.cth_type, "unrecognized")
        self.assertNotIn("Totally Made Up", json.dumps(summary.to_dict()))

    def test_focused_pr_loads_its_handoff(self):
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643")],
            pr=902,
            comment_source=lambda kind, number: [
                _cth(2, created_at="2026-07-25T10:00:00Z", status="PR-open",
                     next_owner="reviewer")
            ],
        )
        self.assertEqual(snapshot.handoff_status.state, HANDOFF_LOADED)
        self.assertEqual(snapshot.focus, ("pr", 902))
        self.assertEqual(snapshot.prs[0].handoff.status, "PR-open")

    def test_unfocused_rows_report_not_loaded_never_none(self):
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643"), _pr(903)],
            pr=902,
            comment_source=lambda kind, number: [],
        )
        other = next(node for node in snapshot.prs if node.number == 903)
        self.assertIsNone(other.handoff)
        self.assertEqual(other.handoff_status.state, HANDOFF_NOT_LOADED)
        self.assertIn("not loaded", render_linkage_page(snapshot))

    def test_no_focus_means_no_thread_is_claimed_handoff_free(self):
        snapshot = _load([_issue(643)], [])
        self.assertEqual(snapshot.handoff_status.state, HANDOFF_NOT_LOADED)
        self.assertIn("thread-scoped", snapshot.handoff_status.reason)

    def test_comment_source_failure_degrades_only_the_handoff(self):
        def _boom(_kind, _number):
            raise RuntimeError("comments 500")

        snapshot = _load(
            [_issue(643)], [_pr(902, body="Closes #643")], pr=902, comment_source=_boom
        )
        self.assertTrue(snapshot.ok)
        self.assertEqual(snapshot.handoff_status.state, HANDOFF_UNAVAILABLE)
        self.assertEqual(snapshot.issues[0].linked_prs, (902,))
        self.assertIn("unavailable", render_linkage_page(snapshot))

    def test_loaded_thread_with_no_cth_says_so_explicitly(self):
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643")],
            pr=902,
            comment_source=lambda kind, number: [{"id": 1, "body": "hi"}],
        )
        self.assertIn(
            "no Canonical Thread Handoff comment found", render_linkage_page(snapshot)
        )


class TestDeepLinks(unittest.TestCase):
    """AC3 — an external Gitea link is emitted only when permitted."""

    def test_deep_links_are_withheld_by_default(self):
        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}):
            snapshot = _load([_issue(643)], [])
            html = render_linkage_page(snapshot)
        self.assertFalse(snapshot.deep_links_enabled)
        self.assertIsNone(snapshot.issues[0].deep_link)
        self.assertIn("Gitea deep links are withheld", html)

    def test_reveal_opt_in_emits_the_link(self):
        with mock.patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}):
            snapshot = _load([_issue(643)], [_pr(902, body="Closes #643")])
            html = render_linkage_page(snapshot)
        self.assertTrue(snapshot.deep_links_enabled)
        self.assertIn("/issues/643", snapshot.issues[0].deep_link)
        self.assertIn("/pulls/902", snapshot.prs[0].deep_link)
        self.assertIn(f'href="{snapshot.issues[0].deep_link}"', html)


class TestRedactionBoundary(unittest.TestCase):
    def test_secret_shaped_title_is_redacted_before_display(self):
        snapshot = _load(
            [_issue(643, title="token=ghp_thisisnotarealsecretvalue0001")], []
        )
        payload = json.dumps(snapshot_to_dict(snapshot))
        self.assertNotIn("ghp_thisisnotarealsecretvalue0001", payload)
        self.assertNotIn(
            "ghp_thisisnotarealsecretvalue0001", render_linkage_page(snapshot)
        )

    def test_handoff_fields_are_redacted(self):
        comment = _cth(
            2, created_at="2026-07-25T10:00:00Z", status="ok", next_owner="reviewer"
        )
        comment["body"] += "\nDecision: password=hunter2hunter2\n"
        snapshot = _load(
            [_issue(643)],
            [_pr(902, body="Closes #643")],
            pr=902,
            comment_source=lambda kind, number: [comment],
        )
        self.assertNotIn("hunter2hunter2", json.dumps(snapshot_to_dict(snapshot)))
        self.assertNotIn("hunter2hunter2", render_linkage_page(snapshot))

    def test_html_escapes_markup_in_a_title(self):
        snapshot = _load([_issue(643, title="<script>alert(1)</script>")], [])
        html = render_linkage_page(snapshot)
        self.assertNotIn("<script>alert(1)</script>", html)
        self.assertIn("&lt;script&gt;", html)


class TestLinkageRoutes(unittest.TestCase):
    def setUp(self):
        self.snapshot = _load(
            [_issue(643, labels=("status:ready",))],
            [_pr(902, body="Closes #643", labels=("status:pr-open",))],
        )
        self.client = TestClient(create_app())

    def test_page_renders_both_tables(self):
        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
            response = self.client.get("/gitea")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Issues → pull requests", response.text)
        self.assertIn("Pull requests → issues", response.text)
        self.assertIn("#643", response.text)

    def test_api_exports_the_same_model(self):
        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
            response = self.client.get("/api/v1/gitea/linkage")
        self.assertEqual(response.status_code, 200)
        payload = response.json()
        self.assertTrue(payload["ok"])
        self.assertEqual(payload["issues"][0]["linked_prs"], [902])
        self.assertEqual(payload["prs"][0]["links"][0]["issue_number"], 643)
        self.assertEqual(payload["schema_version"], 1)

    def test_api_declares_the_evidence_vocabulary(self):
        with mock.patch("webui.app.load_linkage_snapshot", return_value=self.snapshot):
            payload = self.client.get("/api/v1/gitea/linkage").json()
        names = {entry["name"] for entry in payload["evidence_kinds"]}
        self.assertEqual(names, {EVIDENCE_CLOSES, EVIDENCE_BRANCH, EVIDENCE_REFERENCE})

    def test_api_fails_closed_with_a_non_200_when_the_read_failed(self):
        failed = _load([], [], project_id="no-such-project")
        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
            response = self.client.get("/api/v1/gitea/linkage")
        self.assertEqual(response.status_code, 502)
        self.assertFalse(response.json()["ok"])

    def test_page_still_renders_when_the_read_failed(self):
        failed = _load([], [], project_id="no-such-project")
        with mock.patch("webui.app.load_linkage_snapshot", return_value=failed):
            response = self.client.get("/gitea")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Linkage unavailable", response.text)

    def test_query_parameters_reach_the_loader(self):
        with mock.patch(
            "webui.app.load_linkage_snapshot", return_value=self.snapshot
        ) as loader:
            self.client.get("/gitea?project=gitea-tools&state=all&pr=902")
        loader.assert_called_once()
        args, kwargs = loader.call_args
        self.assertEqual(args[0], "gitea-tools")
        self.assertEqual(kwargs["state"], "all")
        self.assertEqual(kwargs["pr"], 902)
        self.assertIsNone(kwargs["issue"])

    def test_surface_stays_read_only(self):
        for path in ("/gitea", "/api/v1/gitea/linkage"):
            with self.subTest(path=path):
                self.assertEqual(self.client.post(path).status_code, 405)

    def test_nav_exposes_the_linkage_page_as_live(self):
        self.assertIn("/gitea", nav_hrefs())
        home = self.client.get("/").text
        self.assertIn('href="/gitea"', home)
        self.assertIn(">Gitea<", home)


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