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>
This commit is contained in:
Claude Opus 5.5
@@ -1,69 +0,0 @@
|
||||
"""Unit tests for mapping config-defined SNMP devices onto a Steward host."""
|
||||
from plugins.snmp.routes import _devices_for_host
|
||||
|
||||
_CFG = [
|
||||
{"name": "core-switch", "host": "192.168.1.1"},
|
||||
{"name": "ups", "host": "192.168.1.10"},
|
||||
{"name": "by-name", "host": "edge-router"},
|
||||
"not-a-dict", # tolerated, skipped
|
||||
{"name": "no-host"}, # no host → never matches
|
||||
]
|
||||
|
||||
|
||||
def test_matches_by_address():
|
||||
out = _devices_for_host(_CFG, "192.168.1.1", "somehost")
|
||||
assert [d["name"] for d in out] == ["core-switch"]
|
||||
|
||||
|
||||
def test_matches_by_name_case_insensitive():
|
||||
out = _devices_for_host(_CFG, "10.0.0.9", "Edge-Router")
|
||||
assert [d["name"] for d in out] == ["by-name"]
|
||||
|
||||
|
||||
def test_no_match_returns_empty():
|
||||
assert _devices_for_host(_CFG, "10.0.0.1", "nope") == []
|
||||
|
||||
|
||||
def test_blank_address_and_name_match_nothing():
|
||||
# A host with no address/name must not match a device with a blank host.
|
||||
assert _devices_for_host(_CFG, "", None) == []
|
||||
|
||||
|
||||
def test_implicit_match_unaffected_when_host_id_passed():
|
||||
# Passing the new host_id arg must not break the legacy implicit match.
|
||||
out = _devices_for_host(_CFG, "192.168.1.1", "somehost", host_id="h-1")
|
||||
assert [d["name"] for d in out] == ["core-switch"]
|
||||
|
||||
|
||||
# ── Explicit binding ─────────────────────────────────────────────────────────
|
||||
|
||||
def test_explicit_host_id_matches_exact_uuid():
|
||||
cfg = [{"name": "sw", "host": "10.0.0.5", "host_id": "uuid-abc"}]
|
||||
out = _devices_for_host(cfg, "10.0.0.9", "other", host_id="uuid-abc")
|
||||
assert [d["name"] for d in out] == ["sw"]
|
||||
|
||||
|
||||
def test_explicit_steward_host_matches_by_name_or_address():
|
||||
cfg = [{"name": "pdu", "host": "10.0.0.5", "steward_host": "Switch01"}]
|
||||
# Bind hits the Host's name despite a different poll target / address.
|
||||
by_name = _devices_for_host(cfg, "10.0.0.9", "switch01", host_id="h9")
|
||||
by_addr = _devices_for_host(cfg, "switch01", "whatever", host_id="h9")
|
||||
assert [d["name"] for d in by_name] == ["pdu"]
|
||||
assert [d["name"] for d in by_addr] == ["pdu"]
|
||||
|
||||
|
||||
def test_explicit_binding_is_exclusive():
|
||||
# Device polls 192.168.1.1 but is explicitly bound to host A by name.
|
||||
cfg = [{"name": "sw", "host": "192.168.1.1", "steward_host": "hostA"}]
|
||||
# Host B *would* match implicitly by the poll target — but must NOT,
|
||||
# because the explicit binding takes over and points only at host A.
|
||||
hostB = _devices_for_host(cfg, "192.168.1.1", "hostB", host_id="b")
|
||||
hostA = _devices_for_host(cfg, "10.0.0.2", "hostA", host_id="a")
|
||||
assert hostB == []
|
||||
assert [d["name"] for d in hostA] == ["sw"]
|
||||
|
||||
|
||||
def test_explicit_host_id_wrong_uuid_does_not_match():
|
||||
cfg = [{"name": "sw", "host": "10.0.0.5", "host_id": "uuid-abc"}]
|
||||
out = _devices_for_host(cfg, "10.0.0.5", "sw", host_id="uuid-zzz")
|
||||
assert out == []
|
||||
@@ -1,44 +1,125 @@
|
||||
"""Unit tests for the SNMP poller (no pysnmp / no network).
|
||||
"""The SNMP poller (milestone 491), against a stand-in for pysnmp.
|
||||
|
||||
The unit lane doesn't install the `snmp` extra, so these exercise the parts that
|
||||
don't need pysnmp: the version→mpModel mapping, that the poller is async, and the
|
||||
graceful "pysnmp missing → empty result" path. Live polling against a real device
|
||||
is verified out-of-band (CI has no SNMP target). Uses asyncio.run() directly so it
|
||||
doesn't depend on the pytest-asyncio mode.
|
||||
The unit lane doesn't install the `snmp` extra, so a fake API answers in
|
||||
pysnmp's shape: one GET per device, answers in request order, an agent's "no
|
||||
such object" as a value of that type, v1's whole-request error with an index,
|
||||
and a timeout as an error indication. Live polling is read from the instance.
|
||||
"""
|
||||
import asyncio
|
||||
import inspect
|
||||
from types import SimpleNamespace as NS
|
||||
|
||||
from plugins.snmp import poller
|
||||
from plugins.snmp.poller import Credentials
|
||||
|
||||
|
||||
def test_poll_device_is_coroutine():
|
||||
# The scheduler awaits it directly (no executor) — it must be async.
|
||||
class NoSuchObject:
|
||||
pass
|
||||
|
||||
|
||||
class _Status:
|
||||
def __init__(self, text):
|
||||
self.text = text
|
||||
|
||||
def __bool__(self):
|
||||
return True
|
||||
|
||||
def prettyPrint(self):
|
||||
return self.text
|
||||
|
||||
|
||||
def _fake_api(answer, calls):
|
||||
"""answer(oids) -> (error_indication, error_status, error_index, values)."""
|
||||
async def get(engine, auth, transport, ctx, *objs, **opts):
|
||||
oids = [o for o in objs]
|
||||
calls.append((auth, oids, opts))
|
||||
ei, es, eidx, values = answer(oids)
|
||||
return ei, es, eidx, [(oid, v) for oid, v in zip(oids, values)]
|
||||
|
||||
class Transport:
|
||||
def __init__(self, addr, **kw):
|
||||
self.addr = addr
|
||||
|
||||
hl = NS(SnmpEngine=lambda: NS(), ContextData=lambda: "ctx", ObjectType=lambda x: x,
|
||||
ObjectIdentity=lambda oid: oid, UdpTransportTarget=Transport,
|
||||
CommunityData=lambda c, mpModel: ("community", c, mpModel),
|
||||
UsmUserData=lambda user, **kw: ("usm", user, kw))
|
||||
auth = NS(**{name: name for name in list(poller.AUTH_PROTOCOLS.values()) + list(poller.PRIV_PROTOCOLS.values())})
|
||||
return NS(hl=hl, auth=auth, get=get, walk=None, async_transport=False)
|
||||
|
||||
|
||||
def _poll(monkeypatch, answer, oids, creds=None):
|
||||
calls = []
|
||||
monkeypatch.setattr(poller, "_pysnmp_available", lambda: True)
|
||||
monkeypatch.setattr(poller, "_api", lambda: _fake_api(answer, calls))
|
||||
result = asyncio.run(poller.poll_device("10.0.0.2", 161, creds or Credentials(community="public"), oids))
|
||||
return result, calls
|
||||
|
||||
|
||||
def test_poll_device_is_a_coroutine():
|
||||
assert inspect.iscoroutinefunction(poller.poll_device)
|
||||
|
||||
|
||||
def test_mp_model_maps_version_to_int():
|
||||
assert poller._mp_model("1") == 0 # SNMP v1
|
||||
assert poller._mp_model("2c") == 1 # SNMP v2c
|
||||
assert poller._mp_model("2") == 1 # anything non-"1" → v2c model
|
||||
def test_one_get_carries_every_oid_and_answers_come_back_by_key(monkeypatch):
|
||||
result, calls = _poll(monkeypatch, lambda oids: (None, None, None, [4200.0, NoSuchObject(), "eth0"]),
|
||||
[("1.1", "uptime"), ("1.2", "fan"), ("1.3", "label")])
|
||||
assert len(calls) == 1 and calls[0][1] == ["1.1", "1.2", "1.3"] and calls[0][2] == {"lookupMib": False}
|
||||
assert calls[0][0] == ("community", "public", 1)
|
||||
assert result.readings == {"uptime": 4200.0} and result.missing == ["fan", "label"] and result.error is None
|
||||
|
||||
|
||||
def test_poll_device_without_pysnmp_returns_empty(monkeypatch):
|
||||
# When the optional dep is absent, polling is disabled gracefully (no raise).
|
||||
def test_v1_drops_the_oid_it_does_not_know_and_asks_again(monkeypatch):
|
||||
def answer(oids):
|
||||
if "1.2" in oids:
|
||||
return None, _Status("noSuchName"), oids.index("1.2") + 1, []
|
||||
return None, None, None, [float(len(o)) for o in oids]
|
||||
result, calls = _poll(monkeypatch, answer, [("1.1", "a"), ("1.2", "b"), ("1.33", "c")],
|
||||
Credentials(version="1", community="public"))
|
||||
assert len(calls) == 2 and calls[0][0] == ("community", "public", 0)
|
||||
assert result.readings == {"a": 3.0, "c": 4.0} and result.missing == ["b"]
|
||||
|
||||
|
||||
def test_no_reply_says_what_to_check(monkeypatch):
|
||||
result, _ = _poll(monkeypatch, lambda oids: ("No SNMP response received before timeout", None, None, []),
|
||||
[("1.1", "uptime")])
|
||||
assert result.readings == {} and result.error.startswith("No reply from 10.0.0.2:161 within 5 s")
|
||||
assert "the community" in result.error
|
||||
|
||||
|
||||
def test_an_answer_without_any_reading_is_an_error(monkeypatch):
|
||||
result, _ = _poll(monkeypatch, lambda oids: (None, None, None, [NoSuchObject()]), [("1.1", "x")])
|
||||
assert result.error == "The device answered but has none of these readings." and result.missing == ["x"]
|
||||
|
||||
|
||||
def test_v3_signs_in_as_its_user_with_the_chosen_protocols(monkeypatch):
|
||||
creds = Credentials(version="3", user="mon", auth="sha256", auth_key="authpass1",
|
||||
priv="aes128", priv_key="privpass1")
|
||||
_, calls = _poll(monkeypatch, lambda oids: (None, None, None, [1.0]), [("1.1", "x")], creds)
|
||||
kind, user, kw = calls[0][0]
|
||||
assert (kind, user) == ("usm", "mon")
|
||||
assert kw == {"authKey": "authpass1", "privKey": "privpass1",
|
||||
"authProtocol": "usmHMAC192SHA256AuthProtocol", "privProtocol": "usmAesCfb128Protocol"}
|
||||
|
||||
|
||||
def test_without_pysnmp_the_poll_says_so(monkeypatch):
|
||||
monkeypatch.setattr(poller, "_pysnmp_available", lambda: False)
|
||||
out = asyncio.run(
|
||||
poller.poll_device("192.0.2.1", 161, "public", "2c", [{"oid": "1.3.6.1.2.1.1.3.0"}])
|
||||
)
|
||||
assert out == {}
|
||||
result = asyncio.run(poller.poll_device("10.0.0.2", 161, Credentials(), [("1.1", "x")]))
|
||||
assert result.readings == {} and "pysnmp" in result.error
|
||||
|
||||
|
||||
def test_errors_read_as_what_to_check():
|
||||
v3 = Credentials(version="3", user="mon")
|
||||
assert "doesn't know the v3 user “mon”" in poller.explain("Unknown USM user", "h", 161, v3)
|
||||
assert "authentication" in poller.explain("Wrong SNMP PDU digest", "h", 161, v3)
|
||||
assert "the v3 user “mon”" in poller.explain("No SNMP response received before timeout", "h", 161, v3)
|
||||
assert poller.explain("Something odd", "h", 161, v3) == "SNMP error from h:161: Something odd"
|
||||
|
||||
|
||||
# --- _close_engine: the fd-leak guard ---------------------------------------
|
||||
# A fresh SnmpEngine per poll leaks its UDP socket unless closed; _close_engine
|
||||
# must release it across both pysnmp API shapes. These fakes stand in for the
|
||||
# engine since the unit lane has no pysnmp.
|
||||
# must release it across both pysnmp API shapes.
|
||||
|
||||
def test_close_engine_uses_engine_level_close_7x():
|
||||
# canonical pysnmp 7.x: close method lives directly on the engine.
|
||||
calls = []
|
||||
|
||||
class Engine:
|
||||
@@ -50,7 +131,6 @@ def test_close_engine_uses_engine_level_close_7x():
|
||||
|
||||
|
||||
def test_close_engine_falls_back_to_dispatcher_6x():
|
||||
# pysnmp-lextudio 6.2.x: no engine-level close; go through the dispatcher.
|
||||
calls = []
|
||||
|
||||
class Dispatcher:
|
||||
@@ -65,12 +145,9 @@ def test_close_engine_falls_back_to_dispatcher_6x():
|
||||
|
||||
|
||||
def test_close_engine_never_raises():
|
||||
# A failing close must not break the poll loop.
|
||||
class Engine:
|
||||
def close_dispatcher(self):
|
||||
raise OSError("boom")
|
||||
|
||||
poller._close_engine(Engine()) # no exception
|
||||
|
||||
# No close path at all (defensive) is also fine.
|
||||
poller._close_engine(Engine())
|
||||
poller._close_engine(object())
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
"""What an SNMP device is asked for and how each answer reads (milestone 491)."""
|
||||
import pytest
|
||||
|
||||
from plugins.snmp.readings import (device_readings, format_value, interface_readings, parse_custom,
|
||||
parse_interfaces, stored_unit)
|
||||
|
||||
|
||||
def test_interface_numbers_read_as_a_list():
|
||||
assert parse_interfaces("1, 2 5-7,2") == [1, 2, 5, 6, 7]
|
||||
assert parse_interfaces("") == []
|
||||
for bad in ("eth0", "7-3", "1-100"):
|
||||
with pytest.raises(ValueError):
|
||||
parse_interfaces(bad)
|
||||
|
||||
|
||||
def test_interfaces_read_64_bit_counters_except_on_v1():
|
||||
v2 = {r.key: r for r in interface_readings([3], "2c")}
|
||||
v1 = {r.key: r for r in interface_readings([3], "1")}
|
||||
assert set(v2) == {"if3_status", "if3_in", "if3_out"}
|
||||
assert v2["if3_in"].oid == "1.3.6.1.2.1.31.1.1.1.6.3" and v2["if3_in"].counter
|
||||
assert v1["if3_in"].oid == "1.3.6.1.2.1.2.2.1.10.3"
|
||||
assert stored_unit(v2["if3_out"]) == "bits_per_second" and not v2["if3_status"].counter
|
||||
|
||||
|
||||
def test_a_custom_line_is_an_oid_a_name_and_a_unit():
|
||||
got = parse_custom("# a comment\n.1.3.6.1.4.1.9.1.0 fan_speed\n1.3.6.1.4.1.9.2.0 inlet_temp celsius\n")
|
||||
assert [(r.oid, r.key, r.name, r.unit) for r in got] == [
|
||||
("1.3.6.1.4.1.9.1.0", "fan_speed", "Fan speed", "number"),
|
||||
("1.3.6.1.4.1.9.2.0", "inlet_temp", "Inlet temp", "celsius")]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line, says", [
|
||||
("1.3.6.1", "OID name"),
|
||||
("sysUpTime uptime", "is not an OID"),
|
||||
("1.3.6.1.2 Bad-Name", "must start with a letter"),
|
||||
("1.3.6.1.2 temp furlongs", "is not a unit"),
|
||||
])
|
||||
def test_a_custom_line_that_does_not_read_says_which_and_why(line, says):
|
||||
with pytest.raises(ValueError, match="Line 1") as exc:
|
||||
parse_custom(line)
|
||||
assert says in str(exc.value)
|
||||
|
||||
|
||||
def test_presets_come_in_order_and_names_may_not_repeat():
|
||||
keys = [r.key for r in device_readings(["ups", "system", "interfaces"], [1], "", "2c")]
|
||||
assert keys[:4] == ["uptime", "if1_status", "if1_in", "if1_out"] and "battery_charge" in keys
|
||||
with pytest.raises(ValueError, match="uptime"):
|
||||
device_readings(["system"], [], "1.3.6.1.2.1.1.3.0 uptime", "2c")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("unit, value, text", [
|
||||
("centiseconds", 105_120_000, "12d 4h"),
|
||||
("percent", 99.6, "100%"),
|
||||
("volts", 121.0, "121 V"),
|
||||
("decivolts", 1203, "120.3 V"),
|
||||
("minutes", 45, "45m"),
|
||||
("kilobytes", 2048, "2.0 MiB"),
|
||||
("bits_per_second", 12_345_678, "12.3 Mbit/s"),
|
||||
("bits_per_second", 950, "950 bit/s"),
|
||||
("if_status", 2, "down"),
|
||||
("ups_battery", 3, "low"),
|
||||
("if_status", 99, "code 99"),
|
||||
("number", 41.25, "41.25"),
|
||||
("number", None, "—"),
|
||||
])
|
||||
def test_a_value_reads_in_its_unit(unit, value, text):
|
||||
assert format_value(unit, value) == text
|
||||
@@ -0,0 +1,118 @@
|
||||
"""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
|
||||
@@ -1,39 +1,96 @@
|
||||
"""The SNMP list says each device's state with the shared badge and shows its
|
||||
readings as stat tiles (#5126)."""
|
||||
"""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 _device_state
|
||||
from plugins.snmp.routes import devices_for_host
|
||||
from tests._templates import render_plugin_template
|
||||
from tests.test_style_tokens import hex_colours
|
||||
|
||||
|
||||
async def test_devices_wear_a_status_badge_and_readings_are_tiles(app):
|
||||
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 = [
|
||||
{"name": "ups", "host": "10.0.0.9", "oids": [{"oid": "1.3.6", "label": "battery_charge_pct"},
|
||||
{"oid": "1.3.7", "label": "in_octets"}],
|
||||
"readings": {"battery_charge_pct": 97.0, "in_octets": 2_500_000.0}, "reach": "up",
|
||||
"at": datetime.now(timezone.utc) - timedelta(seconds=30)},
|
||||
{"name": "switch", "host": "10.0.0.11", "oids": [], "readings": {"x": 1.0}, "reach": "stale",
|
||||
"at": datetime.now(timezone.utc) - timedelta(hours=3)},
|
||||
{"name": "printer", "host": "10.0.0.10", "oids": [], "readings": {}, "reach": "pending", "at": None},
|
||||
_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/"):
|
||||
page = await render_plugin_template(app, "snmp", "snmp/index.html", devices=devices, poll_interval=60)
|
||||
|
||||
assert '<span class="badge badge-ok">Reachable</span>' in page
|
||||
assert '<span class="badge badge-warn">No data yet</span>' in page
|
||||
# Old readings are not "reachable": they say how old they are (#5210).
|
||||
assert 'badge badge-warn" title="The last reading is from' in page and ">No answer</span>" in page
|
||||
assert ">3h ago</time>" in page
|
||||
assert '<span class="stat-tile-value">97</span>' in page and '<span class="stat-tile-value">2.50M</span>' in page
|
||||
assert 'href="/plugins/snmp/device/ups"' in page
|
||||
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)
|
||||
|
||||
|
||||
def test_a_device_is_reachable_only_while_its_readings_are_recent():
|
||||
now = datetime.now(timezone.utc)
|
||||
entry = {"readings": {"x": 1.0}, "at": now - timedelta(seconds=150)}
|
||||
assert _device_state(entry, 60, now)["reach"] == "up" # within 3 polls
|
||||
assert _device_state(entry, 30, now)["reach"] == "stale" # 5 polls without an answer
|
||||
assert _device_state(None, 60, now) == {"readings": {}, "at": None, "reach": "pending"}
|
||||
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) == []
|
||||
|
||||
@@ -1,56 +1,37 @@
|
||||
"""The dashboard SNMP widget (#5255): a verdict, devices that stopped answering
|
||||
first, status in words, and key readings with their units."""
|
||||
"""The dashboard SNMP widget (#5255, milestone 491): a verdict, devices not
|
||||
answering first with since when, status in words, two readings in units."""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from plugins.snmp.routes import key_readings, snmp_reading
|
||||
from tests._templates import render_plugin_template
|
||||
|
||||
|
||||
@pytest.mark.parametrize("label, value, says", [
|
||||
("uptime_centisecs", 105_120_000, ("uptime", "12d 4h")),
|
||||
("battery_charge_pct", 99.6, ("battery charge", "100%")),
|
||||
("input_voltage_dV", 1203, ("input voltage", "120.3 V")),
|
||||
("if1_in_octets", 1536, ("if1 in", "1.5 KiB total")),
|
||||
("battery_status", 2, ("battery status", "2")),
|
||||
("temperature", 41.25, ("temperature", "41.25")),
|
||||
])
|
||||
def test_a_reading_reads_with_its_unit(label, value, says):
|
||||
assert snmp_reading(label, value) == says
|
||||
|
||||
|
||||
def test_key_readings_prefer_gauges_to_counters_in_config_order():
|
||||
oids = [{"oid": "a", "label": "if1_in_octets"}, {"oid": "b", "label": "uptime_centisecs"},
|
||||
{"oid": "c", "label": "if1_out_octets"}, {"oid": "1.2.3"}]
|
||||
readings = {"if1_in_octets": 10, "uptime_centisecs": 6000, "if1_out_octets": 20, "1.2.3": 7}
|
||||
assert [n for n, _ in key_readings(oids, readings)] == ["uptime", "1.2.3"]
|
||||
assert [n for n, _ in key_readings(oids[:1] + oids[2:3], readings)] == ["if1 in", "if1 out"]
|
||||
assert key_readings(oids, {}) == []
|
||||
|
||||
|
||||
def _dev(name, reach, minutes_ago=None, key=()):
|
||||
at = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago) if minutes_ago is not None else None
|
||||
return {"name": name, "reach": reach, "at": at, "readings": {}, "key": list(key)}
|
||||
def _dev(name, reach, ok_minutes=None, key=()):
|
||||
now = datetime.now(timezone.utc)
|
||||
return {"id": f"id-{name}", "name": name, "reach": reach, "says": "No reply.",
|
||||
"polled_at": now - timedelta(seconds=30),
|
||||
"ok_at": now - timedelta(minutes=ok_minutes) if ok_minutes is not None else None,
|
||||
"key": [{"name": n, "text": t} for n, t in key]}
|
||||
|
||||
|
||||
async def test_a_silent_device_leads_with_since_when(app):
|
||||
devices = [_dev("ups", "stale", 14, [("battery charge", "100%")]), _dev("printer", "pending"),
|
||||
_dev("switch", "up", 1, [("uptime", "12d 4h")])]
|
||||
devices = [_dev("ups", "down", 14, [("Charge", "100%")]), _dev("printer", "pending"),
|
||||
_dev("switch", "up", 1, [("Uptime", "12d 4h")])]
|
||||
async with app.test_request_context("/plugins/snmp/widget"):
|
||||
page = await render_plugin_template(app, "snmp", "snmp/widget.html", devices=devices, more=0,
|
||||
counts={"stale": 1, "pending": 1, "up": 1}, total=3)
|
||||
counts={"down": 1, "stale": 0, "pending": 1, "up": 1}, total=3)
|
||||
page = " ".join(page.split())
|
||||
assert "ups: no reply since" in page and ">14m ago<" in page
|
||||
assert "ups isn't answering." in page and "no reply since" in page and ">14m ago<" in page
|
||||
assert page.count("<th>Device</th>") == 1
|
||||
assert page.index('title="ups"') < page.index('title="printer"') < page.index('title="switch"')
|
||||
assert "no data yet" in page and "uptime</span> 12d 4h" in page
|
||||
assert "detail →" not in page and "○" not in page
|
||||
assert "not polled yet" in page and "Uptime</span> 12d 4h" in page
|
||||
|
||||
|
||||
async def test_all_answering_says_so(app):
|
||||
async with app.test_request_context("/plugins/snmp/widget"):
|
||||
page = await render_plugin_template(app, "snmp", "snmp/widget.html",
|
||||
devices=[_dev("switch", "up", 1)], more=0,
|
||||
counts={"stale": 0, "pending": 0, "up": 1}, total=1)
|
||||
counts={"down": 0, "stale": 0, "pending": 0, "up": 1}, total=1)
|
||||
empty = await render_plugin_template(app, "snmp", "snmp/widget.html", devices=[], more=0,
|
||||
counts={}, total=0)
|
||||
assert "All 1 device answering." in page and "—" in page
|
||||
assert "No SNMP devices yet." in empty
|
||||
|
||||
Reference in New Issue
Block a user