shopdb-flask/tests/test_plugins/test_supply_history.py

"""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

import pytest

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_keeps_its_real_countdown():
    """Only zero is empty.

    This used to floor everything at or below 5% to daysleft 0, so 1% draining
    a tenth of a point a day was reported as gone. It has ten days in it, and
    ten days is the only thing the row exists to say. Flooring it put the
    cartridge beside ones that genuinely are empty with no way back to the
    difference.
    """
    # 3.0 down to 1.0 at a tenth of a point a day: a real, slow, measured drop.
    slow = [round(3.0 - 0.1 * i, 2) for i in range(21)]
    result = analyse(series(slow), currentlevel=1)

    assert result['daysleft'] == pytest.approx(10, abs=0.1)


def test_a_cartridge_with_hours_left_is_not_reported_as_empty():
    """The other half of the floor: int() truncated the division, so anything
    under a full day became 0 and read as empty. A cartridge with six hours in
    it still prints."""
    result = analyse(series([100, 75, 50, 25]), currentlevel=6)

    assert 0 < result['daysleft'] < 1
    assert round(result['daysleft'] * 24) == 6


def test_an_actual_zero_is_still_empty():
    """The half that must not regress. Zero means spent, and spent is ordered
    first."""
    result = analyse(series([2.0, 1.7, 1.4, 1.1]), currentlevel=0)

    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=0)
    healthy = analyse(series([100, 80, 60, 40]))

    assert empty['daysleft'] < healthy['daysleft']


def test_an_empty_live_level_is_actionable_without_any_history():
    """A printer new to Zabbix still reports it is out of toner."""
    result = analyse([], currentlevel=0)

    assert result['daysleft'] == 0
    assert result['reason'] is None


def test_a_low_live_level_with_no_history_cannot_be_counted_down():
    """A rate needs two readings. At 1% with none, there is nothing to divide.

    Previously the 5% floor answered this by asserting daysleft 0, which was
    right by accident and wrong in general - it said "empty" about a level
    nobody had watched move. It reports no estimate and says why; the report
    lists it under "No estimate yet" rather than on the order list.
    """
    result = analyse([], currentlevel=1)

    assert result['daysleft'] is None
    assert result['reason'] == 'no history'


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
    # The old endpoint slope gives ~0.88 for this series, so a threshold of 1.0
    # passed with the bug still in. The median gives ~0.2.
    assert rate < 0.5, 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


def test_a_multi_poll_outage_is_one_dip_not_a_cartridge_change():
    """A door open or a supply out of the machine spans several polls.

    These items are often polled every few minutes, so an outage covers more
    than one reading. Removing only SINGLE readings left the original failure
    in place for any cartridge above NEW_CARTRIDGE_LEVEL: 0 -> 90 clears the
    rise and lands near full, so it scored as a swap.
    """
    points = normalise(series([90, 0, 0, 90, 88, 86], hours=5 / 60))
    assert [level for _, level in points] == [90.0, 90.0, 88.0, 86.0]
    assert find_replacements(points) == []


def test_a_long_absence_is_not_deleted_as_noise():
    """Days at zero is a real empty period, not a bad poll.

    Only a BRIEF multi-reading dip is removed. Deleting a level that stayed
    down for days would hide exactly the outage someone needs to see.
    """
    points = normalise(series([90, 0, 0, 90, 88, 86]))
    assert 0.0 in [level for _, level in points]


def test_a_swap_of_a_nearly_full_cartridge_is_not_eaten_by_the_dip_filter():
    """95, then 5, then 100 is a swap, not a dip that recovered.

    Comparing the absolute difference treated it as a recovery, because 100 and
    95 are close, and deleted the evidence: the run then spanned two cartridges
    and the replacement went uncounted. Toner only falls, so a recovery comes
    back at or BELOW where it left - a new cartridge comes back higher.
    """
    points = normalise(series([98, 95, 5, 100, 96, 92]))
    assert 5.0 in [level for _, level in points]
    assert len(find_replacements(points)) == 1


def test_consumption_between_two_swaps_is_not_mistaken_for_a_dip():
    """Swap to full, print for days, swap again: same shape, different thing.

    The readings between the swaps are far below both bounding levels, which is
    what a dip looks like. What separates them is elapsed time.
    """
    points = normalise(series([80, 60, 40, 20, 100, 80, 60, 95, 70]))
    assert [level for _, level in points] == [80.0, 60.0, 40.0, 20.0,
                                              100.0, 80.0, 60.0, 95.0, 70.0]
    assert len(find_replacements(points)) == 2


def test_a_rise_that_stops_mid_range_is_not_a_cartridge_change():
    """Isolates the near-full rule from the dip filter.

    50 -> 75 clears the rise and is NOT a recovering dip (75 comes back higher
    than the 60 before it), so the dip filter leaves it alone. Only the rule
    that a swap must LAND near full rejects it. A new cartridge does not read
    75 percent.
    """
    points = normalise(series([60, 50, 75, 73, 71]))
    assert 50.0 in [level for _, level in points]
    assert find_replacements(points) == []