mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-22 03:49:18 +02:00
maziggy
7ea4410b21
The pre-print deficit warning from #720 only ran inside the PrintModal submit flow. Both the green ▶ button on a staged queue row (POST /queue/{id}/start) and the Virtual Printer queue-mode intake bypassed it — auto_dispatch=True VP intakes would dispatch unsupervised onto spools that physically can't complete the print. Extracted the deficit check into backend/app/services/filament_deficit.py (single source of truth, both internal inventory and Spoolman modes). POST /queue/{id}/start returns 409 with a structured deficit payload unless ?skip_filament_check=true. The dispatch scheduler runs the same check before each _start_print; a deficit promotes the item to manual_start + sets a new filament_short flag (idempotent migration on print_queue). The flag clears automatically on the next tick when the operator swaps a spool to one with enough material. Frontend ▶ catches the 409 and opens a confirm modal showing each shorted slot's required vs remaining grams; the row now renders a yellow "Insufficient filament" badge when filament_short is set. Translated across all 9 locales.
120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Scheduler pre-dispatch filament-deficit guard tests (#1496).
|
|
|
|
``PrintScheduler._block_on_filament_deficit`` is the gate that keeps an
|
|
auto_dispatch=True VP intake (or any other scheduler-driven dispatch) from
|
|
sending a print onto a spool that can't satisfy it. On a deficit it
|
|
promotes the item to manual_start; when a previously-flagged item's spool
|
|
is now adequate it clears the flag so the next tick dispatches.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.models.print_queue import PrintQueueItem
|
|
from backend.app.services.filament_deficit import FilamentDeficit
|
|
from backend.app.services.print_scheduler import PrintScheduler
|
|
|
|
|
|
@pytest.fixture
|
|
def scheduler():
|
|
"""A fresh scheduler instance — internal state is not exercised."""
|
|
return PrintScheduler()
|
|
|
|
|
|
@pytest.fixture
|
|
def queue_item(db_session, printer_factory):
|
|
"""Helper to drop a queue item the helper can mutate."""
|
|
|
|
async def _make(**overrides):
|
|
printer = await printer_factory()
|
|
defaults = {
|
|
"printer_id": printer.id,
|
|
"status": "pending",
|
|
"manual_start": False,
|
|
"filament_short": False,
|
|
}
|
|
defaults.update(overrides)
|
|
item = PrintQueueItem(**defaults)
|
|
db_session.add(item)
|
|
await db_session.commit()
|
|
await db_session.refresh(item)
|
|
return item
|
|
|
|
return _make
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_blocks_on_deficit_promotes_to_manual_start(scheduler, db_session, queue_item):
|
|
item = await queue_item()
|
|
with patch(
|
|
"backend.app.services.print_scheduler.compute_deficit_for_queue_item",
|
|
AsyncMock(
|
|
return_value=[
|
|
FilamentDeficit(
|
|
slot_id=1,
|
|
ams_id=0,
|
|
tray_id=0,
|
|
filament_type="PLA",
|
|
required_grams=270.0,
|
|
remaining_grams=200.0,
|
|
),
|
|
]
|
|
),
|
|
):
|
|
blocked = await scheduler._block_on_filament_deficit(db_session, item)
|
|
|
|
assert blocked is True
|
|
await db_session.refresh(item)
|
|
assert item.manual_start is True
|
|
assert item.filament_short is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clears_stale_flag_when_deficit_resolves(scheduler, db_session, queue_item):
|
|
"""Previously-flagged item whose spool was swapped is unblocked."""
|
|
item = await queue_item(filament_short=True, manual_start=False)
|
|
with patch(
|
|
"backend.app.services.print_scheduler.compute_deficit_for_queue_item",
|
|
AsyncMock(return_value=[]),
|
|
):
|
|
blocked = await scheduler._block_on_filament_deficit(db_session, item)
|
|
|
|
assert blocked is False
|
|
await db_session.refresh(item)
|
|
assert item.filament_short is False
|
|
assert item.manual_start is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_deficit_no_op(scheduler, db_session, queue_item):
|
|
"""Happy path — no deficit, no flag changes, dispatch proceeds."""
|
|
item = await queue_item()
|
|
with patch(
|
|
"backend.app.services.print_scheduler.compute_deficit_for_queue_item",
|
|
AsyncMock(return_value=[]),
|
|
):
|
|
blocked = await scheduler._block_on_filament_deficit(db_session, item)
|
|
|
|
assert blocked is False
|
|
await db_session.refresh(item)
|
|
assert item.filament_short is False
|
|
assert item.manual_start is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_helper_exception_does_not_wedge_dispatch(scheduler, db_session, queue_item):
|
|
"""A flaky deficit check (e.g. Spoolman timeout) must not block dispatch."""
|
|
item = await queue_item()
|
|
with patch(
|
|
"backend.app.services.print_scheduler.compute_deficit_for_queue_item",
|
|
AsyncMock(side_effect=RuntimeError("network down")),
|
|
):
|
|
blocked = await scheduler._block_on_filament_deficit(db_session, item)
|
|
|
|
assert blocked is False
|
|
await db_session.refresh(item)
|
|
assert item.filament_short is False
|