131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
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_iterate_message_is_preserved_on_queued_job() -> None:
|
|
storage = FakeStorage()
|
|
container = SimpleNamespace(storage=storage, gitea=FakeGitea())
|
|
event = _event_from_payload(
|
|
"delivery",
|
|
payload(
|
|
"/agent iterate\n\nkeep the API stable\nlimit changes to the parser",
|
|
is_pull=True,
|
|
),
|
|
)
|
|
assert event is not None
|
|
|
|
response = await _handle_command(container, event)
|
|
|
|
assert response.status_code == 202
|
|
assert len(storage.jobs) == 1
|
|
assert storage.jobs[0].message == (
|
|
"keep the API stable\nlimit changes to the parser"
|
|
)
|
|
|
|
|
|
async def test_non_command_is_ignored() -> None:
|
|
container = SimpleNamespace(storage=FakeStorage(), gitea=FakeGitea())
|
|
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)
|