FabledSteward/tests/plugins/snmp/test_readings.py

"""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", "if3_in_errors", "if3_out_errors",
                       "if3_in_discards", "if3_out_discards"}
    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
    assert v2["if3_in_errors"].oid == "1.3.6.1.2.1.2.2.1.14.3" and stored_unit(v2["if3_in_errors"]) == "per_second"


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"),
    ("bytes_per_second", 3 * 1024 * 1024, "3.0 MiB/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) == "—"


def test_cpu_waiting_on_disk_is_the_wait_counter_s_share_of_the_cpu_s():
    """#5997: asked for as its five tick counters, kept as one percentage."""
    linux = {r.key: r for r in device_readings(["linux"], [], "", "2c")}
    wait = linux["cpu_iowait"]
    assert wait.oid == "" and stored_unit(wait) == "percent"
    assert [i.oid for i in wait.inputs] == [f"1.3.6.1.4.1.2021.11.{n}.0" for n in (50, 51, 52, 53, 54)]
    assert all(i.counter for i in wait.inputs)
    ticks = {"cpu_ticks_user": 100.0, "cpu_ticks_nice": 0.0, "cpu_ticks_system": 50.0,
             "cpu_ticks_idle": 800.0, "cpu_ticks_wait": 50.0}
    assert wait.combine(ticks) == 5.0
    assert wait.combine(dict.fromkeys(ticks, 0.0)) is None             # no ticks between two polls
    assert linux["memory_available"].combine(
        {"memory_free_kb": 1000.0, "memory_buffers_kb": 200.0, "memory_cached_kb": 3000.0}) == 4200.0


def test_a_synology_disk_and_volume_read_their_io_and_the_disk_its_reconnects():
    tables = {t.id: t for t in TABLES["synology"]}
    io = table_readings(tables["synology_disk_io"], [{"index": "1", "label": "sata1",
                                                      "columns": ["read", "writes", "load"]}])
    assert [(r.oid, r.key, r.name, stored_unit(r)) for r in io] == [
        ("1.3.6.1.4.1.6574.101.1.1.12.1", "storageio1_read", "sata1 read", "bytes_per_second"),
        ("1.3.6.1.4.1.6574.101.1.1.6.1", "storageio1_writes", "sata1 writes", "per_second"),
        ("1.3.6.1.4.1.6574.101.1.1.8.1", "storageio1_load", "sata1 load", "percent")]
    assert tables["synology_disk_io"].keeps("sata2") and not tables["synology_disk_io"].keeps("md2")
    assert tables["synology_volume_io"].columns[0].oid == "1.3.6.1.4.1.6574.102.1.1.12"
    disks = {c.key: c.oid for c in tables["synology_disks"].columns}
    assert disks["retries"] == "1.3.6.1.4.1.6574.2.1.1.8" and disks["identify_fails"] == "1.3.6.1.4.1.6574.2.1.1.10"