diff --git a/backend/app/main.py b/backend/app/main.py index ce9d8a2dc..c5de3e101 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -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") diff --git a/backend/tests/unit/test_layer_timelapse_expected_archive.py b/backend/tests/unit/test_layer_timelapse_expected_archive.py index a118bb8f6..4d7318fa9 100644 --- a/backend/tests/unit/test_layer_timelapse_expected_archive.py +++ b/backend/tests/unit/test_layer_timelapse_expected_archive.py @@ -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