feat(mcp): the UniFi network: WAN, devices, clients, events, alarms and config (#5839)
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>
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
"""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")
|
||||
Reference in New Issue
Block a user