public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-18 11:56:01 +02:00
Code Issues Packages Projects Releases Wiki Activity

Give the two stock alerts something that fires them (issue #2955) (#3196)

Browse Source
This commit is contained in:
maziggy
2026-10-01 13:01:21 +02:00
parent dc80b6a91b 318430f0f4
commit 1400e50a0d
9 changed files with 2018 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/core/database.py
+35
View File
@@ -5128,6 +5128,12 @@ async def run_migrations(conn):
# on fresh installs only — this covers databases whose table predates it.
await _migrate_location_ha_sensor_unique_binding(conn)
# Migration: the stock alert templates name the colour and subtype (#2955).
# The forecast groups by colour, so two colours of one product would
# otherwise send the same message. A plain UPDATE guarded on the old text, so
# a template an admin has edited is left alone.
await _migrate_stock_alert_template_sku_variables(conn)
# Migration: supplier master list + spool assignments (#2988).
# create_all() covers fresh installs; this covers upgrades.
await _migrate_create_supplier_tables(conn)
@@ -5422,6 +5428,35 @@ async def _migrate_rename_ha_sensor_alert_template(conn) -> None:
)
_STOCK_ALERT_TEMPLATE_BODIES = {
"stock_reorder_alert": (
"{material} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
"{material} {subtype} {color} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
),
"stock_break_alert": (
"{material} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
"{material} {subtype} {color} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
),
}
async def _migrate_stock_alert_template_sku_variables(conn) -> None:
"""Give the stock alert templates {subtype} and {color} (#2955).
Updates a body only while it is still the shipped default, so an admin's own
wording is kept.
"""
from sqlalchemy import text
for event_type, (old, new) in _STOCK_ALERT_TEMPLATE_BODIES.items():
await conn.execute(
text(
"UPDATE notification_templates SET body_template = :new WHERE event_type = :et AND body_template = :old"
),
{"new": new, "et": event_type, "old": old},
)
async def _migrate_location_ha_sensor_unique_binding(conn) -> None:
"""Unique index on location_ha_sensors (location_id, entity_id) (#2824).
backend/app/models/notification_template.py
+2 -2
View File
@@ -235,13 +235,13 @@ DEFAULT_TEMPLATES = [
"event_type": "stock_reorder_alert",
"name": "Stock Reorder Alert",
"title_template": "Reorder Alert: {material}",
"body_template": "{material} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
"body_template": "{material} {subtype} {color} ({brand}) has reached the reorder point.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Days left: {days_left}d\nReorder now to avoid a stock break.",
},
{
"event_type": "stock_break_alert",
"name": "Stock Break Alert",
"title_template": "Stock Break Risk: {material}",
"body_template": "{material} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
"body_template": "{material} {subtype} {color} ({brand}) will run out before replenishment arrives.\nStock: {stock_g}g | Rate: {rate_g_day}g/day | Lead time: {lead_time_days}d\nOnly {days_left}d of stock remaining — order immediately.",
},
# User email notification templates (sent to the print job owner).
# Names include " Email" so they aren't confused with the provider-level
backend/app/schemas/notification_template.py
+46
View File
@@ -101,6 +101,29 @@ EVENT_VARIABLES: dict[str, list[str]] = {
"queue_job_skipped": ["printer", "job_name", "reason", "timestamp", "app_name"],
"queue_job_failed": ["printer", "job_name", "reason", "timestamp", "app_name"],
"queue_completed": ["completed_count", "timestamp", "app_name"],
"stock_reorder_alert": [
"material",
"subtype",
"brand",
"color",
"stock_g",
"rate_g_day",
"days_left",
"timestamp",
"app_name",
],
"stock_break_alert": [
"material",
"subtype",
"brand",
"color",
"stock_g",
"rate_g_day",
"days_left",
"lead_time_days",
"timestamp",
"app_name",
],
# User management notifications
"user_created": ["username", "password", "login_url", "app_name", "timestamp"],
"password_reset": ["username", "password", "login_url", "app_name", "timestamp"],
@@ -303,6 +326,29 @@ SAMPLE_DATA: dict[str, dict[str, str]] = {
"timestamp": "2024-01-15 18:30",
"app_name": "Bambuddy",
},
"stock_reorder_alert": {
"material": "PLA",
"subtype": "Matte",
"brand": "Bambu Lab",
"color": "Charcoal",
"stock_g": "420",
"rate_g_day": "18.5",
"days_left": "22",
"timestamp": "2024-01-15 14:30",
"app_name": "Bambuddy",
},
"stock_break_alert": {
"material": "PLA",
"subtype": "Matte",
"brand": "Bambu Lab",
"color": "Charcoal",
"stock_g": "160",
"rate_g_day": "18.5",
"days_left": "8",
"lead_time_days": "14",
"timestamp": "2024-01-15 14:30",
"app_name": "Bambuddy",
},
# User management notifications
"user_created": {
"username": "john_doe",
backend/app/services/notification_service.py
+26 -2
View File
@@ -2589,15 +2589,31 @@ class NotificationService:
rate_g_day: float,
days_left: int,
db: AsyncSession,
*,
subtype: str | None = None,
color: str | None = None,
skip_break_subscribers: bool = False,
):
"""Fire when an inventory SKU reaches its reorder point."""
"""Fire when an inventory SKU reaches its reorder point.
``subtype`` and ``color`` are what tell apart the messages for two colours of one
product, which the forecast (grouped by colour as well) reports separately.
A SKU in stock break has also reached its reorder point. ``skip_break_subscribers``
leaves out the providers that have the break alert on, so the producer can send
the reorder event for a break SKU without telling those providers twice.
"""
providers = await self._get_providers_for_event(db, "on_stock_reorder_alert", None)
if skip_break_subscribers:
providers = [p for p in providers if not p.on_stock_break_alert]
if not providers:
return
variables = {
"material": material,
"subtype": subtype or "",
"brand": brand or "",
"color": color or "",
"stock_g": f"{stock_g:.0f}",
"rate_g_day": f"{rate_g_day:.1f}",
"days_left": str(days_left),
@@ -2615,15 +2631,23 @@ class NotificationService:
days_left: int,
lead_time_days: int,
db: AsyncSession,
*,
subtype: str | None = None,
color: str | None = None,
):
"""Fire when a stock break is detected (stock runs out before lead time)."""
"""Fire when a stock break is detected (stock runs out before lead time).
``subtype`` and ``color`` as for :meth:`on_stock_reorder_alert`.
"""
providers = await self._get_providers_for_event(db, "on_stock_break_alert", None)
if not providers:
return
variables = {
"material": material,
"subtype": subtype or "",
"brand": brand or "",
"color": color or "",
"stock_g": f"{stock_g:.0f}",
"rate_g_day": f"{rate_g_day:.1f}",
"days_left": str(days_left),
backend/app/services/print_scheduler.py
+310 -1
View File
@@ -30,7 +30,7 @@ from backend.app.models.settings import Settings
from backend.app.models.smart_plug import SmartPlug
from backend.app.models.spool_assignment import SpoolAssignment
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
from backend.app.services import drying_preflight, print_dispatch_context
from backend.app.services import drying_preflight, print_dispatch_context, stock_forecast
from backend.app.services.bambu_ftp import (
FtpFailureReport,
UploadCancelled,
@@ -111,6 +111,17 @@ def _ams_slot_label(ams_id: int, tray_id: int) -> str:
# low spool does not need answering at that resolution (#2913).
_FILAMENT_LOW_MIN_INTERVAL = 30.0
# Minimum seconds between stock forecast checks (#2955). The forecast is in whole
# days, so a finer check would only repeat the same answer.
_STOCK_FORECAST_MIN_INTERVAL = 3600.0
# The settings row that holds which stock alerts have already been sent (#2955).
_STOCK_ALERTS_SETTING_KEY = "stock_alerts_notified"
# The Inventory page asks for this many usage records and the forecast panel
# builds its rates from them; the alert reads the same window so the two agree.
_STOCK_FORECAST_HISTORY_LIMIT = 5000
def _remaining_percent(label_weight: int | float | None, weight_used: float | None) -> float | None:
"""Remaining filament as a percentage of the label weight.
@@ -1081,6 +1092,18 @@ class PrintScheduler:
self._notified_filament_low: set[tuple[int, int, int, int]] = set()
# Earliest monotonic time the next low-filament check may run (#2913).
self._filament_low_next_check: float = 0.0
# SKU key -> (condition, events told) for a SKU that is alerting (#2955): the
# condition is "reorder" or "break", the events are those that had a
# subscriber when it was sent. Absent means not alerting. Cleared when the
# condition clears, so it can alert again, and when nobody wants either
# event. Mirrored to one settings row (_STOCK_ALERTS_SETTING_KEY), because the
# first check runs as soon as Bambuddy starts and a restart would otherwise
# re-send one message per SKU that is still low.
self._notified_stock_alerts: dict[stock_forecast.SkuKey, tuple[str, frozenset[str]]] = {}
# The JSON last read from or written to that row; None until it has been read.
self._stock_alerts_persisted: str | None = None
# Earliest monotonic time the next stock forecast check may run (#2955).
self._stock_forecast_next_check: float = 0.0
# Printers with a "running" scheduled drying row (#2638). Rebuilt from the
# DB on every _check_scheduled_dryings call so route-side cancels show up.
# Auto-drying's stop-all branches must not stop or untrack these printers;
@@ -1420,6 +1443,7 @@ class PrintScheduler:
inflight_printers = {pid for (_task, pid) in self._inflight.values() if pid is not None}
await self._check_auto_drying(db, [], inflight_printers)
await self._check_filament_low(db)
await self._check_stock_forecast(db)
return bool(self._inflight)
logger.info(
@@ -2205,6 +2229,10 @@ class PrintScheduler:
# the spools in the printers stopped mattering.
await self._check_filament_low(db)
# Stock forecast: alert when a filament SKU reaches its reorder point or
# is about to run out before a replenishment could arrive (#2955).
await self._check_stock_forecast(db)
# Keep the loop on the fast interval while any upload is in flight so
# a slot freed mid-tick refills within seconds rather than after the
# 30 s idle sleep (#2602). Selecting anything this pass (launched or
@@ -4554,6 +4582,287 @@ class PrintScheduler:
except Exception as e:
logger.warning("Low-filament notification failed for slot %s: %s", key, e)
async def _check_stock_forecast(self, db: AsyncSession) -> None:
"""Alert when a filament SKU reaches its reorder point or is about to break (#2955).
``on_stock_reorder_alert`` and ``on_stock_break_alert`` have had a column,
a schema field, a template and a UI toggle, and nothing that computes the
condition -- the Forecast panel does it in the browser, so with no page
open nothing could ever alert. This runs the same arithmetic
(``stock_forecast``) from the scheduler loop.
An alert is sent when a SKU *moves into* a condition, once, and again
after the condition has cleared. A SKU that worsens from reorder to break
alerts again as a break, because the panel treats the two as exclusive; one
that eases from break back to reorder does not alert again.
A SKU with alerts snoozed is treated as not alerting, so un-snoozing it
while it is still low tells you.
Time-gated to hourly for the reasons ``_check_filament_low`` is gated: the
loop runs every 3 s while an upload is in flight, and the whole spool
collection is read over HTTP in Spoolman mode. Both events default to off
on every provider, so nothing is read unless a provider wants one.
"""
now = time.monotonic()
if now < self._stock_forecast_next_check:
return
self._stock_forecast_next_check = now + _STOCK_FORECAST_MIN_INTERVAL
try:
if self._stock_alerts_persisted is None:
await self._load_stock_alerts(db)
wanted = {
"reorder": bool(await notification_service._get_providers_for_event(db, "on_stock_reorder_alert")),
"break": bool(await notification_service._get_providers_for_event(db, "on_stock_break_alert")),
}
if not any(wanted.values()):
# Nobody wants either event. Forget what was sent, so whoever
# switches one on is told about the SKUs that are low now.
self._notified_stock_alerts.clear()
await self._save_stock_alerts(db)
return
forecasts = await self._stock_forecasts(db)
if forecasts is None:
# Spoolman unreachable. Deliberately not clearing the notified
# set: a brief outage would otherwise re-alert every SKU when it
# came back.
return
await self._emit_stock_alerts(db, forecasts, wanted)
await self._save_stock_alerts(db)
except Exception as e:
logger.warning("Stock forecast check failed: %s", e, exc_info=True)
def _serialize_stock_alerts(self) -> str:
"""The notified set as JSON, in a fixed order so an unchanged set compares equal."""
rows = sorted([list(key), kind, sorted(events)] for key, (kind, events) in self._notified_stock_alerts.items())
return json.dumps(rows)
async def _load_stock_alerts(self, db: AsyncSession) -> None:
"""Read the notified set saved by the last run, once per process.
A missing or unreadable row is an empty set: the worst that does is one
repeat of what a restart used to cause every time.
"""
raw = (await db.execute(select(Settings).where(Settings.key == _STOCK_ALERTS_SETTING_KEY))).scalar_one_or_none()
loaded: dict[stock_forecast.SkuKey, tuple[str, frozenset[str]]] = {}
if raw and raw.value:
try:
for key, kind, events in json.loads(raw.value):
if len(key) == 4 and kind in ("reorder", "break"):
loaded[tuple(str(part) for part in key)] = (kind, frozenset(str(e) for e in events))
except (ValueError, TypeError):
logger.warning("Ignoring unreadable %s setting", _STOCK_ALERTS_SETTING_KEY)
loaded = {}
self._notified_stock_alerts = loaded
self._stock_alerts_persisted = self._serialize_stock_alerts()
async def _save_stock_alerts(self, db: AsyncSession) -> None:
"""Write the notified set to its settings row if it changed since it was last read or written."""
serialized = self._serialize_stock_alerts()
if serialized == self._stock_alerts_persisted:
return
from backend.app.core.db_dialect import upsert_setting
await upsert_setting(db, Settings, _STOCK_ALERTS_SETTING_KEY, serialized)
await db.commit()
self._stock_alerts_persisted = serialized
async def _stock_forecasts(self, db: AsyncSession) -> stock_forecast.ForecastMap | None:
"""Forecast every active SKU in whichever inventory mode is on. None if it cannot be read."""
from backend.app.models.filament_sku_settings import FilamentSkuSettings
global_lead_time = max(0, await self._get_int_setting(db, "forecast_global_lead_time_days", default=0))
sku_rows = (await db.execute(select(FilamentSkuSettings))).scalars().all()
sku_settings = {
stock_forecast.sku_key(row.material, row.subtype, row.brand, row.color_name): stock_forecast.SkuSettings(
lead_time_days=row.lead_time_days,
safety_margin_value=row.safety_margin_value,
safety_margin_unit=row.safety_margin_unit,
alerts_snoozed=bool(row.alerts_snoozed),
)
for row in sku_rows
}
if await self._get_bool_setting(db, "spoolman_enabled"):
spools = await self._stock_spools_spoolman()
if spools is None:
return None
# Spoolman owns the usage in this mode and Bambuddy's own history
# table holds nothing for those spools, so the rate is always the
# delta rate. Bambuddy's table is deliberately not consulted: its ids
# are local spool ids, and a Spoolman id that happens to match one
# (left over from before Spoolman was switched on) would borrow an
# unrelated spool's history.
history: dict[int, list[stock_forecast.UsageRecord]] = {}
else:
spools = await self._stock_spools_internal(db)
history = await self._stock_usage_history(db)
return stock_forecast.forecast_all(spools, history, sku_settings, global_lead_time, utcnow_naive())
async def _stock_spools_internal(self, db: AsyncSession) -> list[stock_forecast.StockSpool]:
from backend.app.models.spool import Spool
rows = (await db.execute(select(Spool).where(Spool.archived_at.is_(None)))).scalars().all()
return [
stock_forecast.StockSpool(
id=spool.id,
material=spool.material,
subtype=spool.subtype,
brand=spool.brand,
color_name=spool.color_name,
label_weight=float(spool.label_weight or 0),
weight_used=float(spool.weight_used or 0),
weight_used_baseline=float(spool.weight_used_baseline or 0),
created_at=spool.created_at,
)
for spool in rows
]
async def _stock_usage_history(self, db: AsyncSession) -> dict[int, list[stock_forecast.UsageRecord]]:
from backend.app.models.spool_usage_history import SpoolUsageHistory
rows = (
await db.execute(
select(SpoolUsageHistory.spool_id, SpoolUsageHistory.created_at, SpoolUsageHistory.weight_used)
.order_by(SpoolUsageHistory.created_at.desc())
.limit(_STOCK_FORECAST_HISTORY_LIMIT)
)
).all()
history: dict[int, list[stock_forecast.UsageRecord]] = {}
for spool_id, created_at, weight_used in rows:
if created_at is None:
continue
history.setdefault(spool_id, []).append(stock_forecast.UsageRecord(created_at, float(weight_used or 0)))
return history
async def _stock_spools_spoolman(self) -> list[stock_forecast.StockSpool] | None:
"""Spoolman's spools in the forecast's shape, or None when Spoolman cannot be reached.
``get_all_spools`` leaves archived spools out unless asked, which is the
exclusion the internal query applies explicitly. ``_map_spoolman_spool``
is the same mapping the Inventory page (and so the panel) reads.
"""
from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool
from backend.app.services.spoolman import SpoolmanUnavailableError, get_spoolman_client
client = await get_spoolman_client()
if client is None:
logger.debug("Stock forecast skipped, no Spoolman client (spoolman_enabled without a URL?)")
return None
try:
raw_spools = await client.get_all_spools()
except SpoolmanUnavailableError as e:
logger.debug("Stock forecast skipped, Spoolman unreachable: %s", e)
return None
spools: list[stock_forecast.StockSpool] = []
for raw in raw_spools:
try:
mapped = _map_spoolman_spool(raw)
except ValueError:
continue
created_at = None
if mapped.get("created_at"):
try:
created_at = datetime.fromisoformat(str(mapped["created_at"]).replace("Z", "+00:00"))
except ValueError:
created_at = None
spools.append(
stock_forecast.StockSpool(
id=mapped["id"],
material=mapped["material"],
subtype=mapped.get("subtype"),
brand=mapped.get("brand"),
color_name=mapped.get("color_name"),
label_weight=float(mapped.get("label_weight") or 0),
weight_used=float(mapped.get("weight_used") or 0),
weight_used_baseline=float(mapped.get("weight_used_baseline") or 0),
created_at=created_at,
color_name_is_synthesized=bool(mapped.get("color_name_is_synthesized")),
)
)
return spools
async def _emit_stock_alerts(
self,
db: AsyncSession,
forecasts: stock_forecast.ForecastMap,
wanted: dict[str, bool],
) -> None:
"""Send the notifications for each SKU that has moved into a stock alert condition.
A SKU in break has also reached its reorder point, so both events apply to
it: the break event goes to providers that have it on, and the reorder event
to the providers that have only that one on. A provider with both on gets the
break and no second message.
What is remembered per SKU is the condition and which events had a
subscriber when it was sent. An event with no subscriber is forgotten, so
switching it on later reports the SKUs already in that condition.
"""
active_events = {event for event, on in wanted.items() if on}
for key, (forecast, spool) in forecasts.items():
kind: str | None = None
if not forecast.snoozed:
if forecast.stock_break_alert:
kind = "break"
elif forecast.reorder_alert:
kind = "reorder"
applicable = {"break": {"break", "reorder"}, "reorder": {"reorder"}}.get(kind or "", set())
events = applicable & active_events
if not events:
self._notified_stock_alerts.pop(key, None)
continue
previous = self._notified_stock_alerts.get(key)
told = (previous[1] & active_events) if previous else frozenset()
# Trimmed to the events this condition can send. That is what makes easing from
# break back to reorder silent (the reorder event was told with the break) and
# worsening again news again (the break event is no longer remembered).
self._notified_stock_alerts[key] = (kind, frozenset((told | events) & applicable))
# Recorded before sending, for the reason _emit_filament_low gives: a
# provider that is down should lose this alert, not repeat it every hour.
to_send = events - told
# Both flags imply a positive rate and so a day count.
days_left = forecast.days_remaining if forecast.days_remaining is not None else 0
rate = forecast.daily_rate_g if forecast.daily_rate_g is not None else 0.0
color = None if spool.color_name_is_synthesized else spool.color_name
try:
if "break" in to_send:
await notification_service.on_stock_break_alert(
spool.material,
spool.brand,
forecast.remaining_g,
rate,
days_left,
forecast.effective_lead_time_days,
db,
subtype=spool.subtype,
color=color,
)
if "reorder" in to_send:
await notification_service.on_stock_reorder_alert(
spool.material,
spool.brand,
forecast.remaining_g,
rate,
days_left,
db,
subtype=spool.subtype,
color=color,
skip_break_subscribers=kind == "break",
)
except Exception as e:
logger.warning("Stock %s notification failed for %s: %s", kind, key, e)
# A SKU with no spools left is no longer alerting.
for key in [k for k in self._notified_stock_alerts if k not in forecasts]:
del self._notified_stock_alerts[key]
async def _check_auto_drying(
self,
db: AsyncSession,
backend/app/services/stock_forecast.py
+284
View File
@@ -0,0 +1,284 @@
"""Reorder-point forecast for inventory SKUs (#2955).
This is the arithmetic ``frontend/src/components/ForecastPanel.tsx`` runs in the
browser, as pure functions, so something that is not a browser can decide when
a SKU has reached its reorder point or is about to break. ``test_stock_forecast_2955.py``
pins the two to the same numbers on the same inputs.
The pieces, in the order the panel applies them:
* Spools are grouped by SKU -- ``(material, subtype, brand, color_name)``.
* The daily rate comes from usage history when there are at least two distinct
days of it (weighted by a 30-day half-life, so recent prints dominate), and
from consumption since the spool baseline over the age of the oldest spool
otherwise.
* ``reorder point = rate * lead time + safety stock``, where the safety stock
is a statistical term (``Z_95 * sigma * sqrt(lead time)``) plus a margin the
user sets in days or grams.
* A *stock break* is "the stock runs out before a replenishment can arrive"; a
*reorder* is "the reorder point has been reached". They are exclusive: a SKU
that is already breaking is not also reported as reorderable.
Nothing here touches the database. The caller supplies spools, their usage
history and the settings, and decides what to do with the answer.
"""
from __future__ import annotations
import math
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import date, datetime, timezone
# One-sided 95 % z-score, as in the panel.
Z_95 = 1.65
# Prints from this many days ago count half as much as one from today.
_HALF_LIFE_DAYS = 30.0
_LAMBDA = math.log(2) / _HALF_LIFE_DAYS
DEFAULT_SAFETY_MARGIN_VALUE = 14
DEFAULT_SAFETY_MARGIN_UNIT = "days"
# With no measured spread the panel assumes 20 % of the rate, and with no rate at
# all it assumes 5 g/day so a margin given in days still means something.
_FALLBACK_SIGMA_FRACTION = 0.2
_FALLBACK_RATE_G_PER_DAY = 5
_SECONDS_PER_DAY = 86400.0
SkuKey = tuple[str, str, str, str]
def sku_key(material: str, subtype: str | None, brand: str | None, color_name: str | None) -> SkuKey:
"""The grouping key. ``None`` and an empty string are the same SKU, as in the panel."""
return (material, subtype or "", brand or "", color_name or "")
@dataclass(frozen=True)
class StockSpool:
"""The fields of a spool the forecast reads, whichever inventory mode it came from."""
id: int
material: str
subtype: str | None
brand: str | None
color_name: str | None
label_weight: float
weight_used: float
weight_used_baseline: float
created_at: datetime | None
# True when color_name is not a name the spool carries but the subtype standing in
# for one (Spoolman mode does this so the list has no blank colours). The SKU key
# keeps it, as the panel does; anything shown to a person should not.
color_name_is_synthesized: bool = False
@property
def key(self) -> SkuKey:
return sku_key(self.material, self.subtype, self.brand, self.color_name)
@property
def remaining_g(self) -> float:
return max(0.0, self.label_weight - self.weight_used)
@property
def consumed_g(self) -> float:
"""Consumed since the baseline, so "Reset usage to 0" restarts the forecast (#1390)."""
return max(0.0, self.weight_used - self.weight_used_baseline)
@dataclass(frozen=True)
class UsageRecord:
"""One consumption event: how much a print took from a spool, and when."""
created_at: datetime
weight_used: float
@dataclass(frozen=True)
class SkuSettings:
"""A SKU's user-set overrides. The defaults are the panel's for a SKU with no row."""
lead_time_days: int = 0
safety_margin_value: float = DEFAULT_SAFETY_MARGIN_VALUE
safety_margin_unit: str = DEFAULT_SAFETY_MARGIN_UNIT
alerts_snoozed: bool = False
@dataclass(frozen=True)
class SkuForecast:
key: SkuKey
remaining_g: float
daily_rate_g: float | None
effective_lead_time_days: int
reorder_point_g: float
days_remaining: int | None
days_until_reorder_point: int | None
stock_break_alert: bool
reorder_alert: bool
snoozed: bool
def _utc(moment: datetime) -> datetime:
"""A naive timestamp is UTC: that is what the database stores."""
return moment.replace(tzinfo=timezone.utc) if moment.tzinfo is None else moment.astimezone(timezone.utc)
def history_rate(records: Sequence[UsageRecord], now: datetime) -> tuple[float, float] | None:
"""Daily consumption rate and its standard deviation from usage history.
Records are summed per UTC calendar day, so concurrent multi-spool prints on
one day count together. Each day after the first gives one observation: the
grams printed that day over the gap since the previous day with any usage,
weighted by ``exp(-lambda * age)``. Returns None with fewer than two
distinct days -- there is no gap to measure -- and the caller falls back to
:func:`delta_rate`.
"""
if len(records) < 2:
return None
by_day: dict[date, float] = {}
for record in records:
day = _utc(record.created_at).date()
by_day[day] = by_day.get(day, 0.0) + record.weight_used
if len(by_day) < 2:
return None
now = _utc(now)
days = sorted(by_day.items())
observations: list[tuple[float, float]] = [] # (rate, weight)
for i in range(1, len(days)):
elapsed_days = max((days[i][0] - days[i - 1][0]).days, 1)
midnight = datetime(days[i][0].year, days[i][0].month, days[i][0].day, tzinfo=timezone.utc)
age_days = (now - midnight).total_seconds() / _SECONDS_PER_DAY
observations.append((days[i][1] / elapsed_days, math.exp(-_LAMBDA * age_days)))
total_weight = sum(w for _, w in observations)
if total_weight == 0:
return None
mean = sum(r * w for r, w in observations) / total_weight
variance = sum(w * (r - mean) ** 2 for r, w in observations) / total_weight
return mean, math.sqrt(variance)
def delta_rate(spools: Sequence[StockSpool], now: datetime) -> float | None:
"""Daily rate from consumption since baseline over the age of the oldest spool.
The fallback when there is not enough usage history, which is always the
case in Spoolman mode: Spoolman owns the usage there and Bambuddy's own
``spool_usage_history`` table holds nothing for those spools. Returns None
with nothing consumed, or with under a day of age to divide by.
"""
total_used = sum(s.consumed_g for s in spools)
if total_used == 0:
return None
now = _utc(now)
oldest = now
for spool in spools:
if spool.created_at is not None:
created = _utc(spool.created_at)
if created < oldest:
oldest = created
age_days = (now - oldest).total_seconds() / _SECONDS_PER_DAY
if age_days < 1:
return None
return total_used / age_days
def forecast_sku(
spools: Sequence[StockSpool],
history_by_spool: Mapping[int, Sequence[UsageRecord]],
settings: SkuSettings,
global_lead_time_days: int,
now: datetime,
) -> SkuForecast:
"""Forecast one SKU from its spools. ``spools`` must be non-empty and share a key."""
lead_time = max(global_lead_time_days, settings.lead_time_days)
remaining = sum(s.remaining_g for s in spools)
# Only history from spools that were never reset: a reset spool's earlier
# events have no anchor and would inflate the rate. A spool with no
# baseline is clean and keeps its records.
history: list[UsageRecord] = []
for spool in spools:
if spool.weight_used_baseline == 0:
history.extend(history_by_spool.get(spool.id, ()))
rate: float | None = None
std_dev: float | None = None
measured = history_rate(history, now)
if measured is not None:
rate, std_dev = measured
else:
rate = delta_rate(spools, now)
sigma = std_dev if std_dev is not None else (rate * _FALLBACK_SIGMA_FRACTION if rate is not None else 0.0)
statistical_safety_g = Z_95 * sigma * math.sqrt(lead_time)
if settings.safety_margin_unit == "g":
margin_g = settings.safety_margin_value
elif rate is not None:
margin_g = rate * settings.safety_margin_value
else:
margin_g = settings.safety_margin_value * _FALLBACK_RATE_G_PER_DAY
safety_stock_g = statistical_safety_g + margin_g
reorder_point_g = rate * lead_time + safety_stock_g if rate is not None else 0.0
days_remaining: int | None = None
days_until_rop: int | None = None
if rate is not None and rate > 0:
days_remaining = math.floor(remaining / rate)
days_until_rop = math.floor((remaining - reorder_point_g) / rate)
stock_break = days_remaining is not None and lead_time > 0 and days_remaining <= lead_time
reorder = not stock_break and days_until_rop is not None and days_until_rop <= 0
return SkuForecast(
key=spools[0].key,
remaining_g=remaining,
daily_rate_g=rate,
effective_lead_time_days=lead_time,
reorder_point_g=reorder_point_g,
days_remaining=days_remaining,
days_until_reorder_point=days_until_rop,
stock_break_alert=stock_break,
reorder_alert=reorder,
snoozed=settings.alerts_snoozed,
)
ForecastMap = dict[SkuKey, tuple[SkuForecast, StockSpool]]
"""Each SKU's forecast with the first spool of its group, which names it."""
def forecast_all(
spools: Sequence[StockSpool],
history_by_spool: Mapping[int, Sequence[UsageRecord]],
sku_settings: Mapping[SkuKey, SkuSettings],
global_lead_time_days: int,
now: datetime,
) -> ForecastMap:
"""Forecast every SKU present in ``spools``.
Returns each SKU's forecast with the first spool of its group, which is what
a caller needs to name the SKU (material, subtype, brand, colour) without a
second lookup. The caller passes only spools that count as stock -- archived
ones excluded -- because what is archived differs by inventory mode.
A colour-specific SKU with no settings row of its own falls back to the
colourless row, which is where settings saved before colour became part of
the key still live.
"""
groups: dict[SkuKey, list[StockSpool]] = {}
for spool in spools:
groups.setdefault(spool.key, []).append(spool)
out: ForecastMap = {}
for key, members in groups.items():
settings = sku_settings.get(key)
if settings is None and key[3] != "":
settings = sku_settings.get((key[0], key[1], key[2], ""))
out[key] = (
forecast_sku(members, history_by_spool, settings or SkuSettings(), global_lead_time_days, now),
members[0],
)
return out
backend/tests/unit/services/test_stock_alert_producer_2955.py
+989
View File
@@ -0,0 +1,989 @@
"""The two Inventory stock alerts have a producer now (issue #2955).
``on_stock_reorder_alert`` and ``on_stock_break_alert`` had a column, a schema
field, a route, a template and a UI toggle -- and, until #2945, could not even be
saved -- but nothing that decides a SKU has reached its reorder point: the Forecast
panel does that in the browser, so with no page open nothing could ever alert.
``PrintScheduler._check_stock_forecast`` runs the panel's arithmetic
(``stock_forecast``) from the scheduler loop.
The arithmetic itself is pinned against the panel in ``test_stock_forecast_2955``;
these tests are about *when* it alerts and what it sends.
The scenarios use the delta rate (grams consumed over the age of the spool)
because it needs no usage history: a 1000 g spool created 100 days ago with
``used`` grams consumed runs at ``used / 100`` g/day. With a 6 day lead time and
the default 14 day margin:
used 920 -> 80 g left, 9.2 g/day, 8 days of stock, reorder point ~90 g : reorder
used 960 -> 40 g left, 9.6 g/day, 4 days of stock : break
used 500 -> 500 g left, 5 g/day, 100 days of stock : nothing
"""
import json
import logging
import time
from datetime import datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.models.filament_sku_settings import FilamentSkuSettings
from backend.app.models.notification import NotificationProvider
from backend.app.models.settings import Settings
from backend.app.models.spool import Spool
from backend.app.services.print_scheduler import _STOCK_ALERTS_SETTING_KEY, PrintScheduler
from backend.app.services.spoolman import SpoolmanUnavailableError
from backend.app.utils.local_time import utcnow_naive
REORDER_USED = 920.0
BREAK_USED = 960.0
QUIET_USED = 500.0
async def _spool(
db,
*,
used: float,
color: str | None = "Black",
subtype: str | None = "Basic",
brand: str | None = "Bambu Lab",
age_days: int = 100,
archived: bool = False,
baseline: float = 0.0,
) -> Spool:
spool = Spool(
material="PLA",
subtype=subtype,
brand=brand,
color_name=color,
label_weight=1000,
core_weight=250,
weight_used=used,
weight_used_baseline=baseline,
created_at=utcnow_naive() - timedelta(days=age_days),
archived_at=utcnow_naive() if archived else None,
)
spool.k_profiles = []
spool.assignments = []
db.add(spool)
await db.flush()
return spool
async def _sku(db, *, lead: int = 6, color: str | None = "Black", snoozed: bool = False, margin: int = 14):
row = FilamentSkuSettings(
material="PLA",
subtype="Basic",
brand="Bambu Lab",
color_name=color,
lead_time_days=lead,
safety_margin_value=margin,
safety_margin_unit="days",
alerts_snoozed=snoozed,
)
db.add(row)
await db.flush()
return row
@pytest.fixture
def scheduler():
return PrintScheduler()
async def _pass(scheduler, db):
"""Run one check, stepping over the hourly gate (which has its own test)."""
scheduler._stock_forecast_next_check = 0.0
await scheduler._check_stock_forecast(db)
@pytest.fixture
def notify():
"""The service, replaced, with a provider that wants both events.
Every test about the producer assumes someone wants the alerts; the guard in
front of it has its own tests further down, against the real provider query.
"""
with patch("backend.app.services.print_scheduler.notification_service") as ns:
ns.on_stock_reorder_alert = AsyncMock()
ns.on_stock_break_alert = AsyncMock()
ns._get_providers_for_event = AsyncMock(return_value=[MagicMock()])
yield ns
# -- a restart does not start it over ----------------------------------------
async def _stored(db) -> list | None:
"""What the settings row holds, parsed. None when there is no row."""
from sqlalchemy import select
row = (await db.execute(select(Settings).where(Settings.key == _STOCK_ALERTS_SETTING_KEY))).scalar_one_or_none()
return json.loads(row.value) if row else None
@pytest.mark.asyncio
async def test_a_fresh_scheduler_does_not_re_send_what_is_already_stored(db_session, scheduler, notify):
"""The first check runs as soon as Bambuddy starts. Without the stored set, every update
re-sent one message per SKU that was still low."""
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
restarted = PrintScheduler()
await _pass(restarted, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
notify.on_stock_break_alert.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_stored_reorder_still_lets_the_worsening_to_a_break_through_after_a_restart(
db_session, scheduler, notify
):
await _sku(db_session)
spool = await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
spool.weight_used = BREAK_USED
await db_session.commit()
restarted = PrintScheduler()
await _pass(restarted, db_session)
notify.on_stock_break_alert.assert_awaited_once()
@pytest.mark.asyncio
async def test_a_sku_that_cleared_while_it_was_down_alerts_again_after_a_restart(db_session, scheduler, notify):
await _sku(db_session)
spool = await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
spool.weight_used = QUIET_USED
await db_session.commit()
await _pass(PrintScheduler(), db_session)
spool.weight_used = REORDER_USED
await db_session.commit()
await _pass(PrintScheduler(), db_session)
assert notify.on_stock_reorder_alert.await_count == 2
@pytest.mark.asyncio
async def test_the_stored_set_drops_a_sku_that_no_longer_exists(db_session, scheduler, notify):
await _sku(db_session)
spool = await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
assert len(await _stored(db_session)) == 1
spool.archived_at = utcnow_naive()
await db_session.commit()
await _pass(PrintScheduler(), db_session)
assert await _stored(db_session) == []
@pytest.mark.asyncio
async def test_the_stored_set_is_emptied_when_nobody_wants_either_event(db_session, scheduler, real_providers):
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
provider_row = NotificationProvider(
name="p", provider_type="ntfy", config="{}", enabled=True, on_stock_reorder_alert=True
)
db_session.add(provider_row)
await db_session.commit()
await _pass(scheduler, db_session)
assert len(await _stored(db_session)) == 1
provider_row.on_stock_reorder_alert = False
await db_session.commit()
await _pass(PrintScheduler(), db_session)
assert await _stored(db_session) == []
@pytest.mark.asyncio
async def test_an_unchanged_set_is_not_written_again(db_session, scheduler, notify):
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
with patch("backend.app.core.db_dialect.upsert_setting", AsyncMock()) as upsert:
await _pass(scheduler, db_session)
await _pass(PrintScheduler(), db_session)
upsert.assert_not_awaited()
@pytest.mark.asyncio
async def test_an_unreadable_stored_set_is_treated_as_empty(db_session, scheduler, notify):
db_session.add(Settings(key=_STOCK_ALERTS_SETTING_KEY, value="not json {"))
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_awaited_once()
assert len(await _stored(db_session)) == 1
# -- when it alerts ----------------------------------------------------------
@pytest.mark.asyncio
async def test_a_sku_at_its_reorder_point_alerts_once_with_its_numbers(db_session, scheduler, notify):
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_break_alert.assert_not_awaited()
notify.on_stock_reorder_alert.assert_awaited_once()
material, brand, stock_g, rate, days_left, _db = notify.on_stock_reorder_alert.await_args.args
assert (material, brand) == ("PLA", "Bambu Lab")
assert stock_g == pytest.approx(80.0)
assert rate == pytest.approx(9.2, rel=1e-3)
assert days_left == 8
kwargs = notify.on_stock_reorder_alert.await_args.kwargs
assert kwargs == {"subtype": "Basic", "color": "Black", "skip_break_subscribers": False}
@pytest.mark.asyncio
async def test_a_sku_that_runs_out_before_the_lead_time_alerts_as_a_break(db_session, scheduler, notify):
await _sku(db_session)
await _spool(db_session, used=BREAK_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_break_alert.assert_awaited_once()
args = notify.on_stock_break_alert.await_args.args
assert args[2] == pytest.approx(40.0)
assert args[4] == 4 # days left
assert args[5] == 6 # lead time
assert notify.on_stock_break_alert.await_args.kwargs == {"subtype": "Basic", "color": "Black"}
# A breaking SKU has also reached its reorder point: the reorder event goes too, minus the
# providers that have the break alert on (the service applies that filter; see below).
notify.on_stock_reorder_alert.assert_awaited_once()
assert notify.on_stock_reorder_alert.await_args.kwargs["skip_break_subscribers"] is True
@pytest.mark.asyncio
async def test_a_sku_with_plenty_of_stock_stays_quiet(db_session, scheduler, notify):
await _sku(db_session)
await _spool(db_session, used=QUIET_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_not_awaited()
notify.on_stock_break_alert.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_sku_that_has_not_been_used_has_no_rate_and_stays_quiet(db_session, scheduler, notify):
await _sku(db_session)
await _spool(db_session, used=0.0)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_not_awaited()
notify.on_stock_break_alert.assert_not_awaited()
@pytest.mark.asyncio
async def test_the_global_lead_time_counts(db_session, scheduler, notify):
"""No SKU row at all: the lead time is the global setting, and 40 g at 9.6 g/day is 4 days."""
db_session.add(Settings(key="forecast_global_lead_time_days", value="30"))
await _spool(db_session, used=BREAK_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_break_alert.assert_awaited_once()
assert notify.on_stock_break_alert.await_args.args[5] == 30
@pytest.mark.asyncio
async def test_usage_history_beats_the_delta_rate_when_there_is_enough_of_it(db_session, scheduler, notify):
"""300 g used over 100 days is 3 g/day and 700 g left is months of stock. But the
last three days took 300 g each: the history rate says two days, so it is a break.
Also the only test that reads ``spool_usage_history`` through the scheduler.
"""
from backend.app.models.spool_usage_history import SpoolUsageHistory
await _sku(db_session)
spool = await _spool(db_session, used=300.0)
for days_ago in (3, 2, 1):
db_session.add(
SpoolUsageHistory(
spool_id=spool.id, weight_used=300.0, created_at=utcnow_naive() - timedelta(days=days_ago)
)
)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_break_alert.assert_awaited_once()
@pytest.mark.asyncio
async def test_history_of_a_reset_spool_is_left_out(db_session, scheduler, notify):
"""Same history, but the spool was reset (baseline set): its earlier events have no
anchor, so the rate falls back to the delta rate -- 3 g/day, nothing to alert."""
from backend.app.models.spool_usage_history import SpoolUsageHistory
await _sku(db_session)
spool = await _spool(db_session, used=300.0, baseline=100.0)
for days_ago in (3, 2, 1):
db_session.add(
SpoolUsageHistory(
spool_id=spool.id, weight_used=300.0, created_at=utcnow_naive() - timedelta(days=days_ago)
)
)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_break_alert.assert_not_awaited()
notify.on_stock_reorder_alert.assert_not_awaited()
# -- once per transition -----------------------------------------------------
@pytest.mark.asyncio
async def test_does_not_repeat_on_the_next_pass(db_session, scheduler, notify):
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
await _pass(scheduler, db_session)
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_awaited_once()
@pytest.mark.asyncio
async def test_re_arms_once_the_condition_clears(db_session, scheduler, notify):
"""Restocked (a fresh roll brings the SKU back above the line), then low again."""
await _sku(db_session)
spool = await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
spool.weight_used = QUIET_USED
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
spool.weight_used = REORDER_USED
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 2
@pytest.mark.asyncio
async def test_worsening_from_reorder_to_break_alerts_again_as_a_break(db_session, scheduler, notify):
await _sku(db_session)
spool = await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
spool.weight_used = BREAK_USED
await db_session.commit()
await _pass(scheduler, db_session)
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_awaited_once()
notify.on_stock_break_alert.assert_awaited_once()
@pytest.mark.asyncio
async def test_easing_from_break_to_reorder_is_not_announced_again(db_session, scheduler, notify):
"""Ordered a few rolls' worth, so it is no longer a break but still under the reorder point.
"Reached the reorder point" straight after "order immediately" is the opposite news."""
await _sku(db_session)
spool = await _spool(db_session, used=BREAK_USED)
await db_session.commit()
await _pass(scheduler, db_session)
spool.weight_used = REORDER_USED
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_break_alert.await_count == 1
assert notify.on_stock_reorder_alert.await_count == 1 # the break's own, from before it eased
# Worse again is news again.
spool.weight_used = BREAK_USED
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_break_alert.await_count == 2
assert notify.on_stock_reorder_alert.await_count == 1
@pytest.mark.asyncio
async def test_two_colours_of_one_product_alert_separately_and_say_which(db_session, scheduler, notify):
"""The SKU is material/subtype/brand/colour, so the messages must differ by colour."""
await _sku(db_session, color="Black")
await _sku(db_session, color="White")
await _spool(db_session, used=REORDER_USED, color="Black")
await _spool(db_session, used=REORDER_USED, color="White")
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 2
colours = {call.kwargs["color"] for call in notify.on_stock_reorder_alert.await_args_list}
assert colours == {"Black", "White"}
@pytest.mark.asyncio
async def test_a_sku_that_disappears_and_comes_back_low_alerts_again(db_session, scheduler, notify):
"""The roll runs out and is archived, so the SKU has no spools; a new roll is later low.
The old alert must not silence the new roll: nothing has cleared the SKU's
entry, because it was not in the forecast to be cleared.
"""
await _sku(db_session)
spool = await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
spool.archived_at = utcnow_naive()
await db_session.commit()
await _pass(scheduler, db_session)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 2
@pytest.mark.asyncio
async def test_spools_of_one_sku_are_added_together(db_session, scheduler, notify):
"""Two half-full spools are one SKU with 1000 g, not two SKUs with 500 g each."""
await _sku(db_session)
await _spool(db_session, used=REORDER_USED) # 80 g left, on its own a reorder
await _spool(db_session, used=QUIET_USED) # 500 g left; together 580 g at 14.2 g/day, 40 days
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_not_awaited()
notify.on_stock_break_alert.assert_not_awaited()
# -- snooze and archive ------------------------------------------------------
@pytest.mark.asyncio
async def test_a_snoozed_sku_does_not_alert_and_alerts_when_unsnoozed(db_session, scheduler, notify):
row = await _sku(db_session, snoozed=True)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_not_awaited()
row.alerts_snoozed = False
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_awaited_once()
@pytest.mark.asyncio
async def test_a_colour_with_no_row_of_its_own_uses_the_colourless_row(db_session, scheduler, notify):
"""Settings saved before colour became part of the key live on the colourless row."""
await _sku(db_session, color=None, snoozed=True)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_not_awaited()
@pytest.mark.asyncio
async def test_an_archived_spool_is_not_stock(db_session, scheduler, notify):
"""It would supply 900 g of stock the SKU does not have."""
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await _spool(db_session, used=0.0, archived=True)
await db_session.commit()
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_awaited_once()
assert notify.on_stock_reorder_alert.await_args.args[2] == pytest.approx(80.0)
# -- failure and cost --------------------------------------------------------
@pytest.mark.asyncio
async def test_a_failing_provider_does_not_break_the_pass(db_session, scheduler, notify):
await _sku(db_session)
await _sku(db_session, color="White")
await _spool(db_session, used=REORDER_USED, color="Black")
await _spool(db_session, used=REORDER_USED, color="White")
await db_session.commit()
notify.on_stock_reorder_alert.side_effect = [RuntimeError("provider down"), None]
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 2
# The failed alert is lost, not retried every hour: that repetition is what the
# debounce is for, and a provider that is down is the case where it would never stop.
notify.on_stock_reorder_alert.side_effect = None
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 2
@pytest.mark.asyncio
async def test_a_second_pass_inside_the_interval_does_no_work(db_session, scheduler, notify):
"""The gate. run() re-enters the checks every 3 s while an upload is in flight."""
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
with patch.object(scheduler, "_stock_forecasts", new_callable=AsyncMock) as work:
work.return_value = {}
await scheduler._check_stock_forecast(db_session)
wait = scheduler._stock_forecast_next_check - time.monotonic()
await scheduler._check_stock_forecast(db_session)
await scheduler._check_stock_forecast(db_session)
assert work.await_count == 1
# Hourly: the forecast is in whole days, so a finer check repeats the same answer.
assert 3500 < wait <= 3600
# -- the provider guard ------------------------------------------------------
async def _provider(db, *, enabled: bool = True, reorder: bool = False, brk: bool = False) -> None:
db.add(
NotificationProvider(
name=f"p-{enabled}-{reorder}-{brk}",
provider_type="ntfy",
config="{}",
enabled=enabled,
on_stock_reorder_alert=reorder,
on_stock_break_alert=brk,
)
)
await db.flush()
@pytest.fixture
def real_providers():
"""The real service, with only the two sends replaced, so the guard runs its real query."""
from backend.app.services.print_scheduler import notification_service
with (
patch.object(notification_service, "on_stock_reorder_alert", new_callable=AsyncMock) as reorder,
patch.object(notification_service, "on_stock_break_alert", new_callable=AsyncMock) as brk,
):
yield reorder, brk
@pytest.mark.asyncio
async def test_no_work_when_no_provider_wants_either_event(db_session, scheduler, real_providers):
"""Both toggles default to off, so on most installs nobody wants this.
Unguarded, every install read its whole spool collection every hour (in
Spoolman mode, over HTTP) for alerts that were switched off. The two
providers are the two ways of not wanting them: enabled with both events off,
and the events on but the provider disabled.
"""
await _provider(db_session, enabled=True)
await _provider(db_session, enabled=False, reorder=True, brk=True)
await _spool(db_session, used=REORDER_USED)
await db_session.commit()
with patch.object(scheduler, "_stock_forecasts", new_callable=AsyncMock) as work:
await _pass(scheduler, db_session)
work.assert_not_awaited()
@pytest.mark.asyncio
async def test_an_event_nobody_wants_is_not_recorded_as_sent(db_session, scheduler, real_providers):
"""Only the break event is on, and the SKU is at its reorder point. Switching
the reorder event on later must report it -- it was never sent."""
reorder, brk = real_providers
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
provider_row = NotificationProvider(
name="p", provider_type="ntfy", config="{}", enabled=True, on_stock_break_alert=True
)
db_session.add(provider_row)
await db_session.commit()
await _pass(scheduler, db_session)
reorder.assert_not_awaited()
brk.assert_not_awaited()
provider_row.on_stock_reorder_alert = True
await db_session.commit()
await _pass(scheduler, db_session)
reorder.assert_awaited_once()
@pytest.mark.asyncio
async def test_a_reorder_only_provider_still_hears_about_a_break(db_session, scheduler, real_providers):
"""A SKU that is breaking has certainly reached its reorder point, which is what the
Reorder Alert toggle promises. Without this, the worst state is the one a provider
with only that toggle on is never told about."""
reorder, brk = real_providers
await _sku(db_session)
await _spool(db_session, used=BREAK_USED)
await _provider(db_session, reorder=True)
await db_session.commit()
await _pass(scheduler, db_session)
reorder.assert_awaited_once()
assert reorder.await_args.kwargs["skip_break_subscribers"] is True
brk.assert_not_awaited() # nobody has it on
@pytest.mark.asyncio
async def test_switching_the_break_alert_on_later_reports_the_break(db_session, scheduler, real_providers):
"""The reorder-only provider was told when the SKU broke. A provider that turns the break
alert on afterwards is told too -- and the reorder-only one is not told a second time."""
reorder, brk = real_providers
await _sku(db_session)
await _spool(db_session, used=BREAK_USED)
await _provider(db_session, reorder=True)
await db_session.commit()
await _pass(scheduler, db_session)
assert reorder.await_count == 1
await _provider(db_session, brk=True)
await db_session.commit()
await _pass(scheduler, db_session)
brk.assert_awaited_once()
assert reorder.await_count == 1
@pytest.mark.asyncio
async def test_switching_one_event_off_and_on_again_reports_it_while_the_other_stays_on(
db_session, scheduler, real_providers
):
"""A wants reorders, B wants breaks, and the SKU is breaking: both were told. B switches
the break alert off (A keeps the check running) and on again: B is told again."""
reorder, brk = real_providers
await _sku(db_session)
await _spool(db_session, used=BREAK_USED)
await _provider(db_session, reorder=True)
b_provider = NotificationProvider(
name="b", provider_type="ntfy", config="{}", enabled=True, on_stock_break_alert=True
)
db_session.add(b_provider)
await db_session.commit()
await _pass(scheduler, db_session)
assert (reorder.await_count, brk.await_count) == (1, 1)
b_provider.on_stock_break_alert = False
await db_session.commit()
await _pass(scheduler, db_session)
b_provider.on_stock_break_alert = True
await db_session.commit()
await _pass(scheduler, db_session)
assert brk.await_count == 2
assert reorder.await_count == 1
@pytest.mark.asyncio
async def test_a_break_alert_only_provider_is_not_told_about_a_reorder(db_session, scheduler, real_providers):
reorder, brk = real_providers
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
await _provider(db_session, brk=True)
await db_session.commit()
await _pass(scheduler, db_session)
reorder.assert_not_awaited()
brk.assert_not_awaited()
@pytest.mark.asyncio
async def test_a_provider_with_both_on_gets_the_break_and_no_second_message(db_session, scheduler):
"""Through the real service: A has both toggles on, B only reorder. A break SKU goes to A as
a break and to B as a reorder -- and A does not also get the reorder."""
from backend.app.services.notification_service import NotificationService
from backend.app.services.print_scheduler import notification_service
await _sku(db_session)
await _spool(db_session, used=BREAK_USED)
both = NotificationProvider(
name="both",
provider_type="ntfy",
config="{}",
enabled=True,
on_stock_reorder_alert=True,
on_stock_break_alert=True,
)
only_reorder = NotificationProvider(
name="only-reorder", provider_type="ntfy", config="{}", enabled=True, on_stock_reorder_alert=True
)
db_session.add_all([both, only_reorder])
await db_session.commit()
sent: list[tuple[str, list[str]]] = []
async def record(providers, title, message, db, event_type, **kwargs):
sent.append((event_type, sorted(p.name for p in providers)))
with (
patch.object(notification_service, "_send_to_providers", side_effect=record),
patch.object(notification_service, "_build_message_from_template", AsyncMock(return_value=("t", "b"))),
):
await _pass(scheduler, db_session)
assert isinstance(notification_service, NotificationService)
assert sorted(sent) == [("stock_break_alert", ["both"]), ("stock_reorder_alert", ["only-reorder"])]
@pytest.mark.asyncio
async def test_switching_every_event_off_and_on_again_reports_what_is_low_now(db_session, scheduler, real_providers):
reorder, _brk = real_providers
await _sku(db_session)
await _spool(db_session, used=REORDER_USED)
provider_row = NotificationProvider(
name="p", provider_type="ntfy", config="{}", enabled=True, on_stock_reorder_alert=True
)
db_session.add(provider_row)
await db_session.commit()
await _pass(scheduler, db_session)
assert reorder.await_count == 1
provider_row.on_stock_reorder_alert = False
await db_session.commit()
await _pass(scheduler, db_session)
provider_row.on_stock_reorder_alert = True
await db_session.commit()
await _pass(scheduler, db_session)
assert reorder.await_count == 2
# -- Spoolman mode -----------------------------------------------------------
def _spoolman_spool(spool_id: int, used: float, *, color: str = "Black", age_days: int = 100) -> dict:
"""A raw Spoolman spool, in the shape _map_spoolman_spool reads."""
registered = (utcnow_naive() - timedelta(days=age_days)).isoformat() + "Z"
return {
"id": spool_id,
"remaining_weight": 1000 - used,
"used_weight": used,
"registered": registered,
"filament": {
"id": 1,
"name": "PLA Basic",
"material": "PLA",
"weight": 1000,
"color_name": color,
"vendor": {"name": "Bambu Lab"},
},
}
def _spoolman_client(spools, *, unavailable: bool = False):
client = MagicMock()
if unavailable:
client.get_all_spools = AsyncMock(side_effect=SpoolmanUnavailableError("Cannot reach Spoolman"))
else:
client.get_all_spools = AsyncMock(return_value=spools)
return client
@pytest.mark.asyncio
async def test_spoolman_mode_alerts_from_one_collection_read(db_session, scheduler, notify):
db_session.add(Settings(key="spoolman_enabled", value="true"))
await db_session.commit()
client = _spoolman_client([_spoolman_spool(7, used=REORDER_USED), _spoolman_spool(8, used=0.0, color="White")])
with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
await _pass(scheduler, db_session)
# Black has 80 g at 9.2 g/day and no SKU row: lead time 0, so a reorder needs the
# 14 day margin -- 80 g is under the ~128 g that gives.
notify.on_stock_reorder_alert.assert_awaited_once()
assert notify.on_stock_reorder_alert.await_args.kwargs["color"] == "Black"
client.get_all_spools.assert_awaited_once()
@pytest.mark.asyncio
async def test_an_unreachable_spoolman_stays_quiet_and_does_not_forget(db_session, scheduler, notify, caplog):
"""An outage is an ordinary state for a third-party service. It must not raise, and it
must not clear what was sent: the SKU would alert again the moment Spoolman came back."""
db_session.add(Settings(key="spoolman_enabled", value="true"))
await db_session.commit()
spools = [_spoolman_spool(7, used=REORDER_USED)]
with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client(spools))):
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
down = _spoolman_client([], unavailable=True)
with (
caplog.at_level(logging.WARNING, logger="backend.app.services.print_scheduler"),
patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=down)),
):
await _pass(scheduler, db_session)
# An ordinary state for a third-party service, not a traceback every hour.
assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client(spools))):
await _pass(scheduler, db_session)
assert notify.on_stock_reorder_alert.await_count == 1
@pytest.mark.asyncio
async def test_a_colour_spoolman_made_up_from_the_subtype_is_not_sent_as_a_colour(db_session, scheduler, notify):
"""With no colour of its own, Spoolman mode shows the subtype in its place so the list has
no blank colours. In a message that reads "PLA Basic Basic", so the colour is left out."""
db_session.add(Settings(key="spoolman_enabled", value="true"))
await db_session.commit()
raw = _spoolman_spool(7, used=REORDER_USED)
del raw["filament"]["color_name"]
raw["filament"]["name"] = "Basic"
with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=_spoolman_client([raw]))):
await _pass(scheduler, db_session)
notify.on_stock_reorder_alert.assert_awaited_once()
assert notify.on_stock_reorder_alert.await_args.kwargs["color"] is None
@pytest.mark.asyncio
async def test_spoolman_mode_does_not_borrow_a_local_spools_history(db_session, scheduler, notify):
"""A Spoolman id that matches a local spool id (left from before Spoolman was switched on)
must not pick up that spool's usage history. Here the local spool 1 has a history that would
make this a break; Spoolman's spool 1 is quiet by its own delta rate."""
from backend.app.models.spool_usage_history import SpoolUsageHistory
local = await _spool(db_session, used=300.0)
for days_ago in (3, 2, 1):
db_session.add(
SpoolUsageHistory(
spool_id=local.id, weight_used=300.0, created_at=utcnow_naive() - timedelta(days=days_ago)
)
)
db_session.add(Settings(key="spoolman_enabled", value="true"))
await _sku(db_session)
await db_session.commit()
client = _spoolman_client([_spoolman_spool(local.id, used=300.0)])
with patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)):
await _pass(scheduler, db_session)
notify.on_stock_break_alert.assert_not_awaited()
notify.on_stock_reorder_alert.assert_not_awaited()
# -- the service -------------------------------------------------------------
@pytest.mark.asyncio
async def test_the_events_reach_a_provider_with_subtype_and_colour():
"""Both events pass subtype and colour through to the template variables a provider is sent."""
from backend.app.services.notification_service import NotificationService
service = NotificationService()
provider = MagicMock()
provider.id = 1
with (
patch.object(service, "_get_providers_for_event", new_callable=AsyncMock) as mock_get,
patch.object(service, "_send_to_providers", new_callable=AsyncMock) as mock_send,
patch.object(service, "_build_message_from_template", new_callable=AsyncMock) as mock_build,
):
mock_get.return_value = [provider]
mock_build.return_value = ("t", "b")
await service.on_stock_reorder_alert(
"PLA", "Bambu Lab", 80.0, 9.2, 8, AsyncMock(), subtype="Matte", color="Charcoal"
)
await service.on_stock_break_alert(
"PLA", "Bambu Lab", 40.0, 9.6, 4, 6, AsyncMock(), subtype="Matte", color="Charcoal"
)
assert [c.args[1] for c in mock_get.await_args_list] == ["on_stock_reorder_alert", "on_stock_break_alert"]
assert mock_send.await_count == 2
for call in mock_send.await_args_list:
variables = call.kwargs["variables"]
assert variables["subtype"] == "Matte"
assert variables["color"] == "Charcoal"
assert variables["material"] == "PLA"
@pytest.mark.asyncio
async def test_subtype_and_colour_are_optional_so_the_old_call_shape_still_works():
"""Neither method had a caller, but the new arguments are keyword-only with defaults regardless."""
from backend.app.services.notification_service import NotificationService
service = NotificationService()
with (
patch.object(service, "_get_providers_for_event", new_callable=AsyncMock, return_value=[MagicMock()]),
patch.object(service, "_send_to_providers", new_callable=AsyncMock) as mock_send,
patch.object(service, "_build_message_from_template", new_callable=AsyncMock, return_value=("t", "b")),
):
await service.on_stock_reorder_alert("PLA", None, 80.0, 9.2, 8, AsyncMock())
variables = mock_send.await_args.kwargs["variables"]
assert variables["subtype"] == ""
assert variables["color"] == ""
@pytest.mark.asyncio
async def test_a_spoolman_registered_date_with_a_trailing_z_still_gives_the_spool_an_age(db_session, scheduler, notify):
"""Python 3.10's fromisoformat rejects a trailing "Z" (pyproject still allows 3.10), which
silently dropped the spool's age and with it the delta rate. Simulated here so the test
means the same on every interpreter."""
class _Py310Datetime(datetime):
@classmethod
def fromisoformat(cls, value):
if value.endswith("Z"):
raise ValueError(f"Invalid isoformat string: {value!r}")
return super().fromisoformat(value)
db_session.add(Settings(key="spoolman_enabled", value="true"))
await db_session.commit()
client = _spoolman_client([_spoolman_spool(7, used=REORDER_USED)])
with (
patch("backend.app.services.print_scheduler.datetime", _Py310Datetime),
patch("backend.app.services.spoolman.get_spoolman_client", AsyncMock(return_value=client)),
):
await _pass(scheduler, db_session)
# 920 g in 100 days is 9.2 g/day: a reorder. With the age dropped the rate falls back to a
# different number and this SKU is not at its reorder point.
notify.on_stock_reorder_alert.assert_awaited_once()
assert notify.on_stock_reorder_alert.await_args.args[3] == pytest.approx(9.2, abs=0.05)
backend/tests/unit/services/test_stock_forecast_2955.py
+234
View File
@@ -0,0 +1,234 @@
"""The backend forecast against ForecastPanel.tsx, on the same inputs (#2955).
``stock_forecast`` is the panel's arithmetic moved out of the browser so the
stock alerts have something to run on. The two exist side by side until the
panel reads the backend's values, so these tests pin them: ``EXPECTED`` was
produced by running the panel's own functions (``computeHistoryRate``,
``computeDeltaRate`` and the block that turns them into a reorder point and the
two alert flags, copied verbatim from ``ForecastPanel.tsx``) under Node with
``Date.now()`` fixed to ``NOW`` and ``TZ=UTC``, on the scenarios below. A change
to either side that moves a number shows up here.
``TZ=UTC`` matters for exactly one thing: the panel parses a timezone-less
``created_at`` as browser-local time, the backend reads it as UTC, and the delta
rate divides by an age in days, so the two differ by the browser's offset. The
database stores UTC, so UTC is what the timestamps mean.
"""
from datetime import datetime, timedelta, timezone
import pytest
from backend.app.services.stock_forecast import (
SkuSettings,
StockSpool,
UsageRecord,
delta_rate,
forecast_all,
forecast_sku,
history_rate,
sku_key,
)
NOW = datetime(2026, 9, 30, tzinfo=timezone.utc)
def _ago(days: float) -> datetime:
"""A naive UTC timestamp ``days`` before NOW, the shape the database returns."""
return (NOW - timedelta(days=days)).replace(tzinfo=None)
def _spool(spool_id: int, used: float, age_days: float, baseline: float = 0) -> StockSpool:
return StockSpool(
id=spool_id,
material="PLA",
subtype="Basic",
brand="Bambu Lab",
color_name="Black",
label_weight=1000,
weight_used=used,
weight_used_baseline=baseline,
created_at=_ago(age_days),
)
def _history(*events: tuple[int, float, float]) -> dict[int, list[UsageRecord]]:
"""(spool id, days ago, grams) triples, grouped by spool the way the caller does."""
out: dict[int, list[UsageRecord]] = {}
for spool_id, ago, grams in events:
out.setdefault(spool_id, []).append(UsageRecord(created_at=_ago(ago), weight_used=grams))
return out
# name -> (spools, history, settings, global lead time). The same scenarios, in the
# same order, as the Node run that produced EXPECTED.
SCENARIOS = {
# History rate, no settings row, lead time from the global setting.
"A": (
[_spool(1, 600, 60), _spool(2, 0, 10)],
_history((1, 40, 120), (1, 25, 90), (1, 12, 200), (1, 3, 150)),
None,
7,
),
# No history: the delta rate, with the margin given in days.
"B": ([_spool(1, 300, 20)], {}, SkuSettings(lead_time_days=14, safety_margin_value=10), 0),
# A reset spool's history is left out; the clean spool's is kept.
"C": (
[_spool(1, 700, 50, baseline=200), _spool(2, 300, 30)],
_history((1, 20, 400), (1, 10, 50), (2, 20, 100), (2, 6, 100), (2, 1, 100)),
SkuSettings(lead_time_days=5),
0,
),
# Stock break from the delta rate: 100 g left, 30 days of lead time.
"D": ([_spool(1, 900, 30)], {}, SkuSettings(lead_time_days=30), 0),
# Margin in grams.
"E": (
[_spool(1, 500, 90)],
_history((1, 30, 80), (1, 20, 60), (1, 10, 100)),
SkuSettings(lead_time_days=10, safety_margin_value=150, safety_margin_unit="g"),
0,
),
# Nothing consumed: no rate, so no dates and no alerts.
"F": ([_spool(1, 0, 40)], {}, None, 7),
# Younger than a day: the delta rate is not measurable. The global lead time wins over the SKU's.
"G": ([_spool(1, 200, 0.5)], {}, SkuSettings(lead_time_days=3), 9),
# Under the reorder point with more than the lead time of stock left: a reorder.
"H": (
[_spool(1, 850, 100)],
_history((1, 50, 200), (1, 40, 200), (1, 20, 200)),
SkuSettings(lead_time_days=6, safety_margin_value=3),
0,
),
# The same history with less left: a stock break, not also a reorder.
"I": (
[_spool(1, 930, 100)],
_history((1, 50, 200), (1, 40, 200), (1, 20, 200)),
SkuSettings(lead_time_days=6, safety_margin_value=3),
0,
),
# Two spools of one SKU: the delta rate divides by the age of the oldest.
"J": ([_spool(1, 400, 400), _spool(2, 100, 5)], {}, SkuSettings(lead_time_days=10), 0),
}
EXPECTED = {
"A": {
"remaining_g": 1400,
"daily_rate_g": 13.57710201710711,
"effective_lead_time_days": 7,
"reorder_point_g": 304.32790456992615,
"days_remaining": 103,
"days_until_reorder_point": 80,
"stock_break_alert": False,
"reorder_alert": False,
},
"B": {
"remaining_g": 700,
"daily_rate_g": 15,
"effective_lead_time_days": 14,
"reorder_point_g": 378.521204064531,
"days_remaining": 46,
"days_until_reorder_point": 21,
"stock_break_alert": False,
"reorder_alert": False,
},
"C": {
"remaining_g": 1000,
"daily_rate_g": 13.942344991570476,
"effective_lead_time_days": 5,
"reorder_point_g": 288.583334452561,
"days_remaining": 71,
"days_until_reorder_point": 51,
"stock_break_alert": False,
"reorder_alert": False,
},
"D": {
"remaining_g": 100,
"daily_rate_g": 30,
"effective_lead_time_days": 30,
"reorder_point_g": 1374.2245331930114,
"days_remaining": 3,
"days_until_reorder_point": -43,
"stock_break_alert": True,
"reorder_alert": False,
},
"E": {
"remaining_g": 500,
"daily_rate_g": 8.230026663902231,
"effective_lead_time_days": 10,
"reorder_point_g": 242.666532290416,
"days_remaining": 60,
"days_until_reorder_point": 31,
"stock_break_alert": False,
"reorder_alert": False,
},
"F": {
"remaining_g": 1000,
"daily_rate_g": None,
"effective_lead_time_days": 7,
"reorder_point_g": 0,
"days_remaining": None,
"days_until_reorder_point": None,
"stock_break_alert": False,
"reorder_alert": False,
},
"G": {
"remaining_g": 800,
"daily_rate_g": None,
"effective_lead_time_days": 9,
"reorder_point_g": 0,
"days_remaining": None,
"days_until_reorder_point": None,
"stock_break_alert": False,
"reorder_alert": False,
},
"H": {
"remaining_g": 150,
"daily_rate_g": 13.864882095643093,
"effective_lead_time_days": 6,
"reorder_point_g": 144.46457585381498,
"days_remaining": 10,
"days_until_reorder_point": 0,
"stock_break_alert": False,
"reorder_alert": True,
},
"I": {
"remaining_g": 70,
"daily_rate_g": 13.864882095643093,
"effective_lead_time_days": 6,
"reorder_point_g": 144.46457585381498,
"days_remaining": 5,
"days_until_reorder_point": -6,
"stock_break_alert": True,
"reorder_alert": False,
},
"J": {
"remaining_g": 1500,
"daily_rate_g": 1.25,
"effective_lead_time_days": 10,
"reorder_point_g": 31.304439534819455,
"days_remaining": 1200,
"days_until_reorder_point": 1174,
"stock_break_alert": False,
"reorder_alert": False,
},
}
@pytest.mark.parametrize("name", sorted(SCENARIOS))
def test_backend_matches_the_panels_numbers(name):
spools, history, settings, global_lead = SCENARIOS[name]
got = forecast_sku(spools, history, settings or SkuSettings(), global_lead, NOW)
expected = EXPECTED[name]
for field, want in expected.items():
have = getattr(got, field)
if isinstance(want, float):
assert have == pytest.approx(want, rel=1e-9), field
else:
assert have == want, field
def test_the_scenarios_cover_both_alerts_and_neither():
"""The pin means little if no scenario ever alerts."""
flags = {(e["reorder_alert"], e["stock_break_alert"]) for e in EXPECTED.values()}
assert flags == {(False, False), (True, False), (False, True)}
backend/tests/unit/test_stock_alert_template_migration_2955.py
+92
View File
@@ -0,0 +1,92 @@
"""The stock alert templates name the colour and subtype (#2955).
The forecast groups by colour, so two colours of one product would otherwise send
the same message. The migration rewrites a template body IF AND ONLY IF it is
still the shipped default -- an admin who edited the wording keeps it, the same
guard as the ha_sensor_alert rename (#2824).
"""
from __future__ import annotations
import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from backend.app.core.database import _STOCK_ALERT_TEMPLATE_BODIES, _migrate_stock_alert_template_sku_variables
from backend.app.models.notification_template import DEFAULT_TEMPLATES
@pytest.fixture
async def engine():
"""In-memory SQLite with just the notification_templates table."""
from backend.app.models.notification_template import NotificationTemplate
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
async with engine.begin() as conn:
await conn.run_sync(NotificationTemplate.__table__.create)
try:
yield engine
finally:
await engine.dispose()
async def _insert(conn, event_type: str, body: str) -> None:
await conn.execute(
text(
"INSERT INTO notification_templates (event_type, name, title_template, body_template, is_default) "
"VALUES (:et, 'n', 't', :b, 1)"
),
{"et": event_type, "b": body},
)
async def _body(conn, event_type: str) -> str:
return (
await conn.execute(
text("SELECT body_template FROM notification_templates WHERE event_type = :et"), {"et": event_type}
)
).scalar_one()
@pytest.mark.parametrize("event_type", sorted(_STOCK_ALERT_TEMPLATE_BODIES))
async def test_rewrites_the_shipped_default(engine, event_type):
old, _new = _STOCK_ALERT_TEMPLATE_BODIES[event_type]
async with engine.begin() as conn:
await _insert(conn, event_type, old)
async with engine.begin() as conn:
await _migrate_stock_alert_template_sku_variables(conn)
body = await _body(conn, event_type)
assert "{subtype}" in body
assert "{color}" in body
@pytest.mark.parametrize("event_type", sorted(_STOCK_ALERT_TEMPLATE_BODIES))
async def test_leaves_an_edited_template_alone(engine, event_type):
async with engine.begin() as conn:
await _insert(conn, event_type, "Running low on {material}!")
async with engine.begin() as conn:
await _migrate_stock_alert_template_sku_variables(conn)
assert await _body(conn, event_type) == "Running low on {material}!"
@pytest.mark.parametrize("event_type", sorted(_STOCK_ALERT_TEMPLATE_BODIES))
async def test_the_rewrite_is_exactly_what_a_fresh_install_gets(engine, event_type):
"""The migrated body and the DEFAULT_TEMPLATES body must not drift apart."""
default = next(t for t in DEFAULT_TEMPLATES if t["event_type"] == event_type)
assert _STOCK_ALERT_TEMPLATE_BODIES[event_type][1] == default["body_template"]
async def test_running_it_twice_changes_nothing(engine):
old, new = _STOCK_ALERT_TEMPLATE_BODIES["stock_reorder_alert"]
async with engine.begin() as conn:
await _insert(conn, "stock_reorder_alert", old)
for _ in range(2):
async with engine.begin() as conn:
await _migrate_stock_alert_template_sku_variables(conn)
async with engine.begin() as conn:
assert await _body(conn, "stock_reorder_alert") == new