56 lines
2.1 KiB
Python
56 lines
2.1 KiB
Python
import pytest
|
|
|
|
from ._helpers import make_event, register_agent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incident_key_ownership_rejected(app_client, ui_auth_headers):
|
|
agent_1_headers = await register_agent(app_client, ui_auth_headers, "agent-1")
|
|
agent_2_headers = await register_agent(app_client, ui_auth_headers, "agent-2")
|
|
|
|
ev1 = make_event(status="critical", exit_code=2, incident_key="shared-key")
|
|
r1 = await app_client.post(
|
|
"/api/v1/events", json={"agent_id": "agent-1", "events": [ev1]}, headers=agent_1_headers
|
|
)
|
|
assert r1.status_code == 202
|
|
|
|
ev2 = make_event(status="critical", exit_code=2, incident_key="shared-key")
|
|
r2 = await app_client.post(
|
|
"/api/v1/events", json={"agent_id": "agent-2", "events": [ev2]}, headers=agent_2_headers
|
|
)
|
|
assert r2.status_code == 202
|
|
assert r2.json() == {"accepted": 0, "deduplicated": 0}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_incident_key_too_long(app_client, ui_auth_headers):
|
|
auth_headers = await register_agent(app_client, ui_auth_headers)
|
|
ev = make_event(status="critical", exit_code=2, incident_key="x" * 257)
|
|
r = await app_client.post(
|
|
"/api/v1/events", json={"agent_id": "agent-1", "events": [ev]}, headers=auth_headers
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_events_query_cursor_pagination(app_client, auth_headers, ui_auth_headers):
|
|
auth_headers = await register_agent(app_client, ui_auth_headers)
|
|
evs = [make_event(check_id=f"chk-{i}") for i in range(5)]
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": "agent-1", "events": evs}, headers=auth_headers
|
|
)
|
|
r = await app_client.get("/api/v1/events/query?limit=2", headers=ui_auth_headers)
|
|
assert r.status_code == 200
|
|
page1 = r.json()
|
|
assert len(page1["items"]) == 2
|
|
assert page1["next_cursor"]
|
|
|
|
r2 = await app_client.get(
|
|
f"/api/v1/events/query?limit=2&cursor={page1['next_cursor']}", headers=ui_auth_headers
|
|
)
|
|
page2 = r2.json()
|
|
assert len(page2["items"]) == 2
|
|
ids1 = {i["event_id"] for i in page1["items"]}
|
|
ids2 = {i["event_id"] for i in page2["items"]}
|
|
assert ids1.isdisjoint(ids2)
|