d109314123
The toner report says what is empty now. It could not say what to order, and nothing recorded how fast anything drains - every level read was cached for five minutes and then discarded. Zabbix has been keeping the history all along; we simply never asked. One history.get gives both answers, because a cartridge only goes DOWN while it is in use: a rise is a replacement. Count the rises and you have how many cartridges a printer has been through; fit a slope to the readings SINCE the last rise and you have days-to-empty. Fitting across a replacement averages a spent cartridge with a fresh one and describes neither. Sorted by days left, which is the point. A cartridge at 60% dropping 5% a day needs ordering before one sitting at 8% that has not moved in months, and a level-sorted list ranks those backwards. It refuses to guess. Too few readings, a level that has not moved enough - many printers report in 10% steps and sit on a plateau for a fortnight - or a recent replacement each produce no estimate and say which. Those printers are listed separately rather than sorted in as 0 or as 999, since a printer without an estimate is neither urgent nor safe. Estimates show what they rest on, because "9 days from 21 days of readings" and "9 days from 2 readings" are not the same claim. A separate report card, not an extension of the toner report: that one is an exceptions list a tech acts on today, this is an ordering view read monthly, and the history query is heavier than the live read it would have slowed down. The analysis is pure arithmetic over a list of readings, so the 14 tests cover the noise wobble, the plateau, the swap, junk rows and division by zero without needing Zabbix. Zabbix being unreachable is reported as such rather than rendering an empty table that reads as "nothing is due".
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""Tests for toner burn-rate and replacement counting.
|
|
|
|
The analysis is pure arithmetic over a list of readings, so it is tested that
|
|
way - no Zabbix, no fixtures. What matters is that it refuses to guess: the
|
|
cases where an estimate would be dishonest are the ones most likely to reach a
|
|
purchasing decision.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from plugins.printers.services.supply_history import (
|
|
analyse, burn_rate, current_run, find_replacements, normalise, soonest,
|
|
)
|
|
|
|
START = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
def series(levels, hours=24):
|
|
"""[(clock, value)] one reading per `hours`, as Zabbix returns them."""
|
|
return [(str(int((START + timedelta(hours=i * hours)).timestamp())), str(v))
|
|
for i, v in enumerate(levels)]
|
|
|
|
|
|
def test_counts_one_replacement_per_upward_step():
|
|
points = normalise(series([80, 60, 40, 20, 100, 80, 60, 95, 70]))
|
|
|
|
assert len(find_replacements(points)) == 2
|
|
|
|
|
|
def test_ignores_a_wobble_that_is_not_a_replacement():
|
|
"""SNMP rounding and a gauge settling both nudge a reading upward."""
|
|
points = normalise(series([60, 58, 61, 57, 55]))
|
|
|
|
assert find_replacements(points) == []
|
|
|
|
|
|
def test_estimate_uses_only_the_current_cartridge():
|
|
"""Fitting across a swap averages a spent cartridge with a fresh one."""
|
|
points = normalise(series([90, 60, 30, 100, 90, 80, 70]))
|
|
|
|
run = current_run(points)
|
|
|
|
assert [level for _, level in run] == [100, 90, 80, 70]
|
|
|
|
|
|
def test_days_left_from_a_steady_drain():
|
|
# 10 points a day apart, 5% a day, ending at 55%
|
|
result = analyse(series([100, 95, 90, 85, 80, 75, 70, 65, 60, 55]))
|
|
|
|
assert result['burnrateperday'] == 5.0
|
|
assert result['daysleft'] == 11 # 55 / 5
|
|
assert result['reason'] is None
|
|
assert result['basisdays'] == 9.0
|
|
|
|
|
|
def test_no_estimate_from_two_readings():
|
|
"""Two points through a coarse gauge can prove any rate at all."""
|
|
result = analyse(series([100, 90]))
|
|
|
|
assert result['daysleft'] is None
|
|
assert result['reason'] == 'not enough history yet'
|
|
|
|
|
|
def test_no_estimate_while_the_gauge_has_not_moved():
|
|
"""A printer reporting in 10% steps sits on a plateau for a fortnight."""
|
|
result = analyse(series([70, 70, 70, 70, 70, 70]))
|
|
|
|
assert result['daysleft'] is None
|
|
assert result['reason'] == 'level has not moved enough to estimate'
|
|
|
|
|
|
def test_a_recent_replacement_says_so_rather_than_guessing():
|
|
result = analyse(series([40, 30, 20, 10, 100, 98]))
|
|
|
|
assert result['daysleft'] is None
|
|
assert result['reason'] == 'replaced recently'
|
|
assert result['replacements'] == 1
|
|
assert result['lastreplaced'] is not None
|
|
|
|
|
|
def test_empty_history_is_reported_not_crashed():
|
|
result = analyse([])
|
|
|
|
assert result['reason'] == 'no history'
|
|
assert result['currentlevel'] is None
|
|
assert result['replacements'] == 0
|
|
|
|
|
|
def test_junk_rows_are_dropped_not_fatal():
|
|
points = normalise([('notaclock', '50'), ('1780000000', 'n/a'),
|
|
('1780000000', '50')])
|
|
|
|
assert len(points) == 1
|
|
|
|
|
|
def test_days_left_never_goes_negative():
|
|
result = analyse(series([20, 15, 10, 5, 0]))
|
|
|
|
assert result['daysleft'] == 0
|
|
|
|
|
|
def test_printer_sorts_by_its_soonest_supply():
|
|
"""A colour MFP is as urgent as its most pressing cartridge."""
|
|
supplies = [{'daysleft': 40}, {'daysleft': 6}, {'daysleft': None}]
|
|
|
|
assert soonest(supplies) == 6
|
|
|
|
|
|
def test_soonest_is_none_when_nothing_can_be_estimated():
|
|
assert soonest([{'daysleft': None}, {'daysleft': None}]) is None
|
|
|
|
|
|
def test_burn_rate_needs_elapsed_time():
|
|
"""Several readings in the same second is not a rate."""
|
|
clock = str(int(START.timestamp()))
|
|
points = normalise([(clock, '90'), (clock, '80'), (clock, '70'), (clock, '60')])
|
|
|
|
assert burn_rate(points) is None
|
|
|
|
|
|
def test_forecast_endpoint_says_so_when_zabbix_is_absent(client, db):
|
|
"""A dashboard must be told the data is missing, not shown an empty list
|
|
that reads as 'nothing runs out soon'."""
|
|
response = client.get('/api/printers/supplies/forecast')
|
|
|
|
assert response.status_code == 200
|
|
body = response.get_json()['data']
|
|
assert body['available'] is False
|
|
assert 'not configured' in body['reason']
|
|
assert body['printers'] == []
|