bvandeusenandClaude Opus 5.5
2d88315bb1
get_network gives the WAN, clients, Wi-Fi networks, speed tests and devices by status; list_devices, list_clients (everyone seen, or the newest poll's top talkers), list_network_events, list_alarms and get_network_config (networks, port forwards, firewall rules) read what the poller stored. get_overview names the WAN down while the poll is current, a device offline or degraded that the controller still reports, and active alarms. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
58 lines
2.6 KiB
Python
58 lines
2.6 KiB
Python
"""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")
|