A supply nobody could read is unknown, not empty

tests/test_plugins/test_supply_unknown_level.py

@@ -0,0 +1,163 @@
+"""A supply nobody could read is unknown, not empty.
+
+Reported as "CSF04 now reports all 0%", on the printer's own page as well as the
+reports. Every supply on one host reading zero at the same moment is not what a
+set of cartridges does; it is what a host that stopped answering looks like.
+
+`getsuppliesbyip` turned every unreadable item into 0:
+
+    try:
+        level = int(float(item.get("lastvalue", 0)))
+    except (ValueError, TypeError):
+        level = 0
+
+An item that has never collected returns lastvalue as an EMPTY STRING, so
+float('') raised and the handler substituted zero. `lastclock` was fetched in
+the same request and never read by anything, so a reading from three weeks ago
+was presented as current.
+
+Zero is a legitimate level meaning "spent", which is what made the substitution
+dangerous rather than merely wrong: it reached `level <= EMPTY_LEVEL` in the
+forecast, which set daysleft to 0, which put the printer on the order list. The
+report asked purchasing to buy a full set of cartridges for a printer nobody had
+measured.
+
+None means unknown, and it has to stay None the whole way out.
+"""
+
+import time
+
+import pytest
+
+from plugins.printers.services.zabbix_service import ZabbixService
+from plugins.printers.services.supply_parts import classifysupply
+from plugins.printers.services.supply_history import analyse, EMPTY_LEVEL
+
+
+def item(lastvalue, lastclock=None, name='Black Cartridge'):
+    """One Zabbix item.get row, shaped as the API returns it."""
+    if lastclock is None:
+        lastclock = int(time.time()) - 600
+    return {'itemid': '1', 'name': name,
+            'lastvalue': lastvalue, 'lastclock': str(lastclock)}
+
+
+# --- reading one item -------------------------------------------------------
+
+def test_an_item_that_never_collected_is_unknown_not_zero():
+    """The reported case. Zabbix returns '' for an item with no data."""
+    level, lastseen = ZabbixService._readlevel(item(''))
+    assert level is None
+    assert lastseen is None
+
+
+def test_a_missing_lastvalue_is_unknown_not_zero():
+    level, _ = ZabbixService._readlevel({'lastvalue': None, 'lastclock': '0'})
+    assert level is None
+
+
+def test_an_unparseable_lastvalue_is_unknown_not_zero():
+    level, _ = ZabbixService._readlevel(item('n/a'))
+    assert level is None
+
+
+def test_a_real_zero_is_still_zero():
+    """The whole point of the distinction: an empty cartridge must survive it.
+
+    If this returned None, an actually-spent cartridge would stop being ordered.
+    """
+    level, _ = ZabbixService._readlevel(item('0'))
+    assert level == 0
+
+
+def test_a_normal_reading_comes_back_with_its_clock():
+    when = int(time.time()) - 300
+    level, lastseen = ZabbixService._readlevel(item('47', lastclock=when))
+    assert level == 47
+    assert lastseen == when
+
+
+def test_a_float_reading_is_truncated_to_an_int():
+    level, _ = ZabbixService._readlevel(item('47.8'))
+    assert level == 47
+
+
+# --- staleness --------------------------------------------------------------
+
+def test_a_reading_older_than_the_window_is_unknown():
+    """lastclock was fetched and then ignored, so a host that went quiet weeks
+    ago kept presenting its last level as current."""
+    stale = int(time.time()) - (ZabbixService.STALE_AFTER_SECONDS + 60)
+    level, lastseen = ZabbixService._readlevel(item('47', lastclock=stale))
+    assert level is None
+    assert lastseen == stale, 'the age must survive, so the UI can say how old'
+
+
+def test_a_reading_inside_the_window_is_kept():
+    recent = int(time.time()) - (ZabbixService.STALE_AFTER_SECONDS - 3600)
+    level, _ = ZabbixService._readlevel(item('47', lastclock=recent))
+    assert level == 47
+
+
+def test_a_value_with_no_clock_is_trusted():
+    """A reading that cannot be aged is not thereby wrong. Discarding it would
+    lose real levels wherever lastclock is absent."""
+    level, lastseen = ZabbixService._readlevel(
+        {'lastvalue': '47', 'lastclock': '0'})
+    assert level == 47
+    assert lastseen is None
+
+
+# --- classification ---------------------------------------------------------
+
+def test_an_unknown_level_classifies_as_unknown():
+    result = classifysupply(None, 'Black Cartridge', 'HP')
+    assert result['status'] == 'unknown'
+    assert result['remaining'] is None
+
+
+def test_an_unknown_level_is_not_critical():
+    """'critical' is a claim someone acts on. It may not be made about a supply
+    nobody measured."""
+    assert classifysupply(None, 'Black Cartridge', 'HP')['status'] != 'critical'
+
+
+def test_a_real_zero_is_still_critical():
+    assert classifysupply(0, 'Black Cartridge', 'HP')['status'] == 'critical'
+
+
+def test_an_unknown_waste_cartridge_does_not_invert_into_full():
+    """Waste is scored as 100 - level. With level 0 standing in for unknown,
+    an unreadable waste cartridge read as 100% full, which is the alarm state."""
+    result = classifysupply(None, 'Waste Toner Container', 'HP')
+    assert result['status'] == 'unknown'
+    assert result['remaining'] is None
+
+
+# --- the forecast -----------------------------------------------------------
+
+def test_an_unknown_level_with_no_history_gets_no_countdown():
+    """0 reached `level <= EMPTY_LEVEL` and became daysleft 0 - the step that
+    put CSF04 on the order list."""
+    result = analyse([], currentlevel=None)
+    assert result['daysleft'] is None
+    assert result['reason'] == 'no history'
+
+
+def test_a_real_zero_with_no_history_still_counts_as_due_now():
+    assert analyse([], currentlevel=0)['daysleft'] == 0
+    assert EMPTY_LEVEL >= 0
+
+
+def test_an_unknown_level_falls_back_to_the_last_real_reading():
+    """History is evidence the live read lacks. Preferring it beats inventing a
+    level, and beats throwing the cartridge off the report.
+
+    Clock is unix seconds and level falls over time, which is the only direction
+    toner goes; a rising series would read as a cartridge change.
+    """
+    now = int(time.time())
+    points = [(now - n * 86400, 50 + n * 2) for n in range(20, -1, -1)]
+    result = analyse(points, currentlevel=None)
+    assert result['currentlevel'] == 50, 'the newest stored reading'
+    assert result['daysleft'] is not None, 'a forecast is still possible'