initial impl

tests/test_webhook.py

@@ -0,0 +1,111 @@
+import hashlib
+import hmac
+import json
+from types import SimpleNamespace
+
+from agentci.api.webhook import _event_from_payload, _handle_command, valid_signature
+
+
+class FakeStorage:
+    def __init__(self) -> None:
+        self.jobs = []
+        self.deliveries: set[str] = set()
+
+    async def enqueue(self, delivery_id, job):
+        if delivery_id in self.deliveries:
+            return False
+        self.deliveries.add(delivery_id)
+        self.jobs.append(job)
+        return True
+
+    async def record_delivery(self, delivery_id, _comment_id):
+        if delivery_id in self.deliveries:
+            return False
+        self.deliveries.add(delivery_id)
+        return True
+
+    async def set_job_comment(self, *_args):
+        return None
+
+
+class FakeGitea:
+    def __init__(self, permitted: bool = True) -> None:
+        self.permitted = permitted
+        self.comments: list[str] = []
+
+    async def has_write_permission(self, *_args):
+        return self.permitted
+
+    async def create_comment(self, _owner, _repo, _number, body):
+        self.comments.append(body)
+        return len(self.comments)
+
+
+def payload(body: str, *, is_pull: bool = False) -> dict:
+    value = {
+        "action": "created",
+        "comment": {"id": 8, "body": body, "user": {"login": "alice"}},
+        "repository": {
+            "name": "repo",
+            "owner": {"login": "org"},
+        },
+        "issue": {"number": 4},
+        "is_pull": is_pull,
+    }
+    if is_pull:
+        value["pull_request"] = {"number": 4}
+    return value
+
+
+def test_extracts_pull_request_event() -> None:
+    event = _event_from_payload("delivery", payload("/agent fix now", is_pull=True))
+    assert event is not None
+    assert event.pr_number == 4
+    assert event.target_key == "org/repo:pr:4"
+
+
+async def test_authorized_command_is_queued() -> None:
+    storage = FakeStorage()
+    gitea = FakeGitea()
+    container = SimpleNamespace(storage=storage, gitea=gitea)
+    event = _event_from_payload("delivery", payload("/agent plan consider migrations"))
+    assert event is not None
+    response = await _handle_command(container, event)
+    assert response.status_code == 202
+    assert len(storage.jobs) == 1
+    assert "queued" in gitea.comments[0]
+
+
+async def test_unauthorized_command_is_rejected_and_deduplicated() -> None:
+    storage = FakeStorage()
+    gitea = FakeGitea(permitted=False)
+    container = SimpleNamespace(storage=storage, gitea=gitea)
+    event = _event_from_payload("delivery", payload("/agent implement"))
+    assert event is not None
+    await _handle_command(container, event)
+    await _handle_command(container, event)
+    assert storage.jobs == []
+    assert len(gitea.comments) == 1
+    assert "write permission" in gitea.comments[0]
+
+
+async def test_non_command_does_not_query_permission() -> None:
+    class ExplodingGitea(FakeGitea):
+        async def has_write_permission(self, *_args):
+            raise AssertionError("permission lookup should not run")
+
+    container = SimpleNamespace(storage=FakeStorage(), gitea=ExplodingGitea())
+    event = _event_from_payload("delivery", payload("ordinary discussion"))
+    assert event is not None
+    response = await _handle_command(container, event)
+    assert response.status_code == 204
+
+
+def test_rejects_bad_signature() -> None:
+    assert not valid_signature(b"secret", b"{}", "bad")
+
+
+def test_accepts_valid_signature() -> None:
+    body = json.dumps(payload("ordinary comment")).encode()
+    signature = hmac.new(b"secret", body, hashlib.sha256).hexdigest()
+    assert valid_signature(b"secret", body, signature)