public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
bambuddy/backend/tests/unit/test_scheduler_scheduled_drying.py
T
Ben Hamilton (Ben Gertzfield)MartinNYHCmaziggy
 d37ce94f81 Feature: Scheduled drying (#2703) 
* feat: add ScheduledDrying model for delayed drying runs (#2638)

* Release the printer when a scheduled dry ends (#2638)

_check_scheduled_dryings marks a printer as drying in _drying_in_progress,
which is shared with auto-drying. Auto-drying prunes that map in
_sync_drying_state(), but that call sits behind its enabled check, and this is
the first writer that runs whether auto-drying is on or not. With it off --
the default -- nothing dropped the entry short of a print being dispatched to
the same printer, so the next scheduled run parked on "already_drying"
forever and queue_drying_block held that printer's prints too. A nightly
off-peak dry with no printing in between is exactly the workflow this feature
is for: night one worked, every night after it silently did not. The check now
releases what it acquired, covering both a run that ends mid-pass and one
cancelled through the route between passes.

The retention prune ran on every pass. Issuing the DELETE is what opens a
write transaction, this method is called every 3s while the queue dispatches,
and rows only become prunable a week after they finish, so it is now gated to
hourly on a monotonic stamp -- with the first pass after a restart still
reaping whatever the dead process left behind.

Both drying paths now pick the blocking dry_sf_reason through one rule.
The immediate endpoint quoted whichever code the firmware listed first while
the scheduler prioritised power over retract, so one blocked AMS read two ways
depending on which button you pressed. drying_preflight.primary_reason_code
holds the order and both call it, including the flame button's tooltip, which
had no wording for filament at the outlet at all and sent those users to the
generic "can't start drying right now".

scheduled_drying joins the model list in init_db. The table was already
created -- importing the package registers it -- but it was the only model
relying on that indirection.

Tests: the release (completion and route-cancel), the prune throttle, the
shared reason rule, the tooltip priority, and four driving the real check_queue,
which nothing covered before -- a pass with no rows still dispatching prints, a
due row dispatching, and a failed row not stalling the queue behind it. Each
one fails against the code it guards.

---------

Co-authored-by: MartinNYHC <martin@bambuddy.cool>
Co-authored-by: maziggy <mz@v8w.de>
2026-08-15 13:40:21 +02:00

620 lines
25 KiB
Python
Raw Permalink Blame History

"""Tests for PrintScheduler scheduled-drying dispatch (#2638)."""
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from sqlalchemy import select
from backend.app.models.scheduled_drying import ScheduledDrying
from backend.app.services.print_scheduler import (
SCHEDULED_DRYING_PRUNE_INTERVAL_SECONDS,
SCHEDULED_DRYING_RETENTION_DAYS,
PrintScheduler,
)
def _utcnow_naive() -> datetime:
return datetime.now(timezone.utc).replace(tzinfo=None)
# Above the X1C drying minimum, so the shared preflight lets dispatch through.
DRYING_CAPABLE_FIRMWARE = "01.09.00.00"
def _mock_state(ams_id=0, dry_time=0, dry_sf_reason=None, firmware=DRYING_CAPABLE_FIRMWARE):
state = MagicMock()
state.firmware_version = firmware
state.raw_data = {"ams": [{"id": ams_id, "dry_time": dry_time, "dry_sf_reason": dry_sf_reason or []}]}
return state
async def _make_row(db_session, printer_factory, **kwargs):
printer = await printer_factory()
defaults = {"printer_id": printer.id, "ams_id": 0, "temp": 65, "duration_hours": 8}
defaults.update(kwargs)
row = ScheduledDrying(**defaults)
db_session.add(row)
await db_session.commit()
await db_session.refresh(row)
return row
@pytest.fixture
def scheduler():
return PrintScheduler()
@pytest.mark.asyncio
async def test_future_start_after_not_dispatched(scheduler, db_session, printer_factory):
row = await _make_row(db_session, printer_factory, start_after=_utcnow_naive() + timedelta(hours=2))
with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "pending"
@pytest.mark.asyncio
async def test_due_row_dispatches_and_goes_running(scheduler, db_session, printer_factory):
row = await _make_row(
db_session, printer_factory, start_after=_utcnow_naive() - timedelta(minutes=1), filament="PETG"
)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_called_once_with(
row.printer_id, 0, 65, 8, mode=1, filament="PETG", rotate_tray=False
)
await db_session.refresh(row)
assert row.status == "running"
assert row.started_at is not None
assert scheduler._drying_in_progress.get(row.printer_id)
@pytest.mark.asyncio
async def test_null_start_after_dispatches_immediately(scheduler, db_session, printer_factory):
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "running"
@pytest.mark.asyncio
async def test_busy_printer_stays_pending_with_reason(scheduler, db_session, printer_factory):
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=False),
):
mock_pm.get_status.return_value = _mock_state()
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "pending"
assert row.waiting_reason == "printer_busy"
@pytest.mark.asyncio
async def test_offline_printer_stays_pending(scheduler, db_session, printer_factory):
row = await _make_row(db_session, printer_factory, start_after=None)
with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
mock_pm.get_status.return_value = None
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "pending"
assert row.waiting_reason == "printer_offline"
@pytest.mark.asyncio
async def test_ams_blocked_stays_pending(scheduler, db_session, printer_factory):
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(dry_sf_reason=[2])
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "pending"
assert row.waiting_reason == "ams_blocked"
@pytest.mark.asyncio
async def test_retract_block_gets_its_own_waiting_reason(scheduler, db_session, printer_factory):
"""Code 3 is user-actionable (retract the filament), so it says so rather
than bucketing into the generic blocked message.
"""
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(dry_sf_reason=[3])
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "pending"
assert row.waiting_reason == "ams_retract_filament"
@pytest.mark.asyncio
async def test_power_block_outranks_retract(scheduler, db_session, printer_factory):
"""Both blocking at once: power is the one that has to be fixed first."""
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(dry_sf_reason=[3, 8])
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.waiting_reason == "ams_power_required"
@pytest.mark.asyncio
@pytest.mark.parametrize("code", [1, 8])
async def test_power_block_gets_its_own_waiting_reason(scheduler, db_session, printer_factory, code):
"""A run the user has to unblock says so, rather than waiting silently."""
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(dry_sf_reason=[code])
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "pending"
assert row.waiting_reason == "ams_power_required"
@pytest.mark.asyncio
async def test_screen_only_model_fails_instead_of_dispatching(scheduler, db_session, printer_factory):
"""A P1S acks the publish and ignores it; dispatching would silently self-cancel."""
printer = await printer_factory(model="P1S")
row = ScheduledDrying(printer_id=printer.id, ams_id=0, temp=65, duration_hours=8)
db_session.add(row)
await db_session.commit()
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "failed"
assert row.error_message
assert row.completed_at is not None
@pytest.mark.asyncio
async def test_firmware_below_minimum_fails(scheduler, db_session, printer_factory):
row = await _make_row(db_session, printer_factory, start_after=None)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(firmware="01.05.00.00")
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "failed"
assert row.error_message
assert row.completed_at is not None
@pytest.mark.asyncio
async def test_empty_filament_backfills_from_loaded_tray(scheduler, db_session, printer_factory):
"""Matches the immediate endpoint, which sends the loaded type rather than PLA."""
row = await _make_row(db_session, printer_factory, start_after=None, filament="")
state = _mock_state()
state.raw_data["ams"][0]["tray"] = [{"tray_type": ""}, {"tray_type": "PETG"}]
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = state
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
assert mock_pm.send_drying_command.call_args.kwargs["filament"] == "PETG"
await db_session.refresh(row)
assert row.filament == "PETG"
@pytest.mark.asyncio
async def test_empty_filament_falls_back_to_pla(scheduler, db_session, printer_factory):
await _make_row(db_session, printer_factory, start_after=None, filament="")
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
assert mock_pm.send_drying_command.call_args.kwargs["filament"] == "PLA"
@pytest.mark.asyncio
async def test_finished_rows_pruned_after_retention(scheduler, db_session, printer_factory):
printer = await printer_factory()
stale = ScheduledDrying(
printer_id=printer.id,
ams_id=0,
temp=65,
duration_hours=8,
status="completed",
completed_at=_utcnow_naive() - timedelta(days=SCHEDULED_DRYING_RETENTION_DAYS + 1),
)
recent = ScheduledDrying(
printer_id=printer.id,
ams_id=0,
temp=65,
duration_hours=8,
status="cancelled",
completed_at=_utcnow_naive() - timedelta(hours=1),
)
db_session.add_all([stale, recent])
await db_session.commit()
with patch("backend.app.services.print_scheduler.printer_manager"):
await scheduler._check_scheduled_dryings(db_session)
remaining = (await db_session.execute(select(ScheduledDrying.id))).scalars().all()
assert stale.id not in remaining
assert recent.id in remaining
@pytest.mark.asyncio
async def test_prune_does_not_run_on_every_pass(scheduler, db_session, printer_factory):
"""The prune is throttled, because issuing the DELETE is what starts a
write transaction and this method runs every 3s while the queue dispatches.
Rows only become prunable a week after they finish, so nothing is lost by
waiting an hour to reap them."""
printer = await printer_factory()
async def _stale_row() -> ScheduledDrying:
row = ScheduledDrying(
printer_id=printer.id,
ams_id=0,
temp=65,
duration_hours=8,
status="completed",
completed_at=_utcnow_naive() - timedelta(days=SCHEDULED_DRYING_RETENTION_DAYS + 1),
)
db_session.add(row)
await db_session.commit()
await db_session.refresh(row)
return row
with patch("backend.app.services.print_scheduler.printer_manager"):
# First pass after a restart always prunes, so rows left behind by the
# process that died are still reaped.
first = await _stale_row()
await scheduler._check_scheduled_dryings(db_session)
assert first.id not in (await db_session.execute(select(ScheduledDrying.id))).scalars().all()
# A second pass moments later leaves an equally stale row alone.
second = await _stale_row()
await scheduler._check_scheduled_dryings(db_session)
assert second.id in (await db_session.execute(select(ScheduledDrying.id))).scalars().all()
# ...and reaps it once the interval has elapsed.
scheduler._last_scheduled_drying_prune -= SCHEDULED_DRYING_PRUNE_INTERVAL_SECONDS
await scheduler._check_scheduled_dryings(db_session)
assert second.id not in (await db_session.execute(select(ScheduledDrying.id))).scalars().all()
@pytest.mark.asyncio
async def test_a_finished_run_releases_the_printer_without_auto_drying(scheduler, db_session, printer_factory):
"""_drying_in_progress must not outlive the run that set it.
Auto-drying's _sync_drying_state() prunes that map, but it sits behind the
enabled check, so on a default install (both auto-drying modes off) it never
runs. Nothing else drops the entry unless a print is dispatched to the same
printer, and a nightly off-peak dry with no printing in between is exactly
the case this feature is for: the second night's run would sit on
"already_drying" forever, and with queue_drying_block on the printer would
stop taking prints as well.
"""
row = await _make_row(
db_session, printer_factory, start_after=_utcnow_naive() - timedelta(minutes=1), duration_hours=1
)
printer_id = row.printer_id
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "running"
assert printer_id in scheduler._drying_in_progress
# The firmware has stopped reporting a dry_time, well past the duration.
row.started_at = _utcnow_naive() - timedelta(hours=2)
await db_session.commit()
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(dry_time=0)
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "completed"
assert printer_id not in scheduler._drying_in_progress
# And the next night's run still dispatches.
tomorrow = ScheduledDrying(
printer_id=printer_id,
ams_id=0,
temp=65,
duration_hours=8,
start_after=_utcnow_naive() - timedelta(minutes=1),
)
db_session.add(tomorrow)
await db_session.commit()
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(tomorrow)
assert tomorrow.status == "running"
assert tomorrow.waiting_reason is None
@pytest.mark.asyncio
async def test_a_route_cancel_releases_the_printer(scheduler, db_session, printer_factory):
"""The DELETE route flips the row to cancelled in the database and sends the
stop, but knows nothing about the scheduler's in-memory map. The next pass
has to notice the run is gone and release the printer, or it stays marked as
drying until a restart."""
row = await _make_row(db_session, printer_factory, start_after=_utcnow_naive() - timedelta(minutes=1))
printer_id = row.printer_id
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
assert printer_id in scheduler._drying_in_progress
# What DELETE /scheduled-dryings/{id} leaves behind.
row.status = "cancelled"
row.completed_at = _utcnow_naive()
await db_session.commit()
with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
mock_pm.get_status.return_value = _mock_state()
await scheduler._check_scheduled_dryings(db_session)
assert printer_id not in scheduler._drying_in_progress
assert printer_id not in scheduler._scheduled_drying_printer_ids
@pytest.mark.asyncio
async def test_running_completes_after_duration(scheduler, db_session, printer_factory):
row = await _make_row(
db_session,
printer_factory,
status="running",
duration_hours=1,
started_at=_utcnow_naive() - timedelta(minutes=58), # >= 90% of 1h
)
with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
mock_pm.get_status.return_value = _mock_state(dry_time=0)
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "completed"
assert row.completed_at is not None
@pytest.mark.asyncio
async def test_running_interrupted_by_print_requeues(scheduler, db_session, printer_factory):
row = await _make_row(
db_session,
printer_factory,
status="running",
duration_hours=8,
started_at=_utcnow_naive() - timedelta(minutes=30), # well past grace, far from done
)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=False),
):
mock_pm.get_status.return_value = _mock_state(dry_time=0)
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "pending"
assert row.started_at is None
assert row.waiting_reason == "interrupted"
@pytest.mark.asyncio
async def test_running_stopped_while_idle_cancels(scheduler, db_session, printer_factory):
"""A stop on an idle printer is deliberate; the row must not resurrect."""
row = await _make_row(
db_session,
printer_factory,
status="running",
duration_hours=8,
started_at=_utcnow_naive() - timedelta(minutes=30),
)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state(dry_time=0)
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "cancelled"
assert row.completed_at is not None
@pytest.mark.asyncio
async def test_running_within_grace_untouched(scheduler, db_session, printer_factory):
row = await _make_row(
db_session,
printer_factory,
status="running",
duration_hours=8,
started_at=_utcnow_naive() - timedelta(seconds=30), # inside 120 s grace
)
with patch("backend.app.services.print_scheduler.printer_manager") as mock_pm:
mock_pm.get_status.return_value = _mock_state(dry_time=0)
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "running"
@pytest.mark.asyncio
async def test_scheduled_drying_survives_auto_drying_stop_all(scheduler, db_session, printer_factory):
"""Regression (#2638): a running scheduled drying must not be stopped or
untracked by _check_auto_drying's stop-all branch, even in the default
config where both auto-drying toggles are off. Before the fix, the two
features co-owned _drying_in_progress and auto-drying would stop/pop any
printer it didn't start drying on itself.
"""
row = await _make_row(db_session, printer_factory, start_after=_utcnow_naive() - timedelta(minutes=1))
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
await db_session.refresh(row)
assert row.status == "running"
assert scheduler._drying_in_progress.get(row.printer_id)
assert row.printer_id in scheduler._scheduled_drying_printer_ids
mock_pm.reset_mock()
with patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=False)):
# Default config: queue_drying_enabled and ambient_drying_enabled both off.
await scheduler._check_auto_drying(db_session, [], set())
mock_pm.send_drying_command.assert_not_called()
await db_session.refresh(row)
assert row.status == "running"
assert scheduler._drying_in_progress.get(row.printer_id)
assert row.printer_id in scheduler._scheduled_drying_printer_ids
@pytest.mark.asyncio
async def test_second_pending_row_for_same_printer_does_not_dispatch(scheduler, db_session, printer_factory):
"""Regression (#2638): two pending rows for the same printer must not both
dispatch in the same tick; the second should see the first's dispatch and
stay pending.
"""
printer = await printer_factory()
past = _utcnow_naive() - timedelta(minutes=1)
row1 = ScheduledDrying(printer_id=printer.id, ams_id=0, temp=65, duration_hours=8, start_after=past)
row2 = ScheduledDrying(printer_id=printer.id, ams_id=0, temp=60, duration_hours=6, start_after=past)
db_session.add_all([row1, row2])
await db_session.commit()
await db_session.refresh(row1)
await db_session.refresh(row2)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_called_once()
await db_session.refresh(row1)
await db_session.refresh(row2)
# Ordered dispatch: same start_after, so the row created first wins.
assert row1.status == "running"
assert row2.status == "pending"
assert row2.waiting_reason == "already_drying"
@pytest.mark.asyncio
async def test_earliest_start_after_dispatches_first(scheduler, db_session, printer_factory):
"""Two rows due on one printer: the earlier schedule starts, not an arbitrary one."""
printer = await printer_factory()
now = _utcnow_naive()
later = ScheduledDrying(
printer_id=printer.id, ams_id=0, temp=65, duration_hours=8, start_after=now - timedelta(minutes=1)
)
sooner = ScheduledDrying(
printer_id=printer.id, ams_id=0, temp=60, duration_hours=6, start_after=now - timedelta(hours=3)
)
# Inserted later-first so row order alone cannot produce the right answer.
db_session.add_all([later, sooner])
await db_session.commit()
await db_session.refresh(later)
await db_session.refresh(sooner)
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=True),
):
mock_pm.get_status.return_value = _mock_state()
mock_pm.send_drying_command.return_value = True
await scheduler._check_scheduled_dryings(db_session)
mock_pm.send_drying_command.assert_called_once_with(printer.id, 0, 60, 6, mode=1, filament="PLA", rotate_tray=False)
await db_session.refresh(later)
await db_session.refresh(sooner)
assert sooner.status == "running"
assert later.status == "pending"
@pytest.mark.asyncio
async def test_malformed_ams_id_does_not_throw_while_running(scheduler, db_session, printer_factory):
"""_update_running_scheduled_drying runs inside check_queue: a throw here
would cost the whole pass, print dispatch included, on every tick.
"""
row = await _make_row(
db_session,
printer_factory,
status="running",
started_at=_utcnow_naive() - timedelta(minutes=30),
)
state = MagicMock()
state.firmware_version = DRYING_CAPABLE_FIRMWARE
state.raw_data = {"ams": [{"id": "not-a-number", "dry_time": 120}]}
with (
patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
patch.object(scheduler, "_is_printer_idle", return_value=False),
):
mock_pm.get_status.return_value = state
await scheduler._check_scheduled_dryings(db_session)
# No matching unit means no dry_time; the printer is busy, so it re-queues.
await db_session.refresh(row)
assert row.status == "pending"
assert row.waiting_reason == "interrupted"
Reference in New Issue View Git Blame Copy Permalink