284 lines
11 KiB
Python
284 lines
11 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app import audit
|
|
from app.constants import (RETENTION_ALERTS_DAYS, RETENTION_AUDIT_DAYS, RETENTION_MUTES_DAYS,
|
|
RETENTION_STATE_LOG_DAYS)
|
|
from app.retention import (COUNT_KEYS, Cutoffs, RetentionBusy, RetentionPolicy, RetentionRepository,
|
|
RetentionRunner, RetentionService, RunRecord, compute_cutoffs, policy_from,
|
|
should_run, validate_values)
|
|
from tests.prompt_support import make_env
|
|
|
|
NOW = 1_800_000_000
|
|
DAY = 86400
|
|
OLD, NEW = 100, NOW
|
|
|
|
|
|
def seed(db):
|
|
for ts in (OLD, OLD + 1, NEW):
|
|
alert_id = db.insert(
|
|
"INSERT INTO alerts(ts, host, service, kind, state, message_group) VALUES (?,?,?,?,?,?)",
|
|
(ts, "h", "s", "problem", 2, "g"))
|
|
for target in ("a", "b"):
|
|
db.insert("INSERT INTO deliveries(alert_id, backend, target, message_id) VALUES (?,?,?,?)",
|
|
(alert_id, "telegram", target, "1"))
|
|
for ts in (OLD, NEW):
|
|
db.insert("INSERT INTO agent_runs(ts, backend, sender, command, question, answer) VALUES (?,?,?,?,?,?)",
|
|
(ts, "telegram", "1", "ask", "q", "a"))
|
|
db.insert("INSERT INTO state_log(host, service, state, ts) VALUES (?,?,?,?)", ("h", "s", 2, ts))
|
|
db.insert("INSERT INTO audit(ts, user, action, target) VALUES (?,?,?,?)", (ts, "u", "x", "t"))
|
|
db.insert("INSERT INTO mutes(host_glob, service_glob, expires_at, created_by, created_at, source) "
|
|
"VALUES (?,?,?,?,?,?)", ("h", "*", ts, "u", 1, "ui"))
|
|
db.insert("INSERT INTO service_states(host, service, state, last_state_change) VALUES (?,?,?,?)",
|
|
("h", "s", 2, OLD))
|
|
db.insert("INSERT INTO listener_state(key, value) VALUES (?,?)", ("telegram.offset", "5"))
|
|
|
|
|
|
def count(db, table):
|
|
return db.query(f"SELECT COUNT(*) FROM {table}")[0][0]
|
|
|
|
|
|
def cutoffs_between():
|
|
return Cutoffs(*([NOW - DAY] * 5))
|
|
|
|
|
|
def test_cutoffs_from_defaults():
|
|
policy = policy_from(lambda key, default, minimum: default)
|
|
assert policy == RetentionPolicy(90, 90, 30, 365, 7)
|
|
cutoffs = compute_cutoffs(policy, NOW)
|
|
assert cutoffs == Cutoffs(NOW - 90 * DAY, NOW - 90 * DAY, NOW - 30 * DAY, NOW - 365 * DAY, NOW - 7 * DAY)
|
|
|
|
|
|
def test_policy_from_store_clamps_and_never_disables():
|
|
_, store, _, _ = make_env()
|
|
store.set(RETENTION_ALERTS_DAYS, "0", "t")
|
|
store.set(RETENTION_AUDIT_DAYS, "999999", "t")
|
|
store.set(RETENTION_STATE_LOG_DAYS, "abc", "t")
|
|
store.set(RETENTION_MUTES_DAYS, "-5", "t")
|
|
policy = policy_from(store.get_int)
|
|
assert policy.alerts_days == 1 and policy.audit_days == 3650
|
|
assert policy.state_log_days == 30 and policy.mutes_days == 1 and policy.agent_runs_days == 90
|
|
|
|
|
|
def test_validate_values():
|
|
assert validate_values({}) == []
|
|
assert validate_values({RETENTION_ALERTS_DAYS: "1", RETENTION_MUTES_DAYS: "365"}) == []
|
|
for key, bad in ((RETENTION_ALERTS_DAYS, "0"), (RETENTION_ALERTS_DAYS, "-1"), (RETENTION_ALERTS_DAYS, "x"),
|
|
(RETENTION_ALERTS_DAYS, "3651"), (RETENTION_MUTES_DAYS, "366"), (RETENTION_AUDIT_DAYS, "1.5")):
|
|
assert len(validate_values({key: bad})) == 1, (key, bad)
|
|
|
|
|
|
def test_purge_deletes_per_table_with_counts():
|
|
db, _, _, _ = make_env()
|
|
seed(db)
|
|
counts = RetentionRepository(db).purge(cutoffs_between())
|
|
assert counts == {"alerts": 2, "deliveries": 4, "agent_runs": 1, "state_log": 1, "audit": 1, "mutes": 1}
|
|
assert set(counts) == set(COUNT_KEYS)
|
|
assert (count(db, "alerts"), count(db, "deliveries"), count(db, "agent_runs")) == (1, 2, 1)
|
|
assert (count(db, "state_log"), count(db, "mutes")) == (1, 1)
|
|
assert db.query("SELECT COUNT(*) FROM audit WHERE ts=?", (OLD,))[0][0] == 0
|
|
|
|
|
|
def test_cascade_removes_deliveries_of_deleted_alerts_only():
|
|
db, _, _, _ = make_env()
|
|
seed(db)
|
|
RetentionRepository(db).purge(cutoffs_between())
|
|
assert db.query("SELECT COUNT(*) FROM deliveries WHERE alert_id NOT IN (SELECT id FROM alerts)")[0][0] == 0
|
|
assert count(db, "deliveries") == 2
|
|
|
|
|
|
def test_service_states_and_listener_state_are_never_deleted():
|
|
db, _, _, _ = make_env()
|
|
seed(db)
|
|
RetentionRepository(db).purge(Cutoffs(*([NOW + 10 * DAY] * 5)))
|
|
assert (count(db, "service_states"), count(db, "listener_state")) == (1, 1)
|
|
assert (count(db, "alerts"), count(db, "mutes"), count(db, "audit")) == (0, 0, 0)
|
|
|
|
|
|
def test_boundary_row_is_kept():
|
|
db, _, _, _ = make_env()
|
|
db.insert("INSERT INTO audit(ts, user, action, target) VALUES (?,?,?,?)", (500, "u", "x", "t"))
|
|
assert RetentionRepository(db).purge(Cutoffs(*([500] * 5)))["audit"] == 0
|
|
|
|
|
|
def test_batching_with_pause_between_batches():
|
|
db, _, _, _ = make_env()
|
|
for i in range(5):
|
|
db.insert("INSERT INTO state_log(host, service, state, ts) VALUES (?,?,?,?)", ("h", "s", 1, i))
|
|
alert_id = db.insert(
|
|
"INSERT INTO alerts(ts, host, service, kind, state, message_group) VALUES (?,?,?,?,?,?)",
|
|
(i, "h", "s", "problem", 1, "g"))
|
|
db.insert("INSERT INTO deliveries(alert_id, backend, target, message_id) VALUES (?,?,?,?)",
|
|
(alert_id, "telegram", "t", "1"))
|
|
pauses = []
|
|
repo = RetentionRepository(db, batch_size=2, pause=0.5, sleep=pauses.append)
|
|
counts = repo.purge(cutoffs_between())
|
|
assert counts["state_log"] == 5 and counts["alerts"] == 5 and counts["deliveries"] == 5
|
|
assert pauses == [0.5] * 4
|
|
assert count(db, "state_log") == 0 and count(db, "alerts") == 0 and count(db, "deliveries") == 0
|
|
|
|
|
|
def test_checkpoint_runs_on_database():
|
|
db, _, _, _ = make_env()
|
|
RetentionRepository(db).checkpoint()
|
|
|
|
|
|
def make_service(db, policy=None, clock=lambda: float(NOW), repo=None):
|
|
return RetentionService(repo or RetentionRepository(db), lambda: policy or RetentionPolicy(90, 90, 30, 365, 7),
|
|
audit.recorder(db), clock)
|
|
|
|
|
|
def test_run_writes_audit_state_and_log(caplog):
|
|
db, _, _, _ = make_env()
|
|
seed(db)
|
|
db.insert("INSERT INTO audit(ts, user, action, target) VALUES (?,?,?,?)", (1, "u", "ancient", "t"))
|
|
service = make_service(db, RetentionPolicy(1, 1, 1, 1, 1))
|
|
with caplog.at_level(logging.INFO, logger="app.retention.service"):
|
|
record = service.run("manual", "alice")
|
|
assert record.ts == NOW and record.origin == "manual" and record.counts["alerts"] == 2
|
|
assert service.last_run() == record
|
|
rows = db.query("SELECT user, target, old_value, new_value FROM audit WHERE action='retention.run'")
|
|
assert len(rows) == 1 and rows[0]["user"] == "alice" and rows[0]["target"] == "retention"
|
|
assert json.loads(rows[0]["new_value"]) == record.counts
|
|
assert any("Aufbewahrung" in r.getMessage() and "alerts" in r.getMessage() for r in caplog.records)
|
|
|
|
|
|
def test_run_overwrites_single_state_row():
|
|
db, _, _, _ = make_env()
|
|
service = make_service(db)
|
|
service.run("auto")
|
|
service.run("manual", "alice")
|
|
assert count(db, "retention_state") == 1 and service.last_run().origin == "manual"
|
|
|
|
|
|
def test_checkpoint_failure_is_only_logged(caplog):
|
|
class BrokenCheckpoint(RetentionRepository):
|
|
def checkpoint(self):
|
|
raise RuntimeError("busy")
|
|
|
|
db, _, _, _ = make_env()
|
|
service = make_service(db, repo=BrokenCheckpoint(db))
|
|
with caplog.at_level(logging.WARNING):
|
|
record = service.run()
|
|
assert record.origin == "auto" and any("wal_checkpoint" in r.getMessage() for r in caplog.records)
|
|
|
|
|
|
def test_concurrent_run_is_rejected():
|
|
db, _, _, _ = make_env()
|
|
service = make_service(db)
|
|
service._lock.acquire()
|
|
with pytest.raises(RetentionBusy):
|
|
service.run()
|
|
service._lock.release()
|
|
service.run()
|
|
|
|
|
|
def test_last_run_is_none_initially():
|
|
db, _, _, _ = make_env()
|
|
assert RetentionRepository(db).last_run() is None
|
|
|
|
|
|
def at(day, hour, minute=0):
|
|
return datetime(2026, 11, day, hour, minute)
|
|
|
|
|
|
def stamp(moment):
|
|
return moment.timestamp()
|
|
|
|
|
|
@pytest.mark.parametrize("now, last, startup, expected", [
|
|
(at(6, 2, 59), None, False, False),
|
|
(at(6, 3, 0), None, False, True),
|
|
(at(6, 2, 59), None, True, True),
|
|
(at(6, 10), stamp(at(6, 3, 5)), True, False),
|
|
(at(6, 10), stamp(at(6, 3, 5)), False, False),
|
|
(at(6, 1), stamp(at(5, 3, 5)), True, True),
|
|
(at(6, 1), stamp(at(5, 3, 5)), False, False),
|
|
(at(6, 3, 1), stamp(at(5, 23, 59)), False, True),
|
|
])
|
|
def test_should_run(now, last, startup, expected):
|
|
assert should_run(now, last, startup) is expected
|
|
|
|
|
|
class Moment:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def __call__(self):
|
|
return self.value
|
|
|
|
|
|
class FakeService:
|
|
def __init__(self, clock, error=None):
|
|
self.clock, self.error, self.runs, self.record = clock, error, [], None
|
|
|
|
def last_run(self):
|
|
return self.record
|
|
|
|
def run(self, origin):
|
|
self.runs.append(origin)
|
|
if self.error:
|
|
raise self.error
|
|
self.record = RunRecord(int(self.clock().timestamp()), origin, {})
|
|
return self.record
|
|
|
|
|
|
async def test_runner_runs_once_per_calendar_day():
|
|
clock = Moment(at(6, 10))
|
|
service = FakeService(clock)
|
|
runner = RetentionRunner(service, clock)
|
|
assert await runner.tick(startup=True) is True
|
|
clock.value = at(6, 11)
|
|
assert await runner.tick(startup=False) is False
|
|
clock.value = at(7, 2, 30)
|
|
assert await runner.tick(startup=False) is False
|
|
clock.value = at(7, 3, 5)
|
|
assert await runner.tick(startup=False) is True
|
|
assert service.runs == ["auto", "auto"]
|
|
|
|
|
|
async def test_restart_on_same_day_does_not_delete_again():
|
|
clock = Moment(at(6, 10))
|
|
service = FakeService(clock)
|
|
assert await RetentionRunner(service, clock).tick(startup=True)
|
|
assert await RetentionRunner(service, clock).tick(startup=True) is False
|
|
assert len(service.runs) == 1
|
|
|
|
|
|
async def test_failed_run_is_not_retried_the_same_day():
|
|
clock = Moment(at(6, 10))
|
|
service = FakeService(clock, error=RuntimeError("db"))
|
|
runner = RetentionRunner(service, clock)
|
|
assert await runner.tick(startup=True) is True
|
|
clock.value = at(6, 12)
|
|
assert await runner.tick(startup=False) is False
|
|
clock.value = at(7, 4)
|
|
assert await runner.tick(startup=False) is True
|
|
assert len(service.runs) == 2
|
|
|
|
|
|
async def test_busy_service_is_tolerated():
|
|
clock = Moment(at(6, 10))
|
|
runner = RetentionRunner(FakeService(clock, error=RetentionBusy()), clock)
|
|
assert await runner.tick(startup=True) is True
|
|
|
|
|
|
async def test_run_loop_waits_start_delay_then_checks_periodically():
|
|
clock = Moment(at(6, 10))
|
|
service = FakeService(clock)
|
|
sleeps = []
|
|
|
|
async def sleep(seconds):
|
|
sleeps.append(seconds)
|
|
if len(sleeps) == 3:
|
|
raise asyncio.CancelledError
|
|
|
|
runner = RetentionRunner(service, clock, sleep, start_delay=5.0, interval=30.0)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await runner.run()
|
|
assert sleeps == [5.0, 30.0, 30.0] and service.runs == ["auto"]
|