test(queue,billing,vp): pin dispatch failures, creator scope and busy-printer holds, aborted-print usage fallbacks, end-to-end print completion and billing, VP MQTT pre-login and rate-limit guards; test schema loads every model

backend/tests/conftest.py

@@ -446,49 +446,11 @@ async def test_engine():
     """Create a test database engine."""
     engine = create_async_engine(TEST_DATABASE_URL, echo=False)
 
-    # Import all models to register them
-    from backend.app.models import (
-        active_print_session,  # noqa: F401
-        ams_history,
-        ams_label,
-        api_key,
-        archive,
-        auth_ephemeral,
-        color_catalog,
-        external_link,
-        filament,
-        group,
-        kprofile_note,
-        maintenance,
-        notification,
-        notification_template,
-        oidc_provider,
-        print_log,
-        print_queue,
-        printer,
-        project,
-        project_bom,
-        scheduled_drying,
-        settings,
-        slot_preset,
-        smart_plug,
-        smart_plug_energy_snapshot,  # noqa: F401
-        sponsor_toast_state,  # noqa: F401
-        spool,
-        spool_assignment,
-        spool_catalog,
-        spool_filament_preset,
-        spool_k_profile,
-        spool_usage_history,
-        spoolbuddy_device,
-        spoolman_k_profile,
-        spoolman_slot_assignment,
-        user,
-        user_email_pref,
-        user_otp_code,
-        user_totp,
-        virtual_printer,
-    )
+    # Every model, as the app's schema has them: a hand-kept list here fell
+    # behind and left four tables out of the SQLite test database.
+    from backend.app.core.database import load_all_models
+
+    load_all_models()
 
     if TEST_PG_URL is not None:
         # The same engine setup as the app's (UTC session, aware datetimes

backend/tests/integration/test_print_complete_end_to_end.py

