3d83806135
The report answers a purchasing question, and it was answering it in seven columns, two tables and a rowspan. What someone actually needs from it is a short list of what to buy. So it opens with that list, grouped by part number with a quantity. Two cartridges of the same part in different printers is a quantity of two, which is the number an order needs and the one a per-printer table made the reader count by hand. It covers what is empty plus what goes within a fortnight - ordering only what is already empty means running empty. There is a copy button, because it ends up pasted into a mail. Below it the cartridges sit in urgency bands rather than in one long list sorted by a number. The question is which pile a thing is in, and a pile that is empty is worth seeing as empty. Everything past "empty" starts collapsed; the order list above already covers the same ground in a tenth of the height. The row is a cartridge now, not a printer, so it can carry its own part number, its own level bar and its own countdown. Nesting supplies under a printer meant opening a printer to find out whether anything on it needed doing. Cartridges with no part mapped are counted on a single line rather than given one each. They cannot be dropped, since that would quietly shorten the order, and they cannot be ordered from here either - the job they represent is mapping them, which is one job however many there are. Bands and the order horizon are decided server-side, next to the arithmetic that produces them, so a heading cannot disagree with what got added to the list. Checked against a fleet of 43 dev printers with real part mappings, driven by a stub Zabbix - live Zabbix is not reachable from the dev box.
270 lines
8.9 KiB
Python
270 lines
8.9 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'] == []
|