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
12d17bfbe7402b52823bdd75f2b3041f56bcb3db
bambuddy/backend/tests/unit/services/test_background_dispatch_watchdog.py
T
maziggy 12d17bfbe7 fix(photo): source finish photo from forced timelapse + cleanup (#1397) 
Bambu's end-gcode lowers the bed at gcode_state=FINISH. Bambuddy's
  live-camera grab captured the bed already dropped, ruining the photo
  framing. Source the photo from a brief Bambu timelapse instead —
  firmware stops timelapse recording AFTER toolhead parks but BEFORE
  bed-drop runs, so the last frame frames the finished print correctly.

  When capture_finish_photo is on AND the user did not opt in to
  timelapse for this print, force timelapse=True at dispatch + mark the
  new PrintArchive.bambuddy_forced_timelapse column. After extraction
  (success or failure), cleanup deletes the locally-attached file,
  clears archive.timelapse_path, and walks the four scanner directories
  (/timelapse, /timelapse/video, /record, /recording) trying FTP DELE
  against the original filename. User-opted-in timelapses pass through
  unchanged.

  Resolver lives at services/background_dispatch.py::resolve_effective_timelapse
  (module-level so the print queue can reuse it). Both dispatch paths
  wired: background_dispatch.py (Print Now / Reprint) AND
  print_scheduler.py:_start_print (the queue). Field testing caught the
  scheduler gap on the first round — AST regression test now asserts
  start_print(timelapse=...) references effective_timelapse, not the raw
  item.timelapse, so a future refactor can't silently drop it.

  Extractor: ffmpeg -i input.mp4 -update 1 -q:v 2 out.jpg. Decoded
  frames overwrite the same output file, so the file left on disk is the
  literal last frame regardless of duration. Bambu records one frame per
  layer-change, so a 16-layer cube produces a 0.6 s timelapse — the
  original -sseof -1.0 approach seeked before the start of the file and
  returned frame 0 (empty bed). Decoding every frame is fine; Bambu
  timelapses are short by construction even on hours-long prints.

  Migration adds bambuddy_forced_timelapse branched on is_sqlite()
  (DEFAULT 0 / DEFAULT FALSE — PG rejects DEFAULT 0 for BOOLEAN).
  Verified live on postgres:16-alpine.

  Photo-task wait_for budget extends 45s -> 75s when timelapse_was_active
  so the notification carries the bed-up photo instead of falling back
  to the live-cam grab on slow links.

  Scope limit, documented in the camera wiki: prints started directly
  on the printer touchscreen / Bambu Handy / Bambu Studio Send bypass
  both dispatch paths, so the override doesn't fire there. Future
  option: mid-print M981 S1 P20000 MQTT toggle in on_print_start.

  Setting description rewritten in all 11 locales to drop the "only
  works when timelapse enabled" caveat (Bambuddy now forces it) and
  explain the kept-or-deleted behaviour.
2026-06-05 13:53:26 +02:00

738 lines
28 KiB
Python
Raw Blame History

"""Regression tests for ``BackgroundDispatchService._verify_print_response``.
The background-dispatch watchdog used to be fire-and-forget — it logged a
warning and force-reconnected MQTT, but the dispatch job had already been
marked successful. The user therefore saw "Print started successfully" while
the printer never actually transitioned (#1042 follow-up). The watchdog now
returns a bool so the caller can fail the dispatch job when the printer
doesn't acknowledge the command, mirroring what `_watchdog_print_start` does
on the queue side.
Both transition signals are accepted: ``state`` advancing past ``pre_state``
*or* ``subtask_id`` advancing past ``pre_subtask_id`` — H2D firmware can sit
at FINISH for ~50 s after accepting ``project_file`` while echoing the new
subtask_id back almost immediately (#1078).
"""
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from backend.app.services.background_dispatch import BackgroundDispatchService
def _status(state: str, subtask_id: str | None = None, gcode_file: str | None = None):
"""Minimal stand-in for PrinterState — only the fields the watchdog reads."""
return SimpleNamespace(state=state, subtask_id=subtask_id, gcode_file=gcode_file)
class TestReturnsTrueOnPickup:
@pytest.mark.asyncio
async def test_returns_true_on_state_change(self):
get_status = MagicMock(return_value=_status("RUNNING", "OLD_SUBTASK"))
with patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.3,
poll_interval=0.05,
)
assert result is True
@pytest.mark.asyncio
async def test_returns_true_on_subtask_id_change_even_if_state_still_finish(self):
"""#1078: H2D keeps state=FINISH for ~50 s after accepting project_file
but flips subtask_id immediately. Must be accepted as a pickup signal."""
get_status = MagicMock(return_value=_status("FINISH", "NEW_SUBTASK_12345"))
with patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="H2D",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK_99999",
timeout=0.3,
poll_interval=0.05,
)
assert result is True
class TestReturnsFalseOnTimeout:
@pytest.mark.asyncio
async def test_returns_false_when_neither_state_nor_subtask_id_changes(self):
"""The exact #1042 scenario: P1S sits in FAILED with HMS pending,
accepts the MQTT publish, never transitions. Watchdog must report
failure so the caller fails the dispatch job."""
get_status = MagicMock(return_value=_status("FINISH", "OLD_SUBTASK"))
client = MagicMock()
get_client = MagicMock(return_value=client)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.2,
poll_interval=0.05,
)
assert result is False
client.force_reconnect_stale_session.assert_called_once()
@pytest.mark.asyncio
async def test_returns_false_on_finish_to_idle_user_dismissed_prompt(self):
"""Regression for #1370 in the direct-dispatch path: when pre_state is
FINISH and the printer transitions to IDLE during the verifier window,
that's the user dismissing a post-print prompt — NOT acceptance of our
project_file. The original ``state != pre_state`` check incorrectly
returned True on this transition, so the dispatch job was marked
successful even though no print was running. Must now report failure
so the caller raises RuntimeError and the user sees the actual error.
"""
get_status = MagicMock(return_value=_status("IDLE", "OLD_SUBTASK"))
client = MagicMock()
get_client = MagicMock(return_value=client)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.2,
poll_interval=0.05,
)
assert result is False, (
"FINISH -> IDLE is the user dismissing a screen prompt, not the "
"printer accepting project_file — verifier must report failure (#1370)"
)
@pytest.mark.asyncio
async def test_returns_true_on_each_active_print_state(self):
"""Counterpart to the #1370 fix: transitions into the active-print
state set ARE valid "command landed" signals. PREPARE / SLICING /
RUNNING / PAUSE all return True.
"""
for active_state in ("PREPARE", "SLICING", "RUNNING", "PAUSE"):
get_status = MagicMock(return_value=_status(active_state, "OLD_SUBTASK"))
with patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="IDLE",
pre_subtask_id="OLD_SUBTASK",
timeout=0.2,
poll_interval=0.05,
)
assert result is True, (
f"transition IDLE -> {active_state} must be treated as a valid 'command landed' signal"
)
@pytest.mark.asyncio
async def test_returns_false_when_pre_subtask_id_none_and_state_unchanged(self):
"""Backward-compat: callers without a captured pre_subtask_id (e.g. the
printer never reported one) must still get the timeout failure path
based on state alone."""
get_status = MagicMock(return_value=_status("FINISH", "ANYTHING"))
get_client = MagicMock(return_value=None)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id=None,
timeout=0.2,
poll_interval=0.05,
)
assert result is False
@pytest.mark.asyncio
async def test_subtask_id_none_post_dispatch_does_not_count_as_change(self):
"""If the printer transiently reports subtask_id=None during the
watchdog window (e.g. mid-reconnect), that must not be treated as
"advanced past pre_subtask_id" — otherwise we'd false-pass and mark
a never-started print as successful."""
get_status = MagicMock(return_value=_status("FINISH", None))
get_client = MagicMock(return_value=None)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.2,
poll_interval=0.05,
)
assert result is False
class TestDisconnectHandling:
@pytest.mark.asyncio
async def test_disconnect_does_not_short_circuit_window(self):
"""A momentary ``get_status() is None`` (brief MQTT disconnect mid-window)
must not immediately fail the dispatch — the printer may reconnect and
still produce a valid transition before timeout. Falsely failing on the
first missed tick is the previous bug class we're moving away from."""
# First call: disconnected. Second call onward: reconnected and transitioned.
get_status = MagicMock(side_effect=[None, _status("RUNNING")])
with patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.3,
poll_interval=0.05,
)
assert result is True
assert get_status.call_count >= 2
@pytest.mark.asyncio
async def test_disconnect_for_full_window_returns_false(self):
"""Persistent disconnect for the full window is treated as failure.
Better to false-fail and let the user retry than to false-succeed and
leave them watching an idle printer (#1042)."""
get_status = MagicMock(return_value=None)
get_client = MagicMock(return_value=None)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1S",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.2,
poll_interval=0.05,
)
assert result is False
class TestDefaults:
def test_default_timeout_matches_queue_watchdog(self):
"""Queue and background watchdogs need the same 90 s default to give
slow H2D FINISH→PREPARE transitions the same headroom on both paths."""
import inspect
sig = inspect.signature(BackgroundDispatchService._verify_print_response)
assert sig.parameters["timeout"].default == 90.0
class TestGcodeFileDiscriminator:
"""#1150 vs #887/#936 discriminator: skip the forced reconnect when the
printer's gcode_file changed since pre-dispatch (project_file landed,
printer is parsing slowly — reconnecting mid-parse causes 0500_4003).
Reconnect when gcode_file is unchanged (publish was silently swallowed —
half-broken session needs the original recovery)."""
@pytest.mark.asyncio
async def test_skips_reconnect_when_gcode_file_changed(self):
get_status = MagicMock(
return_value=_status("FINISH", "OLD_SUBTASK", gcode_file="/new.3mf"),
)
client = MagicMock()
get_client = MagicMock(return_value=client)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
result = await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1P",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
pre_gcode_file="/old.3mf",
timeout=0.2,
poll_interval=0.05,
)
assert result is False
client.force_reconnect_stale_session.assert_not_called()
@pytest.mark.asyncio
async def test_reconnects_when_gcode_file_unchanged(self):
# The half-broken-session case (#887/#936): publish was dropped, so
# the printer is still showing the previous file. Reconnect to clear
# the broken paho QoS-1 queue.
get_status = MagicMock(
return_value=_status("FINISH", "OLD_SUBTASK", gcode_file="/old.3mf"),
)
client = MagicMock()
get_client = MagicMock(return_value=client)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1P",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
pre_gcode_file="/old.3mf",
timeout=0.2,
poll_interval=0.05,
)
client.force_reconnect_stale_session.assert_called_once()
@pytest.mark.asyncio
async def test_skips_reconnect_when_pre_gcode_file_was_none(self):
# Printer just connected (pre_gcode_file=None) and now reports a
# file — that's a clear "command landed" signal too.
get_status = MagicMock(
return_value=_status("FINISH", "OLD_SUBTASK", gcode_file="/new.3mf"),
)
client = MagicMock()
get_client = MagicMock(return_value=client)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1P",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
pre_gcode_file=None,
timeout=0.2,
poll_interval=0.05,
)
client.force_reconnect_stale_session.assert_not_called()
@pytest.mark.asyncio
async def test_reconnects_when_no_pre_gcode_file_arg_supplied(self):
# Backward-compat: callers that don't pass pre_gcode_file at all
# (everything but our updated dispatch sites) must still get the
# original reconnect-on-timeout behaviour. Here pre_gcode_file
# defaults to None and the printer's current gcode_file is also
# None → publish_landed=False → reconnect.
get_status = MagicMock(
return_value=_status("FINISH", "OLD_SUBTASK", gcode_file=None),
)
client = MagicMock()
get_client = MagicMock(return_value=client)
with (
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
get_status,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_client",
get_client,
),
):
await BackgroundDispatchService._verify_print_response(
printer_id=42,
printer_name="P1P",
pre_state="FINISH",
pre_subtask_id="OLD_SUBTASK",
timeout=0.2,
poll_interval=0.05,
)
client.force_reconnect_stale_session.assert_called_once()
# ---------------------------------------------------------------------------
# Integration tests: the call sites in _run_reprint_archive and
# _run_print_library_file must (a) await the watchdog instead of fire-and-
# forget, (b) raise RuntimeError on watchdog False so _run_active_job marks
# the job failed, (c) rollback the library-file flow's freshly-created
# archive on timeout. Heavy mocking — the goal is to verify the new wiring,
# not to re-test the dependencies.
# ---------------------------------------------------------------------------
from contextlib import asynccontextmanager # noqa: E402
from unittest.mock import AsyncMock # noqa: E402
from backend.app.services.background_dispatch import ( # noqa: E402
ActiveDispatchState,
PrintDispatchJob,
)
def _make_session_factory(db_mock):
"""Build an async-session factory whose context manager yields ``db_mock``.
Mirrors the ``async with async_session() as db`` shape used by both
``_run_*`` methods so the test can intercept ``db.rollback`` / ``db.scalar``.
"""
@asynccontextmanager
async def _factory():
yield db_mock
return _factory
def _printer_namespace():
return SimpleNamespace(
id=10,
name="P1S",
ip_address="1.2.3.4",
access_code="abc",
model="P1S",
)
def _make_dispatch_job(kind: str = "reprint_archive") -> PrintDispatchJob:
return PrintDispatchJob(
id=1,
kind=kind,
source_id=99,
source_name="Test.gcode.3mf",
printer_id=10,
printer_name="P1S",
options={},
requested_by_user_id=None,
requested_by_username=None,
)
@pytest.fixture
def reprint_archive_mocks(tmp_path):
"""Mock harness for ``_run_reprint_archive`` covering every external
dependency up to (and including) ``start_print``. The watchdog is left
real so the caller can patch ``_verify_print_response`` per-test."""
archive_file = tmp_path / "test.3mf"
archive_file.write_bytes(b"fake-3mf-content")
archive = SimpleNamespace(
id=99,
filename="Test.gcode.3mf",
file_path=str(archive_file),
)
db = MagicMock()
db.scalar = AsyncMock(return_value=_printer_namespace())
db.rollback = AsyncMock()
archive_service = MagicMock()
archive_service.get_archive = AsyncMock(return_value=archive)
return {
"archive": archive,
"archive_file": archive_file,
"db": db,
"archive_service": archive_service,
"session_factory": _make_session_factory(db),
}
@pytest.fixture
def library_file_mocks(tmp_path):
"""Mock harness for ``_run_print_library_file`` — separate from the
reprint fixture because the library flow creates its archive via
``archive_service.archive_print(...)`` rather than fetching one."""
src_file = tmp_path / "lib_src.3mf"
src_file.write_bytes(b"fake-3mf-content")
lib_file = SimpleNamespace(
id=22,
filename="cube.gcode.3mf",
file_path=str(src_file.relative_to(tmp_path)),
)
lib_file.active = staticmethod(lambda: lib_file) # mimic LibraryFile.active() chainable
new_archive = SimpleNamespace(id=500, filename="cube.gcode.3mf", file_path=str(src_file))
db = MagicMock()
db.scalar = AsyncMock() # configured per-test
db.flush = AsyncMock()
db.commit = AsyncMock()
db.rollback = AsyncMock()
archive_service = MagicMock()
archive_service.archive_print = AsyncMock(return_value=new_archive)
return {
"lib_file": lib_file,
"src_file": src_file,
"new_archive": new_archive,
"db": db,
"archive_service": archive_service,
"session_factory": _make_session_factory(db),
}
class TestReprintArchiveDispatchWiring:
"""Verify ``_run_reprint_archive`` (a) awaits the watchdog inline and
(b) raises RuntimeError on False so the dispatch job is marked failed."""
@pytest.mark.asyncio
async def test_raises_runtime_error_when_watchdog_returns_false(self, reprint_archive_mocks):
"""The exact #1042 propagation gap: watchdog detects non-transition,
_run_reprint_archive must surface it as a RuntimeError so the surrounding
_run_active_job marks the job failed (instead of silently completing)."""
from backend.app.services.background_dispatch import BackgroundDispatchService
m = reprint_archive_mocks
service = BackgroundDispatchService()
job = _make_dispatch_job(kind="reprint_archive")
watchdog = AsyncMock(return_value=False)
with (
patch("backend.app.services.background_dispatch.async_session", m["session_factory"]),
patch(
"backend.app.services.background_dispatch.ArchiveService",
return_value=m["archive_service"],
),
patch.object(BackgroundDispatchService, "_verify_print_response", watchdog),
patch(
"backend.app.services.background_dispatch.printer_manager.is_connected",
return_value=True,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
return_value=SimpleNamespace(state="FINISH", subtask_id="OLD_SUBTASK"),
),
patch(
"backend.app.services.background_dispatch.printer_manager.start_print",
return_value=True,
),
patch(
"backend.app.services.background_dispatch.delete_file_async",
new_callable=AsyncMock,
),
patch(
"backend.app.services.background_dispatch.with_ftp_retry",
new_callable=AsyncMock,
return_value=True,
),
patch(
"backend.app.services.background_dispatch.get_ftp_retry_settings",
new_callable=AsyncMock,
return_value=(False, 0, 0, 30.0),
),
patch(
"backend.app.services.background_dispatch.upload_file_async",
new_callable=AsyncMock,
return_value=True,
),
patch(
"backend.app.services.background_dispatch.ws_manager.broadcast",
new_callable=AsyncMock,
),
patch("backend.app.main.register_expected_print"),
# #1397: _resolve_effective_timelapse touches DB + setting layer
# that this watchdog-focused test isn't equipped to mock. Stub
# the whole helper so the dispatch flow proceeds unaffected.
patch.object(
BackgroundDispatchService,
"_resolve_effective_timelapse",
new=AsyncMock(return_value=False),
),
pytest.raises(RuntimeError, match="did not acknowledge print command"),
):
await service._run_reprint_archive(job)
# Watchdog received the captured pre-state and pre_subtask_id.
watchdog.assert_awaited_once()
kwargs = watchdog.await_args.kwargs
args = watchdog.await_args.args
assert "FINISH" in args # pre_state
assert kwargs["pre_subtask_id"] == "OLD_SUBTASK"
@pytest.mark.asyncio
async def test_succeeds_when_watchdog_returns_true(self, reprint_archive_mocks):
"""Happy path: watchdog confirms pickup; _run_reprint_archive returns
without raising. Guards against the wiring accidentally raising on True."""
from backend.app.services.background_dispatch import BackgroundDispatchService
m = reprint_archive_mocks
service = BackgroundDispatchService()
job = _make_dispatch_job(kind="reprint_archive")
with (
patch("backend.app.services.background_dispatch.async_session", m["session_factory"]),
patch(
"backend.app.services.background_dispatch.ArchiveService",
return_value=m["archive_service"],
),
patch.object(
BackgroundDispatchService,
"_verify_print_response",
AsyncMock(return_value=True),
),
patch(
"backend.app.services.background_dispatch.printer_manager.is_connected",
return_value=True,
),
patch(
"backend.app.services.background_dispatch.printer_manager.get_status",
return_value=SimpleNamespace(state="FINISH", subtask_id="OLD_SUBTASK"),
),
patch(
"backend.app.services.background_dispatch.printer_manager.start_print",
return_value=True,
),
patch(
"backend.app.services.background_dispatch.delete_file_async",
new_callable=AsyncMock,
),
patch(
"backend.app.services.background_dispatch.with_ftp_retry",
new_callable=AsyncMock,
return_value=True,
),
patch(
"backend.app.services.background_dispatch.get_ftp_retry_settings",
new_callable=AsyncMock,
return_value=(False, 0, 0, 30.0),
),
patch(
"backend.app.services.background_dispatch.upload_file_async",
new_callable=AsyncMock,
return_value=True,
),
patch(
"backend.app.services.background_dispatch.ws_manager.broadcast",
new_callable=AsyncMock,
),
patch("backend.app.main.register_expected_print"),
# #1397: _resolve_effective_timelapse touches DB + setting layer
# that this watchdog-focused test isn't equipped to mock. Stub
# the whole helper so the dispatch flow proceeds unaffected.
patch.object(
BackgroundDispatchService,
"_resolve_effective_timelapse",
new=AsyncMock(return_value=False),
),
):
await service._run_reprint_archive(job) # must not raise
# Reprint flow does not touch the existing archive — no rollback expected.
m["db"].rollback.assert_not_called()
class TestRunActiveJobMarksFailedOnRuntimeError:
"""End-to-end: a watchdog-driven RuntimeError must reach
`_mark_job_finished(failed=True)` via the existing ``_run_active_job``
catch-all, so the dispatch UI shows a real failure (not "Done")."""
@pytest.mark.asyncio
async def test_runtime_error_from_process_job_marks_failed_with_message(self):
from backend.app.services.background_dispatch import BackgroundDispatchService
service = BackgroundDispatchService()
job = _make_dispatch_job()
# Place the job into _active_jobs so _set_active_message has a target.
service._active_jobs[job.id] = ActiveDispatchState(job=job, message="")
failure_message = (
"Printer did not acknowledge print command — state still FINISH. "
"Check the printer for a pending error (HMS code, plate-clear prompt, "
"SD card) and try again."
)
with (
patch.object(
BackgroundDispatchService,
"_process_job",
AsyncMock(side_effect=RuntimeError(failure_message)),
),
patch.object(
BackgroundDispatchService,
"_mark_job_finished",
new_callable=AsyncMock,
) as mark_finished,
):
await service._run_active_job(job)
mark_finished.assert_awaited_once()
kwargs = mark_finished.await_args.kwargs
assert kwargs["failed"] is True
assert "did not acknowledge print command" in kwargs["message"]
Reference in New Issue View Git Blame Copy Permalink