public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-19 15:54:27 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(timelapse): extract _maybe_start_layer_timelapse + rewrite test

Browse Source 
The CI-only failure on test_layer_timelapse_expected_archive came from
  the test driving the entire on_print_start flow through ~12 patches and
  a MagicMock printer, which behaved differently between Python 3.11 (CI)
  and 3.13 (local) — execution stopped silently somewhere in the
  expected-archive path under CI's pytest-xdist parallelism but completed
  locally.

  Fix root-shape instead of fix the symptom:

  1. Extract the three identical start_session call sites in on_print_start
     (expected-archive promotion at 2030, fallback archive at 2554, fresh
     archive at 2644) into one helper _maybe_start_layer_timelapse() with
     the same external_camera_enabled / external_camera_url guard. The
     three inline blocks had already started drifting (#1353 originally
     only fixed one of them on the first pass) — the helper keeps them
     locked together going forward.

  2. Rewrite the test to call the helper directly. Uses SimpleNamespace
     (strict attribute access) instead of MagicMock (default-truthy), no
     DB mocking, no event loop, no parallel-state surface. Four small
     cases instead of two integration-style ones: enabled→starts,
     disabled→skips, URL-missing→skips, camera_type default 'mjpeg'.
This commit is contained in:
maziggy
2026-05-16 14:35:47 +02:00
parent 18db5ed038
commit 6569d5d1a7
2 changed files with 103 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/main.py
+31 -40
View File
@@ -625,6 +625,31 @@ def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | No
return stored_ams_mapping
def _maybe_start_layer_timelapse(printer, printer_id: int, archive_id: int) -> bool:
"""Start a layer-timelapse session for *archive_id* when the printer has
an external camera configured. Returns True if a session was started.
Three call sites in on_print_start (expected-archive promotion, fallback
archive creation, fresh-archive creation) used to inline this same
if-block; the inline copies kept drifting (#1353 fixed only one of them
on the first pass). Centralising the conditional + call here makes the
contract testable in isolation and keeps the three sites locked in step.
"""
if not (printer.external_camera_enabled and printer.external_camera_url):
return False
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
archive_id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logging.getLogger(__name__).info("Started layer timelapse for printer %s, archive %s", printer_id, archive_id)
return True
def _format_hms_error_summary(hms_errors: list[dict]) -> str | None:
"""Build a human-readable failure reason from MQTT hms_errors for PrintQueueItem.error_message.
@@ -2022,22 +2047,10 @@ async def on_print_start(printer_id: int, data: dict):
_active_prints[(printer_id, f"{subtask_name}.3mf")] = archive.id
# Start timelapse session if external camera is enabled (#1353).
# The two new-archive paths below also call start_session, but
# queue / VP-dispatched prints land here in the expected-archive
# branch and used to skip it entirely — so the timelapse session
# never started, no frames were captured, and the post-print
# stitch silently returned None.
if printer.external_camera_enabled and printer.external_camera_url:
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
archive.id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logger.info("Started layer timelapse for printer %s, expected archive %s", printer_id, archive.id)
# Queue / VP-dispatched prints land here in the expected-archive
# branch and used to skip start_session entirely — frames were
# never captured and the post-print stitch silently returned None.
_maybe_start_layer_timelapse(printer, printer_id, archive.id)
# Inject ams_mapping into usage tracker session — the session was created
# before expected-print promotion, so it may have ams_mapping=None when
@@ -2551,18 +2564,7 @@ async def on_print_start(printer_id: int, data: dict):
logger.info("Created fallback archive %s for %s (no 3MF available)", fallback_archive.id, print_name)
# Start timelapse session if external camera is enabled
if printer.external_camera_enabled and printer.external_camera_url:
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
fallback_archive.id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logger.info("Started layer timelapse for printer %s, archive %s", printer_id, fallback_archive.id)
_maybe_start_layer_timelapse(printer, printer_id, fallback_archive.id)
# Track as active print
_active_prints[(printer_id, fallback_archive.filename)] = fallback_archive.id
@@ -2641,18 +2643,7 @@ async def on_print_start(printer_id: int, data: dict):
logger.info("Created archive %s for %s", archive.id, downloaded_filename)
# Start timelapse session if external camera is enabled
if printer.external_camera_enabled and printer.external_camera_url:
from backend.app.services.layer_timelapse import start_session
start_session(
printer_id,
archive.id,
printer.external_camera_url,
printer.external_camera_type or "mjpeg",
snapshot_url=printer.external_camera_snapshot_url,
)
logger.info("Started layer timelapse for printer %s, archive %s", printer_id, archive.id)
_maybe_start_layer_timelapse(printer, printer_id, archive.id)
# Record starting energy from smart plug if available (#941: persisted column)
await _record_energy_start(archive, printer_id, db, context="auto-archive")
backend/tests/unit/test_layer_timelapse_expected_archive.py
+72 -194
View File
@@ -12,220 +12,98 @@ The expected-archive branch — where reprints and queue dispatch land —
updated the existing archive's status to "printing" but never started a
timelapse session.
Fix: start_session is now called in the expected-archive branch too, guarded
by the same `external_camera_enabled and external_camera_url` check that
the other two paths use.
Fix: the three start_session call sites in on_print_start were unified
behind `_maybe_start_layer_timelapse(printer, printer_id, archive_id)`,
which gates on the same `external_camera_enabled and external_camera_url`
check. Testing the helper directly (instead of driving the whole
on_print_start flow) keeps this regression locked in without dragging in
unrelated side effects (plate detection, DB queries, MQTT relay, etc.).
"""
from unittest.mock import AsyncMock, MagicMock, patch
from types import SimpleNamespace
from unittest.mock import patch
import pytest
from backend.app.main import (
_active_prints,
_expected_print_creators,
_expected_print_registered_at,
_expected_prints,
_print_ams_mappings,
register_expected_print,
)
from backend.app.main import _maybe_start_layer_timelapse
@pytest.fixture(autouse=True)
def _clear_dicts():
"""Clear module-level tracking dicts before and after each test."""
_expected_prints.clear()
_expected_print_registered_at.clear()
_expected_print_creators.clear()
_print_ams_mappings.clear()
_active_prints.clear()
yield
_expected_prints.clear()
_expected_print_registered_at.clear()
_expected_print_creators.clear()
_print_ams_mappings.clear()
_active_prints.clear()
def _build_mocks(*, external_camera_enabled: bool, external_camera_url: str | None):
"""Construct the mock matrix needed to drive on_print_start through the
expected-archive branch. Returns a dict of mock contexts that the test
enters via contextlib.ExitStack.
The session.execute mock returns the printer for the first call (printer
lookup) and the archive row for the second call (expected-archive
re-fetch). The archive row carries a unique filename so the
expected-print key lookup succeeds.
def _make_printer(*, external_camera_enabled: bool, external_camera_url: str | None):
"""Construct a minimal printer-shaped object with exactly the attributes
the helper reads. SimpleNamespace is used over MagicMock so attribute
access raises AttributeError on anything unexpected — keeps the test
honest about which fields the helper actually depends on.
"""
mock_printer = MagicMock()
mock_printer.id = 1
mock_printer.auto_archive = True
mock_printer.external_camera_enabled = external_camera_enabled
mock_printer.external_camera_url = external_camera_url
mock_printer.external_camera_type = "snapshot"
mock_printer.external_camera_snapshot_url = external_camera_url
# Disable plate detection in the mock so on_print_start's plate-detection
# block is skipped entirely. Plate detection isn't the subject under test
# and its real code path tries to capture a frame — which fails differently
# in CI (no ffmpeg) vs. local dev (ffmpeg present), and the CI-only path
# somehow prevents the expected-archive branch's start_session from being
# reached. MagicMock's default attribute access returns a truthy object,
# so without this explicit False the production code enters plate detection.
mock_printer.plate_detection_enabled = False
mock_printer.name = "TestA1"
mock_archive = MagicMock()
mock_archive.id = 42
mock_archive.filename = "Universal_Spirit_level_Holder.3mf"
mock_archive.subtask_id = None
mock_archive.print_time_seconds = None
mock_archive.created_by_id = None
mock_archive.printer_id = 1
mock_archive.print_name = "Universal Spirit Level Holder"
mock_archive.status = "pending"
mock_archive.file_path = "/test/archives/fake.3mf"
return mock_printer, mock_archive
@pytest.mark.asyncio
async def test_expected_archive_path_starts_timelapse_when_external_camera_enabled():
"""Queue/VP-dispatched prints land in the expected-archive branch and must
start the timelapse session there (the #1353 root cause)."""
mock_printer, mock_archive = _build_mocks(
external_camera_enabled=True, external_camera_url="http://camera.local:5000/snapshot.jpg"
return SimpleNamespace(
external_camera_enabled=external_camera_enabled,
external_camera_url=external_camera_url,
external_camera_type="snapshot",
external_camera_snapshot_url=external_camera_url,
)
# Register the expected print so the dispatch flow finds an archive_id.
register_expected_print(1, "Universal_Spirit_level_Holder.3mf", archive_id=42, ams_mapping=[1])
# on_print_start fires many db.execute() calls (settings lookups,
# usage tracker, plate detection, etc) before reaching the expected-
# archive branch. Route on SQL text so each query gets a sensible
# response regardless of order, rather than queuing N mocks.
def execute_router(stmt, *args, **kwargs):
sql = str(stmt).lower()
if "from printers" in sql or "from printer " in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_printer),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_printer]))),
)
if "from print_archives" in sql or "from print_archive" in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_archive),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_archive]))),
)
# Settings, spool assignments, anything else — return empty.
return MagicMock(
scalar_one_or_none=MagicMock(return_value=None),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
)
def test_starts_timelapse_when_external_camera_enabled():
"""Queue/VP-dispatched prints land in the expected-archive branch and must
start the timelapse session there (the #1353 root cause). The helper is
called from all three on_print_start paths (expected-archive promotion,
fallback archive, fresh archive) so testing it once covers all three."""
printer = _make_printer(
external_camera_enabled=True,
external_camera_url="http://camera.local:5000/snapshot.jpg",
)
mock_session = AsyncMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock()
mock_session.execute = AsyncMock(side_effect=execute_router)
mock_session.commit = AsyncMock()
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
started = _maybe_start_layer_timelapse(printer, printer_id=1, archive_id=42)
with (
patch("backend.app.main.async_session") as mock_session_maker,
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.smart_plug_manager") as mock_plug,
patch("backend.app.main.ws_manager") as mock_ws,
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.mqtt_relay") as mock_relay,
patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
patch("backend.app.main._load_objects_from_archive"),
patch("backend.app.main._store_spoolman_print_data", new_callable=AsyncMock),
patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
# The actual subject under test: assert start_session is called.
patch("backend.app.services.layer_timelapse.start_session") as mock_start_session,
):
mock_session_maker.return_value = mock_session
mock_notif.on_print_start = AsyncMock()
mock_plug.on_print_start = AsyncMock()
mock_ws.send_print_start = AsyncMock()
mock_ws.send_archive_updated = AsyncMock()
mock_relay.on_print_start = AsyncMock()
mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
from backend.app.main import on_print_start
await on_print_start(
1,
{
"filename": "Universal_Spirit_level_Holder.3mf",
"subtask_name": "Universal_Spirit_level_Holder",
},
)
mock_start_session.assert_called_once()
# Verify it was called with the archive_id from the expected-print
# registration, not a fresh one — that's the contract.
call_args = mock_start_session.call_args
assert call_args.args[0] == 1, "printer_id must match"
assert call_args.args[1] == 42, "archive_id must come from the expected-print registration"
assert call_args.args[2] == "http://camera.local:5000/snapshot.jpg"
assert call_args.args[3] == "snapshot"
assert started is True
mock_start_session.assert_called_once_with(
1,
42,
"http://camera.local:5000/snapshot.jpg",
"snapshot",
snapshot_url="http://camera.local:5000/snapshot.jpg",
)
@pytest.mark.asyncio
async def test_expected_archive_path_skips_timelapse_when_external_camera_disabled():
def test_skips_timelapse_when_external_camera_disabled():
"""The same guard that the new-archive paths use must hold here: no
external camera → no timelapse session. Otherwise we'd try to capture
from a None URL and crash the print-start flow."""
mock_printer, mock_archive = _build_mocks(external_camera_enabled=False, external_camera_url=None)
printer = _make_printer(external_camera_enabled=False, external_camera_url=None)
mock_archive.filename = "test.3mf"
mock_archive.id = 99
register_expected_print(1, "test.3mf", archive_id=99, ams_mapping=None)
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
started = _maybe_start_layer_timelapse(printer, printer_id=1, archive_id=99)
def execute_router(stmt, *args, **kwargs):
sql = str(stmt).lower()
if "from printers" in sql or "from printer " in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_printer),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_printer]))),
)
if "from print_archives" in sql or "from print_archive" in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_archive),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_archive]))),
)
return MagicMock(
scalar_one_or_none=MagicMock(return_value=None),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
)
assert started is False
mock_start_session.assert_not_called()
mock_session = AsyncMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock()
mock_session.execute = AsyncMock(side_effect=execute_router)
mock_session.commit = AsyncMock()
with (
patch("backend.app.main.async_session") as mock_session_maker,
patch("backend.app.main.notification_service") as mock_notif,
patch("backend.app.main.smart_plug_manager") as mock_plug,
patch("backend.app.main.ws_manager") as mock_ws,
patch("backend.app.main.printer_manager") as mock_pm,
patch("backend.app.main.mqtt_relay") as mock_relay,
patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
patch("backend.app.main._load_objects_from_archive"),
patch("backend.app.main._store_spoolman_print_data", new_callable=AsyncMock),
patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
patch("backend.app.services.layer_timelapse.start_session") as mock_start_session,
):
mock_session_maker.return_value = mock_session
mock_notif.on_print_start = AsyncMock()
mock_plug.on_print_start = AsyncMock()
mock_ws.send_print_start = AsyncMock()
mock_ws.send_archive_updated = AsyncMock()
mock_relay.on_print_start = AsyncMock()
mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
def test_skips_timelapse_when_url_missing_even_if_flag_set():
"""If the toggle is on but the URL field is empty (legacy / half-configured
install), the guard must still hold — calling start_session with an empty
URL would crash downstream when the capture thread tries to fetch frames."""
printer = _make_printer(external_camera_enabled=True, external_camera_url=None)
from backend.app.main import on_print_start
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
started = _maybe_start_layer_timelapse(printer, printer_id=1, archive_id=7)
await on_print_start(1, {"filename": "test.3mf", "subtask_name": "test"})
assert started is False
mock_start_session.assert_not_called()
mock_start_session.assert_not_called()
def test_camera_type_defaults_to_mjpeg_when_unset():
"""external_camera_type defaults to 'mjpeg' in start_session when the
printer column is None — pre-existing contract preserved by the helper."""
printer = SimpleNamespace(
external_camera_enabled=True,
external_camera_url="http://cam/feed",
external_camera_type=None,
external_camera_snapshot_url=None,
)
with patch("backend.app.services.layer_timelapse.start_session") as mock_start_session:
_maybe_start_layer_timelapse(printer, printer_id=2, archive_id=11)
assert mock_start_session.called
call_kwargs = mock_start_session.call_args.kwargs
call_args = mock_start_session.call_args.args
assert call_args[3] == "mjpeg"
assert call_kwargs["snapshot_url"] is None