Files
monitor-shared-utils/tests/test_alert_priority.py
T
dsqlandClaude Code Opus 5.5 f51aed2c9f calculate_priority: resolve non-finite prices to 0 instead of crashing the monitor
tpv-academysports died on 2026-10-08 06:27 UTC and stayed down: six
identical crashes in 8 minutes (each ~2 min after start), then docker's
on-failure:5 gave up. Every one was

    database.py, in calculate_priority
        percentage = int(((original_price - current_price) / original_price) * 100)
    OverflowError: cannot convert float infinity to integer

A monitor builds prices with float() on the retailer's raw field, and
float() accepts "Infinity" / "1e999", so a non-finite price reaches here.
The `not original_price` guard catches None and 0, not inf or nan, and the
exception escapes add_products_to_database -> run_monitor_cycle and kills
the whole process -- one bad product takes the monitor down, and it
crashes again on every restart while that product is listed. Every
monitor that uses this lib has the same exposure.

calculate_priority now returns 0 (the existing "broken prices" bucket)
for a non-finite current/original price, and for a finite pair whose
ratio still overflows (a subnormal original price). Tests cover both;
they fail on the old code (OverflowError, plus ValueError for nan) and
pass now (tests/test_alert_priority.py: 10 passed).

Not changed: the monitor-side float() parsing, which could reject these
prices before they are stored -- worth a look by the monitor owners.

Co-Authored-By: Claude Code Opus 5.5 (ansible-ops, auto-repair) <noreply@anthropic.com>
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-10-08 18:46:59 -04:00

86 lines
3.3 KiB
Python

"""unit test: calculate_priority ranks MSRP-less products instead of burying them
run from the repo root:
uv run --with pytest pytest tests/test_alert_priority.py
pure and offline — passes plain dicts shaped like product documents; touches no
Mongo. The claim query sorts priority descending, so a priority of 0 means an
alert is only drained once everything else is.
"""
from monitor_utils.database import (
NEW_LISTING_PRIORITY,
NO_MSRP_PRIORITY,
calculate_priority,
)
def document(current, original, **overrides) -> dict:
doc = {"_id": "p1", "currentPrice": current, "originalPrice": original}
doc.update(overrides)
return doc
def test_no_msrp_takes_its_own_rank():
assert calculate_priority(document(50.0, 50.0, hasMsrp=False)) == NO_MSRP_PRIORITY
def test_no_msrp_rank_sits_directly_under_the_seventy_band():
"""It must outrank every sub-70% product and lose to every 70%+ one."""
assert calculate_priority(document(25.0, 100.0)) == 7 # 75% off
assert NO_MSRP_PRIORITY < 7
assert NO_MSRP_PRIORITY > calculate_priority(document(45.0, 100.0)) # 55% off
def test_msrp_products_are_unaffected():
assert calculate_priority(document(25.0, 100.0, hasMsrp=True)) == 7
assert calculate_priority(document(50.0, 100.0, hasMsrp=True)) == 5
def test_documents_without_the_field_keep_the_percentage_path():
"""Documents written before hasMsrp existed must not all jump the queue."""
assert calculate_priority(document(50.0, 100.0)) == 5
assert calculate_priority(document(100.0, 100.0)) == 0
def test_broken_prices_still_outrank_nothing():
"""A missing or zero original price is a data defect, not a missing MSRP."""
assert calculate_priority(document(50.0, None, hasMsrp=False)) == 0
assert calculate_priority(document(50.0, 0, hasMsrp=False)) == 0
assert calculate_priority(document(None, 100.0, hasMsrp=False)) == 0
def test_non_finite_prices_resolve_to_zero_instead_of_raising():
"""float("Infinity") from a retailer field once crashed the whole monitor."""
inf, nan = float("inf"), float("nan")
assert calculate_priority(document(50.0, inf)) == 0
assert calculate_priority(document(inf, 100.0)) == 0
assert calculate_priority(document(-inf, 100.0)) == 0
assert calculate_priority(document(nan, 100.0)) == 0
assert calculate_priority(document(50.0, nan)) == 0
assert calculate_priority(document(50.0, inf, newListing=True)) == 0
def test_a_discount_too_large_to_represent_resolves_to_zero():
"""Finite prices can still overflow the ratio (a subnormal original)."""
assert calculate_priority(document(-1e308, 5e-324)) == 0
assert calculate_priority(document(1e308, 5e-324)) == 0
def test_new_listing_is_floored_not_buried():
"""A low-discount new listing sends regardless, so it must not sort last."""
assert (
calculate_priority(document(95.0, 100.0, newListing=True))
== NEW_LISTING_PRIORITY
)
def test_new_listing_keeps_a_higher_percentage_rank():
"""The floor is a floor: a 75%-off new listing still ranks as 75% off."""
assert calculate_priority(document(25.0, 100.0, newListing=True)) == 7
def test_repeat_alerts_keep_the_percentage_path():
assert calculate_priority(document(95.0, 100.0, newListing=False)) == 0
assert calculate_priority(document(95.0, 100.0)) == 0