FabledSteward/tests/integration/test_monitors.py

"""Integration: unified Monitor schema + status source against live Postgres.

Validates the 0022 unification migration applied (monitors/monitor_results
exist, the old per-type tables are gone) and that monitor_status_source rolls
up MonitorResult history into a StatusEntry. Requires STEWARD_DATABASE_URL.
"""
from __future__ import annotations

import asyncio
import os
import uuid
from datetime import datetime, timezone, timedelta

import pytest

pytestmark = pytest.mark.integration

_NEEDS_DB = pytest.mark.skipif(
    not os.environ.get("STEWARD_DATABASE_URL"),
    reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
)


@pytest.fixture
def app():
    if not os.environ.get("STEWARD_DATABASE_URL"):
        pytest.skip("needs Postgres")
    from steward.app import create_app
    return create_app(testing=False)


@_NEEDS_DB
def test_old_monitor_tables_dropped(app):
    from sqlalchemy import text

    async def _go():
        async with app.db_sessionmaker() as s:
            # New tables exist…
            await s.execute(text("SELECT COUNT(*) FROM monitors"))
            await s.execute(text("SELECT COUNT(*) FROM monitor_results"))
            # …old ones are gone (regclass of a missing table is NULL).
            for tbl in ("ping_results", "dns_results", "http_monitors", "http_results"):
                missing = (await s.execute(
                    text("SELECT to_regclass(:t)"), {"t": tbl})).scalar()
                assert missing is None, f"{tbl} should have been dropped"

    asyncio.run(_go())


@_NEEDS_DB
def test_status_source_rolls_up_results(app):
    from sqlalchemy import text
    from steward.models.monitors import Monitor, MonitorResult
    from steward.core.status import monitor_status_source

    mid = str(uuid.uuid4())
    now = datetime.now(timezone.utc)

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                # Clean slate for a deterministic assertion.
                await s.execute(text("DELETE FROM monitor_results"))
                await s.execute(text("DELETE FROM monitors"))
                s.add(Monitor(id=mid, name="custom-http", type="http",
                              target="https://example.test", host_id=None,
                              config_json="{}", enabled=True))
            async with s.begin():
                for i, up in enumerate([True, True, False, True]):
                    s.add(MonitorResult(
                        id=str(uuid.uuid4()), monitor_id=mid,
                        checked_at=now - timedelta(minutes=4 - i),
                        is_up=up, response_ms=10.0 if up else None,
                    ))
            entries = await monitor_status_source(s)
        return entries

    entries = asyncio.run(_go())
    assert len(entries) == 1
    e = entries[0]
    assert e.kind == "http" and e.name == "custom-http"
    assert e.target == "https://example.test"
    assert e.status == "up"                 # latest result is up
    assert len(e.heartbeat) == 4
    assert e.uptime["24h"] == 75.0          # 3 of 4 up


@_NEEDS_DB
def test_last_results_are_the_newest_n_oldest_first_and_host_status_reads_the_last(app):
    """_last_n_results takes each monitor's newest n by a LATERAL probe (#5208);
    host_status reads the newest one per enabled monitor."""
    from steward.core.status import _last_n_results
    from steward.hosts.service import host_status
    from steward.models.hosts import Host
    from steward.models.monitors import Monitor, MonitorResult

    now = datetime.now(timezone.utc)
    hid, a, b, quiet = (str(uuid.uuid4()) for _ in range(4))

    async def _go():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                s.add(Host(id=hid, name="lateral-" + hid[:8], address="10.8.8.8"))
            async with s.begin():
                s.add(Monitor(id=a, name="a-" + a[:8], type="icmp", target="10.8.8.8",
                              host_id=hid, config_json="{}", enabled=True))
                s.add(Monitor(id=b, name="b-" + b[:8], type="icmp", target="10.8.8.9",
                              host_id=hid, config_json="{}", enabled=True))
                s.add(Monitor(id=quiet, name="q-" + quiet[:8], type="icmp", target="10.8.8.7",
                              config_json="{}", enabled=True))
            async with s.begin():
                for i in range(5):    # a: up, up, up, up, down (newest)
                    s.add(MonitorResult(monitor_id=a, checked_at=now - timedelta(minutes=5 - i),
                                        is_up=i < 4))
                s.add(MonitorResult(monitor_id=b, checked_at=now - timedelta(minutes=1),
                                    is_up=True, response_ms=12.0))
            async with s.begin():
                last = await _last_n_results(s, [a, b, quiet], n=3)
                status = await host_status(s, [hid])
            return last, status

    last, status = asyncio.run(_go())
    assert [r.is_up for r in last[a]] == [True, True, False]
    assert last[a][0].checked_at < last[a][-1].checked_at       # oldest first
    assert len(last[b]) == 1 and last[quiet] == []
    assert status[hid]["state"] == "down" and status[hid]["count"] == 2
    assert status[hid]["latency_ms"] == 12.0
    # Down since a's first failure after its last success (#5281).
    assert abs((status[hid]["down_since"] - (now - timedelta(minutes=1))).total_seconds()) < 1