"""Tests for the connected-edge reconciliation that recovers from missed PRINT COMPLETE events (#1542 follow-up). Background: the PRINT COMPLETE MQTT callback is purely reactive to a single state transition (RUNNING → IDLE / FINISH / FAILED). When the printer finishes during an MQTT disconnect window — typical on the A1 line with unstable MQTT keepalives — Bambuddy never observes the transition. If a smart plug then cuts power between completion and the next reconnect, the firmware auto-replays whatever's still on the SD card and produces a ghost print on next power-up. Reporter (#1542 second case) saw this hit 4 out of 4 of his A1s. These tests cover: * `_is_active_archive_stale` — the pure decision function for whether an archive in `status="printing"` should be reconciled given the printer's current state. * `reconcile_stale_active_prints` — the orchestrator that queries the DB, runs the decision function, and synthesises `on_print_complete` for each stale archive. """ from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from backend.app.main import _is_active_archive_stale, _reconciled_end_time, _reconciled_outcome from backend.app.services.bambu_mqtt import HMSError def _state( state: str, *, subtask_id: str = "", subtask_name: str = "", connected: bool = True, progress: float = 0.0, layer_num: int = 0, hms_errors: list | None = None, ) -> SimpleNamespace: """Minimal PrinterState stub for the pure decision function.""" return SimpleNamespace( state=state, subtask_id=subtask_id, subtask_name=subtask_name, connected=connected, raw_data={}, progress=progress, layer_num=layer_num, hms_errors=hms_errors or [], ) def _archive( subtask_id: str | None = "ABC123", filename: str = "ghost.3mf", print_name: str = "ghost", *, started_at: datetime | None = None, print_time_seconds: int | None = None, ) -> SimpleNamespace: """Minimal PrintArchive stub — only the fields the decision function reads.""" return SimpleNamespace( id=42, subtask_id=subtask_id, filename=filename, print_name=print_name, started_at=started_at, print_time_seconds=print_time_seconds, plate_id=None, file_path=None, ) class TestIsActiveArchiveStale: """Decision function — covers all three stale triggers + the intentionally-conservative no-op cases.""" # Trigger 1: printer is in a terminal state. @pytest.mark.parametrize("terminal_state", ["IDLE", "FINISH", "FAILED", "idle", "finish", "failed"]) def test_terminal_state_marks_stale(self, terminal_state): archive = _archive(subtask_id="ABC123") state = _state(terminal_state, subtask_id="ABC123", subtask_name="ghost") is_stale, reason = _is_active_archive_stale(archive, state) assert is_stale is True assert terminal_state.upper() in reason # Trigger 2: printer is running a different subtask_id. def test_subtask_id_changed_marks_stale(self): archive = _archive(subtask_id="OLD_ID") state = _state("RUNNING", subtask_id="NEW_ID", subtask_name="something") is_stale, reason = _is_active_archive_stale(archive, state) assert is_stale is True assert "subtask_id" in reason assert "OLD_ID" in reason assert "NEW_ID" in reason # Trigger 3: printer is running but doesn't know what it's running. def test_empty_subtask_name_marks_stale(self): archive = _archive(subtask_id="ABC123") state = _state("RUNNING", subtask_id="", subtask_name="") is_stale, reason = _is_active_archive_stale(archive, state) assert is_stale is True assert "empty" in reason.lower() or "subtask_name" in reason # Healthy case: same subtask_id, running. def test_matching_running_print_not_stale(self): archive = _archive(subtask_id="ABC123") state = _state("RUNNING", subtask_id="ABC123", subtask_name="ghost") is_stale, _ = _is_active_archive_stale(archive, state) assert is_stale is False # PAUSE is not a stale signal — the print is paused, not ended. def test_paused_print_with_matching_subtask_not_stale(self): archive = _archive(subtask_id="ABC123") state = _state("PAUSE", subtask_id="ABC123", subtask_name="ghost") is_stale, _ = _is_active_archive_stale(archive, state) assert is_stale is False # PREPARE / SLICING are not stale either — pre-print phases. @pytest.mark.parametrize("pre_running_state", ["PREPARE", "SLICING"]) def test_pre_running_states_with_matching_subtask_not_stale(self, pre_running_state): archive = _archive(subtask_id="ABC123") state = _state(pre_running_state, subtask_id="ABC123", subtask_name="ghost") is_stale, _ = _is_active_archive_stale(archive, state) assert is_stale is False # Missing subtask_id on the archive side: don't have evidence either # way, fall through to the empty-subtask_name check. def test_archive_with_no_subtask_id_falls_to_subtask_name_check(self): archive = _archive(subtask_id=None) state = _state("RUNNING", subtask_id="ANYTHING", subtask_name="something") # Subtask_name is populated → not stale, no false positive. is_stale, _ = _is_active_archive_stale(archive, state) assert is_stale is False # Missing subtask_id on both sides: still triggers the empty-subtask_name # branch if the printer doesn't know what it's running. def test_both_subtask_ids_missing_running_with_empty_name_stale(self): archive = _archive(subtask_id=None) state = _state("RUNNING", subtask_id="", subtask_name="") is_stale, _ = _is_active_archive_stale(archive, state) assert is_stale is True # IDLE wins over PRINT-STATE checks — the terminal-state branch fires # first regardless of what the subtask fields look like. def test_idle_state_overrides_matching_subtask(self): archive = _archive(subtask_id="ABC123") state = _state("IDLE", subtask_id="ABC123", subtask_name="ghost") is_stale, reason = _is_active_archive_stale(archive, state) assert is_stale is True assert "IDLE" in reason # #1679: defensive pre-push guard. Even if reconcile gets called against # a PrinterState that's still on construction defaults (state="unknown" # / empty / None, subtask_name=""), the function must NOT report stale — # otherwise the reactive PRINT COMPLETE later creates a duplicate # archive and filament gets double-counted. The on_printer_status_change # caller is the primary fix (gates the reconcile spawn on real state), # but this guard is belt-and-braces for any future caller. @pytest.mark.parametrize("degenerate_state", ["unknown", "UNKNOWN", "Unknown", "", None]) def test_pre_push_state_returns_not_stale_even_with_empty_subtask(self, degenerate_state): archive = _archive(subtask_id="ABC123") state = _state(degenerate_state, subtask_id="", subtask_name="") is_stale, _ = _is_active_archive_stale(archive, state) assert is_stale is False, ( f"state.state={degenerate_state!r} means MQTT hasn't pushed real data yet; " "treating an in-flight archive as stale here causes the #1679 " "duplicate-archive + filament-double-count regression" ) class TestReconcileStaleActivePrints: """Orchestrator-level tests — mock the printer manager + DB session so we can drive the decision flow end-to-end without standing up real fixtures. These cover: * No printer status (disconnected) → no-op, no on_print_complete fired. * No active archives → no-op. * Stale archive → synthesised on_print_complete called with status ``"aborted"`` and the `_reconciled: True` marker so downstream code can distinguish synthetic from real completions. * Non-stale archive → on_print_complete NOT called (no false positive on a healthy in-flight print). * Exception inside on_print_complete must NOT block reconciliation for subsequent archives or crash the caller. """ @pytest.mark.asyncio async def test_no_status_skips_reconciliation(self): from backend.app.main import reconcile_stale_active_prints with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = None count = await reconcile_stale_active_prints(printer_id=1) assert count == 0 @pytest.mark.asyncio async def test_disconnected_status_skips_reconciliation(self): from backend.app.main import reconcile_stale_active_prints with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = _state("RUNNING", connected=False) count = await reconcile_stale_active_prints(printer_id=1) # Disconnected state would be making decisions against cached state — # the connected-edge handler in on_printer_status_change is the only # place that should drive reconciliation. assert count == 0 @pytest.mark.asyncio async def test_no_active_archives_returns_zero(self): from backend.app.main import reconcile_stale_active_prints with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = _state("IDLE") with patch("backend.app.main.async_session") as mock_session: session_ctx = AsyncMock() session_ctx.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: []))) mock_session.return_value.__aenter__.return_value = session_ctx count = await reconcile_stale_active_prints(printer_id=1) assert count == 0 @pytest.mark.asyncio async def test_stale_archive_synthesises_aborted_completion(self): from backend.app.main import reconcile_stale_active_prints stale = _archive(subtask_id="OLD_ID", filename="ghost.3mf", print_name="ghost") with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = _state("IDLE", subtask_id="", subtask_name="") with patch("backend.app.main.async_session") as mock_session: session_ctx = AsyncMock() session_ctx.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [stale]))) mock_session.return_value.__aenter__.return_value = session_ctx with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete: count = await reconcile_stale_active_prints(printer_id=1) assert count == 1 mock_complete.assert_awaited_once() # Verify the synthesised payload shape. args, kwargs = mock_complete.call_args assert args[0] == 1 payload = args[1] assert payload["status"] == "aborted" assert payload["filename"] == "ghost.3mf" assert payload["_reconciled"] is True @pytest.mark.asyncio async def test_non_stale_archive_does_not_synthesise(self): from backend.app.main import reconcile_stale_active_prints healthy = _archive(subtask_id="ABC123") with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = _state("RUNNING", subtask_id="ABC123", subtask_name="ghost") with patch("backend.app.main.async_session") as mock_session: session_ctx = AsyncMock() session_ctx.execute = AsyncMock( return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [healthy])) ) mock_session.return_value.__aenter__.return_value = session_ctx with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete: count = await reconcile_stale_active_prints(printer_id=1) assert count == 0 mock_complete.assert_not_called() @pytest.mark.asyncio async def test_on_print_complete_failure_does_not_block_rest(self): """An exception during one archive's synthesis must not abort reconciliation for the other archives — and must not propagate to the caller (the connected-edge handler is a hot path).""" from backend.app.main import reconcile_stale_active_prints a1 = _archive(subtask_id="A", filename="a.3mf") a1.id = 1 a2 = _archive(subtask_id="B", filename="b.3mf") a2.id = 2 with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = _state("IDLE") with patch("backend.app.main.async_session") as mock_session: session_ctx = AsyncMock() session_ctx.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [a1, a2]))) mock_session.return_value.__aenter__.return_value = session_ctx # First call raises, second call must still happen. mock_complete = AsyncMock(side_effect=[RuntimeError("boom"), None]) with patch("backend.app.main.on_print_complete", new=mock_complete): count = await reconcile_stale_active_prints(printer_id=1) # Only the second archive is recorded as reconciled (first raised). assert count == 1 assert mock_complete.await_count == 2 class TestReconciledOutcome: """#3310: only the printer's FINISH or FAILED for the archive's own job says how the print ended; anything else stays aborted.""" @pytest.mark.parametrize( ("printer_state", "archive_id", "printer_id", "outcome"), [ ("FINISH", "ABC123", "ABC123", "completed"), ("finish", "ABC123", "ABC123", "completed"), ("FAILED", "ABC123", "ABC123", "failed"), # Reported as a number by some firmware. ("FINISH", "123", 123, "completed"), ("IDLE", "ABC123", "ABC123", "aborted"), # A print started from the screen while Bambuddy was down. ("FINISH", "ABC123", "XYZ789", "aborted"), ("FAILED", "ABC123", "XYZ789", "aborted"), ("FINISH", "ABC123", "", "aborted"), ("FINISH", None, "", "aborted"), # "0" is every print started outside the cloud, not one job. ("FINISH", "0", "0", "aborted"), ], ) def test_outcome(self, printer_state, archive_id, printer_id, outcome): archive = _archive(subtask_id=archive_id) state = _state(printer_state, subtask_id=printer_id, subtask_name="ghost") assert _reconciled_outcome(archive, state) == outcome class TestReconciledEndTime: NOW = datetime(2026, 10, 20, 12, 0, tzinfo=timezone.utc) def test_start_plus_estimate(self): archive = _archive(started_at=datetime(2026, 10, 20, 6, 0), print_time_seconds=3 * 3600) assert _reconciled_end_time(archive, self.NOW) == datetime(2026, 10, 20, 9, 0, tzinfo=timezone.utc) def test_never_later_than_now(self): archive = _archive(started_at=datetime(2026, 10, 20, 11, 0), print_time_seconds=5 * 3600) assert _reconciled_end_time(archive, self.NOW) == self.NOW @pytest.mark.parametrize( ("started_at", "estimate"), [(None, 3600), (datetime(2026, 10, 20, 6, 0), None), (datetime(2026, 10, 20, 6, 0), 0)], ) def test_unknown_without_start_or_estimate(self, started_at, estimate): archive = _archive(started_at=started_at, print_time_seconds=estimate) assert _reconciled_end_time(archive, self.NOW) is None READ = "backend.app.utils.threemf_tools.extract_print_time_from_3mf" def _multi_plate(self, plate_id=None): archive = _archive(started_at=datetime(2026, 10, 20, 6, 0), print_time_seconds=5 * 3600) archive.plate_id, archive.file_path = plate_id, "archives/x/part.3mf" return archive def test_the_plate_s_own_estimate(self): """A multi-plate archive carries the whole file's estimate (#1593).""" with patch(self.READ, return_value=1800) as read: ended = _reconciled_end_time(self._multi_plate(plate_id=2), self.NOW) assert ended == datetime(2026, 10, 20, 6, 30, tzinfo=timezone.utc) assert read.call_args.args[1] == 2 def test_the_run_s_plate_over_the_archive_s(self): """A reprint of plate 3 reuses an archive that still says plate 1.""" with patch(self.READ, return_value=1800) as read: _reconciled_end_time(self._multi_plate(plate_id=1), self.NOW, plate_id=3) assert read.call_args.args[1] == 3 def test_no_plate_on_a_single_plate_file(self): with patch(self.READ, return_value=5 * 3600): ended = _reconciled_end_time(self._multi_plate(), self.NOW) assert ended == datetime(2026, 10, 20, 11, 0, tzinfo=timezone.utc) def test_no_plate_on_a_multi_plate_file_is_unknown(self): """The first plate's estimate differs from the sum: which plate printed?""" with patch(self.READ, return_value=1800): assert _reconciled_end_time(self._multi_plate(), self.NOW) is None def test_an_unreadable_file_is_unknown(self): with patch(self.READ, side_effect=OSError("gone")): assert _reconciled_end_time(self._multi_plate(plate_id=2), self.NOW) is None class TestReconcilePayload: async def _payload(self, archive, state): from backend.app.main import reconcile_stale_active_prints with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = state with patch("backend.app.main.async_session") as mock_session: session_ctx = AsyncMock() session_ctx.execute = AsyncMock( return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [archive])) ) mock_session.return_value.__aenter__.return_value = session_ctx with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete: assert await reconcile_stale_active_prints(printer_id=1) == 1 return mock_complete.call_args.args[1] @pytest.mark.asyncio async def test_finished_while_bambuddy_restarted(self): started = datetime.now(timezone.utc) - timedelta(hours=10) archive = _archive(subtask_id="ABC123", started_at=started, print_time_seconds=2 * 3600) payload = await self._payload( archive, _state("FINISH", subtask_id="ABC123", subtask_name="ghost", progress=100, layer_num=250) ) assert payload["status"] == "completed" assert payload["_reconciled"] is True assert payload["_reconciled_completed_at"] == started + timedelta(hours=2) assert (payload["last_progress"], payload["last_layer_num"]) == (100, 250) assert "hms_errors" not in payload @pytest.mark.asyncio async def test_failed_while_bambuddy_restarted(self): error = HMSError(code="0x4057", attr=0x0300_0000, module=0x03, severity=1, full_code="0300000000004057") payload = await self._payload( _archive(subtask_id="ABC123"), _state("FAILED", subtask_id="ABC123", subtask_name="ghost", progress=40, layer_num=80, hms_errors=[error]), ) assert payload["status"] == "failed" assert (payload["last_progress"], payload["last_layer_num"]) == (40, 80) assert payload["hms_errors"] == [error.completion_payload()] assert "_reconciled_completed_at" not in payload @pytest.mark.asyncio async def test_another_job_s_progress_is_not_this_archive_s(self): archive = _archive(subtask_id="ABC123", started_at=datetime.now(timezone.utc), print_time_seconds=3600) payload = await self._payload( archive, _state("FINISH", subtask_id="XYZ789", subtask_name="other", progress=100, layer_num=250) ) assert payload["status"] == "aborted" for key in ("last_progress", "last_layer_num", "hms_errors", "_reconciled_completed_at"): assert key not in payload @pytest.mark.asyncio async def test_the_queue_job_s_plate_reaches_the_estimate(): from backend.app.main import reconcile_stale_active_prints archive = _archive(subtask_id="ABC123", started_at=datetime.now(timezone.utc), print_time_seconds=3600) archive.plate_id = 1 archives = MagicMock(scalars=lambda: MagicMock(all=lambda: [archive])) plates = MagicMock(all=lambda: [(archive.id, 3)]) with patch("backend.app.main.printer_manager") as mock_pm: mock_pm.get_status.return_value = _state("FINISH", subtask_id="ABC123", subtask_name="ghost") with patch("backend.app.main.async_session") as mock_session: session_ctx = AsyncMock() session_ctx.execute = AsyncMock(side_effect=[archives, plates]) mock_session.return_value.__aenter__.return_value = session_ctx with ( patch("backend.app.main._reconciled_end_time", return_value=None) as end_time, patch("backend.app.main.on_print_complete", new=AsyncMock()), ): await reconcile_stale_active_prints(printer_id=1) assert end_time.call_args.args[2] == 3