0c0c7be439
Reported from the floor: two printers showing 81 and 83 percent, both forecast to run out in a fortnight. That is a rate near 5.8 percent a day on a cartridge barely touched. burn_rate required four readings and a 2 point drop, and checked only that some time had passed - not how much. Supply items are commonly polled every few minutes, so four readings can span a quarter of an hour, and a 2 point drop across fifteen minutes extrapolates to nearly 200 percent a day. The report then sorted that confident wrong number into "soon", next to cartridges that really are about to run out. It did not look broken; it looked urgent. A rate now needs two days behind it. Two days is the smallest span that survives a printer's daily rhythm, so one heavy morning does not become the whole picture. Below that the answer is "not enough history yet", which the report already has a home for: band() returns None and the row lands in the "No estimate yet" section with its reason shown, rather than competing for attention with real urgency. This is the third face of the same fault. A phantom replacement truncated the run - "2 replacements in 2.3 days" was the same printer saying so - and the rate was then fitted to whatever short stub remained. The near-full rule and the dip filter stop the truncation; this stops a stub from producing a number at all.
365 lines
13 KiB
Python
365 lines
13 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, band, burn_rate, current_run, find_replacements, normalise,
|
|
orderlist, 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_live_level_drives_the_countdown_not_the_stored_one():
|
|
"""The report shows the live level, so it must count down from that one.
|
|
|
|
History lags a poll, and trends lag an hour. Displaying 20% beside a
|
|
countdown computed from a stored 4% is how a report loses its reader.
|
|
"""
|
|
result = analyse(series([100, 80, 60, 40]), currentlevel=20)
|
|
|
|
assert result['currentlevel'] == 20
|
|
assert result['burnrateperday'] == 20.0 # 60 points over 3 days
|
|
assert result['daysleft'] == 1 # 20 / 20, not 40 / 20
|
|
|
|
|
|
def test_a_nearly_empty_cartridge_reads_as_empty_not_as_weeks_away():
|
|
"""1% draining a tenth of a point a day computes to ten days. It is out."""
|
|
result = analyse(series([2.0, 1.7, 1.4, 1.1]), currentlevel=1)
|
|
|
|
assert result['daysleft'] == 0
|
|
|
|
|
|
def test_empty_sorts_ahead_of_a_fast_healthy_cartridge():
|
|
empty = analyse(series([2.0, 1.7, 1.4, 1.1]), currentlevel=1)
|
|
healthy = analyse(series([100, 80, 60, 40]))
|
|
|
|
assert empty['daysleft'] < healthy['daysleft']
|
|
|
|
|
|
def test_a_low_live_level_is_actionable_without_any_history():
|
|
"""A printer new to Zabbix still reports it is out of toner."""
|
|
result = analyse([], currentlevel=1)
|
|
|
|
assert result['daysleft'] == 0
|
|
assert result['reason'] is None
|
|
|
|
|
|
def test_no_history_and_a_healthy_level_still_says_no_history():
|
|
result = analyse([], currentlevel=80)
|
|
|
|
assert result['daysleft'] is None
|
|
assert result['reason'] == 'no history'
|
|
|
|
|
|
def test_a_live_level_above_the_stored_run_means_it_was_just_swapped():
|
|
"""Trends lag an hour; a cartridge changed in that hour must not inherit
|
|
the rate of the one that came out."""
|
|
result = analyse(series([40, 30, 20, 10]), currentlevel=100)
|
|
|
|
assert result['daysleft'] is None
|
|
assert result['reason'] == 'replaced recently'
|
|
assert result['replacements'] == 1
|
|
|
|
|
|
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 cartridge(daysleft, partnumber='CF258X', color='black', printername='PRN',
|
|
model='HP 428', printerid=1):
|
|
parts = [{'partnumber': partnumber, 'marketingname': None,
|
|
'capacitytier': 'standard'}] if partnumber else []
|
|
return {'daysleft': daysleft, 'partnumbers': parts, 'color': color,
|
|
'supplytype': 'toner', 'model': model, 'printerid': printerid,
|
|
'printername': printername}
|
|
|
|
|
|
def test_bands_split_at_two_weeks_and_a_month():
|
|
assert band(0) == 'empty'
|
|
assert band(1) == 'soon'
|
|
assert band(14) == 'soon'
|
|
assert band(15) == 'month'
|
|
assert band(30) == 'month'
|
|
assert band(31) == 'later'
|
|
|
|
|
|
def test_no_estimate_belongs_to_no_band():
|
|
assert band(None) is None
|
|
|
|
|
|
def test_the_same_part_in_two_printers_is_a_quantity_of_two():
|
|
"""The number purchasing needs, and the one a per-printer table makes you
|
|
count by hand."""
|
|
result = orderlist([cartridge(0, printername='A', printerid=1),
|
|
cartridge(3, printername='B', printerid=2)])
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['quantity'] == 2
|
|
assert [p['printername'] for p in result[0]['printers']] == ['A', 'B']
|
|
|
|
|
|
def test_the_order_list_stops_at_the_horizon():
|
|
"""Ordering only what is already empty means running empty; ordering three
|
|
months out is a stock cupboard."""
|
|
result = orderlist([cartridge(0), cartridge(90, partnumber='W2021A')])
|
|
|
|
assert [item['partnumber'] for item in result] == ['CF258X']
|
|
|
|
|
|
def test_a_cartridge_with_no_part_mapped_is_still_on_the_list():
|
|
"""Dropping it would quietly shorten the order."""
|
|
result = orderlist([cartridge(2, partnumber=None)])
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['partnumber'] is None
|
|
assert result[0]['quantity'] == 1
|
|
|
|
|
|
def test_unmapped_parts_sort_last():
|
|
"""They need a decision before anything can be ordered, so they do not sit
|
|
at the top of a list meant to be read straight down."""
|
|
result = orderlist([cartridge(1, partnumber=None), cartridge(1)])
|
|
|
|
assert result[0]['partnumber'] == 'CF258X'
|
|
assert result[-1]['partnumber'] is None
|
|
|
|
|
|
def test_alternate_capacity_tiers_are_offered_not_counted_separately():
|
|
item = {**cartridge(1), 'partnumbers': [
|
|
{'partnumber': 'CF258A', 'marketingname': None, 'capacitytier': 'standard'},
|
|
{'partnumber': 'CF258X', 'marketingname': None, 'capacitytier': 'high'},
|
|
]}
|
|
|
|
result = orderlist([item])
|
|
|
|
assert len(result) == 1
|
|
assert result[0]['partnumber'] == 'CF258A'
|
|
assert result[0]['alternates'] == ['CF258X']
|
|
|
|
|
|
def test_the_same_part_for_a_different_model_is_ordered_separately():
|
|
"""Two models sharing a part number is a mapping error worth seeing, not a
|
|
quantity to merge."""
|
|
result = orderlist([cartridge(1, model='HP 428'),
|
|
cartridge(1, model='HP M454')])
|
|
|
|
assert len(result) == 2
|
|
|
|
|
|
def test_no_estimate_never_reaches_the_order_list():
|
|
assert orderlist([cartridge(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['cartridges'] == []
|
|
assert body['orderlist'] == []
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Noise that used to read as cartridge changes, and bursts that used to bias
|
|
# the rate for the life of the cartridge. Both were reported from the floor:
|
|
# "5 changes in 90 days, that's hard to believe", and a cartridge that dropped
|
|
# 20 percent in two days then barely moved.
|
|
# --------------------------------------------------------------------------
|
|
|
|
def test_a_poll_returning_zero_is_not_a_cartridge_change():
|
|
"""0 then a real level is an SNMP error or a calibrating printer.
|
|
|
|
The rise is +60, which cleared the old threshold on its own. It does not
|
|
land near full, so it is not a swap.
|
|
"""
|
|
points = normalise(series([70, 60, 0, 60, 55, 50]))
|
|
assert find_replacements(points) == []
|
|
|
|
|
|
def test_a_gauge_ticking_back_up_mid_range_is_not_a_change():
|
|
"""A coarse gauge after a reseat or a power cycle. Lands at 55, not full."""
|
|
points = normalise(series([70, 60, 45, 55, 50, 45]))
|
|
assert find_replacements(points) == []
|
|
|
|
|
|
def test_a_real_swap_is_still_counted():
|
|
"""Near-empty to near-full. The shape a cartridge change actually makes."""
|
|
points = normalise(series([30, 15, 5, 100, 95, 90]))
|
|
assert len(find_replacements(points)) == 1
|
|
|
|
|
|
def test_the_run_starts_at_the_real_swap_not_at_the_noise():
|
|
"""current_run and find_replacements must agree on what a change is.
|
|
|
|
They read the same predicate now; when they did not, a phantom rise reset
|
|
the run and threw away the history the estimate needed.
|
|
"""
|
|
points = normalise(series([90, 0, 85, 80, 75, 70]))
|
|
assert find_replacements(points) == []
|
|
assert len(current_run(points)) == len(points)
|
|
|
|
|
|
def test_an_early_burst_does_not_dominate_the_rate_forever():
|
|
"""20 percent in two days, then a month of almost nothing.
|
|
|
|
The endpoint slope reads the burst forever: (100-75)/30 = 0.83 %/day, so
|
|
the report keeps promising the cartridge runs out long after printing
|
|
stopped. The median sees one fast interval among many quiet ones.
|
|
"""
|
|
levels = [100, 90, 80] + [80 - i * 0.2 for i in range(1, 28)]
|
|
rate = burn_rate(normalise(series(levels)))
|
|
assert rate is not None
|
|
assert rate < 1.0, rate
|
|
detail = analyse(series(levels))
|
|
assert detail['rateunstable'] is True
|
|
|
|
|
|
def test_a_steady_cartridge_is_not_flagged_unstable():
|
|
detail = analyse(series([100, 95, 90, 85, 80, 75, 70]))
|
|
assert detail['burnrateperday'] == 5.0
|
|
assert detail['rateunstable'] is False
|
|
|
|
|
|
def test_a_real_near_empty_reading_before_a_swap_is_kept():
|
|
"""30, 5, 100 is a cartridge run to the end and changed - not a spike.
|
|
|
|
The dip filter must not eat it: the 5 does not RECOVER to 30, it jumps to
|
|
full, which is the shape of a swap rather than of a bad poll.
|
|
"""
|
|
points = normalise(series([40, 30, 5, 100, 95, 90]))
|
|
assert 5.0 in [level for _, level in points]
|
|
assert len(find_replacements(points)) == 1
|
|
|
|
|
|
def test_minutes_of_readings_do_not_forecast_weeks():
|
|
"""Supply items are often polled every few minutes.
|
|
|
|
Four readings a quarter of an hour apart, with a 2 point drop between the
|
|
ends, used to extrapolate to nearly 200 percent a day - so a cartridge
|
|
sitting at 82 percent was forecast to run out in a fortnight. A rate needs
|
|
time behind it, and without it the honest answer is no estimate.
|
|
"""
|
|
minutes = 5
|
|
detail = analyse(series([84, 83, 83, 82], hours=minutes / 60))
|
|
assert detail['burnrateperday'] is None
|
|
assert detail['daysleft'] is None
|
|
assert detail['reason'] == 'not enough history yet'
|
|
assert band(detail['daysleft']) is None # lands in "No estimate yet"
|
|
|
|
|
|
def test_a_run_with_enough_days_still_estimates():
|
|
"""The guard must not silence a genuinely slow, genuinely long run."""
|
|
detail = analyse(series([84, 83, 82, 81, 80, 79]))
|
|
assert detail['burnrateperday'] == 1.0
|
|
assert detail['daysleft'] == 79
|