@@ -0,0 +1,159 @@
+"""A queued print's completion, end to end on the database.
+
+The archive row, the print log entry, the queue item and the wallet charge are
+what the rest of Bambuddy reads afterwards (archives page, statistics, project
+totals, the cost centre's budget). Each was only pinned through mocks of the
+function that writes it; here ``on_print_complete`` runs against real rows and
+only the outside world (websocket, MQTT relay, notifications, plugs, the
+printer's SD card) is mocked; the printer manager knows no printer.
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app import main as main_module
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import CostCenter, WalletTransaction
+from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.app.services.bambu_ftp import DeleteResult
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+RUN = "run-end-to-end"
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    """The completion marks the printer as awaiting a plate clear on the real
+    printer manager; the next test on this worker must not inherit that."""
+    from backend.app.services.printer_manager import printer_manager
+
+    main_module._active_prints.clear()
+    yield
+    main_module._active_prints.clear()
+    printer_manager._awaiting_plate_clear.clear()
+
+
+@pytest.fixture
+async def queued_print(db_session, printer_factory):
+    printer = await printer_factory(name="P")
+    db_session.add(Settings(key="billing_enabled", value="true"))
+    db_session.add(Settings(key="capture_finish_photo", value="false"))
+    user = User(username="payer", password_hash="x", is_active=True)
+    center = CostCenter(name="Lab", is_active=True, is_private=False)
+    db_session.add_all([user, center])
+    await db_session.flush()
+    archive = PrintArchive(
+        printer_id=printer.id,
+        filename="part.3mf",
+        print_name="part",
+        file_path="archives/test/part.3mf",
+        file_size=1,
+        status="printing",
+        cost=5.0,
+        created_by_id=user.id,
+        billing_run_id=RUN,
+    )
+    db_session.add(archive)
+    await db_session.flush()
+    item = PrintQueueItem(
+        printer_id=printer.id,
+        archive_id=archive.id,
+        status="printing",
+        position=1,
+        created_by_id=user.id,
+        cost_center_id=center.id,
+        estimated_cost=5.0,
+        billing_run_id=RUN,
+    )
+    db_session.add(item)
+    await db_session.commit()
+    main_module._active_prints[(printer.id, "part.3mf")] = archive.id
+    return {"printer": printer.id, "user": user.id, "center": center.id, "archive": archive.id, "item": item.id}
+
+
+async def _complete(test_engine, printer_id: int, status: str) -> None:
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    tasks_before = set(asyncio.all_tasks())
+    with (
+        patch("backend.app.main.async_session", maker),
+        patch("backend.app.core.database.async_session", maker),
+        patch("backend.app.main.ws_manager") as ws,
+        patch("backend.app.main.mqtt_relay") as relay,
+        patch("backend.app.main.notification_service") as notif,
+        patch("backend.app.main.smart_plug_manager") as plug,
+        patch("backend.app.services.spoolman_tracking.async_session", maker),
+        # The SD-card cleanup of the uploaded file
+        patch("backend.app.services.bambu_ftp.delete_file_async", AsyncMock(return_value=DeleteResult.DELETED)),
+        patch("backend.app.services.bambu_ftp.list_files_async", AsyncMock(return_value=[])),
+    ):
+        ws.send_print_complete = AsyncMock()
+        ws.send_archive_updated = AsyncMock()
+        ws.broadcast = AsyncMock()
+        relay.on_print_complete = AsyncMock()
+        notif.on_print_complete = AsyncMock()
+        plug.on_print_complete = AsyncMock()
+        await main_module.on_print_complete(
+            printer_id,
+            {"status": status, "filename": "part.gcode", "subtask_name": "part", "timelapse_was_active": False},
+        )
+        spawned = asyncio.all_tasks() - tasks_before
+        if spawned:
+            _done, pending = await asyncio.wait(spawned, timeout=10)
+            # Left running, they would carry on into the next test unpatched
+            assert not pending, f"background work still running: {pending}"
+
+
+async def _rows(db_session, archive_id: int, item_id: int):
+    db_session.expire_all()
+    archive = await db_session.get(PrintArchive, archive_id)
+    item = await db_session.get(PrintQueueItem, item_id)
+    logs = (await db_session.scalars(select(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id))).all()
+    charges = (
+        await db_session.scalars(select(WalletTransaction).where(WalletTransaction.transaction_type == "print_charge"))
+    ).all()
+    return archive, item, logs, charges
+
+
+async def test_a_completed_print(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "completed")
+
+    archive, item, logs, charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert archive.status == "completed"
+    assert item.status == "completed"
+    assert [log.status for log in logs] == ["completed"]
+    assert len(charges) == 1
+    charge = charges[0]
+    assert (charge.user_id, charge.cost_center_id, charge.print_queue_id, charge.print_run_id) == (
+        queued_print["user"],
+        queued_print["center"],
+        queued_print["item"],
+        RUN,
+    )
+    assert abs(charge.amount) == pytest.approx(5.0)
+
+
+async def test_a_failed_print(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "failed")
+
+    archive, item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert archive.status == "failed"
+    assert item.status == "failed"
+    assert [log.status for log in logs] == ["failed"]
+
+
+async def test_a_second_completion_charges_nothing_more(test_engine, db_session, queued_print):
+    """A duplicate completion callback (reconnect replay) is idempotent."""
+    await _complete(test_engine, queued_print["printer"], "completed")
+    main_module._active_prints[(queued_print["printer"], "part.3mf")] = queued_print["archive"]
+    await _complete(test_engine, queued_print["printer"], "completed")
+
+    _archive, _item, _logs, charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert len(charges) == 1

backend/tests/unit/test_cost_tracking.py

@@ -443,6 +443,65 @@ class TestCostCalculation:
         # Cost = 20g / 1000 * 20.0 = 0.40
         assert results[0]["cost"] == 0.40
 
+    async def _failed_after_firmware_reset(self, *, last_progress=0.0, last_layer_num=0, layer_usage=None):
+        """A failed print whose printer already reset progress and layer to 0."""
+        spool = _make_spool(spool_id=1, label_weight=1000, cost_per_kg=20.0)
+        _active_sessions[1] = PrintSession(
+            printer_id=1,
+            print_name="Test",
+            started_at=datetime.now(timezone.utc),
+            tray_remain_start={(0, 0): 80},
+            tray_now_at_start=0,
+        )
+        printer_manager = MagicMock()
+        printer_manager.get_status.return_value = SimpleNamespace(
+            raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]}, progress=0, layer_num=0, tray_now=0
+        )
+        db = _mock_db_sequential([_make_archive(archive_id=10), None, _make_assignment(spool_id=1), spool])
+        cumulative = MagicMock(return_value={0: 1000.0})  # filament 0 = slot 1
+        with (
+            patch("backend.app.core.config.settings") as mock_settings,
+            patch("backend.app.api.routes.settings.get_setting", return_value="15.0"),
+            patch(
+                "backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
+                return_value=[{"slot_id": 1, "used_g": 40.0, "type": "PLA", "color": "#FF0000"}],
+            ),
+            patch("backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf", return_value=layer_usage),
+            patch("backend.app.utils.threemf_tools.get_cumulative_usage_at_layer", cumulative),
+            patch("backend.app.utils.threemf_tools.extract_filament_properties_from_3mf", return_value={}),
+            patch("backend.app.utils.threemf_tools.mm_to_grams", return_value=7.0),
+        ):
+            mock_settings.base_dir = MagicMock()
+            mock_path = MagicMock()
+            mock_path.exists.return_value = True
+            mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
+            results = await on_print_complete(
+                printer_id=1,
+                data={"status": "failed", "last_progress": last_progress, "last_layer_num": last_layer_num},
+                printer_manager=printer_manager,
+                db=db,
+                archive_id=10,
+            )
+        return results, cumulative
+
+    @pytest.mark.asyncio
+    async def test_a_progress_reset_to_zero_falls_back_to_the_last_progress(self):
+        """Firmware zeroes progress on cancel; 0 % would charge nothing."""
+        results, _ = await self._failed_after_firmware_reset(last_progress=40.0)
+
+        # 40 % of 40 g
+        assert results[0]["weight_used"] == 16.0
+
+    @pytest.mark.asyncio
+    async def test_a_layer_reset_to_zero_falls_back_to_the_last_layer(self):
+        results, cumulative = await self._failed_after_firmware_reset(
+            last_progress=40.0, last_layer_num=12, layer_usage={1: {12: 1000.0}}
+        )
+
+        assert cumulative.call_args.args[1] == 12
+        # The per-layer grams, not the 40 % estimate
+        assert results[0]["weight_used"] == 7.0
+
     @pytest.mark.asyncio
     async def test_cost_with_ams_fallback_tracking(self):
         """AMS fallback tracking also calculates cost correctly."""

backend/tests/unit/test_phantom_print_hardening.py

@@ -211,138 +211,5 @@ class TestMultiplePrintingQueueItemsWarning:
         assert "printer 7" in bug_warnings[0].message
 
 
-class TestBusyPrinterSeedingFromPrintingItems:
-    """Regression for the duplicate-dispatch bug observed with quantity>1 batches.
-
-    The old scheduler seeded ``busy_printers`` with an empty set and relied on
-    ``_is_printer_idle()`` to gate dispatch. On H2D / P1 series the MQTT state
-    lags several seconds behind the print command, so the next ``check_queue``
-    tick saw IDLE and dispatched a second queue item onto the same printer —
-    both items ended up in 'printing' status. The fix seeds ``busy_printers``
-    up-front with every printer that already has an item in 'printing' status.
-    """
-
-    @pytest.mark.asyncio
-    async def test_seed_query_returns_printers_with_printing_items(self):
-        """The seeding query must return every printer_id that has a 'printing' item."""
-        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-        import backend.app.models  # noqa: F401
-        from backend.app.core.database import Base
-        from backend.app.models.print_queue import PrintQueueItem
-
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-        async with session_maker() as db:
-            db.add_all(
-                [
-                    PrintQueueItem(printer_id=1, status="printing", position=1, archive_id=10),
-                    PrintQueueItem(printer_id=1, status="pending", position=2, archive_id=10),
-                    PrintQueueItem(printer_id=2, status="printing", position=1, archive_id=11),
-                    PrintQueueItem(printer_id=3, status="pending", position=1, archive_id=12),
-                    PrintQueueItem(printer_id=None, status="pending", position=1, archive_id=13),
-                ]
-            )
-            await db.commit()
-
-            result = await db.execute(
-                select(PrintQueueItem.printer_id)
-                .where(PrintQueueItem.status == "printing")
-                .where(PrintQueueItem.printer_id.is_not(None))
-            )
-            busy_printers = {pid for (pid,) in result.all() if pid is not None}
-
-        assert busy_printers == {1, 2}
-
-        await engine.dispose()
-
-    @pytest.mark.asyncio
-    async def test_seed_query_empty_when_no_printing_items(self):
-        """With only pending items, no printer is considered busy by the query."""
-        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-        import backend.app.models  # noqa: F401
-        from backend.app.core.database import Base
-        from backend.app.models.print_queue import PrintQueueItem
-
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-        async with session_maker() as db:
-            db.add_all(
-                [
-                    PrintQueueItem(printer_id=1, status="pending", position=1, archive_id=10),
-                    PrintQueueItem(printer_id=2, status="completed", position=1, archive_id=11),
-                    PrintQueueItem(printer_id=3, status="failed", position=1, archive_id=12),
-                    PrintQueueItem(printer_id=4, status="cancelled", position=1, archive_id=13),
-                ]
-            )
-            await db.commit()
-
-            result = await db.execute(
-                select(PrintQueueItem.printer_id)
-                .where(PrintQueueItem.status == "printing")
-                .where(PrintQueueItem.printer_id.is_not(None))
-            )
-            busy_printers = {pid for (pid,) in result.all() if pid is not None}
-
-        assert busy_printers == set()
-
-        await engine.dispose()
-
-    @pytest.mark.asyncio
-    async def test_check_queue_skips_printer_with_existing_printing_item(self, caplog):
-        """Simulate the exact observed bug: a pending item targets a printer that already
-        has another queue item in 'printing' status. The scheduler must NOT dispatch the
-        pending item even if the live MQTT state reports IDLE.
-        """
-        from unittest.mock import AsyncMock, patch
-
-        from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
-
-        import backend.app.models  # noqa: F401
-        from backend.app.core.database import Base
-        from backend.app.models.print_queue import PrintQueueItem
-        from backend.app.services.print_scheduler import PrintScheduler
-
-        engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
-        async with engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-        session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
-
-        async with session_maker() as db:
-            db.add_all(
-                [
-                    PrintQueueItem(printer_id=1, status="printing", position=1, archive_id=84),
-                    PrintQueueItem(printer_id=1, status="pending", position=2, archive_id=84),
-                ]
-            )
-            await db.commit()
-
-        scheduler = PrintScheduler()
-        start_print_mock = AsyncMock()
-
-        with (
-            patch("backend.app.services.print_scheduler.async_session", session_maker),
-            patch.object(scheduler, "_get_bool_setting", AsyncMock(return_value=False)),
-            patch.object(scheduler, "_is_printer_idle", return_value=True),
-            patch.object(scheduler, "_check_auto_drying", AsyncMock()),
-            patch.object(scheduler, "_start_print", start_print_mock),
-            patch("backend.app.services.print_scheduler.printer_manager") as mock_pm,
-        ):
-            mock_pm.is_connected.return_value = True
-            await scheduler.check_queue()
-
-        start_print_mock.assert_not_called()
-
-        async with session_maker() as db:
-            rows = (await db.execute(select(PrintQueueItem).order_by(PrintQueueItem.position))).scalars().all()
-            statuses = [r.status for r in rows]
-        assert statuses == ["printing", "pending"]
-
-        await engine.dispose()
+# The busy-printer seed (an item already printing blocks a second dispatch) is
+# pinned on the real check_queue in test_scheduler_check_queue_holds.py.

backend/tests/unit/test_scheduler_check_queue_holds.py

@@ -0,0 +1,97 @@
+"""What check_queue holds back, run on a real database with the dispatch drained.
+
+* An item whose creator may no longer use its printer (#1727) is held with a
+  reason, not dispatched and not failed.
+* A printer that already has an item printing takes no second one, even while
+  it reports IDLE (the phantom-print seed).
+"""
+
+import pytest
+from sqlalchemy import insert, update
+
+from backend.app.models.group import Group, group_printers
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.tests.unit.test_scheduler_busy_reasons_3018 import _scheduler, one_printer  # noqa: F401
+
+
+async def _restricted_creator(session_maker, allowed_printer_id: int) -> int:
+    async with session_maker() as db:
+        db.add(Settings(key="auth_enabled", value="true"))
+        group = Group(name="team", permissions=["queue:create"], restrict_printers=True)
+        user = User(username="member", password_hash="x", is_active=True, groups=[group])
+        db.add_all([group, user])
+        await db.flush()
+        await db.execute(insert(group_printers).values(group_id=group.id, printer_id=allowed_printer_id))
+        await db.commit()
+        return user.id
+
+
+async def _other_printer(session_maker) -> int:
+    async with session_maker() as db:
+        printer = Printer(name="Other", serial_number="OTHER", ip_address="10.0.0.2", access_code="code", model="X1C")
+        db.add(printer)
+        await db.commit()
+        return printer.id
+
+
+class TestACreatorWhoLostThePrinter:
+    @pytest.mark.asyncio
+    async def test_the_item_is_held_with_a_reason(self, one_printer):  # noqa: F811
+        allowed = await _other_printer(one_printer.session_maker)
+        creator = await _restricted_creator(one_printer.session_maker, allowed)
+        item_id = await one_printer.add_item()
+        async with one_printer.session_maker() as db:
+            await db.execute(update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(created_by_id=creator))
+            await db.commit()
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+        async with one_printer.session_maker() as db:
+            item = await db.get(PrintQueueItem, item_id)
+        assert item.status == "pending"
+        assert "no longer has access" in (item.waiting_reason or "")
+
+    @pytest.mark.asyncio
+    async def test_a_creator_who_still_has_it_is_dispatched(self, one_printer):  # noqa: F811
+        creator = await _restricted_creator(one_printer.session_maker, one_printer.printer_id)
+        item_id = await one_printer.add_item()
+        async with one_printer.session_maker() as db:
+            await db.execute(update(PrintQueueItem).where(PrintQueueItem.id == item_id).values(created_by_id=creator))
+            await db.commit()
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+
+        async with one_printer.session_maker() as db:
+            assert (await db.get(PrintQueueItem, item_id)).status != "pending"
+
+
+class TestAPrinterWithAnItemPrinting:
+    @pytest.mark.asyncio
+    async def test_takes_no_second_item_while_it_reports_idle(self, one_printer):  # noqa: F811
+        printing_id = await one_printer.add_item(status="printing", position=0)
+        pending_id = await one_printer.add_item(position=1)
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+            assert not scheduler._inflight
+
+        async with one_printer.session_maker() as db:
+            assert (await db.get(PrintQueueItem, printing_id)).status == "printing"
+            assert (await db.get(PrintQueueItem, pending_id)).status == "pending"
+
+    @pytest.mark.asyncio
+    async def test_without_it_the_same_item_goes_out(self, one_printer):  # noqa: F811
+        """The control: nothing else holds this item back."""
+        pending_id = await one_printer.add_item(position=1)
+
+        async with _scheduler(one_printer, idle=True) as scheduler:
+            await scheduler.check_queue()
+
+        async with one_printer.session_maker() as db:
+            assert (await db.get(PrintQueueItem, pending_id)).status != "pending"

backend/tests/unit/test_scheduler_dispatch_failures.py

@@ -0,0 +1,107 @@
+"""How a dispatch that cannot go ahead ends.
+
+Before the upload: an exhausted cost-centre budget, a printer row that is gone
+and a disconnected printer each fail the item with their own message, upload
+nothing and power the printer off if it was powered on for the job. After the
+upload: a start command the printer refused while it stayed idle fails the
+item, deletes the uploaded file again and sends the failure notification. A
+refusal because the printer turned busy defers instead (#2598, see
+test_scheduler_busy_defer_2598).
+"""
+
+from contextlib import ExitStack
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.services.print_scheduler import PrintScheduler
+from backend.tests.unit.test_scheduler_busy_defer_2598 import _base_patches, _final_item, dispatch_case  # noqa: F401
+
+IDLE = SimpleNamespace(state="IDLE", subtask_id=None, gcode_file=None)
+
+
+async def _dispatch(ctx, *, start_print=None, extra=(), before=None):
+    scheduler = PrintScheduler()
+    upload = AsyncMock(return_value=True)
+    start = start_print or MagicMock(return_value=True)
+    failed = AsyncMock()
+    async with ctx.session_maker() as db:
+        item = await db.get(PrintQueueItem, ctx.ids.item_id)
+        if before is not None:
+            await before(db, item)
+        with ExitStack() as stack:
+            for p in _base_patches(scheduler, ctx, upload, start, MagicMock(return_value=IDLE)):
+                stack.enter_context(p)
+            for p in extra:
+                stack.enter_context(p)
+            stack.enter_context(
+                patch(
+                    "backend.app.services.notification_service.notification_service.on_queue_job_failed",
+                    failed,
+                )
+            )
+            delete = stack.enter_context(
+                patch("backend.app.services.print_scheduler.delete_file_async", AsyncMock(return_value=True))
+            )
+            power_off = scheduler._power_off_if_needed
+            await scheduler._start_print(db, item)
+    return SimpleNamespace(
+        power_off=power_off, upload=upload, start=start, delete=delete, failed=failed, item=await _final_item(ctx)
+    )
+
+
+class TestBeforeTheUpload:
+    @pytest.mark.asyncio
+    async def test_an_exhausted_budget(self, dispatch_case):  # noqa: F811
+        refused = AsyncMock(side_effect=HTTPException(400, "Cost center budget exceeded"))
+
+        result = await _dispatch(
+            dispatch_case, extra=[patch("backend.app.services.print_scheduler.validate_print_budget", refused)]
+        )
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Cost center budget exceeded")
+        result.upload.assert_not_awaited()
+        result.start.assert_not_called()
+        result.power_off.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_a_printer_that_is_gone(self, dispatch_case):  # noqa: F811
+        async def gone(db, item):
+            item.printer_id = 9999
+            await db.commit()
+
+        result = await _dispatch(dispatch_case, before=gone)
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Printer not found")
+        result.upload.assert_not_awaited()
+        result.power_off.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_a_disconnected_printer(self, dispatch_case):  # noqa: F811
+        offline = patch(
+            "backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=False)
+        )
+
+        result = await _dispatch(dispatch_case, extra=[offline])
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Printer not connected")
+        result.upload.assert_not_awaited()
+        result.start.assert_not_called()
+        result.power_off.assert_awaited_once()
+
+
+class TestARefusedStartOnAnIdlePrinter:
+    @pytest.mark.asyncio
+    async def test_fails_the_item_and_removes_the_upload(self, dispatch_case):  # noqa: F811
+        result = await _dispatch(dispatch_case, start_print=MagicMock(return_value=False))
+
+        assert (result.item.status, result.item.error_message) == ("failed", "Failed to send print command to printer")
+        result.upload.assert_awaited_once()
+        uploaded_to = result.upload.await_args.args[3]
+        # One clears the way before the upload, one removes the upload again
+        assert [c.args[2] for c in result.delete.await_args_list] == [uploaded_to, uploaded_to]
+        result.failed.assert_awaited_once()
+        assert result.failed.await_args.kwargs["printer_id"] == dispatch_case.ids.printer_id

backend/tests/unit/test_spoolman_partial_usage_layer_reset.py

@@ -0,0 +1,49 @@
+"""Spoolman partial usage of an aborted print whose printer reset its layer to 0.
+
+The firmware zeroes ``layer_num`` on cancel. The layer captured during the
+print then decides how far the print got; without it the abort reports no
+usage at all (the progress estimate needs a total layer count, which an
+aborted print may not have either).
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services import spoolman_tracking
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("last_layer_num", "charged_at"), [(12, 12), (None, None)])
+async def test_the_captured_layer_stands_in_for_the_reset_one(last_layer_num, charged_at):
+    session = MagicMock()
+    session.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+    session.return_value.__aexit__ = AsyncMock(return_value=False)
+    tracking = SimpleNamespace(
+        layer_usage={"12": {"0": 500.0}},
+        filament_properties={},
+        filament_usage=[{"slot_id": 1, "used_g": 10.0}],
+        ams_trays={},
+        slot_to_tray=[0],
+        tray_remain_start={},
+        archive_id=1,
+        tray_now_at_start=0,
+    )
+    cumulative = MagicMock(return_value={})
+    status = SimpleNamespace(layer_num=0, total_layers=None, raw_data={}, tray_change_log=[])
+
+    with (
+        patch.object(spoolman_tracking, "async_session", session),
+        patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
+        patch("backend.app.services.printer_manager.printer_manager.get_status", MagicMock(return_value=status)),
+        patch.object(spoolman_tracking, "_get_printer_serial", AsyncMock(return_value="SERIAL")),
+        patch.object(spoolman_tracking, "_get_spoolman_client_with_fallback", AsyncMock(return_value=MagicMock())),
+        patch("backend.app.utils.threemf_tools.get_cumulative_usage_at_layer", cumulative),
+    ):
+        await spoolman_tracking._report_partial_usage(1, tracking, last_layer_num=last_layer_num)
+
+    if charged_at is None:
+        cumulative.assert_not_called()
+    else:
+        assert cumulative.call_args.args[1] == charged_at

