79 lines
2.1 KiB
Python
79 lines
2.1 KiB
Python
"""Retention purge service."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.config import Settings
|
|
from app.models import Event, Host, Problem
|
|
from app.services.retention import purge_old_data
|
|
|
|
|
|
def _host(db, name: str) -> Host:
|
|
h = Host(
|
|
hostname=name,
|
|
os_family="linux",
|
|
product="ssh-monitor",
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(h)
|
|
db.flush()
|
|
return h
|
|
|
|
|
|
def test_purge_old_events_and_resolved_problems(db_session):
|
|
h = _host(db_session, "retention-host")
|
|
now = datetime.now(timezone.utc)
|
|
old_event = now - timedelta(days=120)
|
|
old_problem = now - timedelta(days=200)
|
|
_ev = Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=h.id,
|
|
occurred_at=old_event,
|
|
received_at=old_event,
|
|
category="agent",
|
|
type="agent.heartbeat",
|
|
severity="info",
|
|
title="old",
|
|
summary="s",
|
|
payload={},
|
|
)
|
|
db_session.add(_ev)
|
|
db_session.add(
|
|
Problem(
|
|
host_id=h.id,
|
|
title="resolved old",
|
|
summary="s",
|
|
severity="info",
|
|
status="resolved",
|
|
rule_id="rule:test",
|
|
fingerprint="fp-ret-old",
|
|
event_count=1,
|
|
last_seen_at=old_problem,
|
|
created_at=old_problem,
|
|
updated_at=old_problem,
|
|
)
|
|
)
|
|
db_session.add(
|
|
Problem(
|
|
host_id=h.id,
|
|
title="open stays",
|
|
summary="s",
|
|
severity="high",
|
|
status="open",
|
|
rule_id="rule:test",
|
|
fingerprint="fp-ret-open",
|
|
event_count=1,
|
|
last_seen_at=now,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
db_session.commit()
|
|
|
|
settings = Settings(sac_events_retention_days=90, sac_problems_retention_days=180)
|
|
stats = purge_old_data(db_session, settings)
|
|
assert stats["events_deleted"] == 1
|
|
assert stats["problems_deleted"] == 1
|
|
assert db_session.query(Event).count() == 0
|
|
assert db_session.query(Problem).filter(Problem.status == "open").count() == 1
|