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
dev
bambuddy/backend/tests/unit/test_scheduler_auto_drying.py
T
maziggy 61e2841d16 Correct the drying settings path and the print-drying comment (issue #2518) 
The changelog named Settings -> Print Queue -> Auto-Drying; the toggle
lives in the Queue Auto-Drying card under Settings -> Workflow. The
scheduler comment and a test docstring called print_drying a permission
overlay rather than a trigger, but since #1816 it starts mid-print
cycles on its own regardless of queue state.
2026-09-28 11:40:21 +02:00

2496 lines
108 KiB
Python
Raw Permalink Blame History

"""Tests for the auto-drying feature in the print scheduler.
Covers:
- Conservative drying parameter selection (mixed filaments)
- Drying preset loading (user-configured vs defaults)
- Auto-drying lifecycle: start, humidity stop, minimum drying time
- Auto-drying stop conditions: feature disabled, no scheduled items, per-printer
- Sync drying state after restart
"""
import re
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.services.print_scheduler import (
AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES,
AUTO_DRY_REARM_COOLDOWN_SECONDS,
AUTO_DRY_SUSTAINED_GAP_FLOOR_SECONDS,
PrintScheduler,
)
class TestConservativeDryingParams:
"""Test _get_conservative_drying_params — picks safest temp/duration for mixed filaments."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
def test_single_filament_pla(self, scheduler):
"""Single PLA tray uses PLA preset."""
trays = [{"tray_type": "PLA"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3f", presets)
assert result == (45, 12, "PLA")
def test_mixed_filaments_lowest_temp(self, scheduler):
"""Mixed PLA + ABS: should use PLA's 45°C (lowest), ABS's 12h (longest for n3f)."""
trays = [{"tray_type": "PLA"}, {"tray_type": "ABS"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3f", presets)
temp, hours, _ = result
assert temp == 45 # PLA is lowest
assert hours == 12
def test_mixed_filaments_longest_duration(self, scheduler):
"""Mixed ABS (8h) + PVA (18h) on n3s: should use longest duration."""
trays = [{"tray_type": "ABS"}, {"tray_type": "PVA"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3s", presets)
temp, hours, _ = result
assert temp == 80 # ABS n3s=80, PVA n3s=85 → lowest=80
assert hours == 18 # ABS n3s_hours=8, PVA n3s_hours=18 → longest=18
def test_empty_trays_returns_none(self, scheduler):
"""No loaded trays returns None."""
result = scheduler._get_conservative_drying_params([], "n3f", PrintScheduler.DEFAULT_DRYING_PRESETS)
assert result is None
def test_unknown_filament_skipped(self, scheduler):
"""Unknown filament types are ignored."""
trays = [{"tray_type": "EXOTIC_WOOD"}]
result = scheduler._get_conservative_drying_params(trays, "n3f", PrintScheduler.DEFAULT_DRYING_PRESETS)
assert result is None
def test_filament_type_normalization(self, scheduler):
"""'PLA Basic' should normalize to 'PLA'."""
trays = [{"tray_type": "PLA Basic"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3f", presets)
assert result is not None
assert result[0] == 45 # PLA temp
def test_empty_tray_type_skipped(self, scheduler):
"""Trays with empty tray_type are skipped."""
trays = [{"tray_type": ""}, {"tray_type": "PETG"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3f", presets)
assert result is not None
assert result[2] == "PETG"
def test_n3s_uses_n3s_keys(self, scheduler):
"""AMS-HT (n3s) should use n3s temp and n3s_hours."""
trays = [{"tray_type": "TPU"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3s", presets)
assert result == (75, 18, "TPU") # n3s=75, n3s_hours=18
def test_n3f_uses_n3f_keys(self, scheduler):
"""AMS 2 Pro (n3f) should use n3f temp and n3f_hours."""
trays = [{"tray_type": "TPU"}]
presets = PrintScheduler.DEFAULT_DRYING_PRESETS
result = scheduler._get_conservative_drying_params(trays, "n3f", presets)
assert result == (65, 12, "TPU") # n3f=65, n3f_hours=12
def test_custom_presets(self, scheduler):
"""Custom presets override defaults."""
trays = [{"tray_type": "PLA"}]
custom = {"PLA": {"n3f": 50, "n3s": 50, "n3f_hours": 6, "n3s_hours": 6}}
result = scheduler._get_conservative_drying_params(trays, "n3f", custom)
assert result == (50, 6, "PLA")
class TestCompositesResolveToTheirBaseMaterial:
"""#3067: a composite spool was skipped by auto-drying entirely.
The preset key came from ``tray_type.split()[0].upper()``, which splits on
spaces only -- so "PA6-CF" stayed "PA6-CF", found no row in an 8-key table,
and the tray contributed nothing. Every caller reads "no row" as "nothing to
dry here", so the AMS was passed over on every scheduler pass, silently.
It was never only PA. Of the 41 types a printer can report, 33 had no row
under that rule and 20 of them have a base material sitting right there:
every -CF, -GF and -AERO variant of PLA, PETG, ABS, ASA, PC and PA.
The reporter could still dry the same spool by hand, because the drying
popover has resolved composites since #2774 -- these tests pin the two ends
to the same answer.
"""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.parametrize(
("tray_type", "expected_key"),
[
# The reported spool, and the rest of the polyamide spellings. Bambu
# labels nylon "PA" and spells its own composites out, so none of
# these match a PA row without the alias map.
("PA6-CF", "PA"),
("PA6-GF", "PA"),
("PA12-CF", "PA"),
("PAHT-CF", "PA"),
("PA-CF", "PA"),
("Nylon", "PA"),
# Polyphthalamide is a distinct polymer, not a nylon grade, so this
# one is a judgement: an aromatic polyamide that takes up moisture
# the same way, dried on the hottest row the table has.
("PPA-CF", "PA"),
("PPA-GF", "PA"),
# ...and the variants of everything else, which were equally skipped.
("PLA-CF", "PLA"),
("PLA-GF", "PLA"),
("PLA-AERO", "PLA"),
("PLA-S", "PLA"),
("PETG-CF", "PETG"),
("ABS-GF", "ABS"),
("ASA-CF", "ASA"),
("ASA-AERO", "ASA"),
("PC-CF", "PC"),
# Already worked, and must keep working.
("PLA", "PLA"),
("PLA Basic", "PLA"),
("TPU for AMS", "TPU"),
],
)
def test_the_tray_reaches_its_base_materials_preset(self, scheduler, tray_type, expected_key):
result = scheduler._get_conservative_drying_params(
[{"tray_type": tray_type}], "n3s", PrintScheduler.DEFAULT_DRYING_PRESETS
)
assert result is not None, f"{tray_type} is still skipped by auto-drying"
assert result[2] == expected_key
assert result[0] == PrintScheduler.DEFAULT_DRYING_PRESETS[expected_key]["n3s"]
def test_the_reported_spool_gets_nylons_temperature(self, scheduler):
"""The whole point of resolving it rather than defaulting: PA6-CF wants
PA's 85C on an AMS-HT. Landing on PLA's 45 would run a cycle that dries
nothing, which is worse than the skip it replaces -- it looks like it
worked."""
result = scheduler._get_conservative_drying_params(
[{"tray_type": "PA6-CF"}], "n3s", PrintScheduler.DEFAULT_DRYING_PRESETS
)
assert result == (85, 12, "PA")
@pytest.mark.parametrize("tray_type", ["PPS-CF", "PET-CF", "PEEK", "PP", "PE", "wildly unknown"])
def test_a_material_with_no_base_row_is_still_skipped(self, scheduler, tray_type):
"""Nothing here invents a drying profile. A material with no row and no
alias keeps the behaviour it has today rather than being dried at a
number nobody chose.
This is deliberately where the backend parts company with the drying
popover, which falls back to PLA because a dropdown has to show
something. A scheduler does not.
"""
result = scheduler._get_conservative_drying_params(
[{"tray_type": tray_type}], "n3s", PrintScheduler.DEFAULT_DRYING_PRESETS
)
assert result is None
def test_a_user_row_for_the_exact_type_wins_over_the_base(self, scheduler):
"""Someone who has added PA6-CF to their own table meant it."""
custom = {
**PrintScheduler.DEFAULT_DRYING_PRESETS,
"PA6-CF": {"n3f": 70, "n3s": 90, "n3f_hours": 10, "n3s_hours": 10},
}
result = scheduler._get_conservative_drying_params([{"tray_type": "PA6-CF"}], "n3s", custom)
assert result == (90, 10, "PA6-CF")
def test_a_mixed_load_still_takes_the_coolest_row(self, scheduler):
"""Resolving more types must not disturb the conservative choice: a
PA6-CF spool sharing the unit with PLA still gets PLA's 45C, because
85 would deform the PLA."""
result = scheduler._get_conservative_drying_params(
[{"tray_type": "PA6-CF"}, {"tray_type": "PLA"}], "n3s", PrintScheduler.DEFAULT_DRYING_PRESETS
)
assert result[0] == 45
def test_an_empty_preset_row_still_means_skip(self, scheduler):
"""The table is user-editable JSON and nothing validates a row, so one
can be present and empty. Resolving the key is not the same as having a
preset: the temp/hours reads each fall back to 55C/12h, which would dry
a PLA spool at 55 degrees because somebody left a row blank."""
custom = {**PrintScheduler.DEFAULT_DRYING_PRESETS, "PLA": {}}
assert scheduler._get_conservative_drying_params([{"tray_type": "PLA"}], "n3s", custom) is None
# And it does not quietly fall through to some other row either.
assert scheduler._get_conservative_drying_params([{"tray_type": "PLA-CF"}], "n3s", custom) is None
def test_a_zero_valued_row_is_a_row(self, scheduler):
"""The resolver tests key presence, not truthiness. It is shared with the
chamber-preheat map, where 0 is the correct target for PLA, PETG, TPU and
PVA -- reading those as "no row" would send every one of them to the
catch-all."""
targets = PrintScheduler._bundled_preheat_targets()
assert targets["PLA"] == 0
assert PrintScheduler._resolve_filament_key("PLA", targets) == "PLA"
assert scheduler._target_for_tray_type("PLA", targets) == 0
assert scheduler._target_for_tray_type("PLA-CF", targets) == 0
def test_a_whitespace_only_tray_type_is_not_a_material(self, scheduler):
"""Truthy, and splits to nothing. The old normaliser indexed the split
after testing the string, so this raised IndexError rather than reading
as an empty tray."""
assert PrintScheduler._normalize_filament_type(" ") == ""
result = scheduler._get_conservative_drying_params(
[{"tray_type": " "}], "n3s", PrintScheduler.DEFAULT_DRYING_PRESETS
)
assert result is None
class TestDryingPresets:
"""Test _get_drying_presets — loads user presets from DB or falls back to defaults."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
async def test_default_presets_when_no_setting(self, scheduler):
"""Returns built-in defaults when no DB setting exists."""
db = AsyncMock()
result_mock = MagicMock()
result_mock.scalar_one_or_none.return_value = None
db.execute = AsyncMock(return_value=result_mock)
presets = await scheduler._get_drying_presets(db)
assert presets == PrintScheduler.DEFAULT_DRYING_PRESETS
@pytest.mark.asyncio
async def test_user_presets_from_db(self, scheduler):
"""Returns user-configured presets when saved in DB."""
db = AsyncMock()
setting = MagicMock()
setting.value = '{"PLA": {"n3f": 50, "n3s": 50, "n3f_hours": 6, "n3s_hours": 6}}'
result_mock = MagicMock()
result_mock.scalar_one_or_none.return_value = setting
db.execute = AsyncMock(return_value=result_mock)
presets = await scheduler._get_drying_presets(db)
assert presets["PLA"]["n3f"] == 50
@pytest.mark.asyncio
async def test_invalid_json_falls_back(self, scheduler):
"""Invalid JSON in DB falls back to defaults."""
db = AsyncMock()
setting = MagicMock()
setting.value = "not valid json{{"
result_mock = MagicMock()
result_mock.scalar_one_or_none.return_value = setting
db.execute = AsyncMock(return_value=result_mock)
presets = await scheduler._get_drying_presets(db)
assert presets == PrintScheduler.DEFAULT_DRYING_PRESETS
@pytest.mark.asyncio
async def test_empty_string_falls_back(self, scheduler):
"""Empty string in DB falls back to defaults."""
db = AsyncMock()
setting = MagicMock()
setting.value = ""
result_mock = MagicMock()
result_mock.scalar_one_or_none.return_value = setting
db.execute = AsyncMock(return_value=result_mock)
presets = await scheduler._get_drying_presets(db)
assert presets == PrintScheduler.DEFAULT_DRYING_PRESETS
class TestSyncDryingState:
"""Test _sync_drying_state — syncs in-memory state with actual printer status."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@patch("backend.app.services.print_scheduler.printer_manager")
def test_removes_stopped_printers(self, mock_pm, scheduler):
"""Printers that stopped drying are removed from tracking."""
scheduler._drying_in_progress = {1: time.monotonic()}
state = MagicMock()
state.raw_data = {"ams": [{"dry_time": 0}]}
mock_pm.get_status.return_value = state
scheduler._sync_drying_state()
assert 1 not in scheduler._drying_in_progress
@patch("backend.app.services.print_scheduler.printer_manager")
def test_keeps_active_printers(self, mock_pm, scheduler):
"""Printers still drying remain in tracking."""
ts = time.monotonic()
scheduler._drying_in_progress = {1: ts}
state = MagicMock()
state.raw_data = {"ams": [{"dry_time": 120}]}
mock_pm.get_status.return_value = state
scheduler._sync_drying_state()
assert scheduler._drying_in_progress[1] == ts
@patch("backend.app.services.print_scheduler.printer_manager")
def test_removes_disconnected_printers(self, mock_pm, scheduler):
"""Disconnected printers are removed from tracking."""
scheduler._drying_in_progress = {1: time.monotonic()}
mock_pm.get_status.return_value = None
scheduler._sync_drying_state()
assert 1 not in scheduler._drying_in_progress
class TestPlateHoldDoesNotGateDrying:
"""#2801 — an unacknowledged plate must not stop the AMS heating.
Plate-clear answers "is the bed ready for the next job". It says nothing
about whether filament may be dried, and the gap between a finished print
and the acknowledgment is exactly when drying is most useful: the printer
is free and nobody is waiting on it. Leaving the plate unacknowledged is
also how people hold the queue by hand.
Before this, such a printer landed in the dispatch set, was read as
"currently printing", took the mid-print path -- capped temperature,
(mid-print) in the log -- and bypassed the very gate that was meant to
hold it, while the queue loop tore the cycle down once a tick.
"""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _finished_printer_state():
state = MagicMock()
state.state = "FINISH"
state.firmware_version = "01.03.00.00"
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": "75",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
return state
def _db(self):
db = AsyncMock()
db.execute = AsyncMock(
side_effect=TestAmbientDrying._make_db_side_effect(
{
"queue_drying_enabled": TestAmbientDrying._make_setting("false"),
"ambient_drying_enabled": TestAmbientDrying._make_setting("true"),
"print_drying_enabled": TestAmbientDrying._make_setting("true"),
"ams_humidity_fair": TestAmbientDrying._make_setting("60"),
"queue_drying_block": TestAmbientDrying._make_setting("false"),
"drying_presets": None,
}
)
)
return db
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_finished_printer_with_dirty_plate_dries_at_full_temperature(self, mock_sd, mock_pm, scheduler):
mock_pm.get_status.return_value = self._finished_printer_state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "P2S"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
await scheduler._check_auto_drying(self._db(), [], set())
# 45 degC is the uncapped PLA preset: mid-print would have sent 40.
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_idleness_is_judged_without_the_plate_gate(self, mock_sd, mock_pm, scheduler):
mock_pm.get_status.return_value = self._finished_printer_state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "P2S"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
await scheduler._check_auto_drying(self._db(), [], set())
scheduler._is_printer_idle.assert_called_with(1, require_plate_clear=False)
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_a_printer_about_to_print_is_still_left_alone(self, mock_sd, mock_pm, scheduler):
"""The narrow set keeps its job: an imminent print must not be dried into."""
mock_pm.get_status.return_value = self._finished_printer_state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "P2S"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
await scheduler._check_auto_drying(self._db(), [], {1})
assert not mock_pm.send_drying_command.called
class TestStopDrying:
"""Test _stop_drying — sends stop commands and clears tracking."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_stops_all_ams_units(self, mock_pm, scheduler):
"""Sends stop command to each auto-armed AMS unit that is drying."""
scheduler._drying_in_progress = {1: time.monotonic()}
scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
scheduler._auto_dry_units[(1, 128)] = {"ended_at": None}
state = MagicMock()
state.raw_data = {
"ams": [
{"id": 0, "dry_time": 120},
{"id": 1, "dry_time": 0},
{"id": 128, "dry_time": 60},
]
}
mock_pm.get_status.return_value = state
await scheduler._stop_drying(1)
# Should send stop to AMS 0 and 128, not AMS 1
calls = mock_pm.send_drying_command.call_args_list
assert len(calls) == 2
assert calls[0].args == (1, 0, 0, 0)
assert calls[1].args == (1, 128, 0, 0)
assert 1 not in scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_leaves_cycles_bambuddy_did_not_start(self, mock_pm, scheduler):
"""A hand-started dry on another unit survives (#2801).
One auto-dried unit used to be enough to stop every AMS on the
printer reporting dry_time > 0, which took the user's own cycle with
it. The entry gate only ever knew about cycles Bambuddy began; the
action now matches.
"""
scheduler._drying_in_progress = {1: time.monotonic()}
scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "dry_time": 120}, {"id": 1, "dry_time": 600}]}
mock_pm.get_status.return_value = state
await scheduler._stop_drying(1)
calls = mock_pm.send_drying_command.call_args_list
assert [c.args[1] for c in calls] == [0]
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_stops_nothing_it_cannot_prove_it_started(self, mock_pm, scheduler):
"""After a restart Bambuddy cannot tell its own cycle from a manual one.
_sync_drying_state prunes but never adopts, for exactly this reason, so
a cycle armed before the restart is left running rather than risking a
stop on somebody's manual dry. Tracking is still cleared.
"""
scheduler._drying_in_progress = {1: time.monotonic()}
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
mock_pm.get_status.return_value = state
await scheduler._stop_drying(1)
assert not mock_pm.send_drying_command.called
assert 1 not in scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_clears_tracking_when_no_state(self, mock_pm, scheduler):
"""Clears tracking when printer has no state (disconnected)."""
scheduler._drying_in_progress = {1: time.monotonic()}
mock_pm.get_status.return_value = None
await scheduler._stop_drying(1)
assert 1 not in scheduler._drying_in_progress
class TestMinimumDryingTime:
"""Regression #1892: a running drying cycle must never be stopped by a humidity re-check.
Relative humidity reads low in heated air (the AMS sensor sees ~15-20% within
minutes of the dryer starting even while the filament is still saturated), so a
humidity-based auto-stop would truncate every cycle — manual or Bambuddy-started —
to the old minimum-time floor. Drying is now left to run to its configured
duration; the firmware stops it when the duration elapses.
"""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_no_stop_before_minimum_time(self, mock_sd, mock_pm, scheduler):
"""Drying should NOT stop when humidity drops below threshold shortly after start."""
# Simulate: drying started 5 minutes ago
scheduler._drying_in_progress = {1: time.monotonic() - 300}
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 600,
"humidity_raw": "18",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
# Mock _is_printer_idle and DB
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
# Mock settings: enabled, threshold=21
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("21"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns, printer_id=1))
# Queue item with schedule
item = MagicMock()
item.printer_id = 1
item.scheduled_time = MagicMock() # Has a schedule
item.manual_start = False
await scheduler._check_auto_drying(db, [item], set())
# Should NOT have sent stop command via humidity check — minimum time not elapsed
# The only calls should NOT include the humidity-based stop
for call in mock_pm.send_drying_command.call_args_list:
# If any stop was called, it should NOT be from the humidity path
# (humidity path uses keyword args: temp=0, duration=0, mode=0)
assert call != ((1, 0), {"temp": 0, "duration": 0, "mode": 0}), (
"Humidity-based stop should not fire before minimum drying time"
)
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_no_stop_after_long_elapsed_time(self, mock_sd, mock_pm, scheduler):
"""#1892: drying must NOT stop even long after start with low humidity — let it run."""
# Simulate: drying started 35 minutes ago, humidity reads low (heated air)
scheduler._drying_in_progress = {1: time.monotonic() - 2100}
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 600,
"humidity_raw": "18",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("21"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns, printer_id=1))
item = MagicMock()
item.printer_id = 1
item.scheduled_time = MagicMock()
item.manual_start = False
await scheduler._check_auto_drying(db, [item], set())
# Must NOT send a humidity-based stop — drying is left to run to its duration
for call in mock_pm.send_drying_command.call_args_list:
assert call != ((1, 0), {"temp": 0, "duration": 0, "mode": 0}), (
"Humidity re-check must never stop a running drying cycle (#1892)"
)
@staticmethod
def _make_setting(value):
s = MagicMock()
s.value = value
return s
@staticmethod
def _make_db_side_effect(settings_map, printer_id=1):
"""Create a side_effect for db.execute that returns settings and printers."""
async def side_effect(stmt):
result = MagicMock()
stmt_str = str(stmt)
# Extract bind parameter values (SQLAlchemy uses :key_1 placeholders)
try:
compiled = stmt.compile(compile_kwargs={"literal_binds": False})
param_values = list(compiled.params.values())
except Exception:
param_values = []
# Match settings queries by checking bind parameter values
matched = False
for key, val in settings_map.items():
if key in param_values:
result.scalar_one_or_none.return_value = val
matched = True
break
if not matched:
if "printer" in stmt_str.lower() or "is_active" in stmt_str:
printer = MagicMock()
printer.id = printer_id
printer.is_active = True
scalars_mock = MagicMock()
scalars_mock.__iter__ = MagicMock(return_value=iter([printer]))
result.scalars.return_value = scalars_mock
else:
result.scalar_one_or_none.return_value = None
return result
return side_effect
class TestAutoStopOnFeatureDisabled:
"""Regression: disabling auto-drying in settings should stop active drying sessions."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_stops_drying_when_disabled(self, mock_pm, scheduler):
"""Disabling auto-drying should send stop commands to all drying printers."""
scheduler._drying_in_progress = {1: time.monotonic(), 2: time.monotonic()}
scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
scheduler._auto_dry_units[(2, 0)] = {"ended_at": None}
# Printer 1: drying, Printer 2: drying
def get_status(pid):
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
return state
mock_pm.get_status.side_effect = get_status
db = AsyncMock()
# queue_drying_enabled = false
setting = MagicMock()
setting.value = "false"
result_mock = MagicMock()
result_mock.scalar_one_or_none.return_value = setting
db.execute = AsyncMock(return_value=result_mock)
await scheduler._check_auto_drying(db, [], set())
# Should have sent stop commands
assert mock_pm.send_drying_command.call_count == 2
assert not scheduler._drying_in_progress
class TestAutoStopOnNoScheduledItems:
"""Regression: removing scheduled items should stop auto-drying."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _make_setting(value):
s = MagicMock()
s.value = value
return s
@staticmethod
def _make_db_side_effect(settings_map):
"""Create a side_effect for db.execute that returns settings by key."""
async def side_effect(stmt):
result = MagicMock()
try:
compiled = stmt.compile(compile_kwargs={"literal_binds": False})
param_values = list(compiled.params.values())
except Exception:
param_values = []
for key, val in settings_map.items():
if key in param_values:
result.scalar_one_or_none.return_value = val
return result
result.scalar_one_or_none.return_value = None
return result
return side_effect
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_stops_when_no_scheduled_items(self, mock_pm, scheduler):
"""Auto-drying stops when queue has no scheduled items (queue mode only)."""
scheduler._drying_in_progress = {1: time.monotonic()}
scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
mock_pm.get_status.return_value = state
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# Manual-start items only (no scheduled_time)
item = MagicMock()
item.printer_id = 1
item.scheduled_time = None
item.manual_start = True
await scheduler._check_auto_drying(db, [item], set())
# Should have stopped drying
assert mock_pm.send_drying_command.called
assert not scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_stops_when_empty_queue(self, mock_pm, scheduler):
"""Auto-drying stops when queue is completely empty (queue mode only)."""
scheduler._drying_in_progress = {1: time.monotonic()}
scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
mock_pm.get_status.return_value = state
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], set())
assert mock_pm.send_drying_command.called
assert not scheduler._drying_in_progress
class TestDryingTrackingTimestamps:
"""Test that _drying_in_progress uses timestamps, not booleans."""
def test_initial_state_empty(self):
"""Fresh scheduler has no drying tracked."""
scheduler = PrintScheduler()
assert scheduler._drying_in_progress == {}
def test_timestamp_is_monotonic(self):
"""Tracked values should be monotonic timestamps."""
scheduler = PrintScheduler()
before = time.monotonic()
scheduler._drying_in_progress[1] = time.monotonic()
after = time.monotonic()
assert before <= scheduler._drying_in_progress[1] <= after
def test_timestamp_is_truthy(self):
"""Timestamps are truthy for .get() checks (backward compat with bool pattern)."""
scheduler = PrintScheduler()
scheduler._drying_in_progress[1] = time.monotonic()
assert scheduler._drying_in_progress.get(1)
assert not scheduler._drying_in_progress.get(999)
class _DryingTestBase:
"""Shared helpers for auto-drying integration tests."""
@staticmethod
def _make_setting(value):
s = MagicMock()
s.value = value
return s
@staticmethod
def _make_db_side_effect(settings_map, printer_ids=None):
"""Create a side_effect for db.execute that returns settings by key and printers."""
if printer_ids is None:
printer_ids = [1]
async def side_effect(stmt):
result = MagicMock()
stmt_str = str(stmt)
try:
compiled = stmt.compile(compile_kwargs={"literal_binds": False})
param_values = list(compiled.params.values())
except Exception:
param_values = []
for key, val in settings_map.items():
if key in param_values:
result.scalar_one_or_none.return_value = val
return result
if "printer" in stmt_str.lower() or "is_active" in stmt_str:
printers = []
for pid in printer_ids:
p = MagicMock()
p.id = pid
p.is_active = True
printers.append(p)
scalars_mock = MagicMock()
scalars_mock.__iter__ = MagicMock(return_value=iter(printers))
result.scalars.return_value = scalars_mock
else:
result.scalar_one_or_none.return_value = None
return result
return side_effect
class TestAmbientDrying(_DryingTestBase):
"""Tests for ambient drying mode — drying based on humidity regardless of queue state."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_ambient_dries_idle_printer_without_queue(self, mock_sd, mock_pm, scheduler):
"""Ambient mode starts drying on idle printers even with no queue items."""
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": "75",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# Empty queue — ambient mode should still dry
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
assert 1 in scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_ambient_does_not_dry_below_threshold(self, mock_sd, mock_pm, scheduler):
"""Ambient mode does NOT dry when humidity is below threshold."""
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": "40",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_not_called()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_ambient_off_stops_drying_without_queue(self, mock_pm, scheduler):
"""Disabling ambient drying stops drying on printers without queue items."""
scheduler._drying_in_progress = {1: time.monotonic()}
scheduler._auto_dry_units[(1, 0)] = {"ended_at": None}
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "dry_time": 120}]}
mock_pm.get_status.return_value = state
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("false"),
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], set())
assert mock_pm.send_drying_command.called
assert not scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_ambient_continues_when_queue_empty(self, mock_sd, mock_pm, scheduler):
"""Ambient drying continues even when queue has no scheduled items (unlike queue mode)."""
scheduler._drying_in_progress = {1: time.monotonic() - 100}
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 600,
"humidity_raw": "75",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], set())
# Should NOT have sent stop — humidity still high, drying continues
for call in mock_pm.send_drying_command.call_args_list:
assert call.kwargs.get("mode") != 0, "Should not stop drying in ambient mode with high humidity"
assert 1 in scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_queue_only_does_not_dry_without_scheduled_items(self, mock_sd, mock_pm, scheduler):
"""Queue mode alone does NOT dry printers that have no scheduled queue items."""
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": "75",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# No queue items at all
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_not_called()
class TestBlockForDryingBugFix(_DryingTestBase):
"""Regression: block mode gates NEW drying starts but must leave running dries alone (#1892)."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_block_mode_leaves_active_drying_running(self, mock_sd, mock_pm, scheduler):
"""#1892: a printer already drying in block mode must not be stopped by a humidity re-check."""
# Drying started 35 minutes ago
scheduler._drying_in_progress = {1: time.monotonic() - 2100}
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 600,
"humidity_raw": "30", # Below threshold
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("true"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# Queue item exists for this printer (triggers block mode gate)
item = MagicMock()
item.printer_id = 1
item.scheduled_time = MagicMock()
item.manual_start = False
await scheduler._check_auto_drying(db, [item], set())
# Must NOT stop the running dry — block mode gates new starts, not active cycles
for call in mock_pm.send_drying_command.call_args_list:
assert call != ((1, 0), {"temp": 0, "duration": 0, "mode": 0}), (
"Block mode must not stop an already-running drying cycle (#1892)"
)
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_block_mode_prevents_new_drying_start(self, mock_sd, mock_pm, scheduler):
"""Block mode should still prevent starting NEW drying on printers with pending items."""
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": "75",
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("true"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
item = MagicMock()
item.printer_id = 1
item.scheduled_time = MagicMock()
item.manual_start = False
await scheduler._check_auto_drying(db, [item], set())
# Should NOT start drying — block mode with pending items
mock_pm.send_drying_command.assert_not_called()
class TestResolveHumidityThreshold:
"""Per-filament humidity threshold resolver (#1605).
Resolves the trigger threshold for an AMS unit from the loaded tray types.
Mixed loads use the lowest (most restrictive) value. Empty / unloaded trays
contribute no constraint; falls back to the global ``ams_humidity_fair``
when no per-type overrides are configured.
"""
def test_no_overrides_falls_back_to_global(self):
"""Empty overrides map → caller's global fallback is used verbatim."""
result = PrintScheduler.resolve_humidity_threshold([{"tray_type": "PLA"}], {}, 60)
assert result == 60
def test_single_known_type_uses_override(self):
"""Single PLA tray with override = 50 returns 50."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": "PLA Basic"}],
{"default": 60, "PLA": 50},
60,
)
assert result == 50
def test_a_composite_takes_its_base_materials_threshold(self):
"""Same lookup, same gap (#3067): a PA6-CF spool read as an unknown type
and took the default, so the override the user set for nylon -- the
material most worth a low threshold -- never applied to the spool they
set it for."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": "PA6-CF"}],
{"default": 60, "PA": 20},
60,
)
assert result == 20
def test_a_composite_with_its_own_threshold_row_keeps_it(self):
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": "PETG-CF"}],
{"default": 60, "PETG": 55, "PETG-CF": 40},
60,
)
assert result == 40
def test_mixed_load_picks_lowest(self):
"""Mixed PLA (60) + Nylon (20) → most restrictive = 20."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": "PLA Basic"}, {"tray_type": "PA Glass"}],
{"default": 60, "PLA": 60, "PA": 20},
60,
)
assert result == 20
def test_unknown_type_uses_default_key(self):
"""Tray type not in the map falls back to the 'default' key, not the
caller fallback. Lets the user tune unknown-filament behavior."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": "EXOTIC_WOOD"}],
{"default": 40, "PLA": 60},
999,
)
assert result == 40
def test_empty_tray_slots_skipped(self):
"""Empty tray_type strings (unloaded slots) contribute no constraint."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": ""}, {"tray_type": "PLA"}],
{"default": 30, "PLA": 50},
60,
)
assert result == 50
def test_all_empty_trays_uses_default_key(self):
"""No loaded trays at all → falls back to default key (or fallback if
no overrides). Matches the empty-AMS behavior of the existing alarm
site so an empty AMS still alarms at the user's default rate."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": ""}, {}],
{"default": 30, "PLA": 50},
60,
)
assert result == 30
def test_filament_name_normalized(self):
"""Tray types like 'PLA Basic', 'pla basic' all normalize to 'PLA'."""
result = PrintScheduler.resolve_humidity_threshold(
[{"tray_type": "pla basic"}],
{"default": 60, "PLA": 25},
60,
)
assert result == 25
def test_no_tray_type_field_skipped(self):
"""Missing tray_type field is treated as empty (unloaded)."""
result = PrintScheduler.resolve_humidity_threshold(
[{}, {"tray_type": "ASA"}],
{"default": 60, "ASA": 30},
60,
)
assert result == 30
class TestGetHumidityThresholds:
"""The DB-loading helper for ``ams_humidity_thresholds`` (#1605)."""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@pytest.mark.asyncio
async def test_missing_setting_returns_empty(self, scheduler):
db = AsyncMock()
db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=None)))
result = await scheduler._get_humidity_thresholds(db)
assert result == {}
@pytest.mark.asyncio
async def test_empty_value_returns_empty(self, scheduler):
db = AsyncMock()
setting = MagicMock(value="")
db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=setting)))
result = await scheduler._get_humidity_thresholds(db)
assert result == {}
@pytest.mark.asyncio
async def test_invalid_json_returns_empty(self, scheduler):
db = AsyncMock()
setting = MagicMock(value="not json{")
db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=setting)))
result = await scheduler._get_humidity_thresholds(db)
assert result == {}
@pytest.mark.asyncio
async def test_valid_json_normalizes_keys(self, scheduler):
"""Filament-type keys uppercase; 'default' preserved."""
db = AsyncMock()
setting = MagicMock(value='{"default": 60, "pla": 50, "ASA": 30, "garbage": "x"}')
db.execute = AsyncMock(return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=setting)))
result = await scheduler._get_humidity_thresholds(db)
assert result == {"default": 60, "PLA": 50, "ASA": 30}
class TestMidPrintDrying(_DryingTestBase):
"""Tests for the print_drying_enabled path — drying that runs CONCURRENTLY
with an active print on capable hardware (H2D / H2C / H2S / P2S / X2D / X1C /
A2L / H2D Pro on recent firmware). Distinct from idle drying.
Verifies:
- With the toggle ON and capable hardware, a printer in the busy set is
still evaluated and drying fires at the capped temperature.
- The temperature cap is max(40, preset_temp - 5) — protects spools.
- With the toggle OFF, the existing busy-printer skip still applies.
- With the toggle ON but unsupported firmware, the busy-printer skip
still applies (gated by supports_drying_while_printing).
"""
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _ams_unit(humidity: str = "75"):
return {
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": humidity,
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
def _state(self, firmware: str):
state = MagicMock()
state.raw_data = {"ams": [self._ams_unit()]}
state.firmware_version = firmware
return state
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_running_printer_dries_when_enabled_and_capable(self, mock_pm, scheduler):
"""Toggle ON + capable hardware: running printer dries at capped temp."""
state = self._state("01.03.00.00")
state.state = "RUNNING"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=False)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"print_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# Actually printing (RUNNING), so the mid-print path applies
await scheduler._check_auto_drying(db, [], {1})
# PLA preset is 45 degC for n3f; mid-print cap is max(40, 45-5) = 40
mock_pm.send_drying_command.assert_called_once_with(1, 0, 40, 12, mode=1, filament="PLA")
assert 1 in scheduler._drying_in_progress
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_temp_cap_applied_above_floor(self, mock_pm, scheduler):
"""Higher-temp filament (PETG n3f=65) caps to 60, not floor."""
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": 0,
"humidity_raw": "75",
"dry_sf_reason": [],
"tray": [{"tray_type": "PETG"}],
}
]
}
state.firmware_version = "01.03.00.00"
state.state = "RUNNING"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=False)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"print_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], {1})
# PETG preset 65 -> max(40, 65-5) = 60
mock_pm.send_drying_command.assert_called_once_with(1, 0, 60, 12, mode=1, filament="PETG")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_running_printer_skipped_when_toggle_off(self, mock_sd, mock_pm, scheduler):
"""Toggle OFF: running printer is skipped even on capable hardware."""
mock_pm.get_status.return_value = self._state("01.03.00.00")
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
scheduler._is_printer_idle = MagicMock(return_value=False)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"print_drying_enabled": self._make_setting("false"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], {1})
mock_pm.send_drying_command.assert_not_called()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_running_printer_skipped_when_firmware_too_old(self, mock_pm, scheduler):
"""Toggle ON but firmware below matrix threshold: skip."""
# H2D matrix minimum is 01.03.00.00; this is below
mock_pm.get_status.return_value = self._state("01.02.30.00")
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
scheduler._is_printer_idle = MagicMock(return_value=False)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"print_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], {1})
mock_pm.send_drying_command.assert_not_called()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_running_printer_skipped_when_model_excluded(self, mock_pm, scheduler):
"""Toggle ON, recent firmware, but excluded model (A1): skip."""
mock_pm.get_status.return_value = self._state("99.99.99.99")
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "A1"
scheduler._is_printer_idle = MagicMock(return_value=False)
db = AsyncMock()
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"print_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], {1})
mock_pm.send_drying_command.assert_not_called()
class TestAutoDryRearmGuards(_DryingTestBase):
"""The re-arm loop from #2770.
An AMS reads higher humidity while it is warm than once it has cooled, so a
threshold set inside that band is never satisfied at the moment a cycle
ends. The firmware is separately free to end a cycle early when it decides
the filament is dry. Together those produced five 12-hour cycles armed in
four hours on the reporter's H2D, one of them six seconds after the previous
ended. These tests pin the two guards that break the loop and, just as
importantly, that neither guard ever stops a cycle that is running.
"""
THRESHOLD = "14"
ABOVE = 16
BELOW = 10
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _ams_state(dry_time, humidity):
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": dry_time,
"humidity_raw": str(humidity),
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
return state
def _db(self):
db = AsyncMock()
db.execute = AsyncMock(
side_effect=self._make_db_side_effect(
{
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting(self.THRESHOLD),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
)
)
return db
async def _pass(self, scheduler, mock_pm, db, dry_time, humidity):
"""One 30-second scheduler pass with the AMS in the given state."""
mock_pm.get_status.return_value = self._ams_state(dry_time, humidity)
await scheduler._check_auto_drying(db, [], set())
async def _unproductive_cycle(self, scheduler, mock_pm, db):
"""Arm a cycle, watch it run, then see it end with humidity still high."""
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
await self._pass(scheduler, mock_pm, db, 720, self.ABOVE)
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_does_not_rearm_immediately_after_a_cycle_ends(self, mock_sd, mock_pm, scheduler):
"""The six-second re-arm: one cycle ends, the next pass must not start another."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
await self._unproductive_cycle(scheduler, mock_pm, db)
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
assert mock_pm.send_drying_command.call_count == 1
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.notification_service")
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_suspends_after_repeated_unproductive_cycles(self, mock_sd, mock_pm, mock_notify, scheduler):
"""Past the cooldown the loop would resume, so the counter has to stop it."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
mock_notify.on_ams_drying_suspended = AsyncMock()
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
for _ in range(AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES):
await self._unproductive_cycle(scheduler, mock_pm, db)
# Age the last cycle out of its cooldown so the next arm is allowed.
scheduler._auto_dry_units[(1, 0)]["ended_at"] -= AUTO_DRY_REARM_COOLDOWN_SECONDS + 1
assert mock_pm.send_drying_command.call_count == AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
assert mock_pm.send_drying_command.call_count == AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES
assert scheduler._auto_dry_units[(1, 0)]["suspended"] is True
mock_notify.on_ams_drying_suspended.assert_awaited_once()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.notification_service")
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_suspension_notifies_once_and_stays_put(self, mock_sd, mock_pm, mock_notify, scheduler):
"""A suspension is a state, not a repeating alarm."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
mock_notify.on_ams_drying_suspended = AsyncMock()
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
scheduler._auto_dry_units[(1, 0)] = {
"unproductive": AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES,
"suspended": False,
"ended_at": None,
}
for _ in range(4):
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
mock_pm.send_drying_command.assert_not_called()
assert mock_notify.on_ams_drying_suspended.await_count == 1
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.notification_service")
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_humidity_dropping_lifts_the_suspension(self, mock_sd, mock_pm, mock_notify, scheduler):
"""The reading coming down is the evidence that drying works after all."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
mock_notify.on_ams_drying_suspended = AsyncMock()
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
scheduler._auto_dry_units[(1, 0)] = {
"unproductive": AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES,
"suspended": True,
"ended_at": None,
}
await self._pass(scheduler, mock_pm, db, 0, self.BELOW)
# The judgement is cleared, but the re-arm clock is kept (#2801).
assert not scheduler._auto_dry_units[(1, 0)].get("suspended")
assert not scheduler._auto_dry_units[(1, 0)].get("unproductive")
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_manual_cycle_is_neither_counted_nor_interrupted(self, mock_sd, mock_pm, scheduler):
"""A dry the user started by hand carries no history and is never stopped."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
await self._pass(scheduler, mock_pm, db, 720, self.ABOVE)
await self._pass(scheduler, mock_pm, db, 600, self.ABOVE)
mock_pm.send_drying_command.assert_not_called()
assert (1, 0) not in scheduler._auto_dry_units
# It ends; Bambuddy is free to arm its own cycle, with a clean slate.
await self._pass(scheduler, mock_pm, db, 0, self.ABOVE)
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
assert scheduler._auto_dry_units[(1, 0)]["unproductive"] == 0
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_history_is_dropped_when_the_printer_goes_away(self, mock_sd, mock_pm, scheduler):
"""A printer deleted and re-added must not inherit a suspension."""
mock_pm.get_status.return_value = None
scheduler._auto_dry_units[(1, 0)] = {"unproductive": 5, "suspended": True, "ended_at": None}
scheduler._sync_drying_state()
assert scheduler._auto_dry_units == {}
class TestAutoDryStoppedByBambuddy(_DryingTestBase):
"""A cycle Bambuddy itself cut short must not count against the unit."""
THRESHOLD = "14"
ABOVE = 16
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _ams_state(dry_time, humidity):
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": dry_time,
"humidity_raw": str(humidity),
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
return state
def _db(self):
db = AsyncMock()
db.execute = AsyncMock(
side_effect=self._make_db_side_effect(
{
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting(self.THRESHOLD),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
)
)
return db
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_print_takes_priority_stop_is_not_an_unproductive_cycle(self, mock_sd, mock_pm, scheduler):
"""The queue stopping a dry so a print can start is Bambuddy's own doing.
Counting it would suspend auto-drying on any printer that dries between
jobs often enough -- exactly the install queue-drying exists for.
"""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
mock_pm.get_status.return_value = self._ams_state(0, self.ABOVE)
await scheduler._check_auto_drying(db, [], set())
mock_pm.get_status.return_value = self._ams_state(720, self.ABOVE)
await scheduler._check_auto_drying(db, [], set())
# A print is ready: check_queue stops drying on this printer.
await scheduler._stop_drying(1)
mock_pm.get_status.return_value = self._ams_state(0, self.ABOVE)
await scheduler._check_auto_drying(db, [], set())
assert scheduler._auto_dry_units[(1, 0)]["unproductive"] == 0
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_manual_stop_button_is_not_an_unproductive_cycle(self, mock_sd, mock_pm, scheduler):
"""The Stop button goes straight to printer_manager, bypassing the
scheduler, so the route has to tell the scheduler to forget the cycle."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
mock_pm.get_status.return_value = self._ams_state(0, self.ABOVE)
await scheduler._check_auto_drying(db, [], set())
mock_pm.get_status.return_value = self._ams_state(720, self.ABOVE)
await scheduler._check_auto_drying(db, [], set())
scheduler.forget_auto_dry_cycle(1, 0)
mock_pm.get_status.return_value = self._ams_state(0, self.ABOVE)
await scheduler._check_auto_drying(db, [], set())
assert scheduler._auto_dry_units[(1, 0)]["unproductive"] == 0
class TestAutoDryProgressKeepsItGoing(_DryingTestBase):
"""Suspension must not punish a spool that is genuinely drying, just slowly."""
THRESHOLD = "25"
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _ams_state(dry_time, humidity):
state = MagicMock()
state.raw_data = {
"ams": [
{
"id": 0,
"module_type": "n3f",
"dry_time": dry_time,
"humidity_raw": str(humidity),
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
]
}
state.firmware_version = "01.09.00.00"
return state
def _db(self):
db = AsyncMock()
db.execute = AsyncMock(
side_effect=self._make_db_side_effect(
{
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"ams_humidity_fair": self._make_setting(self.THRESHOLD),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
)
)
return db
async def _cycle(self, scheduler, mock_pm, db, end_humidity):
"""Arm a cycle, run it, and end it at the given reading."""
mock_pm.get_status.return_value = self._ams_state(0, end_humidity)
await scheduler._check_auto_drying(db, [], set())
mock_pm.get_status.return_value = self._ams_state(720, end_humidity)
await scheduler._check_auto_drying(db, [], set())
mock_pm.get_status.return_value = self._ams_state(0, end_humidity)
await scheduler._check_auto_drying(db, [], set())
entry = scheduler._auto_dry_units.get((1, 0))
if entry is not None and isinstance(entry.get("ended_at"), float):
entry["ended_at"] -= AUTO_DRY_REARM_COOLDOWN_SECONDS + 1
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.notification_service")
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_a_reading_that_keeps_falling_is_never_suspended(self, mock_sd, mock_pm, mock_notify, scheduler):
"""40 -> 37 -> 35 -> 33 with a 25% threshold: nowhere near it yet, but
every cycle is working. A humid workshop must not lose auto-drying."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
mock_notify.on_ams_drying_suspended = AsyncMock()
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
for reading in (40, 37, 35, 33, 31, 29):
await self._cycle(scheduler, mock_pm, db, reading)
assert scheduler._auto_dry_units[(1, 0)]["suspended"] is False
mock_notify.on_ams_drying_suspended.assert_not_awaited()
assert mock_pm.send_drying_command.call_count == 6
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.notification_service")
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_a_reading_oscillating_by_one_point_still_suspends(self, mock_sd, mock_pm, mock_notify, scheduler):
"""A plateau with sensor noise — 31, 30, 31, 30 — is not progress.
Comparing against the previous cycle rather than the best-so-far would
read every other cycle as an improvement and loop indefinitely."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
mock_notify.on_ams_drying_suspended = AsyncMock()
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db()
for reading in (31, 30, 31, 30, 31, 30):
await self._cycle(scheduler, mock_pm, db, reading)
assert scheduler._auto_dry_units[(1, 0)]["suspended"] is True
mock_notify.on_ams_drying_suspended.assert_awaited_once()
class TestAmbientDryingSustainedDelay(_DryingTestBase):
"""#2518 — ambient auto-drying waits for CONTINUOUSLY-above-threshold humidity.
``ambient_drying_sustained_minutes`` (default 0 = instant, matching the
pre-#2518 behavior) makes a pure-ambient start wait for the reading to sit
above the threshold for that many minutes straight before drying begins.
The streak lives in ``scheduler._auto_dry_above``, keyed like
``_auto_dry_units`` by (printer_id, ams_id), but deliberately a separate
dict: arming a streak must never look like a Bambuddy-started cycle to
``_auto_dry_units``, which is what ``_stop_drying`` and the manual-cycle
immunity in #2801 key off of.
"""
UNIT_KEY = (1, 0)
@pytest.fixture
def scheduler(self):
return PrintScheduler()
@staticmethod
def _ams_unit(dry_time=0, humidity="75", include_humidity=True):
ams = {
"id": 0,
"module_type": "n3f",
"dry_time": dry_time,
"dry_sf_reason": [],
"tray": [{"tray_type": "PLA"}],
}
if include_humidity:
ams["humidity_raw"] = humidity
return ams
@classmethod
def _state(cls, dry_time=0, humidity="75", include_humidity=True):
state = MagicMock()
state.raw_data = {"ams": [cls._ams_unit(dry_time, humidity, include_humidity)]}
state.firmware_version = "01.09.00.00"
return state
def _db(self, sustained_minutes=None, queue_enabled="false", ambient_enabled="true"):
settings = {
"queue_drying_enabled": self._make_setting(queue_enabled),
"ambient_drying_enabled": self._make_setting(ambient_enabled),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
if sustained_minutes is not None:
settings["ambient_drying_sustained_minutes"] = self._make_setting(str(sustained_minutes))
db = AsyncMock()
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings))
return db
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_setting_absent_dries_immediately(self, mock_sd, mock_pm, scheduler):
"""No ``ambient_drying_sustained_minutes`` row → default 0 → instant dry,
same as ambient mode before #2518."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
await scheduler._check_auto_drying(self._db(sustained_minutes=None), [], set())
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_first_above_threshold_pass_arms_but_does_not_dry(self, mock_sd, mock_pm, scheduler):
"""First observation above threshold arms the streak instead of drying."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
mock_pm.send_drying_command.assert_not_called()
assert self.UNIT_KEY in scheduler._auto_dry_above
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_matured_streak_dries_on_next_pass(self, mock_sd, mock_pm, scheduler):
"""Once the streak's age clears the configured minutes, the next pass dries."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5)
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_not_called()
# Age the streak past the 5-minute requirement, keeping "last" recent
# so the observation-gap guard doesn't treat it as a fresh streak.
scheduler._auto_dry_above[self.UNIT_KEY]["since"] = time.monotonic() - (5 * 60 + 5)
scheduler._auto_dry_above[self.UNIT_KEY]["last"] = time.monotonic() - 5
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_dip_below_threshold_clears_streak(self, mock_sd, mock_pm, scheduler):
"""A below-threshold reading between passes clears the streak — the wait restarts."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5)
mock_pm.get_status.return_value = self._state(humidity="75")
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY in scheduler._auto_dry_above
mock_pm.get_status.return_value = self._state(humidity="40") # below the 60% threshold
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY not in scheduler._auto_dry_above
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_missing_humidity_does_not_clear_armed_streak(self, mock_sd, mock_pm, scheduler):
"""No humidity_raw/humidity in the AMS payload is no-information, not a dip below threshold."""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5)
mock_pm.get_status.return_value = self._state(humidity="75")
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY in scheduler._auto_dry_above
mock_pm.get_status.return_value = self._state(include_humidity=False)
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY in scheduler._auto_dry_above
mock_pm.send_drying_command.assert_not_called()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_manual_cycle_immunity(self, mock_sd, mock_pm, scheduler):
"""Arming a streak must never create a ``_auto_dry_units`` entry — that dict is
reserved for cycles Bambuddy itself started (#2801's manual-cycle immunity). A
unit already drying by the user's own hand (dry_time > 0) with a streak still
present is left alone: no command sent, no entry created, streak untouched.
"""
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5)
# First pass: nothing drying yet, arms the streak.
mock_pm.get_status.return_value = self._state(dry_time=0)
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY not in scheduler._auto_dry_units
assert self.UNIT_KEY in scheduler._auto_dry_above
# A manual cycle is now running on the same unit — dry_time > 0.
mock_pm.get_status.return_value = self._state(dry_time=120)
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_not_called()
assert self.UNIT_KEY not in scheduler._auto_dry_units
assert self.UNIT_KEY in scheduler._auto_dry_above
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_scheduled_queue_item_dries_immediately(self, mock_sd, mock_pm, scheduler):
"""A printer with a scheduled (non-manual) queue item dries instantly —
the sustained wait applies only to pure-ambient starts."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5, queue_enabled="true", ambient_enabled="false")
item = MagicMock()
item.printer_id = 1
item.scheduled_time = MagicMock()
item.manual_start = False
await scheduler._check_auto_drying(db, [item], set())
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_observation_gap_restarts_the_streak(self, mock_sd, mock_pm, scheduler):
"""A streak whose last observation is older than the gap ceiling
can no longer claim "continuously above" — it is treated as fresh and the
pass must not dry, even though 'since' alone clears the required minutes."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
old_since = time.monotonic() - (5 * 60 + 100)
scheduler._auto_dry_above[self.UNIT_KEY] = {
"since": old_since,
"last": time.monotonic() - (AUTO_DRY_SUSTAINED_GAP_FLOOR_SECONDS + 10),
}
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
mock_pm.send_drying_command.assert_not_called()
# The streak was restarted, not left at its stale (matured) age.
assert scheduler._auto_dry_above[self.UNIT_KEY]["since"] > old_since + 90
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_suspension_wins_over_a_matured_streak(self, mock_sd, mock_pm, scheduler):
"""A suspended unit (#2770) stays suspended even once the sustained wait matures."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
scheduler._auto_dry_units[self.UNIT_KEY] = {
"unproductive": AUTO_DRY_MAX_UNPRODUCTIVE_CYCLES,
"suspended": True,
"ended_at": None,
}
scheduler._auto_dry_above[self.UNIT_KEY] = {
"since": time.monotonic() - (5 * 60 + 5),
"last": time.monotonic() - 5,
}
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
mock_pm.send_drying_command.assert_not_called()
assert scheduler._auto_dry_units[self.UNIT_KEY]["suspended"] is True
def test_forget_auto_dry_cycle_pops_the_streak(self, scheduler):
"""``forget_auto_dry_cycle`` (the route-side stop path) also spends the
streak that armed the cycle (#2518). The other cycle-end path — the
in-loop branch inside ``_check_auto_drying`` that pops ``running`` when
firmware reports ``dry_time == 0`` — is covered separately by
``test_in_loop_cycle_end_restarts_the_streak``."""
scheduler._auto_dry_units[self.UNIT_KEY] = {
"running": True,
"unproductive": 0,
"suspended": False,
"ended_at": None,
}
scheduler._auto_dry_above[self.UNIT_KEY] = {"since": time.monotonic(), "last": time.monotonic()}
scheduler.forget_auto_dry_cycle(*self.UNIT_KEY)
assert self.UNIT_KEY not in scheduler._auto_dry_above
assert "running" not in scheduler._auto_dry_units[self.UNIT_KEY]
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_in_loop_cycle_end_restarts_the_streak(self, mock_sd, mock_pm, scheduler):
"""When the in-loop cycle-end branch fires (a Bambuddy-armed cycle whose
``dry_time`` has gone to 0), the matured streak that armed it is spent —
the pass re-arms a FRESH streak rather than keeping the old ``since``.
Without the pop, a matured pre-cycle streak would survive the cycle and
reduce the sustained wait to the re-arm cooldown on every re-arm."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
scheduler._auto_dry_units[self.UNIT_KEY] = {"running": True}
old_since = time.monotonic() - (60 * 60)
scheduler._auto_dry_above[self.UNIT_KEY] = {"since": old_since, "last": time.monotonic() - 5}
before = time.monotonic()
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
# The cycle-end pass never dries (its own ended_at starts the cooldown)...
mock_pm.send_drying_command.assert_not_called()
# ...and the streak now on file must be fresh, not the pre-cycle one.
assert scheduler._auto_dry_above[self.UNIT_KEY]["since"] >= before
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_streak_arms_during_the_rearm_cooldown(self, mock_sd, mock_pm, scheduler):
"""The wait OVERLAPS the 30-minute re-arm cooldown instead of stacking
after it: a pass blocked by the cooldown still arms/advances the streak,
so a streak matured inside the cooldown dries at cooldown expiry."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5)
scheduler._auto_dry_units[self.UNIT_KEY] = {"ended_at": time.monotonic() - 60}
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_not_called()
# The cooldown blocked the start, but the streak armed anyway.
assert self.UNIT_KEY in scheduler._auto_dry_above
# Cooldown over; the streak that matured inside it satisfies the wait.
scheduler._auto_dry_units[self.UNIT_KEY]["ended_at"] = time.monotonic() - (AUTO_DRY_REARM_COOLDOWN_SECONDS + 5)
scheduler._auto_dry_above[self.UNIT_KEY]["since"] = time.monotonic() - (5 * 60 + 5)
scheduler._auto_dry_above[self.UNIT_KEY]["last"] = time.monotonic() - 5
await scheduler._check_auto_drying(db, [], set())
mock_pm.send_drying_command.assert_called_once_with(1, 0, 45, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_mid_print_ambient_start_waits_like_any_ambient_start(self, mock_pm, scheduler):
"""With ambient drying on, a humidity start on a printer that happens to
be printing must serve the sustained wait like one on an idle printer.
(The original exemption here was disproven live — a 2-point threshold
crossing mid-print bought a parked 12h command instantly.)"""
state = self._state()
state.state = "RUNNING"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=False)
state.firmware_version = "01.03.00.00"
settings_returns = {
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"print_drying_enabled": self._make_setting("true"),
"ambient_drying_sustained_minutes": self._make_setting("5"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db = AsyncMock()
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# First pass: the streak arms; no instant start.
await scheduler._check_auto_drying(db, [], {1})
mock_pm.send_drying_command.assert_not_called()
assert self.UNIT_KEY in scheduler._auto_dry_above
# Streak matured: the start fires, at the mid-print capped
# temperature max(40, 45 - 5) = 40.
scheduler._auto_dry_above[self.UNIT_KEY]["since"] = time.monotonic() - (5 * 60 + 5)
scheduler._auto_dry_above[self.UNIT_KEY]["last"] = time.monotonic() - 5
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
await scheduler._check_auto_drying(db, [], {1})
mock_pm.send_drying_command.assert_called_once_with(1, 0, 40, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_mid_print_with_scheduled_item_stays_instant(self, mock_pm, scheduler):
"""The instant exemption keys on the pending schedule — a printer
drying ahead of a scheduled job keeps it whether idle or printing."""
state = self._state()
state.state = "RUNNING"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=False)
state.firmware_version = "01.03.00.00"
settings_returns = {
"queue_drying_enabled": self._make_setting("false"),
"ambient_drying_enabled": self._make_setting("true"),
"print_drying_enabled": self._make_setting("true"),
"ambient_drying_sustained_minutes": self._make_setting("5"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db = AsyncMock()
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
item = MagicMock()
item.printer_id = 1
item.scheduled_time = MagicMock()
item.manual_start = False
await scheduler._check_auto_drying(db, [item], {1})
mock_pm.send_drying_command.assert_called_once_with(1, 0, 40, 12, mode=1, filament="PLA")
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
async def test_stored_wait_is_inert_while_ambient_drying_is_off(self, mock_pm, scheduler):
"""The settings UI hides the wait while ambient drying is off, so a value
left behind must not keep delaying a mid-print start under print_drying
(queue mode on, no scheduled item): that start stays instant, and no
streak entry is written for it."""
state = self._state()
state.state = "RUNNING"
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "H2D"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=False)
state.firmware_version = "01.03.00.00"
settings_returns = {
"queue_drying_enabled": self._make_setting("true"),
"ambient_drying_enabled": self._make_setting("false"),
"print_drying_enabled": self._make_setting("true"),
"ambient_drying_sustained_minutes": self._make_setting("15"),
"ams_humidity_fair": self._make_setting("60"),
"queue_drying_block": self._make_setting("false"),
"drying_presets": None,
}
db = AsyncMock()
db.execute = AsyncMock(side_effect=self._make_db_side_effect(settings_returns))
# A streak left over from when ambient drying was on is dropped too.
scheduler._auto_dry_above[self.UNIT_KEY] = {"since": time.monotonic(), "last": time.monotonic()}
await scheduler._check_auto_drying(db, [], {1})
mock_pm.send_drying_command.assert_called_once_with(1, 0, 40, 12, mode=1, filament="PLA")
assert scheduler._auto_dry_above == {}
@patch("backend.app.services.print_scheduler.printer_manager")
def test_sync_drying_state_prunes_streaks_of_vanished_printers(self, mock_pm, scheduler):
"""A printer that has gone away takes its streak with it — a deleted and
re-added printer starts a fresh wait, and entries do not leak."""
mock_pm.get_status.return_value = None
scheduler._auto_dry_above[(99, 0)] = {"since": time.monotonic(), "last": time.monotonic()}
scheduler._auto_dry_units[(99, 0)] = {"unproductive": 1}
scheduler._sync_drying_state()
assert (99, 0) not in scheduler._auto_dry_above
assert (99, 0) not in scheduler._auto_dry_units
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_dip_below_threshold_logs_the_discard_at_info(self, mock_sd, mock_pm, scheduler, caplog):
"""Discarding an armed streak on a real below-threshold reading is logged
at INFO — it is the only way an operator can tell why an ambient dry
never started."""
import logging
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
db = self._db(sustained_minutes=5)
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY in scheduler._auto_dry_above
mock_pm.get_status.return_value = self._state(humidity="50")
with caplog.at_level(logging.INFO, logger="backend.app.services.print_scheduler"):
await scheduler._check_auto_drying(db, [], set())
assert self.UNIT_KEY not in scheduler._auto_dry_above
assert any("fell back" in rec.getMessage() for rec in caplog.records if rec.levelno == logging.INFO)
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_observation_gap_restart_logs_at_info(self, mock_sd, mock_pm, scheduler, caplog):
"""Restarting a streak on an observation gap is logged at INFO with the
measured gap — the silent twin of the dip reset, and the only way a
user who set a long wait can see why it never matures."""
import logging
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
scheduler._auto_dry_above[self.UNIT_KEY] = {
"since": time.monotonic() - (5 * 60 + 100),
"last": time.monotonic() - (AUTO_DRY_SUSTAINED_GAP_FLOOR_SECONDS + 10),
}
with caplog.at_level(logging.INFO, logger="backend.app.services.print_scheduler"):
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
restarts = [
rec for rec in caplog.records if rec.levelno == logging.INFO and "streak restarted" in rec.getMessage()
]
assert len(restarts) == 1
# The measured gap is in the message, not just the fact of a restart.
assert re.search(r"\d+s observation gap", restarts[0].getMessage())
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_gap_ceiling_scales_with_the_check_interval(self, mock_sd, mock_pm, scheduler):
"""The ceiling is max(4 * check_interval, floor): a scheduler polling
slower than the default must not void streaks it structurally cannot
observe often enough (review on #2895)."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
scheduler._check_interval = 100 # ceiling 400s, well above the 120s floor
scheduler._auto_dry_above[self.UNIT_KEY] = {
"since": time.monotonic() - (5 * 60 + 100),
# Over the floor, under 4 * check_interval: must NOT restart.
"last": time.monotonic() - (AUTO_DRY_SUSTAINED_GAP_FLOOR_SECONDS + 50),
}
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
# The streak survived the slow pass and had already matured.
mock_pm.send_drying_command.assert_called_once()
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_streak_tracking_is_inert_when_the_feature_is_off(self, mock_sd, mock_pm, scheduler):
"""sustained_minutes 0 must not write streak entries — every install
would otherwise pay the bookkeeping for a feature nobody enabled
(review on #2895)."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
await scheduler._check_auto_drying(self._db(sustained_minutes=0), [], set())
mock_pm.send_drying_command.assert_called_once()
assert scheduler._auto_dry_above == {}
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_toggle_off_drops_a_leftover_streak_entry(self, mock_sd, mock_pm, scheduler):
"""An entry armed while the feature was on is dropped once it is off,
so toggling back on later cannot inherit a stale streak."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
scheduler._auto_dry_above[self.UNIT_KEY] = {
"since": time.monotonic() - 30,
"last": time.monotonic() - 1,
}
await scheduler._check_auto_drying(self._db(sustained_minutes=0), [], set())
assert self.UNIT_KEY not in scheduler._auto_dry_above
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_disabling_all_auto_drying_drops_streak_before_early_return(self, mock_sd, mock_pm, scheduler):
"""A quick disable/re-enable must start a fresh wait, even though the
disabled pass returns before visiting individual AMS units."""
mock_pm.get_status.return_value = self._state()
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
mock_pm.send_drying_command.return_value = True
scheduler._is_printer_idle = MagicMock(return_value=True)
scheduler._auto_dry_above[self.UNIT_KEY] = {
"since": time.monotonic() - 360,
"last": time.monotonic() - 1,
}
await scheduler._check_auto_drying(self._db(sustained_minutes=5, ambient_enabled="false"), [], set())
assert self.UNIT_KEY not in scheduler._auto_dry_above
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
mock_pm.send_drying_command.assert_not_called()
assert self.UNIT_KEY in scheduler._auto_dry_above
@pytest.mark.asyncio
@patch("backend.app.services.print_scheduler.printer_manager")
@patch("backend.app.services.print_scheduler.supports_drying", return_value=True)
async def test_firmware_refusal_skips_before_the_wait_logs(self, mock_sd, mock_pm, scheduler, caplog):
"""A unit the firmware refuses to dry (dry_sf_reason set) hits its skip
before the sustained wait, so it never logs "waiting" for a dry it was
never going to get (review nit on #2895)."""
import logging
state = self._state()
state.raw_data["ams"][0]["dry_sf_reason"] = ["8"]
mock_pm.get_status.return_value = state
mock_pm.is_connected.return_value = True
mock_pm.get_model.return_value = "X1C"
scheduler._is_printer_idle = MagicMock(return_value=True)
with caplog.at_level(logging.DEBUG, logger="backend.app.services.print_scheduler"):
await scheduler._check_auto_drying(self._db(sustained_minutes=5), [], set())
messages = [rec.getMessage() for rec in caplog.records]
assert any("cannot dry reasons" in m for m in messages)
assert not any("waiting" in m for m in messages)
Reference in New Issue View Git Blame Copy Permalink