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

"""Hosts list API."""

import uuid
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select

from app.config import get_settings
from app.models import Event, Host, Problem


@pytest.fixture
def host_stale_settings(monkeypatch):
    monkeypatch.setenv("SAC_HEARTBEAT_STALE_MINUTES", "60")
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


def test_hosts_filter_agent_status_stale(client, db_session, jwt_headers, host_stale_settings):
    now = datetime.now(timezone.utc)
    stale_host = Host(
        hostname="stale-pc",
        os_family="windows",
        product="rdp-login-monitor",
        last_seen_at=now,
    )
    fresh_host = Host(
        hostname="fresh-pc",
        os_family="windows",
        product="rdp-login-monitor",
        last_seen_at=now,
    )
    unknown_host = Host(
        hostname="unknown-pc",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=now,
    )
    db_session.add_all([stale_host, fresh_host, unknown_host])
    db_session.flush()

    db_session.add(
        Event(
            event_id=str(uuid.uuid4()),
            host_id=stale_host.id,
            occurred_at=now - timedelta(hours=2),
            received_at=now - timedelta(hours=2),
            category="agent",
            type="agent.heartbeat",
            severity="info",
            title="hb",
            summary="s",
            payload={},
        )
    )
    db_session.add(
        Event(
            event_id=str(uuid.uuid4()),
            host_id=fresh_host.id,
            occurred_at=now - timedelta(minutes=5),
            received_at=now - timedelta(minutes=5),
            category="agent",
            type="agent.heartbeat",
            severity="info",
            title="hb",
            summary="s",
            payload={},
        )
    )
    db_session.commit()

    r = client.get("/api/v1/hosts?agent_status=stale", headers=jwt_headers)
    assert r.status_code == 200
    body = r.json()
    assert body["total"] == 1
    assert body["items"][0]["hostname"] == "stale-pc"
    assert body["items"][0]["agent_status"] == "stale"

    r_all = client.get("/api/v1/hosts", headers=jwt_headers)
    assert r_all.json()["total"] == 3


def test_hosts_search_by_display_name(client, db_session, jwt_headers):
    h = Host(
        hostname="pc-01",
        display_name="UNMS Kalina",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=datetime.now(timezone.utc),
    )
    db_session.add(h)
    db_session.commit()

    r = client.get("/api/v1/hosts?hostname=UNMS", headers=jwt_headers)
    assert r.status_code == 200
    body = r.json()
    assert body["total"] == 1
    assert body["items"][0]["display_name"] == "UNMS Kalina"

    r2 = client.get("/api/v1/hosts?hostname=nomatch", headers=jwt_headers)
    assert r2.json()["total"] == 0


def test_delete_host_removes_events_and_problems(client, db_session, jwt_headers):
    h = Host(
        hostname="test-host",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=datetime.now(timezone.utc),
    )
    db_session.add(h)
    db_session.flush()
    ev = Event(
        event_id=str(uuid.uuid4()),
        host_id=h.id,
        occurred_at=datetime.now(timezone.utc),
        category="agent",
        type="agent.test",
        severity="info",
        title="t",
        summary="s",
        payload={},
    )
    db_session.add(ev)
    db_session.add(
        Problem(
            host_id=h.id,
            title="p",
            summary="s",
            severity="warning",
            status="open",
            rule_id="rule:test",
            fingerprint="fp-del-host",
            event_count=1,
            last_seen_at=datetime.now(timezone.utc),
        )
    )
    db_session.commit()

    host_id = h.id
    r = client.delete(f"/api/v1/hosts/{host_id}", headers=jwt_headers)
    assert r.status_code == 200
    body = r.json()
    assert body["hostname"] == "test-host"
    assert body["deleted_events"] == 1
    assert body["deleted_problems"] == 1
    assert db_session.scalar(select(Host.id).where(Host.id == host_id)) is None


def test_delete_host_404(client, jwt_headers):
    r = client.delete("/api/v1/hosts/999999", headers=jwt_headers)
    assert r.status_code == 404


def test_delete_host_forbidden_for_monitor(client, db_session, jwt_monitor_headers):
    h = Host(
        hostname="monitor-blocked",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=datetime.now(timezone.utc),
    )
    db_session.add(h)
    db_session.commit()

    r = client.delete(f"/api/v1/hosts/{h.id}", headers=jwt_monitor_headers)
    assert r.status_code == 403
    assert db_session.get(Host, h.id) is not None


def test_delete_host_requires_jwt(client, db_session):
    h = Host(
        hostname="no-auth",
        os_family="linux",
        product="ssh-monitor",
        last_seen_at=datetime.now(timezone.utc),
    )
    db_session.add(h)
    db_session.commit()
    r = client.delete(f"/api/v1/hosts/{h.id}")
    assert r.status_code == 401