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


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), 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"}


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"