bvandeusenandClaude Opus 5.5
d065cafd11
A preset can now read a vendor's tables: one row per disk, volume, fan or
sensor. The scheduled poll discovers the rows by walking the device: on its
first poll, after a save, and then hourly, and only while it answers. Each row
becomes readings named by the device's own labels ("Disk 1 status"). Only the
columns a row answered are asked for.
A vendor status, whether a code (Synology) or a word (Asustor), is a state
reading. It is stored as a level (0 OK, 1 busy, 2 warning, 3 failed), so one
alert rule covers any vendor, and shown in the device's own word.
- Synology: system, power and fan status and temperature; per disk: status,
health, temperature and bad sectors; per volume: RAID state (scrubbing,
parity check, rebuild, degraded), free space and size.
- Asustor: per disk: status, SMART and temperature; per volume: status, free
space and size; CPU and system temperature, free memory and fans.
From ADM's ASUSTOR-DISK-MIB and ASUSTOR-SYSTEM-MIB.
- Linux server: plus each physical disk's 1-minute busy % (UCD-DISKIO-MIB).
Migration snmp_002_table_rows adds tables, tables_at and texts to
snmp_devices. Milestone 530.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
50 lines
2.2 KiB
Python
50 lines
2.2 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=[], tables={}, texts={})
|
|
|
|
|
|
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")
|