bvandeusenandClaude Opus 5.5
bd1c32d2a4
The table walk stopped after 64 raw entries, counting the RAM disks, loops,
partitions and md arrays that rows_like then drops, so a Synology's eighth
disk was never reached. The walk is now bounded at 1,024 entries per column;
64 still caps the rows kept.
ADM labels its disks "Disk0".."Disk4" while their indexes run 1..5; the
Asustor disk table now names each row by its index ("Disk 1"). Keys are
unchanged, so history carries on.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
143 lines
6.7 KiB
Python
143 lines
6.7 KiB
Python
"""What an SNMP device is asked for and how each answer reads (milestone 491)."""
|
|
import pytest
|
|
|
|
from plugins.snmp.readings import (TABLES, Column, Reading, Table, device_readings, format_value,
|
|
interface_readings, parse_custom, parse_interfaces, reading_text, state_of,
|
|
stored_unit, table_readings, tables_for)
|
|
|
|
|
|
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"),
|
|
("battery_seconds", 0, "No, on mains"),
|
|
("battery_seconds", 300, "for 5m"),
|
|
("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, "—"),
|
|
("state", 2, "Warning"),
|
|
("rpm", 1450, "1,450 rpm"),
|
|
])
|
|
def test_a_value_reads_in_its_unit(unit, value, text):
|
|
assert format_value(unit, value) == text
|
|
|
|
|
|
_DISKS = Table("d", "disk", "9.2", "Disk", (Column("9.5", "status", "status", "state", states="words"),
|
|
Column("9.6", "temp", "temperature", "celsius")))
|
|
|
|
|
|
def test_a_table_row_is_keyed_by_its_index_and_named_by_its_label():
|
|
got = table_readings(_DISKS, [{"index": "0", "label": "Disk 1", "columns": ["status", "temp"]},
|
|
{"index": "1", "label": "2", "columns": ["temp"]}])
|
|
assert [(r.oid, r.key, r.name, r.unit) for r in got] == [
|
|
("9.5.0", "disk0_status", "Disk 1 status", "state"),
|
|
("9.6.0", "disk0_temp", "Disk 1 temperature", "celsius"),
|
|
("9.6.1", "disk1_temp", "Disk 2 temperature", "celsius")] # a bare bay number gets its noun
|
|
assert got[0].states == "words"
|
|
|
|
|
|
def test_an_asustor_disk_is_named_by_its_bay_not_adms_zero_based_label():
|
|
disks = next(t for t in TABLES["asustor"] if t.id == "asustor_disks")
|
|
got = table_readings(disks, [{"index": "1", "label": "Disk0", "columns": ["status"]}])
|
|
assert [(r.key, r.name) for r in got] == [("disk1_status", "Disk 1 status")]
|
|
|
|
|
|
def test_a_table_without_labels_is_named_by_its_columns():
|
|
fans = Table("f", "", "", "Fan", (Column("9.2", "fan", "Fan speed", "rpm"),))
|
|
[one] = table_readings(fans, [{"index": "1", "label": "", "columns": ["fan"]}])
|
|
assert (one.key, one.name) == ("fan", "Fan speed")
|
|
two = table_readings(fans, [{"index": "1", "label": "", "columns": ["fan"]},
|
|
{"index": "2", "label": "", "columns": ["fan"]}])
|
|
assert [(r.key, r.name) for r in two] == [("fan1", "Fan speed (fan 1)"), ("fan2", "Fan speed (fan 2)")]
|
|
|
|
|
|
def test_rows_come_after_the_presets_only_for_ticked_presets_and_a_custom_name_wins():
|
|
rows = {"synology_volumes": [{"index": "0", "label": "Volume 1", "columns": ["status", "free"]}],
|
|
"asustor_fans": [{"index": "1", "label": "", "columns": ["fan"]}]}
|
|
keys = [r.key for r in device_readings(["system", "synology"], [], "1.3.6.1.9.9 raid0_free bytes", "2c", rows)]
|
|
assert keys[0] == "uptime" and "system_status" in keys and "fan" not in keys
|
|
assert keys.index("raid0_status") > keys.index("cpu_fan") and keys.count("raid0_free") == 1
|
|
assert keys[-1] == "raid0_free" # the custom line, not the row
|
|
assert [t.id for t in tables_for(["asustor", "linux"])] == [
|
|
"ucd_diskio", "asustor_disks", "asustor_volumes", "asustor_temperatures", "asustor_memory", "asustor_fans"]
|
|
|
|
|
|
def test_only_physical_disks_are_read_for_disk_load():
|
|
[diskio] = TABLES["linux"]
|
|
assert diskio.keeps("sda") and diskio.keeps("nvme0n1") and diskio.keeps("sata2")
|
|
assert not diskio.keeps("sda1") and not diskio.keeps("loop0") and not diskio.keeps("md2")
|
|
|
|
|
|
@pytest.mark.parametrize("states, raw, want", [
|
|
("synology_raid", 13, ("Data scrubbing", 1)),
|
|
("synology_raid", 11.0, ("Degraded", 2)),
|
|
("synology_raid", 1, ("Normal", 0)),
|
|
("synology_disk", 5, ("Crashed", 3)),
|
|
("synology_raid", 99, ("code 99", None)),
|
|
("words", " Normal ", ("Normal", 0)),
|
|
("words", "Healthy", ("Healthy", 0)),
|
|
("words", "Unhealthy", ("Unhealthy", 3)),
|
|
("words", "Degraded", ("Degraded", 2)),
|
|
("words", "Resyncing", ("Resyncing", 1)),
|
|
("words", "Broken", ("Broken", 3)),
|
|
("words", "Frobnicated", ("Frobnicated", None)),
|
|
])
|
|
def test_a_vendor_state_reads_as_its_word_and_a_level(states, raw, want):
|
|
assert state_of(Reading("1.1", "s", "S", "state", states=states), raw) == want
|
|
|
|
|
|
def test_a_state_shows_the_device_word_and_falls_back_to_its_level():
|
|
r = Reading("1.1", "raid0_status", "Volume 1 status", "state", states="synology_raid")
|
|
assert reading_text(r, 1.0, {"raid0_status": "Data scrubbing"}) == "Data scrubbing"
|
|
assert reading_text(r, 1.0, {}) == "Busy" and reading_text(r, None, None) == "—"
|