FabledSteward/tests/mcp/test_tools.py

"""The diagnostic tools' pure parts: response shaping, lookup, redaction, ranking.

The queries behind each tool run against real Postgres in
tests/integration/test_mcp_tools.py; these cover the logic that decides what
an agent is told.
"""
from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS
from unittest.mock import MagicMock

import pytest

from steward.mcp.tools import _shape, overview
from steward.mcp.tools.monitors import _config
from steward.models.alerts import AlertOperator
from steward.models.ansible import AnsibleRunStatus
from steward.models.users import UserRole

NOW = datetime(2026, 10, 6, 12, 0, tzinfo=timezone.utc)

MCP4_TOOLS = {
    "get_overview", "list_hosts", "get_host", "get_host_metrics", "query_metrics",
    "list_monitors", "get_monitor", "list_alerts", "get_alert_history",
    "list_alert_rules", "list_maintenance_windows", "list_audit_events",
}


# ── shaping ──────────────────────────────────────────────────────────────────

def test_iso_is_utc_with_z_and_naive_times_are_read_as_utc():
    assert _shape.iso(NOW) == "2026-10-06T12:00:00Z"
    assert _shape.iso(datetime(2026, 10, 6, 12, 0)) == "2026-10-06T12:00:00Z"
    assert _shape.iso(None) is None


def test_age_s_never_negative():
    assert _shape.age_s(NOW - timedelta(seconds=90), NOW) == 90
    assert _shape.age_s(NOW + timedelta(seconds=5), NOW) == 0


def test_clamp_limit():
    assert _shape.clamp_limit(0, 50, 500) == 50
    assert _shape.clamp_limit(-3, 50, 500) == 50
    assert _shape.clamp_limit(10, 50, 500) == 10
    assert _shape.clamp_limit(9999, 50, 500) == 500


def test_bounded_says_when_and_how_it_cut():
    whole = _shape.bounded([1, 2], 5, key="xs")
    assert whole == {"xs": [1, 2], "total": 2, "truncated": False}
    cut = _shape.bounded(list(range(10)), 3, key="xs", more="Filter by `state`.")
    assert cut["xs"] == [0, 1, 2] and cut["total"] == 10 and cut["truncated"] is True
    assert cut["more"] == "Showing 3 of 10. Filter by `state`."


def test_lookback_refuses_an_unknown_range_instead_of_guessing():
    since, key = _shape.lookback("7d")
    assert key == "7d"
    with pytest.raises(ValueError, match="'2w' is not one of"):
        _shape.lookback("2w")


def test_summarize_and_downsample():
    pts = [(NOW + timedelta(minutes=i), float(i)) for i in range(1, 101)]
    s = _shape.summarize(pts, n=10)
    assert (s["points"], s["min"], s["max"], s["last"]) == (100, 1.0, 100.0, 100.0)
    assert s["avg"] == 50.5 and s["p95"] == 95.0
    assert len(s["series"]) == 10
    assert s["series"][0] == ["2026-10-06T12:01:00Z", 5.5]
    assert _shape.summarize([]) == {"points": 0, "series": []}


def _rows(*names):
    return [NS(id=f"id-{i}", name=n) for i, n in enumerate(names)]


def test_pick_by_id_then_case_insensitive_name():
    rows = _rows("nas", "web01")
    assert _shape.pick("id-1", rows, kind="host", list_tool="list_hosts").name == "web01"
    assert _shape.pick("NAS", rows, kind="host", list_tool="list_hosts").id == "id-0"


def test_pick_explains_a_miss_and_an_ambiguity():
    with pytest.raises(ValueError, match=r"Did you mean 'web01'\?.*list_hosts"):
        _shape.pick("web1", _rows("nas", "web01"), kind="host", list_tool="list_hosts")
    with pytest.raises(ValueError, match="2 hosts are named 'nas'.*id-0, id-1"):
        _shape.pick("nas", _rows("nas", "NAS"), kind="host", list_tool="list_hosts")
    with pytest.raises(ValueError, match="pass a host name or id"):
        _shape.pick("", _rows("nas"), kind="host", list_tool="list_hosts")


def test_redact_url_hides_only_the_password():
    assert _shape.redact_url("https://bob:pw@h.test:8443/x?q=1") == "https://bob:***@h.test:8443/x?q=1"
    assert _shape.redact_url("https://h.test/x") == "https://h.test/x"
    assert _shape.redact_url("10.0.0.1") == "10.0.0.1"


def test_monitor_config_redacts_header_values():
    m = NS(config_json=json.dumps({"method": "GET", "headers": {"Authorization": "Bearer x"}}))
    assert _config(m) == {"method": "GET", "headers": {"Authorization": "<redacted>"}}
    assert _config(NS(config_json="not json")) == {}


