bvandeusenandClaude Opus 5.5
eee8469a17
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>
50 lines
2.1 KiB
Python
50 lines
2.1 KiB
Python
"""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")
|