36 lines
1.5 KiB
Python
36 lines
1.5 KiB
Python
import io
|
|
import json
|
|
|
|
from audit_core.ingestion import IngestionApplication, SQLiteEventStore
|
|
|
|
|
|
def invoke(app, payload, *, token="opaque", key="evt-1"):
|
|
raw = json.dumps(payload).encode()
|
|
result = {}
|
|
body = b"".join(app({"PATH_INFO":"/v1/events","REQUEST_METHOD":"POST",
|
|
"CONTENT_LENGTH":str(len(raw)),"wsgi.input":io.BytesIO(raw),
|
|
"HTTP_AUTHORIZATION":f"Bearer {token}","HTTP_IDEMPOTENCY_KEY":key},
|
|
lambda status, headers: result.update(status=status)))
|
|
return result["status"], json.loads(body)
|
|
|
|
|
|
def event():
|
|
return {"id":"evt-1","type":"membership.added","source":"user-engine",
|
|
"subject":"membership-1","tenant":"tenant:friendly:binky",
|
|
"correlation_id":"corr-1","occurred_at":"2026-08-09T00:00:00+00:00",
|
|
"data":{"membership_id":"membership-1"}}
|
|
|
|
|
|
def test_accepts_once_and_replays_idempotently(tmp_path):
|
|
app = IngestionApplication(SQLiteEventStore(str(tmp_path / "events.db")), "opaque")
|
|
assert invoke(app, event())[0].startswith("202")
|
|
status, body = invoke(app, event())
|
|
assert status.startswith("200") and body["status"] == "duplicate"
|
|
|
|
|
|
def test_rejects_auth_secret_fields_and_mismatched_key(tmp_path):
|
|
app = IngestionApplication(SQLiteEventStore(str(tmp_path / "events.db")), "opaque")
|
|
assert invoke(app, event(), token="wrong")[0].startswith("401")
|
|
bad = event(); bad["data"] = {"password": "never"}
|
|
assert invoke(app, bad)[1]["error"] == "secret_shaped_field"
|
|
assert invoke(app, event(), key="other")[1]["error"] == "idempotency_key_mismatch"
|