# ── overview findings ────────────────────────────────────────────────────────

def _monitor(mid, *, enabled=True, host_id=None):
    return NS(id=mid, name=f"mon-{mid}", type="http", target="https://x.test",
              enabled=enabled, host_id=host_id)


def _result(*, up, error=None, tls_days=None):
    return NS(is_up=up, error_msg=error, checked_at=NOW - timedelta(minutes=1),
              tls_expires_at=NOW + timedelta(days=tls_days) if tls_days is not None else None)


def test_monitor_findings():
    mons = [_monitor("a", host_id="h1"), _monitor("b"), _monitor("c"), _monitor("d", enabled=False),
            _monitor("e")]
    states = {
        "a": {"latest": _result(up=False, error="503"), "down_since": NOW - timedelta(hours=2)},
        "b": {"latest": _result(up=True, tls_days=2)},
        "c": {"latest": _result(up=True, tls_days=10)},
        "d": {"latest": _result(up=False)},
        "e": {"latest": _result(up=True, tls_days=60)},
    }
    out = {f["subject"]: f for f in overview.monitor_findings(mons, states, {"h1": "nas"}, NOW)}
    assert set(out) == {"mon-a", "mon-b", "mon-c"}  # disabled and healthy left out
    down = out["mon-a"]
    assert down["severity"] == "critical" and down["kind"] == "monitor_down"
    assert "on nas" in down["summary"] and "503" in down["summary"]
    assert down["since"] == "2026-10-06T10:00:00Z"
    assert down["next"] == 'get_monitor(monitor="a")'
    assert (out["mon-b"]["severity"], out["mon-c"]["severity"]) == ("critical", "warning")
    # The TLS window comes from Settings → Thresholds.
    narrow = {f["subject"] for f in overview.monitor_findings(
        mons, states, {}, NOW, {"warn": 5, "crit": 1})}
    assert narrow == {"mon-a", "mon-b"}


def test_alert_findings_severity_by_state():
    rule = NS(id="r1", name="cpu", metric_name="cpu_pct", operator=AlertOperator.gt,
              threshold=90.0, source_module="host_agent", resource_name="nas")
    rows = [(NS(state=NS(value=s), last_transition_at=NOW, no_data_since=None), rule)
            for s in ("firing", "pending", "acknowledged")]
    sev = {f["kind"]: f["severity"] for f in overview.alert_findings(rows)}
    assert sev == {"alert_firing": "critical", "alert_pending": "warning",
                   "alert_acknowledged": "warning"}


def test_alert_findings_say_when_a_metric_went_silent():
    rule = NS(id="r1", name="cpu", metric_name="cpu_pct", operator=AlertOperator.gt,
              threshold=90.0, source_module="host_agent", resource_name="nas")
    silent = NOW - timedelta(minutes=10)
    quiet, firing = overview.alert_findings([
        (NS(state=NS(value="inactive"), last_transition_at=NOW - timedelta(days=1),
            no_data_since=silent), rule),
        (NS(state=NS(value="firing"), last_transition_at=NOW - timedelta(hours=1),
            no_data_since=silent), rule),
    ])
    assert (quiet["kind"], quiet["severity"]) == ("alert_no_data", "warning")
    assert quiet["since"] == "2026-10-06T09:50:00Z" and "stopped reporting" in quiet["summary"]
    # A firing alert that goes silent stays a firing finding, marked.
    assert (firing["kind"], firing["severity"]) == ("alert_firing", "critical")
    assert "no data since 2026-10-06T09:50:00Z" in firing["summary"]


THRESHOLDS = {"cpu": {"warn": 80, "crit": 90}, "mem": {"warn": 80, "crit": 90},
              "disk": {"warn": 80, "crit": 90}}


def test_agent_findings_use_the_configured_thresholds():
    hosts = [NS(id="h1", name="nas"), NS(id="h2", name="web"), NS(id="h3", name="db")]
    agent = {
        "nas": {"fresh": True, "cpu_pct": 95.0, "mem_used_pct": 85.0, "disk_used_pct_worst": 10.0},
        "web": {"fresh": False, "cpu_pct": 99.0, "last_at": NOW - timedelta(minutes=30)},
    }
    out = overview.agent_findings(hosts, agent, THRESHOLDS)
    got = {(f["subject"], f["kind"]): f["severity"] for f in out}
    assert got == {
        ("nas", "host_cpu_high"): "critical",
        ("nas", "host_mem_high"): "warning",
        # A stale agent's old numbers are not reported as current load.
        ("web", "agent_stale"): "warning",
    }


