PapaTramp
2b76ae8497
- Validate UUID format; log created/duplicate/rejected - Tests for 201, 409, 422; update agent-integration and work-plan - Docs: neutral IDE wording (no product-specific editor names)
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""Ingest idempotency and HTTP status contract."""
|
|
|
|
import uuid
|
|
|
|
from app.services.schema_validate import validate_event_payload
|
|
|
|
VALID_EVENT = {
|
|
"schema_version": "1.0",
|
|
"event_id": "550e8400-e29b-41d4-a716-446655440000",
|
|
"occurred_at": "2026-05-27T10:00:00+03:00",
|
|
"source": {"product": "ssh-monitor", "product_version": "1.2.3-SAC"},
|
|
"host": {"hostname": "test-host", "os_family": "linux"},
|
|
"category": "agent",
|
|
"type": "agent.test",
|
|
"severity": "info",
|
|
"title": "Test",
|
|
"summary": "pytest ingest",
|
|
}
|
|
|
|
|
|
def test_schema_rejects_non_uuid_event_id():
|
|
bad = {**VALID_EVENT, "event_id": "not-a-uuid"}
|
|
errors = validate_event_payload(bad)
|
|
assert errors
|
|
|
|
|
|
def test_schema_rejects_missing_event_id():
|
|
payload = {k: v for k, v in VALID_EVENT.items() if k != "event_id"}
|
|
errors = validate_event_payload(payload)
|
|
assert errors
|
|
|
|
|
|
def test_ingest_created_201(client, auth_headers):
|
|
event_id = str(uuid.uuid4())
|
|
payload = {**VALID_EVENT, "event_id": event_id}
|
|
r = client.post("/api/v1/events", json=payload, headers=auth_headers)
|
|
assert r.status_code == 201
|
|
data = r.json()
|
|
assert data["created"] is True
|
|
assert data["status"] == "created"
|
|
assert data["event_id"] == event_id
|
|
|
|
|
|
def test_ingest_duplicate_409(client, auth_headers):
|
|
event_id = str(uuid.uuid4())
|
|
payload = {**VALID_EVENT, "event_id": event_id}
|
|
assert client.post("/api/v1/events", json=payload, headers=auth_headers).status_code == 201
|
|
r = client.post("/api/v1/events", json=payload, headers=auth_headers)
|
|
assert r.status_code == 409
|
|
data = r.json()
|
|
assert data["created"] is False
|
|
assert data["status"] == "duplicate"
|
|
assert data["event_id"] == event_id
|
|
|
|
|
|
def test_ingest_invalid_payload_422(client, auth_headers):
|
|
r = client.post(
|
|
"/api/v1/events",
|
|
json={"event_id": "bad", "summary": "x"},
|
|
headers=auth_headers,
|
|
)
|
|
assert r.status_code == 422
|
|
assert "schema_errors" in r.json()["detail"]
|