security-alert-center/backend/tests/test_problems.py

"""Problems correlation and API smoke tests."""

import uuid
from datetime import datetime, timedelta, timezone

from app.models import Host, Problem
from tests.test_ingest import VALID_EVENT


def _event_payload(**overrides):
    now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
    base = {
        **VALID_EVENT,
        "event_id": str(uuid.uuid4()),
        "occurred_at": now,
        "severity": "high",
        "type": "ssh.ip.banned",
        "title": "Ban test",
        "summary": "test problem correlation",
    }
    base.update(overrides)
    return base


def test_problem_correlates_two_events(client, auth_headers, jwt_headers):
    e1 = _event_payload()
    e2 = _event_payload(title="Ban test 2")
    assert client.post("/api/v1/events", json=e1, headers=auth_headers).status_code == 201
    assert client.post("/api/v1/events", json=e2, headers=auth_headers).status_code == 201

    r = client.get("/api/v1/problems?status=open", headers=jwt_headers)
    assert r.status_code == 200
    data = r.json()
    assert data["total"] == 1
    problem = data["items"][0]
    assert problem["event_count"] == 2
    assert problem["fingerprint"].startswith("h")
    assert ":tssh.ip.banned:" in problem["fingerprint"]

    detail = client.get(f"/api/v1/problems/{problem['id']}", headers=jwt_headers)
    assert detail.status_code == 200
    assert len(detail.json()["events"]) == 2


def test_problem_ack_and_resolve(client, auth_headers, jwt_headers):
    payload = _event_payload()
    client.post("/api/v1/events", json=payload, headers=auth_headers)
    pid = client.get("/api/v1/problems?status=open", headers=jwt_headers).json()["items"][0]["id"]

    ack = client.post(f"/api/v1/problems/{pid}/ack", headers=jwt_headers)
    assert ack.status_code == 200
    assert ack.json()["status"] == "acknowledged"

    res = client.post(f"/api/v1/problems/{pid}/resolve", headers=jwt_headers)
    assert res.status_code == 200
    assert res.json()["status"] == "resolved"


def test_new_problem_after_resolve(client, auth_headers, jwt_headers):
    p1 = _event_payload()
    client.post("/api/v1/events", json=p1, headers=auth_headers)
    pid = client.get("/api/v1/problems?status=open", headers=jwt_headers).json()["items"][0]["id"]
    client.post(f"/api/v1/problems/{pid}/resolve", headers=jwt_headers)

    p2 = _event_payload(summary="after resolve")
    client.post("/api/v1/events", json=p2, headers=auth_headers)
    open_items = client.get("/api/v1/problems?status=open", headers=jwt_headers).json()["items"]
    assert len(open_items) == 1
    assert open_items[0]["event_count"] == 1
    assert open_items[0]["id"] != pid


def test_problems_created_within_hours(client, db_session, jwt_headers):
    now = datetime.now(timezone.utc)
    h = Host(
        hostname="p-host",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=now,
    )
    db_session.add(h)
    db_session.flush()
    db_session.add(
        Problem(
            host_id=h.id,
            title="recent",
            summary="s",
            severity="warning",
            status="open",
            rule_id="rule:test",
            fingerprint="fp-recent",
            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=h.id,
            title="old",
            summary="s",
            severity="warning",
            status="resolved",
            rule_id="rule:test",
            fingerprint="fp-old",
            event_count=1,
            last_seen_at=now,
            created_at=now - timedelta(days=3),
            updated_at=now - timedelta(days=2),
        )
    )
    db_session.commit()

    r = client.get("/api/v1/problems?created_within_hours=24", headers=jwt_headers)
    assert r.status_code == 200
    body = r.json()
    assert body["total"] == 1
    assert body["items"][0]["title"] == "recent"


def test_problems_resolved_within_hours(client, db_session, jwt_headers):
    now = datetime.now(timezone.utc)
    h = Host(
        hostname="r-host",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=now,
    )
    db_session.add(h)
    db_session.flush()
    db_session.add(
        Problem(
            host_id=h.id,
            title="closed today",
            summary="s",
            severity="warning",
            status="resolved",
            rule_id="rule:test",
            fingerprint="fp-closed-today",
            event_count=1,
            last_seen_at=now,
            created_at=now - timedelta(days=2),
            updated_at=now - timedelta(hours=1),
        )
    )
    db_session.add(
        Problem(
            host_id=h.id,
            title="closed long ago",
            summary="s",
            severity="warning",
            status="resolved",
            rule_id="rule:test",
            fingerprint="fp-closed-old",
            event_count=1,
            last_seen_at=now,
            created_at=now - timedelta(days=10),
            updated_at=now - timedelta(days=5),
        )
    )
    db_session.commit()

    r = client.get("/api/v1/problems?resolved_within_hours=24", headers=jwt_headers)
    assert r.status_code == 200
    body = r.json()
    assert body["total"] == 1
    assert body["items"][0]["title"] == "closed today"