def test_run_findings_quote_the_failures():
    run = NS(id="r", playbook_path="site.yml", source_name="infra",
             status=AnsibleRunStatus.failed, started_at=NOW, finished_at=NOW,
             results={"failures": ["web01: apt lock held"]})
    f = overview.run_findings([run])[0]
    assert f["kind"] == "ansible_run_failed" and "apt lock held" in f["summary"]


def test_rank_puts_critical_first_then_oldest():
    fs = [overview.finding("warning", "k", "w-old", "", since=NOW - timedelta(days=1)),
          overview.finding("critical", "k", "c-new", "", since=NOW - timedelta(minutes=1)),
          overview.finding("critical", "k", "c-old", "", since=NOW - timedelta(hours=1))]
    assert [f["subject"] for f in overview.rank(fs)] == ["c-old", "c-new", "w-old"]


async def test_plugin_overview_sources_skip_disabled_plugins_and_survive_failures(monkeypatch):
    calls = []

    async def good(db):
        calls.append("good")
        return [overview.finding("warning", "container_unhealthy", "redis", "unhealthy")]

    async def broken(db):
        raise RuntimeError("boom")

    async def off(db):
        calls.append("off")
        return []

    class _CM:
        async def __aenter__(self):
            return object()

        async def __aexit__(self, *a):
            return False

    monkeypatch.setattr(overview, "db_session", lambda: _CM())
    server = MagicMock()
    server.overview_sources = {"docker": good, "snmp": broken, "unifi": off}
    server.plugin_enabled = lambda p: p != "unifi"
    found, unavailable = await overview._plugin_findings(server)
    assert [f["subject"] for f in found] == ["redis"]
    assert unavailable == ["snmp"] and calls == ["good"]


def test_register_plugin_picks_up_an_overview_source():
    from steward.mcp.server import StewardMCPServer
    app = MagicMock()
    app.config = {"PLUGINS": {}}
    mcp = StewardMCPServer(app)

    async def mcp_overview(db):
        return []
    mcp.register_plugin("docker", NS(mcp_overview=mcp_overview))
    assert mcp.overview_sources == {"docker": mcp_overview}
    mcp.register_plugin("docker", NS())  # reloaded without it
    assert mcp.overview_sources == {}


# ── registration ─────────────────────────────────────────────────────────────

def test_diagnostic_tools_are_registered_read_only_for_viewers(app):
    mcp = app.mcp_server
    assert MCP4_TOOLS <= set(mcp.tool_roles)
    for name in MCP4_TOOLS:
        assert mcp.tool_roles[name] is UserRole.viewer, name
        assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is True, name
        assert (mcp._tool_manager.get_tool(name).description or "").strip(), f"{name} has no docstring"


def test_get_host_groups_parts_by_kind_and_summarises_cores():
    from steward.mcp.tools.hosts import agent_parts
    parts = {f"core{i}": {"cpu_pct": float(i)} for i in range(38)}
    parts.update({
        "/": {"disk_used_pct": 40.0, "disk_used_bytes": 4.0, "disk_total_bytes": 10.0},
        "/data": {"disk_used_pct": 91.0},
        "net:eth0": {"net_rx_bps": 10.0, "net_tx_bps": 5.0},
        "net:br-1": {"net_rx_bps": 900.0, "net_tx_bps": 0.0},
        "net:eth1": {"net_rx_bps": 0.0, "net_tx_bps": 0.0},
        "diskio:sda": {"disk_read_bps": 1.0, "disk_write_bps": 2.0},
        "temp:Package id 0": {"temp_c": 61.0},
    })
    got = agent_parts(parts, "box", limit=2)
    # Cores are a summary, so they can't crowd out the rest.
    assert got["cpu_cores"]["count"] == 38 and got["cpu_cores"]["max_pct"] == 37.0
    assert [c["core"] for c in got["cpu_cores"]["busiest"]] == [37, 36, 35]
    assert [m["mount"] for m in got["mounts"]["items"]] == ["/data", "/"]  # fullest first
    # Busiest interface first; the cut is said, with how to get the rest.
    nets = got["interfaces"]
    assert [n["interface"] for n in nets["items"]] == ["br-1", "eth0"]
    assert nets["total"] == 3 and nets["truncated"] is True
    assert "parts_limit" in nets["more"] and 'resource="box:<part>"' in nets["more"]
    assert got["disk_io"]["items"][0]["device"] == "sda"
    assert got["temperatures"]["items"][0] == {"sensor": "Package id 0", "part": "temp:Package id 0",
                                               "temp_c": 61.0}
    assert "other" not in got