"""Unit tests for the webhook receiver's gating, dedupe and limits. No network."""
import os
import sys
import threading

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
sys.path.insert(0, os.path.join(ROOT, "pilot"))

import webhook_server as ws  # noqa: E402


def _payload(**over):
    pr = {
        "number": 7,
        "title": "t",
        "body": "b",
        "labels": [{"name": "AI-REVIEW"}],
        "head": {"sha": "a" * 40},
        "base": {"ref": "main"},
    }
    pr.update(over.pop("pr", {}))
    p = {"action": "opened", "pull_request": pr, "repository": {"full_name": "o/r"}}
    p.update(over)
    return p


# ---------------------------------------------------------------------------
# label gating
# ---------------------------------------------------------------------------


def test_labels_have_matches_dicts_and_strings():
    assert ws._labels_have([{"name": "AI-REVIEW"}], "AI-REVIEW")
    assert ws._labels_have(["AI-REVIEW"], "AI-REVIEW")
    assert not ws._labels_have([{"name": "other"}], "AI-REVIEW")
    assert not ws._labels_have(None, "AI-REVIEW")


# ---------------------------------------------------------------------------
# in-flight claim — the check-then-act race around the sha-marker dedupe
# ---------------------------------------------------------------------------


def test_claim_is_exclusive_then_released():
    key = ("o/r", "7", "abc")
    ws._release(key)
    assert ws._claim(key) is True
    assert ws._claim(key) is False
    ws._release(key)
    assert ws._claim(key) is True
    ws._release(key)


def test_claim_is_thread_safe():
    key = ("o/r", "9", "def")
    ws._release(key)
    wins = []
    barrier = threading.Barrier(8)

    def go():
        barrier.wait()
        if ws._claim(key):
            wins.append(1)

    threads = [threading.Thread(target=go) for _ in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert len(wins) == 1
    ws._release(key)


def test_duplicate_delivery_for_same_sha_is_not_reviewed_twice(monkeypatch):
    started = []
    monkeypatch.setattr(ws, "BOT_TOKEN", "tok")

    class FakeThread:
        def __init__(self, target, args, daemon):
            self.args = args

        def start(self):
            started.append(self.args)

    monkeypatch.setattr(ws.threading, "Thread", FakeThread)
    ws._release(("o/r", "7", "a" * 40))

    s1, _ = ws._handle_pull_request(_payload())
    s2, m2 = ws._handle_pull_request(_payload(action="edited"))
    assert s1 == 202
    assert s2 == 200 and "in flight" in m2
    assert len(started) == 1
    ws._release(("o/r", "7", "a" * 40))


def test_base_ref_is_passed_to_the_review_thread(monkeypatch):
    started = []
    monkeypatch.setattr(ws, "BOT_TOKEN", "tok")

    class FakeThread:
        def __init__(self, target, args, daemon):
            self.args = args

        def start(self):
            started.append(self.args)

    monkeypatch.setattr(ws.threading, "Thread", FakeThread)
    ws._release(("o/r", "7", "a" * 40))
    ws._handle_pull_request(_payload(pr={"base": {"ref": "release/v2"}}))
    assert started[0][-1] == "release/v2"
    ws._release(("o/r", "7", "a" * 40))


def test_closed_action_is_ignored(monkeypatch):
    monkeypatch.setattr(ws, "BOT_TOKEN", "tok")
    status, msg = ws._handle_pull_request(_payload(action="closed"))
    assert status == 200
    assert "ignore" in msg


def test_missing_label_is_ignored(monkeypatch):
    monkeypatch.setattr(ws, "BOT_TOKEN", "tok")
    status, msg = ws._handle_pull_request(_payload(pr={"labels": [{"name": "wip"}]}))
    assert status == 200
    assert "AI-REVIEW" in msg


# ---------------------------------------------------------------------------
# concurrency bound
# ---------------------------------------------------------------------------


def test_review_slots_bound_matches_config():
    assert ws.MAX_CONCURRENT >= 1
    assert ws._review_slots._value <= ws.MAX_CONCURRENT


# ---------------------------------------------------------------------------
# AI-USAGE opt-in applied after the triggering label event
# ---------------------------------------------------------------------------


def test_run_review_rereads_labels_for_late_ai_usage(monkeypatch):
    # Labelling a PR AI-REVIEW then AI-USAGE fires two events; the review
    # claims on the first, so the second is deduped and the payload snapshot
    # never sees AI-USAGE. The re-read at review start must recover it.
    seen = {}
    monkeypatch.setattr(ws, "_fetch_current_labels",
                        lambda repo, index: [{"name": "AI-REVIEW"}, {"name": "AI-USAGE"}])
    monkeypatch.setattr(ws, "review_pr", lambda **kw: seen.update(kw) or True)
    ws._run_review(("o/r", "9", "abc1234"), "t", "b", False, "main")
    assert seen["report_usage"] is True


def test_run_review_keeps_usage_off_when_label_absent(monkeypatch):
    seen = {}
    monkeypatch.setattr(ws, "_fetch_current_labels",
                        lambda repo, index: [{"name": "AI-REVIEW"}])
    monkeypatch.setattr(ws, "review_pr", lambda **kw: seen.update(kw) or True)
    ws._run_review(("o/r", "9", "abc1234"), "t", "b", False, "main")
    assert seen["report_usage"] is False


def test_fetch_current_labels_hits_the_versioned_api_path(monkeypatch):
    # GITEA_API is the bare host; the /api/v1 prefix is the caller's job.
    # Getting this wrong 404s silently and the opt-in is lost — which is
    # exactly what shipped the first time.
    seen = {}

    class _Resp:
        def read(self): return b'[{"name": "AI-USAGE"}]'
        def __enter__(self): return self
        def __exit__(self, *a): return False

    def _urlopen(req, timeout=None):
        seen["url"] = req.full_url
        return _Resp()

    monkeypatch.setattr(ws.urllib.request, "urlopen", _urlopen)
    out = ws._fetch_current_labels("o/r", "9")
    assert out == [{"name": "AI-USAGE"}]
    assert seen["url"] == f"{ws.GITEA_API}/api/v1/repos/o/r/issues/9/labels"


def test_label_reread_failure_is_not_fatal(monkeypatch):
    # A dead API must not take the review down with it — the re-read is a
    # best-effort upgrade of an opt-in flag, nothing more.
    def _boom(*a, **k):
        raise OSError("api down")
    monkeypatch.setattr(ws.urllib.request, "urlopen", _boom)
    assert ws._fetch_current_labels("o/r", "9") == []

    seen = {}
    monkeypatch.setattr(ws, "review_pr", lambda **kw: seen.update(kw) or True)
    ws._run_review(("o/r", "9", "abc1234"), "t", "b", False, "main")
    assert seen["report_usage"] is False