304 lines
11 KiB
Python
304 lines
11 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select, update
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from monlet_server.models import Agent, Check, Event, Incident, NotificationOutbox
|
|
from monlet_server.services.detector import LIVENESS_CHECK_ID, _tick
|
|
from monlet_server.settings import reset_settings_cache
|
|
|
|
from ._helpers import make_event, register_agent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detector_transitions(app_client, ui_auth_headers, engine, session):
|
|
await register_agent(app_client, ui_auth_headers, "agent-alive")
|
|
await register_agent(app_client, ui_auth_headers, "agent-stale")
|
|
await register_agent(app_client, ui_auth_headers, "agent-dead")
|
|
|
|
now = datetime.now(UTC)
|
|
await session.execute(
|
|
update(Agent)
|
|
.where(Agent.agent_id == "agent-stale")
|
|
.values(last_seen_at=now - timedelta(seconds=120))
|
|
)
|
|
await session.execute(
|
|
update(Agent)
|
|
.where(Agent.agent_id == "agent-dead")
|
|
.values(last_seen_at=now - timedelta(minutes=10))
|
|
)
|
|
await session.commit()
|
|
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
|
|
statuses = {
|
|
r.agent_id: r.status for r in (await session.execute(select(Agent))).scalars().all()
|
|
}
|
|
assert statuses["agent-alive"] == "alive"
|
|
assert statuses["agent-stale"] == "stale"
|
|
assert statuses["agent-dead"] == "dead"
|
|
|
|
incidents = (
|
|
(await session.execute(select(Incident).where(Incident.check_id == LIVENESS_CHECK_ID)))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
by_agent = {i.agent_id: i for i in incidents}
|
|
assert "agent-alive" not in by_agent
|
|
assert by_agent["agent-stale"].state == "open"
|
|
assert by_agent["agent-stale"].severity == "warning"
|
|
assert by_agent["agent-stale"].summary == "agent is stale"
|
|
assert by_agent["agent-dead"].state == "open"
|
|
assert by_agent["agent-dead"].severity == "critical"
|
|
assert by_agent["agent-dead"].summary == "agent is dead"
|
|
|
|
await session.execute(
|
|
update(Agent).where(Agent.agent_id == "agent-dead").values(last_seen_at=datetime.now(UTC))
|
|
)
|
|
await session.commit()
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
|
|
resolved = (
|
|
await session.execute(
|
|
select(Incident).where(
|
|
Incident.agent_id == "agent-dead",
|
|
Incident.check_id == LIVENESS_CHECK_ID,
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert resolved.state == "resolved"
|
|
assert resolved.resolved_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detector_noop_tick_does_not_rewrite_liveness(
|
|
app_client, ui_auth_headers, engine, session
|
|
):
|
|
"""PH-009: when no agent's liveness status changes, the detector must not
|
|
rewrite the liveness Check rows. We verify by tagging last_observed_at to a
|
|
sentinel before the tick and checking the row was untouched."""
|
|
await register_agent(app_client, ui_auth_headers, "agent-still")
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
# Force a transition first so a liveness Check row exists.
|
|
await session.execute(
|
|
update(Agent)
|
|
.where(Agent.agent_id == "agent-still")
|
|
.values(last_seen_at=datetime.now(UTC) - timedelta(minutes=10))
|
|
)
|
|
await session.commit()
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
row_before = (
|
|
await session.execute(
|
|
select(Check).where(
|
|
Check.agent_id == "agent-still", Check.check_id == LIVENESS_CHECK_ID
|
|
)
|
|
)
|
|
).scalar_one()
|
|
sentinel = datetime(2030, 1, 1, tzinfo=UTC)
|
|
await session.execute(
|
|
update(Check)
|
|
.where(Check.agent_id == "agent-still", Check.check_id == LIVENESS_CHECK_ID)
|
|
.values(last_observed_at=sentinel)
|
|
)
|
|
await session.commit()
|
|
|
|
# Second tick with no liveness change: row must NOT be rewritten.
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
row_after = (
|
|
await session.execute(
|
|
select(Check).where(
|
|
Check.agent_id == "agent-still", Check.check_id == LIVENESS_CHECK_ID
|
|
)
|
|
)
|
|
).scalar_one()
|
|
assert row_after.last_observed_at == sentinel
|
|
assert row_after.last_event_id == row_before.last_event_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detector_owns_liveness_check_and_events(
|
|
app_client, ui_auth_headers, engine, session
|
|
):
|
|
await register_agent(app_client, ui_auth_headers, "agent-flap")
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
|
|
# Tick 1: alive → Check exists with status=ok, one event.
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
check = (
|
|
await session.execute(
|
|
select(Check).where(Check.agent_id == "agent-flap", Check.check_id == LIVENESS_CHECK_ID)
|
|
)
|
|
).scalar_one()
|
|
assert check.status == "ok"
|
|
n_events = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(Event)
|
|
.where(Event.agent_id == "agent-flap", Event.check_id == LIVENESS_CHECK_ID)
|
|
)
|
|
).scalar_one()
|
|
assert n_events == 1
|
|
|
|
# Tick 2: still alive → no new event.
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
n_events = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(Event)
|
|
.where(Event.agent_id == "agent-flap", Event.check_id == LIVENESS_CHECK_ID)
|
|
)
|
|
).scalar_one()
|
|
assert n_events == 1
|
|
|
|
# Push agent into dead → transition event written, Check updates.
|
|
await session.execute(
|
|
update(Agent)
|
|
.where(Agent.agent_id == "agent-flap")
|
|
.values(last_seen_at=datetime.now(UTC) - timedelta(minutes=10))
|
|
)
|
|
await session.commit()
|
|
await _tick(sm)
|
|
session.expire_all()
|
|
check = (
|
|
await session.execute(
|
|
select(Check).where(Check.agent_id == "agent-flap", Check.check_id == LIVENESS_CHECK_ID)
|
|
)
|
|
).scalar_one()
|
|
assert check.status == "critical"
|
|
n_events = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(Event)
|
|
.where(Event.agent_id == "agent-flap", Event.check_id == LIVENESS_CHECK_ID)
|
|
)
|
|
).scalar_one()
|
|
assert n_events == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detector_prunes_outbox(app_client, ui_auth_headers, engine, session, monkeypatch):
|
|
monkeypatch.setenv("MONLET_OUTBOX_RETENTION_MAX_ROWS", "2")
|
|
reset_settings_cache()
|
|
try:
|
|
auth_headers = await register_agent(app_client, ui_auth_headers, "agent-prune")
|
|
critical = make_event(check_id="critical_prune", status="critical", exit_code=2)
|
|
await app_client.post(
|
|
"/api/v1/events",
|
|
json={"agent_id": "agent-prune", "events": [critical]},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
inc = (
|
|
await session.execute(select(Incident).where(Incident.check_id == "critical_prune"))
|
|
).scalar_one()
|
|
now = datetime.now(UTC)
|
|
for _ in range(5):
|
|
session.add(
|
|
NotificationOutbox(
|
|
id=uuid4(),
|
|
incident_id=inc.id,
|
|
notifier="debug",
|
|
event_type="firing",
|
|
state="sent",
|
|
attempts=1,
|
|
next_attempt_at=None,
|
|
payload={},
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
await _tick(sm)
|
|
|
|
terminal_outbox = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(NotificationOutbox)
|
|
.where(NotificationOutbox.state == "sent")
|
|
)
|
|
).scalar_one()
|
|
assert terminal_outbox == 2
|
|
finally:
|
|
reset_settings_cache()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detector_prune_preserves_active_outbox(
|
|
app_client, ui_auth_headers, engine, session, monkeypatch
|
|
):
|
|
"""PH-012: prune must never delete active rows (pending/sending/retry),
|
|
even when retention is exceeded. The keep-N-newest list applies only to
|
|
terminal states; active rows are filtered out of the eligible set."""
|
|
monkeypatch.setenv("MONLET_OUTBOX_RETENTION_MAX_ROWS", "1")
|
|
reset_settings_cache()
|
|
try:
|
|
auth_headers = await register_agent(app_client, ui_auth_headers, "agent-prune-active")
|
|
critical = make_event(check_id="active_prune", status="critical", exit_code=2)
|
|
await app_client.post(
|
|
"/api/v1/events",
|
|
json={"agent_id": "agent-prune-active", "events": [critical]},
|
|
headers=auth_headers,
|
|
)
|
|
inc = (
|
|
await session.execute(select(Incident).where(Incident.check_id == "active_prune"))
|
|
).scalar_one()
|
|
now = datetime.now(UTC)
|
|
# Mix of active and terminal states.
|
|
for state, count in (("sent", 5), ("pending", 3), ("retry", 2), ("sending", 1)):
|
|
for _ in range(count):
|
|
session.add(
|
|
NotificationOutbox(
|
|
id=uuid4(),
|
|
incident_id=inc.id,
|
|
notifier="debug",
|
|
event_type="firing",
|
|
state=state,
|
|
attempts=1,
|
|
next_attempt_at=None,
|
|
payload={},
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
await _tick(sm)
|
|
|
|
# Active rows must be untouched. Ingestion may add extra pending rows
|
|
# for the debug notifier; we only assert ours are still present and
|
|
# that no active rows were deleted by the prune.
|
|
for state, injected in (("pending", 3), ("retry", 2), ("sending", 1)):
|
|
n = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(NotificationOutbox)
|
|
.where(NotificationOutbox.state == state)
|
|
)
|
|
).scalar_one()
|
|
assert n >= injected, (state, n, injected)
|
|
# Terminal rows must be capped to retention=1 (or 2 after notifier worker
|
|
# produces extras from the critical event we ingested; but worker is
|
|
# disabled in tests so we just check terminal pruning).
|
|
sent = (
|
|
await session.execute(
|
|
select(func.count())
|
|
.select_from(NotificationOutbox)
|
|
.where(NotificationOutbox.state == "sent")
|
|
)
|
|
).scalar_one()
|
|
assert sent <= 2 # 1 retained + at most 1 created by detector enqueue
|
|
finally:
|
|
reset_settings_cache()
|