PTah
c8f357007f
- Extend /dashboards/summary; Events filters from query (hostname, severity) Co-authored-by: Cursor <cursoragent@cursor.com>
93 lines
2.5 KiB
Python
93 lines
2.5 KiB
Python
"""Dashboard summary API."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.models import Event, Host, Problem
|
|
|
|
|
|
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 _event(db, host: Host, etype: str, *, received_at: datetime | None = None) -> Event:
|
|
now = received_at or datetime.now(timezone.utc)
|
|
e = Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=host.id,
|
|
occurred_at=now,
|
|
received_at=now,
|
|
category="security",
|
|
type=etype,
|
|
severity="info",
|
|
title=f"t {etype}",
|
|
summary="s",
|
|
payload={},
|
|
)
|
|
db.add(e)
|
|
db.flush()
|
|
return e
|
|
|
|
|
|
def test_dashboard_summary_top_and_problems_24h(client, db_session, jwt_headers):
|
|
h1 = _host(db_session, "web-01")
|
|
h2 = _host(db_session, "db-01")
|
|
now = datetime.now(timezone.utc)
|
|
for _ in range(3):
|
|
_event(db_session, h1, "ssh.login.failed")
|
|
_event(db_session, h2, "agent.heartbeat")
|
|
_event(db_session, h2, "agent.heartbeat")
|
|
|
|
db_session.add(
|
|
Problem(
|
|
host_id=h1.id,
|
|
title="open now",
|
|
summary="s",
|
|
severity="high",
|
|
status="open",
|
|
rule_id="rule:test",
|
|
fingerprint="fp1",
|
|
event_count=1,
|
|
last_seen_at=now,
|
|
created_at=now - timedelta(hours=2),
|
|
updated_at=now - timedelta(hours=2),
|
|
)
|
|
)
|
|
db_session.add(
|
|
Problem(
|
|
host_id=h2.id,
|
|
title="resolved today",
|
|
summary="s",
|
|
severity="warning",
|
|
status="resolved",
|
|
rule_id="rule:test",
|
|
fingerprint="fp2",
|
|
event_count=1,
|
|
last_seen_at=now,
|
|
created_at=now - timedelta(hours=5),
|
|
updated_at=now - timedelta(hours=1),
|
|
)
|
|
)
|
|
db_session.commit()
|
|
|
|
r = client.get("/api/v1/dashboards/summary", headers=jwt_headers)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["events_last_24h"] == 5
|
|
assert body["problems_open"] == 1
|
|
assert body["problems_opened_24h"] == 2
|
|
assert body["problems_resolved_24h"] == 1
|
|
assert body["top_hosts"][0]["hostname"] == "web-01"
|
|
assert body["top_hosts"][0]["count"] == 3
|
|
types = {row["type"]: row["count"] for row in body["top_event_types"]}
|
|
assert types["ssh.login.failed"] == 3
|
|
assert types["agent.heartbeat"] == 2
|