"""The unifi plugin's MCP tools (#5839): what the overview says about the WAN, devices and alarms, and the tools it registers.""" from datetime import datetime, timedelta, timezone from types import SimpleNamespace as NS import pytest import plugins.unifi as unifi_plugin from plugins.unifi import mcp_tools as t from steward.models.users import UserRole NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc) FRESH = timedelta(minutes=5) def _device(name, state, ago=timedelta(minutes=1)): return NS(name=name, mac=f"aa:{name}", model="U6-Pro", device_type="uap", state=state, last_seen=NOW - ago, scraped_at=NOW - ago) def test_a_wan_down_is_critical_only_while_the_poll_is_current(): down = NS(is_up=False, scraped_at=NOW - timedelta(minutes=1)) [f] = t.network_findings(down, [], [], NOW, FRESH) assert f["kind"] == "unifi_wan_down" and f["severity"] == "critical" stale = NS(is_up=False, scraped_at=NOW - timedelta(hours=1)) assert t.network_findings(stale, [], [], NOW, FRESH) == [] assert t.network_findings(NS(is_up=True, scraped_at=NOW), [], [], NOW, FRESH) == [] def test_devices_offline_are_critical_degraded_warn_and_transitions_say_nothing(): found = {f["subject"]: f for f in t.network_findings(None, [ _device("ap-hall", 0), _device("ap-loft", 6), _device("ap-shed", 4), _device("gw", 1), _device("ap-old", 0, ago=timedelta(days=30))], [], NOW, FRESH)} assert set(found) == {"ap-hall", "ap-loft"} assert found["ap-hall"]["severity"] == "critical" and "Offline" in found["ap-hall"]["summary"] assert found["ap-loft"]["severity"] == "warning" and "Heartbeat missed" in found["ap-loft"]["summary"] def test_active_alarms_are_one_finding_naming_the_newest(): alarms = [NS(message="IPS alert: scan from 1.2.3.4", alarm_time=NOW), NS(message="older", alarm_time=NOW - timedelta(hours=1))] [f] = t.network_findings(None, [], alarms, NOW, FRESH) assert f["kind"] == "unifi_alarms" and f["summary"] == "UniFi alarm: IPS alert: scan from 1.2.3.4 and 1 more" def test_the_plugin_registers_its_tools_for_viewers(): got = {} unifi_plugin.register_mcp_tools(lambda fn, role, read_only: got.update({fn.__name__: (role, read_only)})) assert got == {name: (UserRole.viewer, True) for name in ( "get_network", "list_devices", "list_clients", "list_network_events", "list_alarms", "get_network_config")} assert callable(unifi_plugin.mcp_overview) async def test_list_devices_refuses_an_unknown_status(): with pytest.raises(ValueError, match="status must be one of"): await t.list_devices(status="broken")