cd894310dc
- Router layer: HTTP validation + HMAC-SHA256 signature verification - Service layer: Redis stream push (aider_event_service.push_aider_batch_to_stream) - leWOOOgo積木化遵循: Router → Service → Redis - All 6 tests passing (signature validation, batch limits, edge cases)
126 lines
3.3 KiB
Python
126 lines
3.3 KiB
Python
# apps/api/tests/test_aider_events_api.py | 2026-04-20 @ Asia/Taipei
|
|
import hmac
|
|
import hashlib
|
|
import json
|
|
import pytest
|
|
from datetime import datetime, timezone, timedelta
|
|
from unittest.mock import AsyncMock, patch
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from src.api.v1.aider_events import router
|
|
|
|
SECRET = "testsecret_for_unittest_do_not_use_in_prod_" + "x" * 20
|
|
TAIPEI = timezone(timedelta(hours=8))
|
|
|
|
|
|
def _sign(body: bytes, secret: str = SECRET) -> str:
|
|
return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def _ev():
|
|
return {
|
|
"ts": datetime(2026, 4, 20, 10, 0, tzinfo=TAIPEI).isoformat(),
|
|
"session_id": "s1",
|
|
"host": "ogt-mac",
|
|
"type": "session_start",
|
|
"payload": {
|
|
"cwd": "/t/x",
|
|
"model": "elephant",
|
|
"aider_args": [],
|
|
"aider_pid": 1,
|
|
"cli_version": "0.86",
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
monkeypatch.setenv("AIDER_WEBHOOK_SECRET", SECRET)
|
|
|
|
# Patch the service function
|
|
async def mock_push(batch):
|
|
return ["1234-0"] * len(batch.events)
|
|
|
|
with patch("src.api.v1.aider_events.push_aider_batch_to_stream", new_callable=lambda: mock_push):
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/v1")
|
|
yield TestClient(app)
|
|
|
|
|
|
def test_accepts_signed_batch(client):
|
|
body = json.dumps({"events": [_ev()]}).encode()
|
|
r = client.post(
|
|
"/api/v1/aider/events",
|
|
content=body,
|
|
headers={
|
|
"X-Aider-Signature": _sign(body),
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
assert r.status_code == 202, r.text
|
|
data = r.json()
|
|
assert data["accepted"] == 1
|
|
|
|
|
|
def test_rejects_invalid_signature(client):
|
|
body = json.dumps({"events": [_ev()]}).encode()
|
|
r = client.post(
|
|
"/api/v1/aider/events",
|
|
content=body,
|
|
headers={
|
|
"X-Aider-Signature": "sha256=" + "0" * 64,
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_rejects_missing_signature(client):
|
|
body = json.dumps({"events": [_ev()]}).encode()
|
|
r = client.post(
|
|
"/api/v1/aider/events",
|
|
content=body,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_rejects_malformed_event(client):
|
|
body = json.dumps({"events": [{"bad": "payload"}]}).encode()
|
|
r = client.post(
|
|
"/api/v1/aider/events",
|
|
content=body,
|
|
headers={
|
|
"X-Aider-Signature": _sign(body),
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_rejects_oversize_batch(client):
|
|
body = json.dumps({"events": [_ev()] * 51}).encode()
|
|
r = client.post(
|
|
"/api/v1/aider/events",
|
|
content=body,
|
|
headers={
|
|
"X-Aider-Signature": _sign(body),
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_accepts_batch_of_50(client):
|
|
body = json.dumps({"events": [_ev()] * 50}).encode()
|
|
r = client.post(
|
|
"/api/v1/aider/events",
|
|
content=body,
|
|
headers={
|
|
"X-Aider-Signature": _sign(body),
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
assert r.status_code == 202
|
|
assert r.json()["accepted"] == 50
|