PapaTramp
aafb80fa31
Background scan every 5 min opens rule:host_silence and notifies without waiting for the next ingest event. CLI job + optional systemd timer. Co-authored-by: Cursor <cursoragent@cursor.com>
123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
"""Proactive host_silence scan (stale heartbeat without waiting for ingest)."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.config import get_settings
|
|
from app.models import Problem
|
|
from app.services.host_silence_scan import run_host_silence_scan
|
|
from app.services.problem_rules import RULE_HOST_SILENCE
|
|
from tests.test_problem_rules import _ingest
|
|
|
|
|
|
@pytest.fixture
|
|
def scan_settings(monkeypatch):
|
|
monkeypatch.setenv("SAC_HEARTBEAT_STALE_MINUTES", "60")
|
|
monkeypatch.setenv("SAC_HOST_SILENCE_SCAN_ENABLED", "true")
|
|
monkeypatch.setenv("SAC_HOST_SILENCE_SCAN_INTERVAL_MINUTES", "5")
|
|
get_settings.cache_clear()
|
|
yield
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_scan_creates_problem_for_stale_host(db_session, scan_settings):
|
|
hb = _ingest(
|
|
db_session,
|
|
type="agent.heartbeat",
|
|
category="agent",
|
|
severity="info",
|
|
title="hb",
|
|
summary="heartbeat",
|
|
)
|
|
hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2)
|
|
db_session.flush()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
results = run_host_silence_scan(db_session, now=now)
|
|
assert len(results) == 1
|
|
assert results[0].created is True
|
|
assert results[0].problem.rule_id == RULE_HOST_SILENCE
|
|
|
|
problem = db_session.scalar(
|
|
select(Problem).where(
|
|
Problem.host_id == hb.host_id,
|
|
Problem.rule_id == RULE_HOST_SILENCE,
|
|
Problem.status == "open",
|
|
)
|
|
)
|
|
assert problem is not None
|
|
|
|
|
|
def test_scan_does_not_duplicate_open_problem(db_session, scan_settings):
|
|
hb = _ingest(
|
|
db_session,
|
|
type="agent.heartbeat",
|
|
category="agent",
|
|
severity="info",
|
|
title="hb",
|
|
summary="heartbeat",
|
|
)
|
|
hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2)
|
|
db_session.flush()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
first = run_host_silence_scan(db_session, now=now)
|
|
assert first[0].created is True
|
|
|
|
second = run_host_silence_scan(db_session, now=now + timedelta(minutes=5))
|
|
assert len(second) == 1
|
|
assert second[0].created is False
|
|
|
|
problems = db_session.scalars(
|
|
select(Problem).where(
|
|
Problem.host_id == hb.host_id,
|
|
Problem.rule_id == RULE_HOST_SILENCE,
|
|
Problem.status == "open",
|
|
)
|
|
).all()
|
|
assert len(problems) == 1
|
|
|
|
|
|
def test_scan_skips_host_without_heartbeat(db_session, scan_settings):
|
|
_ingest(
|
|
db_session,
|
|
event_id=str(uuid.uuid4()),
|
|
type="ssh.login.success",
|
|
severity="info",
|
|
title="login",
|
|
summary="ok",
|
|
)
|
|
results = run_host_silence_scan(db_session)
|
|
assert results == []
|
|
|
|
|
|
def test_scan_notifies_only_on_create(db_session, scan_settings):
|
|
hb = _ingest(
|
|
db_session,
|
|
type="agent.heartbeat",
|
|
category="agent",
|
|
severity="info",
|
|
title="hb",
|
|
summary="heartbeat",
|
|
)
|
|
hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2)
|
|
db_session.flush()
|
|
|
|
from app.services.host_silence_scan import dispatch_host_silence_notifications
|
|
|
|
with patch("app.services.notify_dispatch.notify_problem") as mock_notify:
|
|
results = run_host_silence_scan(db_session)
|
|
notified = dispatch_host_silence_notifications(db_session, results)
|
|
assert notified == 1
|
|
mock_notify.assert_called_once()
|
|
|
|
with patch("app.services.notify_dispatch.notify_problem") as mock_notify:
|
|
results2 = run_host_silence_scan(db_session)
|
|
notified2 = dispatch_host_silence_notifications(db_session, results2)
|
|
assert notified2 == 0
|
|
mock_notify.assert_not_called()
|