pragent/tests/pilot/test_webhook_server.py

"""Unit tests for the webhook receiver's gating, dedupe and limits. No network."""
import base64
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


# ---------------------------------------------------------------------------
# is_repo_enabled — reads `.pr-review.json` from the PR base ref and parses
# its `enabled` flag. False on any failure (404, parse error, missing field).
# ---------------------------------------------------------------------------


def test_is_repo_enabled_returns_false_when_404(monkeypatch):
    monkeypatch.setattr(
        "webhook_server.gitea_get",
        lambda *a, **kw: (404, b'{"message":"not found"}'),
    )
    assert ws.is_repo_enabled("api", "owner/repo", "main", "tok") is False


def test_is_repo_enabled_returns_true_when_enabled(monkeypatch):
    body = b'{"content":"' + base64.b64encode(b'{"enabled": true}').decode().encode() + b'"}'
    monkeypatch.setattr("webhook_server.gitea_get", lambda *a, **kw: (200, body))
    assert ws.is_repo_enabled("api", "owner/repo", "main", "tok") is True


def test_is_repo_enabled_returns_false_when_disabled(monkeypatch):
    body = b'{"content":"' + base64.b64encode(b'{"enabled": false}').decode().encode() + b'"}'
    monkeypatch.setattr("webhook_server.gitea_get", lambda *a, **kw: (200, body))
    assert ws.is_repo_enabled("api", "owner/repo", "main", "tok") is False


def test_is_repo_enabled_returns_false_when_field_missing(monkeypatch):
    body = b'{"content":"' + base64.b64encode(b'{}').decode().encode() + b'"}'
    monkeypatch.setattr("webhook_server.gitea_get", lambda *a, **kw: (200, body))
    assert ws.is_repo_enabled("api", "owner/repo", "main", "tok") is False