plane-agentd/tests/test_webhook.py

import json
import threading
import time
import unittest
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer

from agentd.webhook import BUSY, Trigger, make_handler, route, sign, verify

SECRET = "plane_wh_test"
PROJECTS = {"GWENT": {"id": "proj-gw"}, "PAD": {"id": "proj-pad"}}


def payload(project="proj-gw", event="issue", action="updated"):
    return {"event": event, "action": action, "webhook_id": "w1", "workspace_id": "ws", "workspace_slug": "kappiee",
            "data": {"id": "i1", "project": project}, "activity": {"field": "state"}}


def plane_body(p):
    return json.dumps(p).encode()   # what Plane signs (json.dumps(payload)) and sends


class SignatureTest(unittest.TestCase):
    def test_ok_bad_missing(self):
        body = plane_body(payload())
        self.assertTrue(verify(SECRET, body, sign(SECRET, body)))
        self.assertTrue(verify(SECRET, body, sign(SECRET, body).upper()))
        self.assertFalse(verify(SECRET, body, sign("other", body)))
        self.assertFalse(verify(SECRET, body + b" ", sign(SECRET, body + b"x")))
        self.assertFalse(verify(SECRET, body, None))
        self.assertFalse(verify(SECRET, body, ""))
        self.assertFalse(verify("", body, sign("", body)))       # no secret configured: reject everything
        self.assertFalse(verify(SECRET, body, "zzé"))         # non-hex garbage

    def test_reencoded_body_matches_plane_serialisation(self):
        p = payload()
        compact = json.dumps(p, separators=(",", ":")).encode()
        self.assertTrue(verify(SECRET, compact, sign(SECRET, plane_body(p))))
        self.assertFalse(verify(SECRET, b"not json", sign(SECRET, plane_body(p))))


class RouteTest(unittest.TestCase):
    def test_route(self):
        self.assertEqual(route(payload(), PROJECTS), "GWENT")
        self.assertEqual(route(payload("proj-pad", "issue_comment", "created"), PROJECTS), "PAD")
        self.assertIsNone(route(payload("proj-other"), PROJECTS))
        self.assertIsNone(route({"event": "issue", "action": "deleted", "data": {"id": "i1"}}, PROJECTS))
        self.assertIsNone(route({"event": "issue"}, PROJECTS))
        self.assertEqual(route({"data": {"project": {"id": "proj-gw"}}}, PROJECTS), "GWENT")


class FakeTrigger:
    def __init__(self):
        self.kicks = []

    def kick(self, reason):
        self.kicks.append(reason)

    def status(self):
        return {"pending": bool(self.kicks)}


class HandlerTest(unittest.TestCase):
    def setUp(self):
        self.trigger, self.lines = FakeTrigger(), []
        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(SECRET, PROJECTS, self.trigger, self.lines.append))
        threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
        self.url = "http://127.0.0.1:%d" % self.httpd.server_address[1]

    def tearDown(self):
        self.httpd.shutdown()
        self.httpd.server_close()

    def post(self, body, sig, path="/plane"):
        headers = {"Content-Type": "application/json", "X-Plane-Delivery": "d-1", "X-Plane-Event": "issue"}
        if sig is not None:
            headers["X-Plane-Signature"] = sig
        req = urllib.request.Request(self.url + path, data=body, headers=headers, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=5) as r:
                return r.status
        except urllib.error.HTTPError as e:
            return e.code

    def test_signed_tenant_delivery_triggers(self):
        body = plane_body(payload())
        self.assertEqual(self.post(body, sign(SECRET, body)), 200)
        self.assertEqual(self.trigger.kicks, ["d-1"])
        self.assertIn("project=GWENT", self.lines[-1])
        self.assertIn("decision=tick", self.lines[-1])

    def test_bad_or_missing_signature_is_401_and_triggers_nothing(self):
        body = plane_body(payload())
        self.assertEqual(self.post(body, sign("wrong", body)), 401)
        self.assertEqual(self.post(body, None), 401)
        self.assertEqual(self.trigger.kicks, [])
        self.assertIn("401 missing", self.lines[-1])

    def test_other_project_is_202(self):
        body = plane_body(payload("proj-other"))
        self.assertEqual(self.post(body, sign(SECRET, body)), 202)
        self.assertEqual(self.trigger.kicks, [])

    def test_wrong_path_and_healthz(self):
        body = plane_body(payload())
        self.assertEqual(self.post(body, sign(SECRET, body), "/other"), 404)
        with urllib.request.urlopen(self.url + "/healthz", timeout=5) as r:
            self.assertEqual((r.status, json.loads(r.read())["ok"]), (200, True))


class FakeRunner:
    def __init__(self, codes=(), hold=0.0):
        self.codes, self.hold, self.calls = list(codes), hold, 0
        self.started = threading.Event()

    def __call__(self):
        self.calls += 1
        self.started.set()
        time.sleep(self.hold)
        return self.codes.pop(0) if self.codes else 0


class TriggerTest(unittest.TestCase):
    def make(self, runner, **kw):
        kw = dict(dict(debounce=0.2, retry=0.1, retry_window=1.0), **kw)
        self.lines = []
        return Trigger(runner, log=self.lines.append, **kw)

    def test_burst_coalesces_into_one_tick(self):
        run = FakeRunner()
        t = self.make(run)
        for i in range(5):
            t.kick("d%d" % i)
        self.assertTrue(t.step(wait=2))
        self.assertEqual(run.calls, 1)
        self.assertIn("5 deliveries", self.lines[0])
        self.assertFalse(t.step(wait=0.4))   # nothing left
        self.assertEqual(run.calls, 1)

    def test_debounce_waits_before_ticking(self):
        run = FakeRunner()
        t = self.make(run, debounce=0.3)
        t0 = time.monotonic()
        t.kick("d1")
        t.step(wait=2)
        self.assertGreaterEqual(time.monotonic() - t0, 0.29)

    def test_delivery_during_a_tick_causes_one_follow_up(self):
        run = FakeRunner(hold=0.3)
        t = self.make(run, debounce=0.05).start()
        t.kick("d1")
        self.assertTrue(run.started.wait(2))
        t.kick("d2")      # arrives while the first tick runs
        t.kick("d3")
        time.sleep(1.0)
        self.assertEqual(run.calls, 2)
        self.assertEqual(t.status()["pending"], False)

    def test_busy_lock_is_retried_until_the_tick_runs(self):
        run = FakeRunner(codes=[BUSY, BUSY, 0])
        t = self.make(run, debounce=0.0)
        t.kick("d1")
        for _ in range(3):
            self.assertTrue(t.step(wait=2))
        self.assertEqual(run.calls, 3)
        self.assertEqual(sum("retry in" in l for l in self.lines), 2)
        self.assertIn("1 deliveries: d1", self.lines[-2])   # the delivery is carried into the retry
        self.assertIsNone(t.status()["retrying_since"])
        self.assertFalse(t.step(wait=0.3))

    def test_busy_gives_up_after_the_window(self):
        run = FakeRunner(codes=[BUSY] * 10)
        t = self.make(run, debounce=0.0, retry=0.1, retry_window=0.35)
        t.kick("d1")
        while t.step(wait=1):
            pass
        self.assertTrue(any("giving up" in l for l in self.lines))
        self.assertLessEqual(run.calls, 5)
        self.assertFalse(t.status()["pending"])


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