Make the toner forecast an order, not a table
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.
This commit is contained in:
@@ -9,7 +9,8 @@ purchasing decision.
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from plugins.printers.services.supply_history import (
|
||||
analyse, burn_rate, current_run, find_replacements, normalise, soonest,
|
||||
analyse, band, burn_rate, current_run, find_replacements, normalise,
|
||||
orderlist, soonest,
|
||||
)
|
||||
|
||||
START = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
||||
@@ -170,6 +171,91 @@ def test_burn_rate_needs_elapsed_time():
|
||||
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'."""
|
||||
@@ -179,4 +265,5 @@ def test_forecast_endpoint_says_so_when_zabbix_is_absent(client, db):
|
||||
body = response.get_json()['data']
|
||||
assert body['available'] is False
|
||||
assert 'not configured' in body['reason']
|
||||
assert body['printers'] == []
|
||||
assert body['cartridges'] == []
|
||||
assert body['orderlist'] == []
|
||||
|
||||
165
tests/test_plugins/test_toner_forecast_report.py
Normal file
165
tests/test_plugins/test_toner_forecast_report.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""Tests for the assembled toner forecast response.
|
||||
|
||||
The arithmetic is covered in test_supply_history; this is about the shape the
|
||||
report is read through - one row per cartridge, sorted into urgency bands, and
|
||||
an order list a person can hand to purchasing. Zabbix is stubbed, because what
|
||||
is being checked is the assembly, not the client.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from plugins.printers.services.zabbix_service import ZabbixService
|
||||
from shopdb.core.models import Asset, AssetType, Communication, CommunicationType
|
||||
from shopdb.core.models import Model, Vendor
|
||||
from plugins.printers.models import ModelSupply, Printer
|
||||
|
||||
|
||||
LEVELS = {
|
||||
# itemid: (name, live level, history of daily readings)
|
||||
'10': ('Black Cartridge', 1, [6, 5, 4, 3]),
|
||||
'11': ('Cyan Cartridge', 40, [100, 90, 80, 70, 60, 50, 40]),
|
||||
'12': ('Yellow Cartridge', 92, [95, 94, 93, 92]),
|
||||
}
|
||||
|
||||
START = 1780000000
|
||||
|
||||
|
||||
class StubZabbix(ZabbixService):
|
||||
"""Answers from LEVELS, so no network and no Zabbix instance."""
|
||||
|
||||
@property
|
||||
def isconfigured(self):
|
||||
return True
|
||||
|
||||
@property
|
||||
def isreachable(self):
|
||||
return True
|
||||
|
||||
def getsuppliesbyip_cached(self, ip):
|
||||
return [{'name': name, 'level': level, 'color': None, 'itemid': itemid}
|
||||
for itemid, (name, level, _) in LEVELS.items()]
|
||||
|
||||
def getlevelhistory(self, itemids, days=90):
|
||||
return {
|
||||
itemid: [(str(START + i * 86400), str(value))
|
||||
for i, value in enumerate(history)]
|
||||
for itemid, (_, _, history) in LEVELS.items()
|
||||
if itemid in [str(i) for i in itemids]
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def printer_with_parts(app, db, monkeypatch):
|
||||
"""One HP printer, three cartridges, two of them with a part on file."""
|
||||
monkeypatch.setattr('plugins.printers.api.asset_routes.ZabbixService',
|
||||
StubZabbix)
|
||||
|
||||
with app.app_context():
|
||||
vendor = Vendor(vendor='HP')
|
||||
db.session.add(vendor)
|
||||
db.session.flush()
|
||||
model = Model(modelnumber='HP M454', vendorid=vendor.vendorid)
|
||||
db.session.add(model)
|
||||
db.session.flush()
|
||||
|
||||
assettype = AssetType.query.filter_by(assettype='printer').first()
|
||||
if not assettype:
|
||||
assettype = AssetType(assettype='printer', pluginname='printers',
|
||||
tablename='printers')
|
||||
db.session.add(assettype)
|
||||
db.session.flush()
|
||||
|
||||
asset = Asset(assetnumber='PRN-1', name='WJ-PRN-0122',
|
||||
assettypeid=assettype.assettypeid)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
db.session.add(Printer(assetid=asset.assetid, vendorid=vendor.vendorid,
|
||||
modelnumberid=model.modelnumberid))
|
||||
|
||||
comtype = CommunicationType.query.filter_by(comtype='IP').first()
|
||||
if not comtype:
|
||||
comtype = CommunicationType(comtype='IP')
|
||||
db.session.add(comtype)
|
||||
db.session.flush()
|
||||
db.session.add(Communication(assetid=asset.assetid,
|
||||
comtypeid=comtype.comtypeid,
|
||||
ipaddress='10.20.30.40', isprimary=True))
|
||||
|
||||
# Black and cyan are mapped; yellow deliberately is not.
|
||||
db.session.add_all([
|
||||
ModelSupply(modelnumberid=model.modelnumberid, supplytype='toner',
|
||||
color='black', capacitytier='standard',
|
||||
partnumber='W2020A'),
|
||||
ModelSupply(modelnumberid=model.modelnumberid, supplytype='toner',
|
||||
color='cyan', capacitytier='standard',
|
||||
partnumber='W2021A'),
|
||||
])
|
||||
db.session.commit()
|
||||
yield
|
||||
|
||||
|
||||
def forecast(client):
|
||||
response = client.get('/api/printers/supplies/forecast?days=90')
|
||||
assert response.status_code == 200
|
||||
return response.get_json()['data']
|
||||
|
||||
|
||||
def test_a_row_is_a_cartridge_not_a_printer(client, printer_with_parts):
|
||||
"""The thing being ordered is the cartridge, so it is the unit of the
|
||||
report. Nested under a printer, a reader had to open a printer to find out
|
||||
whether anything on it needed doing."""
|
||||
data = forecast(client)
|
||||
|
||||
names = sorted(c['name'] for c in data['cartridges'] + data['unestimated'])
|
||||
assert names == ['Black Cartridge', 'Cyan Cartridge', 'Yellow Cartridge']
|
||||
|
||||
|
||||
def test_each_cartridge_carries_its_own_countdown(client, printer_with_parts):
|
||||
"""The defect this replaces: one printer-level figure printed beside every
|
||||
cartridge, so a healthy one wore its neighbour's deadline."""
|
||||
data = forecast(client)
|
||||
bylevel = {c['name']: c for c in data['cartridges']}
|
||||
|
||||
assert bylevel['Black Cartridge']['daysleft'] == 0 # live level 1%
|
||||
assert bylevel['Cyan Cartridge']['daysleft'] == 4 # 40% at 10%/day
|
||||
assert len({c['daysleft'] for c in data['cartridges']}) > 1
|
||||
|
||||
|
||||
def test_cartridges_are_banded_by_urgency(client, printer_with_parts):
|
||||
data = forecast(client)
|
||||
byname = {c['name']: c['band'] for c in data['cartridges']}
|
||||
|
||||
assert byname['Black Cartridge'] == 'empty'
|
||||
assert byname['Cyan Cartridge'] == 'soon'
|
||||
assert data['summary']['bands']['empty'] == 1
|
||||
|
||||
|
||||
def test_the_order_list_names_the_part_and_the_quantity(client, printer_with_parts):
|
||||
data = forecast(client)
|
||||
byparts = {item['partnumber']: item for item in data['orderlist']}
|
||||
|
||||
assert byparts['W2020A']['quantity'] == 1
|
||||
assert byparts['W2020A']['printers'][0]['printername'] == 'WJ-PRN-0122'
|
||||
assert 'W2021A' in byparts
|
||||
|
||||
|
||||
def test_a_healthy_cartridge_stays_off_the_order_list(client, printer_with_parts):
|
||||
"""Yellow at 92% is months away and belongs in the tail, not in an order."""
|
||||
data = forecast(client)
|
||||
|
||||
assert all(item['color'] != 'yellow' for item in data['orderlist'])
|
||||
|
||||
|
||||
def test_cartridges_are_sorted_soonest_first(client, printer_with_parts):
|
||||
data = forecast(client)
|
||||
|
||||
daysleft = [c['daysleft'] for c in data['cartridges']]
|
||||
assert daysleft == sorted(daysleft)
|
||||
|
||||
|
||||
def test_the_horizon_is_published_so_the_page_can_name_it(client, printer_with_parts):
|
||||
"""The heading says "within N days"; N has to come from whatever decided
|
||||
the list, or the two drift apart."""
|
||||
data = forecast(client)
|
||||
|
||||
assert data['horizondays'] == 14
|
||||
Reference in New Issue
Block a user