72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
import hashlib
|
|
import hmac
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from agentci.api.webhook import _event_from_payload, _handle_command, valid_signature
|
|
|
|
|
|
class FakeHost:
|
|
def __init__(self, duplicate: bool = False) -> None:
|
|
self.events = []
|
|
self.duplicate = duplicate
|
|
|
|
async def receive(self, event):
|
|
self.events.append(event)
|
|
state = SimpleNamespace(id="job", receive_sequence=1)
|
|
return SimpleNamespace(state=state, duplicate=self.duplicate)
|
|
|
|
|
|
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
|
|
|
|
|
|
async def test_command_is_forwarded_without_parsing() -> None:
|
|
host = FakeHost()
|
|
event = _event_from_payload(
|
|
"delivery", payload("/agent iterate\n\nkeep raw body", is_pull=True)
|
|
)
|
|
assert event is not None
|
|
response = await _handle_command(SimpleNamespace(state_machine=host), event)
|
|
assert response.status_code == 202
|
|
assert host.events[0].body == "/agent iterate\n\nkeep raw body"
|
|
|
|
|
|
async def test_duplicate_returns_200() -> None:
|
|
event = _event_from_payload("delivery", payload("/agent plan"))
|
|
assert event is not None
|
|
response = await _handle_command(SimpleNamespace(state_machine=FakeHost(True)), event)
|
|
assert response.status_code == 200
|
|
|
|
|
|
async def test_missing_delivery_is_rejected() -> None:
|
|
event = _event_from_payload("", payload("/agent plan"))
|
|
assert event is not None
|
|
with pytest.raises(HTTPException) as raised:
|
|
await _handle_command(SimpleNamespace(state_machine=FakeHost()), event)
|
|
assert raised.value.status_code == 400
|
|
|
|
|
|
async def test_non_command_is_ignored() -> None:
|
|
event = _event_from_payload("delivery", payload("ordinary discussion"))
|
|
assert event is not None
|
|
response = await _handle_command(SimpleNamespace(state_machine=FakeHost()), event)
|
|
assert response.status_code == 204
|
|
|
|
|
|
def test_signature_validation() -> None:
|
|
signature = hmac.new(b"secret", b"{}", hashlib.sha256).hexdigest()
|
|
assert valid_signature(b"secret", b"{}", signature)
|
|
assert not valid_signature(b"secret", b"{}", "bad")
|