119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from monlet_server.models import Incident, NotificationOutbox
|
|
|
|
from ._helpers import make_event, register_agent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_open_escalate_resolve_reopen(app_client, ui_auth_headers, session):
|
|
auth_headers = await register_agent(app_client, ui_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, ui_auth_headers, session):
|
|
auth_headers = await register_agent(app_client, ui_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, ui_auth_headers, session):
|
|
auth_headers = await register_agent(app_client, ui_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"]
|
|
|
|
|
|
def test_status_rank_map_covers_all_pairs():
|
|
"""PH-review: the SQL-side rank is computed by a STORED generated column
|
|
(`incidents.status_rank`). The Python `_STATUS_RANK` map mirrors that CASE
|
|
for cursor encoding only. This test pins the map so the two cannot drift —
|
|
if a new (state, severity) pair is added, both must update."""
|
|
from monlet_server.api.incidents import _STATUS_RANK
|
|
|
|
assert _STATUS_RANK == {
|
|
("open", "critical"): 5,
|
|
("open", "warning"): 4,
|
|
("open", "unknown"): 3,
|
|
("resolved", "critical"): 2,
|
|
("resolved", "warning"): 1,
|
|
("resolved", "unknown"): 0,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_rank_generated_column_matches_python_map(
|
|
app_client, ui_auth_headers, session
|
|
):
|
|
"""End-to-end check: insert one event per (state, severity) pair and
|
|
confirm the DB-side generated column equals the Python map value."""
|
|
from monlet_server.api.incidents import _STATUS_RANK
|
|
|
|
# Open critical incident comes from a single critical event.
|
|
auth_headers = await register_agent(app_client, ui_auth_headers)
|
|
crit = make_event(check_id="crit", status="critical", exit_code=2)
|
|
warn = make_event(check_id="warn", status="warning", exit_code=1)
|
|
await app_client.post(
|
|
"/api/v1/events",
|
|
json={"agent_id": "agent-1", "events": [crit, warn]},
|
|
headers=auth_headers,
|
|
)
|
|
rows = (await session.execute(select(Incident))).scalars().all()
|
|
for r in rows:
|
|
assert r.status_rank == _STATUS_RANK[(r.state, r.severity)]
|