mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-02 04:01:21 +02:00
maziggy
ead6c37147
The provider toggle, schema, template, and NotificationService.on_printer_offline all shipped, but no caller invoked the dispatcher — the offline event was an orphan toggle. Edge detection in on_printer_status_change now schedules a debounced (60s) background task on the connected→disconnected transition; reconnect before the window elapses cancels it. Covers both upstream paths (smart-plug power-off via mark_printer_offline, and MQTT staleness via check_staleness), both of which already route through the status callback. The "back online" channel is the existing print-failure notification on firmware FAILED report — no symmetric on_printer_online needed.
258 lines
11 KiB
Python
258 lines
11 KiB
Python
"""Tests for the connected → disconnected edge that fires the
|
|
`on_printer_offline` notification (#1752).
|
|
|
|
The provider toggle, schema, and dispatcher already existed; what was missing
|
|
was a caller that fires `notification_service.on_printer_offline` when a
|
|
printer goes offline. These tests pin both layers:
|
|
|
|
* `_maybe_notify_printer_offline` — the debounced background task. Must
|
|
fire when the printer is still offline at the end of the window, and
|
|
must NOT fire if the printer reconnected during the window.
|
|
|
|
* Edge detection inside `on_printer_status_change` — schedules the task
|
|
only on the True → False transition, cancels any pending task on
|
|
reconnect, and stays silent on startup (no prior connected state).
|
|
"""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app import main as main_module
|
|
|
|
|
|
def _state(connected: bool, state: str = "IDLE") -> SimpleNamespace:
|
|
"""Minimal PrinterState stub. `state="IDLE"` keeps the reconcile-edge
|
|
branch quiescent (it only fires on `connected=True` with a non-unknown
|
|
state-string, which we exercise separately) but otherwise lets the
|
|
handler thread through without doing extra DB / WS work."""
|
|
return SimpleNamespace(
|
|
connected=connected,
|
|
state=state,
|
|
progress=0,
|
|
layer_num=0,
|
|
temperatures={},
|
|
raw_data={},
|
|
stg_cur=0,
|
|
cooling_fan_speed=0,
|
|
big_fan1_speed=0,
|
|
big_fan2_speed=0,
|
|
chamber_light="",
|
|
active_extruder=0,
|
|
tray_now=0,
|
|
door_open=False,
|
|
subtask_name="",
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_edge_state():
|
|
"""Clear the module-level edge dicts between tests so one test's
|
|
True-edge doesn't leak into the next."""
|
|
main_module._printer_last_connected.clear()
|
|
for task in list(main_module._printer_offline_notify_tasks.values()):
|
|
if not task.done():
|
|
task.cancel()
|
|
main_module._printer_offline_notify_tasks.clear()
|
|
main_module._printer_reconciled_since_connect.clear()
|
|
main_module._last_status_broadcast.clear()
|
|
yield
|
|
main_module._printer_last_connected.clear()
|
|
for task in list(main_module._printer_offline_notify_tasks.values()):
|
|
if not task.done():
|
|
task.cancel()
|
|
main_module._printer_offline_notify_tasks.clear()
|
|
|
|
|
|
class TestMaybeNotifyPrinterOffline:
|
|
"""The debounced background task — fires notification at the end of the
|
|
window only if the printer is still offline."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fires_notification_when_still_offline_after_debounce(self):
|
|
printer = SimpleNamespace(id=1, name="Workshop")
|
|
scalar = MagicMock()
|
|
scalar.scalar_one_or_none.return_value = printer
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(return_value=scalar)
|
|
session_cm = MagicMock()
|
|
session_cm.__aenter__ = AsyncMock(return_value=db)
|
|
session_cm.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
|
|
patch("backend.app.main.printer_manager") as mock_pm,
|
|
patch("backend.app.main.async_session", return_value=session_cm),
|
|
patch("backend.app.main.notification_service") as mock_notif,
|
|
):
|
|
mock_pm.is_connected.return_value = False
|
|
mock_notif.on_printer_offline = AsyncMock()
|
|
|
|
await main_module._maybe_notify_printer_offline(printer_id=1)
|
|
|
|
mock_notif.on_printer_offline.assert_awaited_once_with(1, "Workshop", db)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_does_not_fire_when_printer_reconnected_during_debounce(self):
|
|
with (
|
|
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
|
|
patch("backend.app.main.printer_manager") as mock_pm,
|
|
patch("backend.app.main.notification_service") as mock_notif,
|
|
):
|
|
mock_pm.is_connected.return_value = True
|
|
mock_notif.on_printer_offline = AsyncMock()
|
|
|
|
await main_module._maybe_notify_printer_offline(printer_id=1)
|
|
|
|
mock_notif.on_printer_offline.assert_not_awaited()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_does_not_fire_when_printer_missing_from_db(self):
|
|
scalar = MagicMock()
|
|
scalar.scalar_one_or_none.return_value = None
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(return_value=scalar)
|
|
session_cm = MagicMock()
|
|
session_cm.__aenter__ = AsyncMock(return_value=db)
|
|
session_cm.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with (
|
|
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
|
|
patch("backend.app.main.printer_manager") as mock_pm,
|
|
patch("backend.app.main.async_session", return_value=session_cm),
|
|
patch("backend.app.main.notification_service") as mock_notif,
|
|
):
|
|
mock_pm.is_connected.return_value = False
|
|
mock_notif.on_printer_offline = AsyncMock()
|
|
|
|
await main_module._maybe_notify_printer_offline(printer_id=1)
|
|
|
|
mock_notif.on_printer_offline.assert_not_awaited()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clears_task_entry_after_run(self):
|
|
with (
|
|
patch("backend.app.main.asyncio.sleep", new=AsyncMock()),
|
|
patch("backend.app.main.printer_manager") as mock_pm,
|
|
):
|
|
mock_pm.is_connected.return_value = True # No notification path
|
|
main_module._printer_offline_notify_tasks[1] = MagicMock()
|
|
await main_module._maybe_notify_printer_offline(printer_id=1)
|
|
assert 1 not in main_module._printer_offline_notify_tasks
|
|
|
|
|
|
class TestOfflineEdgeDetection:
|
|
"""Edge detection inside `on_printer_status_change` — only the
|
|
True → False transition schedules a task. Reconnects cancel pending
|
|
tasks. Startup-with-disconnected does not fire."""
|
|
|
|
@staticmethod
|
|
def _patch_handler_deps():
|
|
"""Patch out the heavy side-effects of `on_printer_status_change`
|
|
(MQTT relay, WebSocket broadcast, state serializer) so we can focus
|
|
on edge state."""
|
|
ws_mgr = MagicMock()
|
|
ws_mgr.send_printer_status = AsyncMock()
|
|
relay = MagicMock()
|
|
relay.on_printer_status = AsyncMock()
|
|
pm = MagicMock()
|
|
pm.get_printer.return_value = None # Skip the relay payload branch.
|
|
pm.get_model.return_value = ""
|
|
return ws_mgr, relay, pm
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_call_connected_does_not_schedule(self):
|
|
ws_mgr, relay, pm = self._patch_handler_deps()
|
|
with (
|
|
patch("backend.app.main.ws_manager", ws_mgr),
|
|
patch("backend.app.main.mqtt_relay", relay),
|
|
patch("backend.app.main.printer_manager", pm),
|
|
patch("backend.app.main.spawn_background_task"),
|
|
patch("backend.app.main.printer_state_to_dict", return_value={}),
|
|
):
|
|
await main_module.on_printer_status_change(1, _state(connected=True))
|
|
assert 1 not in main_module._printer_offline_notify_tasks
|
|
assert main_module._printer_last_connected[1] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_first_call_disconnected_does_not_schedule(self):
|
|
"""Startup with an already-offline printer must not fire — there's
|
|
no prior True observation, so we have no edge to trigger on."""
|
|
ws_mgr, relay, pm = self._patch_handler_deps()
|
|
with (
|
|
patch("backend.app.main.ws_manager", ws_mgr),
|
|
patch("backend.app.main.mqtt_relay", relay),
|
|
patch("backend.app.main.printer_manager", pm),
|
|
patch("backend.app.main.spawn_background_task"),
|
|
patch("backend.app.main.printer_state_to_dict", return_value={}),
|
|
):
|
|
await main_module.on_printer_status_change(1, _state(connected=False))
|
|
assert 1 not in main_module._printer_offline_notify_tasks
|
|
assert main_module._printer_last_connected[1] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connected_to_disconnected_schedules_task(self):
|
|
ws_mgr, relay, pm = self._patch_handler_deps()
|
|
with (
|
|
patch("backend.app.main.ws_manager", ws_mgr),
|
|
patch("backend.app.main.mqtt_relay", relay),
|
|
patch("backend.app.main.printer_manager", pm),
|
|
patch("backend.app.main.spawn_background_task"),
|
|
patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()),
|
|
patch("backend.app.main.printer_state_to_dict", return_value={}),
|
|
):
|
|
await main_module.on_printer_status_change(1, _state(connected=True))
|
|
await main_module.on_printer_status_change(1, _state(connected=False))
|
|
|
|
task = main_module._printer_offline_notify_tasks.get(1)
|
|
assert task is not None
|
|
task.cancel()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reconnect_cancels_pending_task(self):
|
|
ws_mgr, relay, pm = self._patch_handler_deps()
|
|
with (
|
|
patch("backend.app.main.ws_manager", ws_mgr),
|
|
patch("backend.app.main.mqtt_relay", relay),
|
|
patch("backend.app.main.printer_manager", pm),
|
|
patch("backend.app.main.spawn_background_task"),
|
|
patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()),
|
|
patch("backend.app.main.printer_state_to_dict", return_value={}),
|
|
):
|
|
await main_module.on_printer_status_change(1, _state(connected=True))
|
|
await main_module.on_printer_status_change(1, _state(connected=False))
|
|
scheduled = main_module._printer_offline_notify_tasks.get(1)
|
|
assert scheduled is not None
|
|
await main_module.on_printer_status_change(1, _state(connected=True))
|
|
# Yield so the cancellation propagates through the event loop.
|
|
await asyncio.sleep(0)
|
|
|
|
assert 1 not in main_module._printer_offline_notify_tasks
|
|
assert scheduled.cancelled() or scheduled.done()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repeated_disconnected_does_not_reschedule(self):
|
|
"""A second False observation while a task is already pending must
|
|
not replace the in-flight task — otherwise the debounce clock
|
|
resets on every status callback and the notification never fires."""
|
|
ws_mgr, relay, pm = self._patch_handler_deps()
|
|
with (
|
|
patch("backend.app.main.ws_manager", ws_mgr),
|
|
patch("backend.app.main.mqtt_relay", relay),
|
|
patch("backend.app.main.printer_manager", pm),
|
|
patch("backend.app.main.spawn_background_task"),
|
|
patch("backend.app.main._maybe_notify_printer_offline", new=AsyncMock()),
|
|
patch("backend.app.main.printer_state_to_dict", return_value={}),
|
|
):
|
|
await main_module.on_printer_status_change(1, _state(connected=True))
|
|
await main_module.on_printer_status_change(1, _state(connected=False))
|
|
first_task = main_module._printer_offline_notify_tasks.get(1)
|
|
await main_module.on_printer_status_change(1, _state(connected=False))
|
|
second_task = main_module._printer_offline_notify_tasks.get(1)
|
|
|
|
assert first_task is second_task
|
|
if first_task is not None:
|
|
first_task.cancel()
|