8b9b9363ee
Two things reported from the floor, one cause each. "5 CHANGES IN 90 DAYS, THAT'S HARD TO BELIEVE." It was. A replacement was any +10 rise between readings, with no check on where it landed, so two shapes that are not swaps scored as swaps: a supply reading 0 or near-0 while it was out of the machine and then reading normally again, and a coarse gauge ticking back up after a reseat or a power cycle. A swap must now also LAND near full, because that is what a new cartridge reads, and a single dip that RECOVERS to roughly where it came from is dropped before anything looks at it. The dip filter keys on shape rather than cause, which is why it holds for all of them - a supply pulled out to be shaken, a door open mid-poll, or a site whose preprocessing maps the Printer MIB's unknown sentinels onto 0. It is NOT a Zabbix timeout: an item that does not answer records nothing rather than writing a zero. A genuine near-empty reading before a real swap does not recover, it jumps to full, so it survives and its swap still counts. find_replacements and current_run now read one predicate. When they disagreed, a phantom rise reset the run and threw away the history the estimate needed - so the bad count was quietly damaging the rate as well, which is why both were wrong at once. Expect replacement counts to FALL and per-cartridge history to lengthen. A BURST BIASED THE RATE FOR THE LIFE OF THE CARTRIDGE. The rate was the slope between the first and last reading of the run, and two endpoints cannot tell "steady" from "burst then stopped". A cartridge that lost 20 percent in two days and then barely moved for a month read as 0.83 percent/day forever after, so the report kept promising it would run out long after printing slowed. It is now the median of the per-interval rates: the burst is one interval among many rather than one of two points. Rising intervals are dropped as noise; flat ones stay in at zero, because a cartridge that did not move is real information. If every interval is flat or rising yet the run dropped overall, it falls back to the whole-run slope rather than reporting nothing. Where the intervals disagree by 5x or more the rate carries a marker and an explanation on hover. The number is still the best estimate available; the flag stops it reading as a measurement. The "Changed" column is "Replacements", and its cell says "2 in 90d" rather than "2 / 90d", which was read as a date, a ratio and a version number.
342 lines
12 KiB
Python
342 lines
12 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
|