bvandeusenandClaude Opus 5.5
9ca2e38fa1
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>
119 lines
5.7 KiB
Python
119 lines
5.7 KiB
Python
"""The SNMP device form, the row it becomes, and what a page says about the
|
|
newest poll (milestone 491)."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace as NS
|
|
|
|
import pytest
|
|
from werkzeug.datastructures import MultiDict
|
|
|
|
from plugins.snmp.readings import Reading
|
|
from plugins.snmp.service import (FormError, RateKeeper, credentials, device_state, form_credentials,
|
|
form_values, parse_form, readings_for)
|
|
from steward.core.crypto import decrypt_secret, init_crypto, is_encrypted
|
|
|
|
|
|
def setup_function():
|
|
init_crypto("unit-test-secret-key")
|
|
|
|
|
|
def _form(**fields):
|
|
presets = fields.pop("presets", ["system"])
|
|
md = MultiDict({"name": "core-switch", "address": "192.168.1.2", "port": "161", "version": "2c", **fields})
|
|
for p in presets:
|
|
md.add("presets", p)
|
|
return md
|
|
|
|
|
|
def test_a_device_is_stored_with_its_community_encrypted():
|
|
values = parse_form(_form(community="s3cret", interfaces="1-2", presets=["system", "interfaces"]))
|
|
assert values["name"] == "core-switch" and values["port"] == 161 and values["version"] == "2c"
|
|
assert values["presets"] == ["system", "interfaces"] and values["interfaces"] == "1-2"
|
|
assert is_encrypted(values["community"]) and decrypt_secret(values["community"]) == "s3cret"
|
|
assert values["v3_user"] == "" and values["host_id"] is None
|
|
|
|
|
|
def test_a_blank_community_is_public_on_a_new_device_and_kept_on_an_old_one():
|
|
assert decrypt_secret(parse_form(_form())["community"]) == "public"
|
|
existing = NS(community="enc:kept", v3_auth_key="", v3_priv_key="")
|
|
assert parse_form(_form(), existing)["community"] == "enc:kept"
|
|
|
|
|
|
@pytest.mark.parametrize("fields, says", [
|
|
({"name": ""}, "name"),
|
|
({"address": ""}, "address"),
|
|
({"port": "99999"}, "port"),
|
|
({"version": "4"}, "version"),
|
|
({"presets": []}, "at least one"),
|
|
({"presets": ["interfaces"]}, "interface numbers"),
|
|
({"custom": "1.3.6 Bad-Name"}, "Line 1"),
|
|
({"version": "3"}, "user name"),
|
|
({"version": "3", "v3_user": "mon", "v3_auth": "none", "v3_priv": "aes128"}, "Privacy needs authentication"),
|
|
({"version": "3", "v3_user": "mon", "v3_auth": "sha", "v3_priv": "none"}, "needs its key"),
|
|
({"version": "3", "v3_user": "mon", "v3_auth": "sha", "v3_auth_key": "short", "v3_priv": "none"},
|
|
"at least 8"),
|
|
])
|
|
def test_a_form_that_cannot_be_saved_says_what_to_fix(fields, says):
|
|
with pytest.raises(FormError, match=says):
|
|
parse_form(_form(**fields))
|
|
|
|
|
|
def test_a_v3_device_keeps_its_user_and_encrypted_keys_and_no_community():
|
|
values = parse_form(_form(version="3", v3_user="mon", v3_auth="sha256", v3_auth_key="authpass1",
|
|
v3_priv="aes128", v3_priv_key="privpass1"))
|
|
assert values["community"] == "" and values["v3_user"] == "mon"
|
|
assert (values["v3_auth"], values["v3_priv"]) == ("sha256", "aes128")
|
|
assert decrypt_secret(values["v3_auth_key"]) == "authpass1" and decrypt_secret(values["v3_priv_key"]) == "privpass1"
|
|
row = NS(name="x", **{k: values[k] for k in ("version", "community", "v3_user", "v3_auth",
|
|
"v3_auth_key", "v3_priv", "v3_priv_key")})
|
|
creds = credentials(row)
|
|
assert (creds.user, creds.auth_key, creds.priv_key) == ("mon", "authpass1", "privpass1")
|
|
|
|
|
|
def test_the_edit_form_never_carries_a_secret_back():
|
|
device = NS(name="ups", address="10.0.0.9", port=161, version="2c", v3_user="", v3_auth="none",
|
|
v3_priv="none", presets=["ups"], interfaces="", custom="", host_id=None,
|
|
community="enc:x", v3_auth_key="", v3_priv_key="")
|
|
v = form_values(device)
|
|
assert v["has_community"] and not v["has_auth_key"]
|
|
assert "enc:x" not in str(v)
|
|
|
|
|
|
def test_test_uses_what_is_typed_and_falls_back_to_the_stored_secret():
|
|
from steward.core.crypto import encrypt_secret
|
|
existing = NS(community=encrypt_secret("stored"), v3_auth_key="", v3_priv_key="")
|
|
address, port, creds = form_credentials(_form(port="1161"), existing)
|
|
assert (address, port, creds.community) == ("192.168.1.2", 1161, "stored")
|
|
assert form_credentials(_form(community="typed"), existing)[2].community == "typed"
|
|
assert form_credentials(_form(), None)[2].community == "public"
|
|
|
|
|
|
def test_a_row_that_no_longer_parses_is_asked_for_nothing():
|
|
row = NS(presets=["system"], interfaces="eth0", custom="", version="2c")
|
|
assert readings_for(row) == []
|
|
|
|
|
|
def _row(**kw):
|
|
return NS(**{"last_polled_at": None, "last_error": None, **kw})
|
|
|
|
|
|
def test_the_newest_poll_decides_what_the_page_says():
|
|
now = datetime(2026, 10, 8, 15, 0, tzinfo=timezone.utc)
|
|
recent = now - timedelta(seconds=30)
|
|
assert device_state(_row(), 60, now).reach == "pending"
|
|
assert device_state(_row(last_polled_at=recent), 60, now).reach == "up"
|
|
down = device_state(_row(last_polled_at=recent, last_error="No reply from x"), 60, now)
|
|
assert (down.reach, down.says) == ("down", "No reply from x")
|
|
assert device_state(_row(last_polled_at=now - timedelta(minutes=5)), 60, now).reach == "stale"
|
|
|
|
|
|
def test_a_counter_becomes_a_rate_and_a_reset_is_skipped():
|
|
keeper = RateKeeper()
|
|
rx = Reading("1.3", "if1_in", "Interface 1 in", "octets")
|
|
t0 = datetime(2026, 10, 8, 15, 0, tzinfo=timezone.utc)
|
|
assert keeper.rate("d1", rx, 1_000, t0) is None # nothing to compare yet
|
|
assert keeper.rate("d1", rx, 7_000, t0 + timedelta(seconds=60)) == 800.0 # 100 B/s as bit/s
|
|
assert keeper.rate("d1", rx, 50, t0 + timedelta(seconds=120)) is None # rebooted or wrapped
|
|
assert keeper.rate("d1", rx, 650, t0 + timedelta(seconds=180)) == 80.0
|
|
keeper.forget("d1")
|
|
assert keeper.rate("d1", rx, 9_999, t0 + timedelta(seconds=240)) is None
|