Files
bvandeusenandClaude Opus 5.5 9ca2e38fa1
CI / lint (push) Successful in 2s
CI / integration (push) Failing after 55s
CI / unit (push) Successful in 1m24s
CI / publish (push) Skipped
feat(snmp): devices you add, test and read on the SNMP page (milestone 491)
The operator, reviewing SNMP on #5214: "this system looks incomplete like
there's something missing from it". The one device shown was plugin.yaml's
example (core-switch 192.168.1.1, community public), polled every minute;
devices could only be defined by editing YAML in the image; a failed poll was
logged at DEBUG and stored nowhere, so the page said "waiting for the next
poll" forever; v3 ran as v2c; counters showed as lifetime totals.

- Devices are rows (snmp_devices, migration snmp_001): address, port,
  v1/v2c community or a v3 user with auth/priv protocols and keys (secrets
  encrypted), presets plus custom readings, an optional host binding, and
  the newest poll's outcome. The YAML device list and its example are gone.
- The SNMP page adds, edits and deletes them (admin, audited, with
  get_audit_entities so the audit log names and links a device). Readings
  come from presets (system uptime, interface traffic by number, UPS-MIB,
  Linux net-snmp) and custom "OID name [unit]" lines.
- One GET per device with every OID, devices polled at once. The outcome is
  written to the row in words ("No reply from … within 5 s. Check the address
  and port, that SNMP is on, and the community"), with readings it lacks.
  v1's whole-request error drops that OID and asks again. SNMPv3 signs in
  with USM (MD5/SHA-1/SHA-2, DES/3DES/AES).
- Test asks the unsaved form's device for its name, description, uptime and
  interfaces by number. It is an admin's explicit action; every page reads
  only stored rows (System 178's ruling).
- Each reading has a unit; a counter is stored as a per-second rate (bytes as
  bit/s), a reset or wrap skipped. The page, widget and host panel say values
  in units and why a device is silent; the device page draws a chart per
  reading on a dated axis.

Tests: readings and presets, the form and its secrets, v3 checks, rates,
the poller against a stand-in pysnmp (one GET, v1 retry, timeout wording,
USM), the pages, and an integration test of the poll writing outcomes and
rates on Postgres.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-08 10:54:11 -04:00

97 lines
6.1 KiB
Python

"""The SNMP pages (milestone 491): devices not answering first and saying why,
readings in their units, the editor, and Test's answer."""
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace as NS
from plugins.snmp.routes import devices_for_host
from tests._templates import render_plugin_template
from tests.test_style_tokens import hex_colours
def _view(name, reach, says="Answering.", readings=(), ok_minutes=None, missing=()):
now = datetime.now(timezone.utc)
return {"id": f"id-{name}", "name": name, "address": "10.0.0.9", "port": 161, "version": "2c",
"reach": reach, "says": says, "polled_at": now - timedelta(seconds=20),
"ok_at": now - timedelta(minutes=ok_minutes) if ok_minutes is not None else None,
"readings": list(readings), "missing": list(missing)}
async def test_a_device_not_answering_says_why_and_since_when(app):
from quart import session
devices = [
_view("ups", "down", "No reply from 10.0.0.9:161 within 5 s. Check the address.", ok_minutes=180,
readings=[{"key": "battery_charge", "name": "Charge", "counter": False, "text": "97%"}]),
_view("printer", "down", "No reply from 10.0.0.10:161 within 5 s."),
_view("switch", "up", readings=[{"key": "if1_in", "name": "Interface 1 in", "counter": True, "text": "—"},
{"key": "uptime", "name": "Uptime", "counter": False, "text": "12d 4h"}],
missing=["Load"]),
_view("new", "pending"),
]
async with app.test_request_context("/plugins/snmp/"):
session["user_role"] = "admin"
page = await render_plugin_template(app, "snmp", "snmp/index.html", devices=devices)
page = " ".join(page.split())
assert page.count('<span class="badge badge-crit">Not answering</span>') == 2
assert "No reply from 10.0.0.9:161 within 5 s. Check the address." in page
assert "Last answered" in page and ">3h ago</time>" in page and "It has never answered." in page
assert "the readings below are from then" in page
assert '<span class="stat-tile-value">97%</span>' in page and '<span class="stat-tile-value">12d 4h</span>' in page
assert "from the second poll" in page # a counter's rate needs two polls
assert "Not answered by this device: Load" in page
assert "Not polled yet: the first poll runs within a minute of saving." in page
assert 'href="/plugins/snmp/devices/new"' in page and 'href="/plugins/snmp/devices/id-ups/edit"' in page
main = page[page.index("<main"):]
assert "font-weight:600" not in main and not hex_colours(main), hex_colours(main)
async def test_a_viewer_sees_no_editing_and_an_empty_page_offers_to_add(app):
from quart import session
async with app.test_request_context("/plugins/snmp/"):
session["user_role"] = "viewer"
page = await render_plugin_template(app, "snmp", "snmp/index.html", devices=[_view("ups", "up")])
session["user_role"] = "admin"
empty = await render_plugin_template(app, "snmp", "snmp/index.html", devices=[])
assert "/edit" not in page and "devices/new" not in page
assert "No SNMP devices yet." in empty and 'href="/plugins/snmp/devices/new"' in empty
assert "plugin.yaml" not in empty
async def test_the_form_never_echoes_a_secret_and_offers_test(app):
from plugins.snmp.readings import INTERFACES_LABEL, PRESETS, PRESET_ORDER, UNITS
presets = [(p, *(INTERFACES_LABEL if p == "interfaces" else PRESETS[p][:2])) for p in PRESET_ORDER]
device = NS(id="d1", name="ups")
v = {"name": "ups", "address": "10.0.0.9", "port": 161, "version": "3", "v3_user": "mon", "v3_auth": "sha256",
"v3_priv": "aes128", "presets": ["ups"], "interfaces": "", "custom": "", "host_id": "h1",
"has_community": False, "has_auth_key": True, "has_priv_key": True}
async with app.test_request_context("/plugins/snmp/devices/d1/edit"):
page = await render_plugin_template(app, "snmp", "snmp/form.html", device=device, v=v, error=None,
hosts=[NS(id="h1", name="nas")], presets=presets, units=UNITS)
assert 'name="v3_auth_key" type="password"' in page and "kept; type to change" in page
assert 'value="sha256" selected' in page and 'value="h1" selected' in page
assert 'value="ups" checked' in page and 'value="system"' in page and 'value="system" checked' not in page
assert 'hx-post="/plugins/snmp/devices/test"' in page and 'name="device_id" value="d1"' in page
assert 'data-confirm="Delete SNMP device ups?' in page
assert "SNMP" in page and 'href="/plugins/snmp/"' in page # the trail back
async def test_test_shows_who_answered_and_the_interface_numbers(app):
async with app.test_request_context("/plugins/snmp/devices/test"):
ok = await render_plugin_template(app, "snmp", "snmp/_test.html", address="10.0.0.2", port=161,
found={"name": "core", "description": "Switch 24G", "uptime_cs": 8_640_000,
"interfaces": [(1, "ge-0/0/1"), (49, "xe-0/1/0")]})
bad = await render_plugin_template(app, "snmp", "snmp/_test.html", address="10.0.0.2", port=161,
found={"error": "No reply from 10.0.0.2:161 within 5 s."})
assert "10.0.0.2 answered." in ok and "Switch 24G" in ok and "1d" in ok
assert '<span class="mono tone-dim">49</span> xe-0/1/0' in ok
assert 'role="alert"' in bad and "No reply from 10.0.0.2:161" in bad
def test_a_device_shows_on_its_bound_host_or_at_its_address():
devices = [NS(name="sw", address="192.168.1.2", host_id=None),
NS(name="ups", address="10.0.0.9", host_id="h-nas"),
NS(name="pdu", address="nas.lan", host_id=None)]
assert [d.name for d in devices_for_host(devices, "h-nas", "192.168.1.50", "NAS.lan")] == ["ups", "pdu"]
assert [d.name for d in devices_for_host(devices, "h-other", "10.0.0.9", "x")] == [] # bound elsewhere
assert [d.name for d in devices_for_host(devices, "h1", "192.168.1.2", None)] == ["sw"]
assert devices_for_host(devices, "h1", "", None) == []