FabledSteward/tests/mcp/test_anomaly_review.py

"""The anomaly review tools' pure parts and refusals (#5837): slot names, the
line an anomaly opens at, how far a series reaches, the plugin events hook,
and calls refused before anything is read or changed.

get_anomaly's circumstances, a bulk review and a settings change run against
Postgres in tests/integration/test_mcp_anomaly_review.py.
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS

import pytest

from steward.core import alerts
from steward.mcp.tools import _events, anomalies
from steward.models.users import UserRole

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


def test_slots_are_named_as_a_reader_would():
    assert anomalies.slot_label("day", 3) == "03:00 UTC"
    assert anomalies.slot_label("week", 0) == "Mon 00:00 UTC"
    assert anomalies.slot_label("week", 6 * 24 + 23) == "Sun 23:00 UTC"


def test_an_anomaly_opens_above_the_larger_of_the_factor_and_the_floor():
    assert anomalies.flag_above(20.0, 10.0) == 30.0       # 20 + floor beats 20 × 1.25
    assert anomalies.flag_above(100.0, 10.0) == 125.0     # the factor beats the floor


def test_a_series_reaches_as_far_as_the_occurrence_lasted_within_one_to_six_hours():
    assert anomalies.series_pad(NOW - timedelta(minutes=20), NOW, NOW) == timedelta(hours=1)
    assert anomalies.series_pad(NOW - timedelta(hours=3), NOW, NOW) == timedelta(hours=3)
    assert anomalies.series_pad(NOW - timedelta(days=2), None, NOW) == timedelta(hours=6)


def test_an_event_says_when_what_and_until():
    e = _events.event(NOW, "downtime", "Rebooted (60 s)", until=NOW + timedelta(minutes=1))
    assert e == {"at": "2026-10-15T12:00:00Z", "until": "2026-10-15T12:01:00Z", "kind": "downtime",
                 "what": "Rebooted (60 s)", "subject": None, "ids": {}}


def test_a_resource_names_its_host_through_the_source_that_feeds_it():
    alerts.agent_fed("test_src", lambda r: r.split(":", 1)[0])
    try:
        assert alerts.host_of_resource("test_src", "nas:/var") == "nas"
        assert alerts.host_of_resource("no_such_source", "nas") is None
    finally:
        alerts._AGENT_FED.pop("test_src", None)


def test_same_host_covers_the_resource_itself_and_its_hosts_other_series():
    alerts.agent_fed("test_src", lambda r: r.split(":", 1)[0])
    try:
        place = _events.Place("test_src", "nas", "h1", "nas")
        assert _events._same_host("test_src", "nas:/var", place)
        assert not _events._same_host("test_src", "other", place)
        bare = _events.Place("traefik", "router-a")
        assert _events._same_host("traefik", "router-a", bare)
        assert not _events._same_host("traefik", "router-b", bare)
    finally:
        alerts._AGENT_FED.pop("test_src", None)


async def test_refusals_change_nothing():
    with pytest.raises(ValueError, match="at least one anomaly"):
        await anomalies.review_anomaly([], "normal")
    with pytest.raises(ValueError, match="At most"):
        await anomalies.review_anomaly([f"id-{i}" for i in range(anomalies.REVIEW_MAX + 1)], "real")
    with pytest.raises(ValueError, match="No floor named bogus"):
        await anomalies.set_anomaly_settings(floors={"bogus": 1})
    with pytest.raises(ValueError, match="0 or more"):
        await anomalies.set_anomaly_settings(floors={"host_cpu_pct": -1})
    with pytest.raises(ValueError, match="Nothing to change"):
        await anomalies.set_anomaly_settings()
    with pytest.raises(ValueError, match="doesn't learn a normal"):
        await anomalies.get_metric_normal("no_source", "x", "no_metric")


def test_anomaly_tools_are_registered_with_their_roles(app):
    mcp = app.mcp_server
    for name, role, read_only in (("list_anomalies", UserRole.viewer, True),
                                  ("get_anomaly", UserRole.viewer, True),
                                  ("get_metric_normal", UserRole.viewer, True),
                                  ("get_anomaly_settings", UserRole.viewer, True),
                                  ("review_anomaly", UserRole.operator, False),
                                  ("set_anomaly_settings", UserRole.admin, False)):
        assert mcp.tool_roles[name] is role, name
        assert mcp._tool_manager.get_tool(name).annotations.read_only_hint is read_only, name


def test_a_plugin_adds_events_through_its_hook(app):
    async def events(db, host_id, host_name, since, until):
        return []

    mcp = app.mcp_server
    mcp.register_plugin("events_test", NS(mcp_events=events))
    try:
        assert mcp.event_sources["events_test"] is events
        mcp.register_plugin("events_test", NS())
        assert "events_test" not in mcp.event_sources
    finally:
        mcp.event_sources.pop("events_test", None)


def test_a_plugin_adds_a_hostless_resources_events_through_its_hook(app):
    async def events(db, source, resource, since, until):
        return []

    mcp = app.mcp_server
    mcp.register_plugin("events_test", NS(mcp_resource_events=events))
    try:
        assert mcp.resource_event_sources["events_test"] is events
        mcp.register_plugin("events_test", NS())
        assert "events_test" not in mcp.resource_event_sources
    finally:
        mcp.resource_event_sources.pop("events_test", None)


def test_steward_not_recording_is_an_event_beside_every_anomaly():
    """A Steward restart between two of its once-a-minute samples (#5862)."""
    ticks = [NOW + timedelta(minutes=m) for m in (0, 1, 2)] + [NOW + timedelta(minutes=4, seconds=10)]
    [gap] = _events.gaps_in(ticks, NOW, NOW + timedelta(hours=1))
    assert gap["kind"] == "steward" and gap["at"] == "2026-10-15T12:02:00Z"
    assert gap["until"] == "2026-10-15T12:04:10Z" and "wasn't recording" in gap["what"]
    assert _events.gaps_in(ticks, NOW + timedelta(minutes=5), NOW + timedelta(hours=1)) == []
    assert _events.gaps_in([NOW + timedelta(minutes=m) for m in range(5)], NOW, NOW + timedelta(hours=1)) == []