"""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())