backend/tests/unit/test_vp_mqtt_server.py

@@ -5,7 +5,7 @@ import asyncio
 import inspect
 import json
 from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -761,3 +761,85 @@ class TestSlicerCommandsTheLogKeepsWhole:
         )
 
         assert '"tray_id": 26' not in text
+
+
+def _client_writer(peer: str = "1.2.3.4") -> MagicMock:
+    writer = MagicMock()
+    writer.write = MagicMock()
+    writer.drain = AsyncMock()
+    writer.close = MagicMock()
+    writer.wait_closed = AsyncMock()
+    writer.get_extra_info = MagicMock(side_effect=lambda name: (peer, 12345) if name == "peername" else None)
+    return writer
+
+
+class TestNothingBeforeConnect:
+    """PUBLISH and SUBSCRIBE before a successful CONNECT reach nothing (V-U4)."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("packet", "handler"),
+        [
+            (bytes([0x30, 0x05, 0x00, 0x01, ord("t"), ord("{"), ord("}")]), "_handle_publish"),
+            (bytes([0x82, 0x06, 0x00, 0x01, 0x00, 0x01, ord("t"), 0x00]), "_handle_subscribe"),
+        ],
+    )
+    async def test_a_packet_before_connect_is_dropped(self, packet, handler):
+        server = _make_server()
+        server._running = True
+        reader = asyncio.StreamReader()
+        connect = _build_connect_payload(keep_alive=60)
+        # The packet first, then a valid CONNECT and the same packet again:
+        # only the one after the CONNECT may reach the handler.
+        reader.feed_data(packet + bytes([0x10, len(connect)]) + connect + packet + bytes([0xE0, 0x00]))
+        reader.feed_eof()
+        server._send_status_report = AsyncMock()
+        handled = AsyncMock()
+
+        with patch.object(server, handler, handled):
+            await server._handle_client(reader, _client_writer())
+
+        assert handled.await_count == 1
+
+
+class TestARateLimitedAddress:
+    """The rate limit refuses the right access code too, through the real handler (V-W2)."""
+
+    @pytest.mark.asyncio
+    async def test_a_correct_connect_is_refused(self):
+        from backend.app.services.virtual_printer.mqtt_server import _AUTH_RATE_LIMIT_MAX_ATTEMPTS
+
+        server = _make_server()
+        server._running = True
+        for _ in range(_AUTH_RATE_LIMIT_MAX_ATTEMPTS):
+            server._record_auth_failure("1.2.3.4")
+        reader = asyncio.StreamReader()
+        connect = _build_connect_payload(keep_alive=60)
+        reader.feed_data(bytes([0x10, len(connect)]) + connect)
+        reader.feed_eof()
+        writer = _client_writer()
+        server._handle_connect = AsyncMock(return_value=(True, 60))
+
+        await server._handle_client(reader, writer)
+
+        server._handle_connect.assert_not_awaited()
+        writer.write.assert_called_once_with(bytes([0x20, 0x02, 0x00, 0x05]))
+        assert not server._clients
+
+    @pytest.mark.asyncio
+    async def test_another_address_still_connects(self):
+        from backend.app.services.virtual_printer.mqtt_server import _AUTH_RATE_LIMIT_MAX_ATTEMPTS
+
+        server = _make_server()
+        server._running = True
+        for _ in range(_AUTH_RATE_LIMIT_MAX_ATTEMPTS):
+            server._record_auth_failure("1.2.3.4")
+        reader = asyncio.StreamReader()
+        connect = _build_connect_payload(keep_alive=60)
+        reader.feed_data(bytes([0x10, len(connect)]) + connect)
+        reader.feed_eof()
+        server._handle_connect = AsyncMock(return_value=(True, 60))
+
+        await server._handle_client(reader, _client_writer(peer="5.6.7.8"))
+
+        server._handle_connect.assert_awaited_once()