feat(mcp): SNMP devices and Prometheus targets: their state, readings and series (#5840)
The snmp plugin registers list_snmp_devices and get_snmp_device; the prometheus plugin registers list_prometheus_targets and get_prometheus_target. Both are viewer-level, read what the poller and scraper stored, and exist only while their plugin is enabled. get_overview names a device or target whose last poll or scrape failed; stale, pending and off are left out. A device's community and v3 keys and a target's token are never returned, and a password in a target URL is masked. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,49 @@
|
||||
"""The snmp plugin's MCP tools (#5840): which devices the overview names, what
|
||||
a device row carries (never its community or v3 keys), and the tools it
|
||||
registers."""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from types import SimpleNamespace as NS
|
||||
|
||||
import pytest
|
||||
|
||||
import plugins.snmp as snmp_plugin
|
||||
from plugins.snmp import mcp_tools as t
|
||||
from steward.models.users import UserRole
|
||||
|
||||
NOW = datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _device(name, *, polled=timedelta(seconds=30), error=None):
|
||||
return NS(id=f"id-{name}", name=name, address="10.0.0.9", port=161, version="2c",
|
||||
community="s3cret", v3_user="", v3_auth_key="authkey", v3_priv_key="privkey",
|
||||
presets=[], interfaces="", custom=None, host_id=None,
|
||||
last_polled_at=NOW - polled if polled else None, last_ok_at=NOW - timedelta(hours=1),
|
||||
last_error=error, missing=[])
|
||||
|
||||
|
||||
def test_only_a_device_whose_poll_failed_is_an_overview_finding():
|
||||
views = [t._device_view(d, {}, NOW, 60) | {"_ok_at": d.last_ok_at} for d in (
|
||||
_device("ups", error="Timed out"), _device("switch"),
|
||||
_device("printer", polled=timedelta(days=1), error="Timed out"), _device("nas", polled=None))]
|
||||
[f] = t.device_findings(views)
|
||||
assert f["kind"] == "snmp_device_down" and f["subject"] == "ups" and "Timed out" in f["summary"]
|
||||
assert f["next"] == 'get_snmp_device(device="ups")'
|
||||
|
||||
|
||||
def test_a_device_row_never_carries_its_secrets():
|
||||
row = t._device_view(_device("ups"), {}, NOW, 60)
|
||||
text = repr(row)
|
||||
assert "s3cret" not in text and "authkey" not in text and "privkey" not in text
|
||||
assert row["reach"] == "up" and row["device_id"] == "id-ups"
|
||||
|
||||
|
||||
def test_the_plugin_registers_its_tools_for_viewers():
|
||||
got = {}
|
||||
snmp_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 ("list_snmp_devices", "get_snmp_device")}
|
||||
assert callable(snmp_plugin.mcp_overview)
|
||||
|
||||
|
||||
async def test_list_snmp_devices_refuses_an_unknown_reach():
|
||||
with pytest.raises(ValueError, match="reach must be one of"):
|
||||
await t.list_snmp_devices(reach="broken")
|
||||
Reference in New Issue
Block a user