FabledSteward/tests/integration/test_mcp_anomaly_review.py

"""Integration (#5837): understanding and judging a metric anomaly through
MCP, against real Postgres.

A host's CPU anomaly with two occurrences. Around the newest one, the host
had an alert, a container death, a reboot, an Ansible run and a memory
anomaly; an earlier anomaly on the same metric was marked normal with a note.
get_anomaly brings all of that back; review_anomaly marks two at once;
get_metric_normal reads the learned slots; set_anomaly_settings changes a
floor and the review window, and the running app follows.

Plugin rows are seeded with plain SQL (a second import of a plugin's models
raises "Table already defined").
"""
from __future__ import annotations

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

import pytest

pytestmark = pytest.mark.integration


@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)


def test_an_anomaly_with_its_circumstances_judged_through_mcp(app):
    from sqlalchemy import select, text
    from steward.core.api_tokens import Caller
    from steward.core.norms import slot_of
    from steward.mcp import _context
    from steward.mcp.tools import anomalies as tools
    from steward.models.alerts import AlertEvent, AlertOperator, AlertRule
    from steward.models.ansible import AnsibleRun, AnsibleRunStatus
    from steward.models.audit import AuditEvent
    from steward.models.hosts import Host
    from steward.models.metrics import PluginMetric
    from steward.models.norms import MetricAnomaly, MetricAnomalySpan, MetricNorm
    from steward.models.users import User, UserRole

    tag = uuid.uuid4().hex[:8]
    host_name, other_host = f"an-{tag}", f"an-other-{tag}"
    now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
    new_start, new_end = now - timedelta(hours=3), now - timedelta(hours=2)
    old_start, old_end = now - timedelta(hours=26), now - timedelta(hours=25, minutes=30)
    ids = {}

    def anomaly(resource, metric, start, end, **kw):
        return MetricAnomaly(source_module="host_agent", resource_name=resource, metric_name=metric,
                             started_at=start, last_started_at=start, ended_at=end,
                             occurrences=kw.pop("occurrences", 1), peak_level=kw.pop("peak", 90.0), peak_at=start, usual_hold=20.0,
                             usual_peak=35.0, **kw)

    async def _seed():
        async with app.db_sessionmaker() as s:
            async with s.begin():
                user = User(username=f"an-{tag}", email=f"an-{tag}@test", password_hash="x",
                            role=UserRole.admin)
                host, other = Host(name=host_name, address="10.9.9.7"), Host(name=other_host, address="10.9.9.8")
                s.add_all([user, host, other])
                await s.flush()
                ids.update(user=user.id, host=host.id)

                cpu = anomaly(host_name, "cpu_pct", old_start, new_end, occurrences=2, peak=93.0)
                cpu.last_started_at = new_start
                before = anomaly(host_name, "cpu_pct", now - timedelta(days=9), now - timedelta(days=9, hours=-1),
                                 review="normal", reviewed_by="op", note="nightly backup", peak=70.0)
                mem = anomaly(host_name, "mem_used_pct", new_start + timedelta(minutes=10), new_end, peak=95.0)
                far = anomaly(other_host, "cpu_pct", new_start, new_end)
                s.add_all([cpu, before, mem, far])
                await s.flush()
                ids.update(cpu=cpu.id, before=before.id, mem=mem.id)
                s.add_all([
                    MetricAnomalySpan(anomaly_id=cpu.id, started_at=old_start, ended_at=old_end, peak_level=88.0),
                    MetricAnomalySpan(anomaly_id=cpu.id, started_at=new_start, ended_at=new_end, peak_level=93.0),
                    MetricAnomalySpan(anomaly_id=mem.id, started_at=new_start + timedelta(minutes=10),
                                      ended_at=new_end, peak_level=95.0),
                    MetricAnomalySpan(anomaly_id=far.id, started_at=new_start, ended_at=new_end, peak_level=90.0),
                    MetricNorm(source_module="host_agent", resource_name=host_name, metric_name="cpu_pct",
                               slot=slot_of(new_start, "day"), kind="day", usual_hold=20.0, usual_peak=35.0,
                               days_seen=12, updated_at=now),
                ])
                for minutes in range(-240, -60, 5):
                    at = now + timedelta(minutes=minutes)
                    s.add(PluginMetric(source_module="host_agent", resource_name=host_name,
                                       metric_name="cpu_pct", recorded_at=at,
                                       value=90.0 if new_start <= at < new_end else 18.0))

                rule = AlertRule(name=f"CPU high {tag}", source_module="host_agent", resource_name=host_name,
                                 metric_name="cpu_pct", operator=AlertOperator.gt, threshold=85,
                                 created_by=user.id)
                s.add(rule)
                await s.flush()
                s.add(AlertEvent(rule_id=rule.id, from_state="ok", to_state="firing", metric_value=91.0,
                                 transitioned_at=new_start + timedelta(minutes=20)))
                s.add(AnsibleRun(playbook_path="maintenance/backup.yml", source_name="steward-builtin",
                                 status=AnsibleRunStatus.success, started_at=new_start - timedelta(minutes=10),
                                 finished_at=new_start + timedelta(minutes=5), params={"check": False},
                                 results={"hosts": {host_name: {"ok": 3}}, "failures": []}))
                await s.execute(text(
                    "INSERT INTO docker_events (id, host_id, container_name, event, detail, at) "
                    "VALUES (:id, :h, 'web', 'die', 'exit 137', :at)"),
                    {"id": str(uuid.uuid4()), "h": host.id, "at": new_start + timedelta(minutes=15)})
                await s.execute(text(
                    "INSERT INTO host_downtime (id, host_id, kind, started_at, ended_at, created_at) "
                    "VALUES (:id, :h, 'reboot', :s, :e, now())"),
                    {"id": str(uuid.uuid4()), "h": host.id, "s": new_end - timedelta(minutes=5),
                     "e": new_end + timedelta(minutes=1)})

    async def _go():
        await _seed()
        _context._app_ctx.set(app)
        _context._caller_ctx.set(Caller(user_id=ids["user"], username=f"an-{tag}",
                                        role=UserRole.admin, token_id=9))

        got = await tools.get_anomaly(ids["cpu"])
        assert got["place"] == {"host": host_name, "host_id": ids["host"]}
        assert got["occurrences_total"] == 2 and got["truncated"] is False
        newest, oldest = got["occurrences"]
        assert newest["held_s"] == 3600 and newest["peak_level"] == 93.0
        assert newest["usual"]["usual_hold"] == 20.0 and newest["usual"]["flag_above"] >= 25.0
        kinds = {e["kind"] for e in newest["around"]}
        assert {"alert", "ansible_run", "container", "downtime", "anomaly"} <= kinds
        whats = " | ".join(e["what"] for e in newest["around"])
        assert "web die (exit 137)" in whats and "maintenance/backup.yml" in whats
        assert any(e["kind"] == "anomaly" and e["ids"]["anomaly_id"] == ids["mem"] for e in newest["around"])
        assert newest["anomalies_elsewhere"] >= 1
        assert not [e for e in oldest["around"] if e["kind"] in ("container", "ansible_run")]
        points = got["series"]["points"]
        assert max(p[1] for p in points) == 90.0 and min(p[1] for p in points) == 18.0
        assert any(p[2] == 20.0 for p in points)
        [prior] = got["precedent"]
        assert prior["anomaly_id"] == ids["before"] and prior["note"] == "nightly backup"

        normal = await tools.get_metric_normal("host_agent", host_name, "cpu_pct")
        assert normal["learned"] is True and normal["kind"] == "day"
        assert normal["floor_setting"] == "host_cpu_pct"
        [slot] = normal["slots"]
        assert slot["usual_hold"] == 20.0 and slot["label"].endswith(":00 UTC")
        with pytest.raises(ValueError, match="doesn't learn a normal"):
            await tools.get_metric_normal("host_agent", host_name, "nope_pct")

        with pytest.raises(ValueError, match="Nothing was marked"):
            await tools.review_anomaly([ids["cpu"], "no-such-id"], "real")
        marked = await tools.review_anomaly([ids["cpu"], ids["mem"]], "real", note="the backup job hung")
        assert {m["anomaly_id"] for m in marked["marked"]} == {ids["cpu"], ids["mem"]}
        assert marked["reviewed_by"] == f"an-{tag} via MCP"

        try:
            changed = await tools.set_anomaly_settings(review_days=21, floors={"host_cpu_pct": 15})
            assert changed["review_days"] == 21
            floor = {f["name"]: f for f in changed["floors"]}["host_cpu_pct"]
            assert floor["value"] == 15.0 and "host_agent/cpu_pct" in floor["metrics"]
            assert app.config["THRESHOLDS"]["floors"]["host_cpu_pct"] == 15.0
        finally:
            await tools.set_anomaly_settings(review_days=14, floors={"host_cpu_pct": 10})

        async with app.db_sessionmaker() as s:
            audits = list((await s.execute(select(AuditEvent.action).where(
                AuditEvent.user_id == ids["user"]).order_by(AuditEvent.timestamp))).scalars())
            rows = {a.id: a for a in (await s.execute(select(MetricAnomaly).where(
                MetricAnomaly.id.in_([ids["cpu"], ids["mem"]])))).scalars()}
        assert audits.count("metric_anomaly.marked_real") == 2 and audits.count("settings.saved") == 2
        assert {a.review for a in rows.values()} == {"real"}
        assert {a.note for a in rows.values()} == {"the backup job hung"}

    asyncio.run(_go())