PTah
81 lines
2.6 KiB
Python
81 lines
2.6 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_rdp_lifecycle_201(client, auth_headers):
|
|
event_id = str(uuid.uuid4())
|
|
payload = {
|
|
**VALID_EVENT,
|
|
"event_id": event_id,
|
|
"source": {"product": "rdp-login-monitor", "product_version": "1.2.11-SAC"},
|
|
"host": {"hostname": "K6A-DC3", "os_family": "windows"},
|
|
"type": "agent.lifecycle",
|
|
"title": "RDP login monitor started",
|
|
"summary": "Мониторинг запущен на K6A-DC3, версия 1.2.11-SAC",
|
|
"details": {"lifecycle": "started"},
|
|
}
|
|
r = client.post("/api/v1/events", json=payload, headers=auth_headers)
|
|
assert r.status_code == 201
|
|
assert r.json()["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"]
|