80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from monlet_server.models import Incident, NotificationOutbox
|
|
|
|
from ._helpers import make_event, make_heartbeat
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_escalate_resolve_reopen(app_client, auth_headers, session):
|
|
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
|
|
a = "agent-1"
|
|
|
|
# warning -> open warning
|
|
ev1 = make_event(status="warning", exit_code=1)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": a, "events": [ev1]}, headers=auth_headers
|
|
)
|
|
inc = (await session.execute(select(Incident))).scalar_one()
|
|
assert inc.state == "open"
|
|
assert inc.severity == "warning"
|
|
|
|
# critical -> escalated, still single open
|
|
ev2 = make_event(status="critical", exit_code=2)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": a, "events": [ev2]}, headers=auth_headers
|
|
)
|
|
session.expire_all()
|
|
incs = (await session.execute(select(Incident))).scalars().all()
|
|
assert len(incs) == 1
|
|
assert incs[0].severity == "critical"
|
|
|
|
# ok -> resolved
|
|
ev3 = make_event(status="ok", exit_code=0)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": a, "events": [ev3]}, headers=auth_headers
|
|
)
|
|
session.expire_all()
|
|
incs = (await session.execute(select(Incident))).scalars().all()
|
|
assert len(incs) == 1
|
|
assert incs[0].state == "resolved"
|
|
|
|
# critical again -> new open
|
|
ev4 = make_event(status="critical", exit_code=2)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": a, "events": [ev4]}, headers=auth_headers
|
|
)
|
|
incs = (await session.execute(select(Incident).order_by(Incident.opened_at))).scalars().all()
|
|
assert len(incs) == 2
|
|
states = {i.state for i in incs}
|
|
assert states == {"open", "resolved"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outbox_skipped_when_notifications_disabled(app_client, auth_headers, session):
|
|
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
|
|
ev = make_event(status="critical", exit_code=2, notifications_enabled=False)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": "agent-1", "events": [ev]}, headers=auth_headers
|
|
)
|
|
rows = (await session.execute(select(NotificationOutbox))).scalars().all()
|
|
assert rows == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_outbox_firing_and_resolved(app_client, auth_headers, session):
|
|
await app_client.post("/api/v1/heartbeat", json=make_heartbeat(), headers=auth_headers)
|
|
a = "agent-1"
|
|
ev_open = make_event(status="critical", exit_code=2)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": a, "events": [ev_open]}, headers=auth_headers
|
|
)
|
|
ev_res = make_event(status="ok", exit_code=0)
|
|
await app_client.post(
|
|
"/api/v1/events", json={"agent_id": a, "events": [ev_res]}, headers=auth_headers
|
|
)
|
|
rows = (await session.execute(select(NotificationOutbox))).scalars().all()
|
|
types = sorted(r.event_type for r in rows)
|
|
assert types == ["firing", "resolved"]
|