bambuddy/backend/tests/unit/services/test_virtual_printer.py

"""Unit tests for Virtual Printer services.

Tests the virtual printer manager, FTP server, and SSDP server components.
"""

import asyncio
import json
import zipfile
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest


def _write_3mf_with_filaments(file_path: Path, filaments: list[dict], plate_index: int = 1) -> None:
    """Build a minimal 3MF zip with `Metadata/slice_info.config` carrying the
    given per-slot filament entries. Each `filaments` dict needs `id`, `type`,
    `color`, `used_g`. Used by the #1188 VP queue-mode tests below."""
    filament_xml = "".join(
        f'<filament id="{f["id"]}" type="{f["type"]}" color="{f["color"]}" '
        f'used_g="{f["used_g"]}" tray_info_idx="{f.get("tray_info_idx", "")}"/>'
        for f in filaments
    )
    config = (
        '<?xml version="1.0" encoding="utf-8"?>'
        "<config>"
        f'<plate><metadata key="index" value="{plate_index}"/>'
        f"{filament_xml}"
        "</plate>"
        "</config>"
    )
    with zipfile.ZipFile(file_path, "w") as zf:
        zf.writestr("Metadata/slice_info.config", config)
        # Plate gcode is referenced for plate-id detection in the VP path —
        # presence is enough; contents don't matter.
        zf.writestr(f"Metadata/plate_{plate_index}.gcode", "; gcode\n")


class TestVirtualPrinterInstance:
    """Tests for VirtualPrinterInstance class."""

    @pytest.fixture
    def instance(self, tmp_path):
        """Create a VirtualPrinterInstance with test defaults."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        return VirtualPrinterInstance(
            vp_id=1,
            name="TestPrinter",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )

    # ========================================================================
    # Tests for instance properties
    # ========================================================================

    def test_instance_stores_parameters(self, instance):
        """Verify constructor stores parameters correctly."""
        assert instance.id == 1
        assert instance.name == "TestPrinter"
        assert instance.mode == "archive"
        assert instance.model == "C11"
        assert instance.access_code == "12345678"
        assert instance.serial_suffix == "391800001"

    def test_instance_serial_property(self, instance):
        """Verify serial is generated from model prefix + suffix."""
        # C11 = P1P, prefix = 01S00A
        assert instance.serial == "01S00A391800001"

    def test_instance_serial_x1c(self, tmp_path):
        """Verify X1C serial uses correct prefix."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=2,
            name="X1C",
            mode="archive",
            model="BL-P001",
            access_code="12345678",
            serial_suffix="391800002",
            base_dir=tmp_path,
        )
        assert inst.serial == "00M00A391800002"

    def test_instance_is_proxy_false(self, instance):
        """Verify is_proxy is False for non-proxy mode."""
        assert instance.is_proxy is False

    def test_instance_is_proxy_true(self, tmp_path):
        """Verify is_proxy is True for proxy mode."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=3,
            name="Proxy",
            mode="proxy",
            model="C11",
            access_code="",
            serial_suffix="391800003",
            target_printer_ip="192.168.1.100",
            base_dir=tmp_path,
        )
        assert inst.is_proxy is True

    def test_instance_is_running_with_active_tasks(self, instance):
        """Verify is_running is True when tasks are active."""
        mock_task = MagicMock()
        mock_task.done.return_value = False
        instance._tasks = [mock_task]
        assert instance.is_running is True

    def test_instance_is_running_with_no_tasks(self, instance):
        """Verify is_running is False when no tasks."""
        assert instance.is_running is False

    def test_instance_creates_directories(self, instance, tmp_path):
        """Verify instance creates upload and cert directories."""
        assert (tmp_path / "uploads" / "1").exists()
        assert (tmp_path / "uploads" / "1" / "cache").exists()
        assert (tmp_path / "certs" / "1").exists()

    # ========================================================================
    # Tests for status
    # ========================================================================

    def test_get_status_returns_correct_format(self, instance):
        """Verify get_status returns expected fields."""
        instance._pending_files = {"file1.3mf": Path("/tmp/file1.3mf")}  # nosec B108
        mock_task = MagicMock(done=MagicMock(return_value=False))
        instance._tasks = [mock_task]

        status = instance.get_status()
        assert status["running"] is True
        assert status["pending_files"] == 1

    def test_get_status_not_running(self, instance):
        """Verify get_status when no tasks."""
        status = instance.get_status()
        assert status["running"] is False
        assert status["pending_files"] == 0

    # ========================================================================
    # Tests for file handling
    # ========================================================================

    @pytest.mark.asyncio
    async def test_on_file_received_adds_to_pending(self, instance):
        """Verify received file is added to pending list in review mode."""
        instance.mode = "review"

        file_path = Path("/tmp/test.3mf")  # nosec B108

        with patch.object(instance, "_queue_file", new_callable=AsyncMock) as mock_queue:
            await instance.on_file_received(file_path, "192.168.1.100")

            assert "test.3mf" in instance._pending_files
            mock_queue.assert_called_once()

    @pytest.mark.asyncio
    async def test_on_file_received_archives_immediately(self, instance):
        """Verify file is archived in immediate mode."""
        file_path = Path("/tmp/test.3mf")  # nosec B108

        with patch.object(instance, "_archive_file", new_callable=AsyncMock) as mock_archive:
            await instance.on_file_received(file_path, "192.168.1.100")

            mock_archive.assert_called_once_with(file_path, "192.168.1.100")

    @pytest.mark.asyncio
    async def test_on_file_received_signals_FINISH_to_slicer(self, instance):
        """Regression #1280: when a slicer's Print flow uploads to a non-proxy VP,
        the VP must transition gcode_state PREPARE → FINISH so the slicer's
        in-flight-job lock releases. Going PREPARE → IDLE wedges Orca at
        "Downloading...(0%)" and blocks the next dispatch with "busy with
        another print job".

        Send-flow slicers don't watch the post-upload state, so this is a
        no-op behavior change for them.
        """
        instance.mode = "archive"
        instance._mqtt = MagicMock()
        instance._mqtt.set_gcode_state = MagicMock()
        file_path = Path("/tmp/test.3mf")  # nosec B108

        with patch.object(instance, "_archive_file", new_callable=AsyncMock):
            await instance.on_file_received(file_path, "192.168.1.100")

        instance._mqtt.set_gcode_state.assert_called_once_with("FINISH", filename="test.3mf", prepare_percent="100")

    @pytest.mark.asyncio
    async def test_on_file_received_non_3mf_does_not_touch_state(self, instance):
        """Non-3MF uploads (e.g., a job's auxiliary files) must not transition
        the visible state — the slicer is only tracking the .3mf upload."""
        instance.mode = "archive"
        instance._mqtt = MagicMock()
        instance._mqtt.set_gcode_state = MagicMock()
        file_path = Path("/tmp/test.gcode")  # nosec B108

        with patch.object(instance, "_archive_file", new_callable=AsyncMock):
            await instance.on_file_received(file_path, "192.168.1.100")

        instance._mqtt.set_gcode_state.assert_not_called()

    @pytest.mark.asyncio
    async def test_on_print_command_schedules_finish_release_non_proxy(self, instance):
        """#1658: Bambu Studio 2.7.x flipped the slicer's Send flow to
        FTP → FTP → MQTT project_file. Under that order the synthetic
        project_file ack overwrites the FINISH set by #1280 in
        ``on_file_received`` back to PREPARE, leaving the slicer's
        "Downloading" modal stuck forever. Re-firing FINISH a moment after
        the ack releases the modal.
        """
        instance.mode = "archive"
        instance._mqtt = MagicMock()
        instance._mqtt.set_gcode_state = MagicMock()

        with patch.object(instance, "_delayed_finish_release", new_callable=AsyncMock) as mock_delayed:
            await instance.on_print_command("test.3mf", {"command": "project_file"})

        mock_delayed.assert_called_once()
        # First positional arg should be the filename; second is the delay seconds.
        args = mock_delayed.call_args.args
        assert args[0] == "test.3mf"
        assert isinstance(args[1], int | float)

    @pytest.mark.asyncio
    async def test_on_print_command_proxy_mode_does_not_reschedule_finish(self, instance):
        """Proxy mode hands push_status straight from the real printer through
        the bridge. Re-firing a synthetic FINISH would clobber a real
        PREPARE / RUNNING transition coming back from the printer, so the
        scheduler is exempt for proxy mode."""
        instance.mode = "proxy"
        instance._mqtt = MagicMock()

        with patch.object(instance, "_delayed_finish_release", new_callable=AsyncMock) as mock_delayed:
            await instance.on_print_command("test.3mf", {"command": "project_file"})

        mock_delayed.assert_not_called()

    @pytest.mark.asyncio
    async def test_on_print_command_no_mqtt_does_not_schedule(self, instance):
        """If the MQTT server isn't running yet (transient race during boot),
        the scheduler must skip silently — no AttributeError, no orphan task."""
        instance.mode = "queue"
        instance._mqtt = None

        # Should not raise.
        await instance.on_print_command("test.3mf", {"command": "project_file"})
        assert instance._finish_release_task is None

    @pytest.mark.asyncio
    async def test_schedule_finish_release_cancels_previous_timer(self, instance):
        """A slicer that fires project_file twice (e.g. retry after a transient
        FTP hiccup) must only result in one FINISH transition — the earlier
        in-flight timer is cancelled when the next one is scheduled."""
        instance.mode = "queue"
        instance._mqtt = MagicMock()

        instance._schedule_finish_release("first.3mf", delay=10.0)
        first_task = instance._finish_release_task
        assert first_task is not None

        instance._schedule_finish_release("second.3mf", delay=10.0)
        second_task = instance._finish_release_task

        assert second_task is not first_task
        # Give the loop one tick so the cancelled task settles.
        await asyncio.sleep(0)
        assert first_task.cancelled() or first_task.done()
        # Clean up the still-pending second task so the test doesn't leak it.
        second_task.cancel()
        try:
            await second_task
        except asyncio.CancelledError:
            pass

    @pytest.mark.asyncio
    async def test_delayed_finish_release_sets_finish_state(self, instance):
        """End-to-end: after the delay elapses, set_gcode_state is called with
        FINISH and prepare_percent=100, matching the wire-format the slicer's
        Print flow consumes to release "Downloading"."""
        instance._mqtt = MagicMock()
        instance._mqtt.set_gcode_state = MagicMock()

        await instance._delayed_finish_release("queued.3mf", delay=0.0)

        instance._mqtt.set_gcode_state.assert_called_once_with("FINISH", filename="queued.3mf", prepare_percent="100")

    @pytest.mark.asyncio
    async def test_on_print_command_no_filename_does_not_schedule(self, instance):
        """A project_file command without a subtask_name (defensive — real
        slicers always send one) must not schedule a no-op FINISH that would
        carry an empty filename on the next 1 Hz push."""
        instance.mode = "queue"
        instance._mqtt = MagicMock()

        with patch.object(instance, "_delayed_finish_release", new_callable=AsyncMock) as mock_delayed:
            await instance.on_print_command("", {"command": "project_file"})

        mock_delayed.assert_not_called()

    @pytest.mark.asyncio
    async def test_archive_file_skips_non_3mf(self, instance):
        """Verify non-3MF files are skipped and cleaned up."""
        instance._session_factory = MagicMock()
        instance._pending_files["verify_job"] = Path("/tmp/verify_job")  # nosec B108

        with patch("pathlib.Path.unlink"):
            await instance._archive_file(Path("/tmp/verify_job"), "192.168.1.100")  # nosec B108

            assert "verify_job" not in instance._pending_files

    @pytest.mark.asyncio
    async def test_archive_file_broadcasts_archive_created(self, tmp_path):
        """#1282: VP immediate-mode archives must broadcast archive_created so
        the Archives page refreshes without a tab switch. Real-printer prints
        get this via main.py's MQTT print_start handler; the VP path used to
        skip the broadcast entirely."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=30,
            name="ImmediateBroadcast",
            mode="archive",
            model="C12",
            access_code="12345678",
            serial_suffix="391800030",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 99
        mock_archive.printer_id = None
        mock_archive.filename = "test.3mf"
        mock_archive.print_name = "test"
        mock_archive.status = "archived"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.core.websocket.ws_manager.send_archive_created",
                new_callable=AsyncMock,
            ) as mock_broadcast,
        ):
            await inst._archive_file(file_path, "192.168.1.100")

        mock_broadcast.assert_awaited_once()
        payload = mock_broadcast.await_args.args[0]
        assert payload["id"] == 99
        assert payload["filename"] == "test.3mf"
        assert payload["status"] == "archived"

    # ========================================================================
    # Tests for auto_dispatch
    # ========================================================================

    def test_auto_dispatch_defaults_to_true(self, tmp_path):
        """Verify auto_dispatch defaults to True when not specified."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=10,
            name="DefaultDispatch",
            mode="queue",
            model="C11",
            access_code="12345678",
            serial_suffix="391800010",
            base_dir=tmp_path,
        )
        assert inst.auto_dispatch is True

    @pytest.mark.asyncio
    async def test_add_to_print_queue_with_auto_dispatch_on(self, tmp_path):
        """Verify queue items have manual_start=False when auto_dispatch=True."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        added_items = []

        def capture_add(item):
            added_items.append(item)

        mock_db.add = MagicMock(side_effect=capture_add)
        mock_db.commit = AsyncMock()

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=11,
            name="AutoDispatchOn",
            mode="queue",
            model="C11",
            access_code="12345678",
            serial_suffix="391800011",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        # Create a temp 3mf file
        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.manual_start is False

    @pytest.mark.asyncio
    async def test_add_to_print_queue_broadcasts_archive_created(self, tmp_path):
        """#1282: VP queue-mode uploads must broadcast archive_created so the
        Archives page picks up the new entry live. Pre-fix the page only
        refreshed when the user manually switched tabs."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        mock_db.add = MagicMock()
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=31,
            name="QueueBroadcast",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800031",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 77
        mock_archive.printer_id = None
        mock_archive.filename = "test.3mf"
        mock_archive.print_name = "test"
        mock_archive.status = "archived"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.core.websocket.ws_manager.send_archive_created",
                new_callable=AsyncMock,
            ) as mock_broadcast,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        mock_broadcast.assert_awaited_once()
        payload = mock_broadcast.await_args.args[0]
        assert payload["id"] == 77
        assert payload["print_name"] == "test"
        assert payload["status"] == "archived"

    @pytest.mark.asyncio
    async def test_add_to_print_queue_with_auto_dispatch_off(self, tmp_path):
        """Verify queue items have manual_start=True when auto_dispatch=False."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        added_items = []

        def capture_add(item):
            added_items.append(item)

        mock_db.add = MagicMock(side_effect=capture_add)
        mock_db.commit = AsyncMock()

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=12,
            name="AutoDispatchOff",
            mode="queue",
            model="C11",
            access_code="12345678",
            serial_suffix="391800012",
            auto_dispatch=False,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        # Create a temp 3mf file
        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.manual_start is True

    @pytest.mark.asyncio
    async def test_add_to_print_queue_gcode_injection_on(self, tmp_path):
        """#1516: queue items opt into injection when the VP has gcode_injection=True."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        added_items = []
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=13,
            name="InjectOn",
            mode="queue",
            model="C11",
            access_code="12345678",
            serial_suffix="391800013",
            gcode_injection=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].gcode_injection is True

    @pytest.mark.asyncio
    async def test_add_to_print_queue_gcode_injection_off_by_default(self, tmp_path):
        """#1516: queue items do NOT inject when the VP leaves gcode_injection at its default."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        added_items = []
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=14,
            name="InjectOff",
            mode="queue",
            model="C11",
            access_code="12345678",
            serial_suffix="391800014",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].gcode_injection is False

    @pytest.mark.asyncio
    async def test_add_to_print_queue_uses_workflow_defaults_from_settings(self, tmp_path):
        """#1235: VP queue-mode constructed PrintQueueItem without specifying
        bed_levelling / flow_cali / vibration_cali / layer_inspect / timelapse,
        so SQLAlchemy applied the column-level defaults and ignored the user's
        workflow preferences entirely. Every print sent from the slicer to the
        VP came through with the OPPOSITE of what the workflow page said,
        forcing the user to edit each queue item by hand.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=22,
            name="DefaultsTest",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800022",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        # The reporter set every workflow default to the OPPOSITE of the model's
        # column default. Pre-fix the column defaults won; with the fix the
        # settings values must flow through to the queue item exactly as stored.
        settings_map = {
            "virtual_printer_archive_name_source": None,
            # Legacy boolean-string rows still coerce (false->off, true->on).
            "default_bed_levelling": "false",  # tri-state default: auto
            "default_flow_cali": "true",  # tri-state default: auto
            "default_vibration_cali": "false",  # model default: True
            "default_layer_inspect": "true",  # model default: False
            "default_timelapse": "true",  # model default: False
        }

        async def fake_get_setting(_db, key):
            return settings_map.get(key)

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new=fake_get_setting,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.bed_levelling == "off", "default_bed_levelling=false must flow through"
        assert queue_item.flow_cali == "on", "default_flow_cali=true must flow through"
        assert queue_item.vibration_cali is False, "default_vibration_cali=false must flow through"
        assert queue_item.layer_inspect is True, "default_layer_inspect=true must flow through"
        assert queue_item.timelapse is True, "default_timelapse=true must flow through"

    @pytest.mark.asyncio
    async def test_add_to_print_queue_falls_back_to_schema_defaults_when_unset(self, tmp_path):
        """#1235 fallback: when no workflow setting is in the DB, the queue
        item should use the AppSettings (Pydantic) defaults — same values
        the user sees in the workflow page on a fresh install.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=23,
            name="FreshInstallDefaults",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800023",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,  # No settings → fall back to schema defaults
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        # These must match the AppSettings (Pydantic) defaults in schemas/settings.py
        assert queue_item.bed_levelling == "auto"
        assert queue_item.flow_cali == "auto"
        assert queue_item.vibration_cali is True
        assert queue_item.layer_inspect is False
        assert queue_item.timelapse is False

    @pytest.mark.asyncio
    async def test_add_to_print_queue_inherits_slicer_print_options(self, tmp_path):
        """#1403: VP-queue items used to fall back to `default_timelapse` even
        though the slicer's MQTT `project_file` command carries the user's
        actual choice. Capture-via-`on_print_command` flow lets the user's
        slicer toggle reach the queue item.

        Settings here have timelapse OFF; the slicer's MQTT capture has it ON.
        After the fix the queue item must reflect the slicer's choice.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=24,
            name="SlicerInherits",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800024",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        # Pre-populate the capture as if MQTT `project_file` arrived already.
        # Settings (below) deliberately have timelapse OFF — only the slicer
        # capture should drive the resulting queue item.
        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "timelapse": True,
                "bed_leveling": False,  # Note: MQTT field is single-L `bed_leveling`
                "flow_cali": True,
                "vibration_cali": False,
                "layer_inspect": True,
            },
        )

        settings_map = {
            "virtual_printer_archive_name_source": None,
            "default_bed_levelling": "true",
            "default_flow_cali": "false",
            "default_vibration_cali": "true",
            "default_layer_inspect": "false",
            "default_timelapse": "false",
        }

        async def fake_get_setting(_db, key):
            return settings_map.get(key)

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new=fake_get_setting,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.timelapse is True, "Slicer's timelapse=True must override settings.default_timelapse=False"
        assert queue_item.bed_levelling == "off", "Slicer's bed_leveling=False must override default_bed_levelling"
        assert queue_item.flow_cali == "on"
        assert queue_item.vibration_cali is False
        assert queue_item.layer_inspect is True
        # Capture is consumed — no lingering state for the next print of the same name.
        assert file_path.name not in inst._slicer_print_options

    @pytest.mark.asyncio
    async def test_add_to_print_queue_coerces_slicer_integer_zero_one(self, tmp_path):
        """#1403: H-family firmwares carry calibration flags as integers
        (0/1) rather than booleans. The capture must coerce both shapes so
        H-family-sliced jobs through the VP queue work the same as P1/X1.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=25,
            name="SlicerIntegers",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800025",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "timelapse": 1, "bed_leveling": 0, "flow_cali": 1},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.timelapse is True, "integer 1 must coerce to True"
        assert queue_item.bed_levelling == "off", "integer 0 must coerce to off"
        assert queue_item.flow_cali == "on"

    @pytest.mark.asyncio
    async def test_add_to_print_queue_captures_slicer_auto_from_int_companion(self, tmp_path):
        """The slicer's tri-state rides on the int companion (auto_bed_leveling /
        extrude_cali_flag). When the slicer picks "Auto" it sends bed_leveling
        false + auto_bed_leveling 2; the VP must record "auto", not "off".
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=26,
            name="SlicerAuto",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800026",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "bed_leveling": False,
                "auto_bed_leveling": 2,
                "flow_cali": False,
                "extrude_cali_flag": 2,
            },
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.bed_levelling == "auto", "auto_bed_leveling=2 must record 'auto'"
        assert queue_item.flow_cali == "auto", "extrude_cali_flag=2 must record 'auto'"

    @pytest.mark.asyncio
    async def test_add_to_print_queue_populates_required_filament_types(self, tmp_path):
        """#1188: VP queue-mode used to create PrintQueueItems with no
        filament fields, so the scheduler fell through to model-only matching
        and dispatched onto whatever printer was free regardless of loaded
        colour. ``required_filament_types`` is populated unconditionally
        (cheap, helps the scheduler validate type even without
        ``force_color_match``) — pin that contract here."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=21,
            name="Reqs",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800021",
            auto_dispatch=True,
            queue_force_color_match=False,  # off → only required_filament_types
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "multi.3mf"
        _write_3mf_with_filaments(
            file_path,
            [
                {"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "12.3"},
                {"id": "2", "type": "PETG", "color": "#000000", "used_g": "4.5"},
                # used_g=0 → not actually consumed by this plate, must be ignored
                {"id": "3", "type": "ABS", "color": "#FF0000", "used_g": "0"},
            ],
            plate_index=1,
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "multi"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        # Type-only fallback always populated. Sorted, deduped, no zero-use ABS.
        assert queue_item.required_filament_types is not None
        assert json.loads(queue_item.required_filament_types) == ["PETG", "PLA"]
        # Setting off → no force_color_match overrides leaked.
        assert queue_item.filament_overrides is None

    @pytest.mark.asyncio
    async def test_add_to_print_queue_force_color_match_writes_overrides(self, tmp_path):
        """#1188 core fix: when the per-VP ``queue_force_color_match`` toggle
        is on, every consumed slot lands as a ``filament_overrides`` entry
        with ``force_color_match: true``. This is the field the scheduler
        keys on (``print_scheduler.py:512``) — without it, slot-by-slot
        type+color matching never runs."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=22,
            name="ForceColor",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800022",
            auto_dispatch=True,
            queue_force_color_match=True,  # on
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "forced.3mf"
        _write_3mf_with_filaments(
            file_path,
            [
                {"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "10.0"},
                {"id": "2", "type": "PLA", "color": "#FF00FF", "used_g": "5.0"},
            ],
            plate_index=1,
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "forced"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        assert queue_item.filament_overrides is not None
        overrides = json.loads(queue_item.filament_overrides)
        assert overrides == [
            {"slot_id": 1, "type": "PLA", "color": "#FFFFFF", "tray_info_idx": "", "force_color_match": True},
            {"slot_id": 2, "type": "PLA", "color": "#FF00FF", "tray_info_idx": "", "force_color_match": True},
        ]
        # required_filament_types still populated alongside overrides.
        assert json.loads(queue_item.required_filament_types) == ["PLA"]

    @pytest.mark.asyncio
    async def test_add_to_print_queue_force_color_match_carries_tray_info_idx(self, tmp_path):
        """#2650: the force override must carry the 3MF's ``tray_info_idx`` so the
        scheduler can tell Bambu PLA variants apart (Basic GFA00 / Matte GFA01 /
        Silk GFA06) — they all report ``tray_type == "PLA"`` with the same colour,
        so type+colour alone dispatches onto the wrong variant."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=24,
            name="Variant",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800024",
            auto_dispatch=True,
            queue_force_color_match=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "variant.3mf"
        _write_3mf_with_filaments(
            file_path,
            [
                # White PLA Matte — same colour as Basic/Silk, distinguished only by idx.
                {"id": "1", "type": "PLA", "color": "#FFFFFF", "used_g": "10.0", "tray_info_idx": "GFA01"},
            ],
            plate_index=1,
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "variant"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        overrides = json.loads(added_items[0].filament_overrides)
        assert overrides == [
            {"slot_id": 1, "type": "PLA", "color": "#FFFFFF", "tray_info_idx": "GFA01", "force_color_match": True},
        ]

    @pytest.mark.asyncio
    async def test_add_to_print_queue_force_color_match_skips_when_3mf_unparseable(self, tmp_path):
        """A malformed or fake-bytes 3MF must not crash the upload path —
        we just write the queue item with no filament fields and let the
        scheduler fall back to model-only matching (the pre-#1188 default).
        Regression guard for the existing fake-bytes happy-path tests."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=23,
            name="Unparseable",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800023",
            auto_dispatch=True,
            queue_force_color_match=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "bad.3mf"
        file_path.write_bytes(b"not a real 3mf zip")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "bad"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        queue_item = added_items[0]
        # No filament data extractable → both fields stay None (graceful
        # fallback to model-only scheduling).
        assert queue_item.required_filament_types is None
        assert queue_item.filament_overrides is None

    # ========================================================================
    # Tests for archive_name_source setting (#1152)
    # ========================================================================

    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        ("setting_value", "expected_prefer_filename"),
        [
            ("filename", True),
            ("metadata", False),
            (None, False),  # Default when setting unset
            ("", False),  # Defensive: empty string is not "filename"
        ],
    )
    async def test_archive_file_passes_prefer_filename_per_setting(
        self, tmp_path, setting_value, expected_prefer_filename
    ):
        """_archive_file reads `virtual_printer_archive_name_source` and forwards
        prefer_filename_for_name=True only when it equals 'filename' (#1152)."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=20,
            name="NameSource",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800020",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "user-renamed-job.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "user-renamed-job"

        archive_print_mock = AsyncMock(return_value=mock_archive)

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=setting_value,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                archive_print_mock,
            ),
        ):
            await inst._archive_file(file_path, "192.168.1.100")

        assert archive_print_mock.await_count == 1
        kwargs = archive_print_mock.await_args.kwargs
        assert kwargs.get("prefer_filename_for_name") is expected_prefer_filename

    # ========================================================================
    # Tests for failure-path cleanup (#audit-R2-1)
    # ========================================================================
    #
    # All three file handlers (_archive_file, _queue_file, _add_to_print_queue)
    # previously only popped _pending_files and unlinked the temp file on the
    # success branch. Failure paths leaked the marker (blocking same-name
    # retries via the FTP layer) and the temp file on disk. The cleanup must
    # ALWAYS run, even when archival / queue insert raises.

    @pytest.mark.asyncio
    async def test_archive_file_failure_path_pops_pending_and_unlinks(self, tmp_path):
        """When the archive layer raises, `_pending_files[filename]` must still
        be popped and the temp file must be unlinked. Otherwise the FTP layer's
        same-name retry guard would silently reject the slicer's next attempt
        and the upload_dir would accumulate ghost files."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=40,
            name="ArchiveFailCleanup",
            mode="archive",
            model="C12",
            access_code="12345678",
            serial_suffix="391800040",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        file_path = tmp_path / "cleanup-archive.3mf"
        file_path.write_bytes(b"fake3mf")
        inst._pending_files[file_path.name] = file_path

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                side_effect=RuntimeError("archive blew up"),
            ),
        ):
            await inst._archive_file(file_path, "192.168.1.100")

        assert file_path.name not in inst._pending_files
        assert not file_path.exists()

    @pytest.mark.asyncio
    async def test_queue_file_failure_path_pops_pending_and_unlinks(self, tmp_path):
        """Same invariant for _queue_file: a DB error during PendingUpload
        insert must not leak the in-flight marker or the temp file."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        # Commit raises — emulating a DB connectivity error.
        mock_db.add = MagicMock()
        mock_db.commit = AsyncMock(side_effect=RuntimeError("db unreachable"))
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=41,
            name="QueueFailCleanup",
            mode="review",
            model="C12",
            access_code="12345678",
            serial_suffix="391800041",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        file_path = tmp_path / "cleanup-queue.3mf"
        file_path.write_bytes(b"fake3mf")
        inst._pending_files[file_path.name] = file_path

        await inst._queue_file(file_path, "192.168.1.100")

        assert file_path.name not in inst._pending_files
        assert not file_path.exists()

    @pytest.mark.asyncio
    async def test_add_to_print_queue_failure_path_pops_pending_and_unlinks(self, tmp_path):
        """Same invariant for _add_to_print_queue: a DB error or archive
        failure must not leak the in-flight marker or the temp file."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        mock_db.add = MagicMock()
        mock_db.commit = AsyncMock()
        mock_db.execute = AsyncMock(side_effect=RuntimeError("queue insert blew up"))
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=42,
            name="DispatchFailCleanup",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800042",
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        file_path = tmp_path / "cleanup-dispatch.3mf"
        file_path.write_bytes(b"fake3mf")
        inst._pending_files[file_path.name] = file_path

        with patch(
            "backend.app.api.routes.settings.get_setting",
            new_callable=AsyncMock,
            return_value=None,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert file_path.name not in inst._pending_files
        assert not file_path.exists()

    # ========================================================================
    # Test for position=MAX+1 (audit-R2)
    # ========================================================================

    @pytest.mark.asyncio
    async def test_add_to_print_queue_position_picks_max_plus_one(self, tmp_path):
        """VP-queue items previously got hardcoded `position=1`, colliding
        with existing items at position 1 and producing non-deterministic
        execution order. Now the position is chosen by `MAX(position)+1`
        against the target queue, matching the canonical `POST /print-queue/`
        path."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        # Capture the inserted PrintQueueItem so we can assert on .position.
        added_items: list = []

        class _RecordingDb:
            def __init__(self):
                self.add = lambda item: added_items.append(item)
                self.commit = AsyncMock()

            async def execute(self, query):  # noqa: ARG002
                """Return a stub result whose `.scalar()` reports the existing
                MAX(position) for the target. Returning 7 means the new item
                should land at 8."""
                result = MagicMock()
                result.scalar = MagicMock(return_value=7)
                return result

        mock_db = _RecordingDb()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=43,
            name="PositionMaxPlusOne",
            mode="queue",
            model="C12",
            access_code="12345678",
            serial_suffix="391800043",
            target_printer_id=99,
            auto_dispatch=True,
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        file_path = tmp_path / "next-position.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 555
        mock_archive.printer_id = None
        mock_archive.filename = "next-position.3mf"
        mock_archive.print_name = "next-position"
        mock_archive.status = "archived"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.core.websocket.ws_manager.send_archive_created",
                new_callable=AsyncMock,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        # One queue item was added.
        assert len(added_items) == 1
        queue_item = added_items[0]
        # Position = max(7) + 1 = 8 — NOT the legacy hardcoded 1.
        assert queue_item.position == 8

    @pytest.mark.asyncio
    async def test_add_to_print_queue_multi_plate_send_all_enqueues_one_per_plate(self, tmp_path):
        """#1733: BambuStudio / OrcaSlicer "Send All" of a multi-plate project
        uploads ONE 3MF containing every plate. Pre-fix only the first plate
        index was extracted and one queue item was created; plates 2..N were
        silently dropped. Post-fix every `<plate>` block in `slice_info.config`
        produces its own PrintQueueItem with the correct ``plate_id``, sharing
        the same backing archive, with consecutive positions for plate-order
        execution.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []

        class _RecordingDb:
            def __init__(self):
                # Capture inserted items as they're added; assign a fake .id
                # on flush so the manager's logger doesn't see None.
                self._next_id = 1000

                def _add(item):
                    added_items.append(item)

                self.add = _add
                self.commit = AsyncMock()

            async def execute(self, query):  # noqa: ARG002
                """Return MAX(position) = 0 so plate items land at 1, 2, 3."""
                result = MagicMock()
                result.scalar = MagicMock(return_value=0)
                return result

            async def flush(self):
                # Mimic the FK populate so queue_item.id is available after add().
                for item in added_items:
                    if getattr(item, "id", None) is None:
                        item.id = self._next_id
                        self._next_id += 1

        mock_db = _RecordingDb()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=44,
            name="MultiPlateSendAll",
            mode="queue",
            model="O1D",  # H2D — matches the live VP H2D-1 Proxy in #1733
            access_code="12345678",
            serial_suffix="391800044",
            target_printer_id=1,
            auto_dispatch=False,  # manual_start, mirrors the live VP
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        # Build a 3MF with three plates baked into slice_info.config —
        # mirrors what BambuStudio / OrcaSlicer's "Send All" puts on the wire.
        file_path = tmp_path / "Cube.gcode.3mf"
        _write_3mf_with_filaments(
            file_path, [{"id": 1, "type": "PLA", "color": "#000000", "used_g": "15.61"}], plate_index=1
        )
        # Append plate 2 and 3 blocks to slice_info.config to mimic Send All.
        with zipfile.ZipFile(file_path, "r") as zf:
            existing = zf.read("Metadata/slice_info.config").decode()
        # Inject two additional <plate> blocks (indices 2 and 3) inside <config>.
        multi_plate_config = existing.replace(
            "</config>",
            (
                '<plate><metadata key="index" value="2"/>'
                '<filament id="2" type="PETG" color="#FB0207" used_g="14.45"/>'
                "</plate>"
                '<plate><metadata key="index" value="3"/>'
                '<filament id="3" type="PLA" color="#FFFFFF" used_g="12.10"/>'
                "</plate>"
                "</config>"
            ),
        )
        # Repack the zip with the expanded slice_info.config.
        import io as _io

        buf = _io.BytesIO()
        with zipfile.ZipFile(file_path, "r") as src, zipfile.ZipFile(buf, "w") as dst:
            for name in src.namelist():
                if name == "Metadata/slice_info.config":
                    dst.writestr(name, multi_plate_config)
                else:
                    dst.writestr(name, src.read(name))
            # Plate-2 and plate-3 gcode payloads so `extract_filament_requirements`
            # has something to read for each — contents irrelevant, presence matters.
            dst.writestr("Metadata/plate_2.gcode", "; plate 2 gcode\n")
            dst.writestr("Metadata/plate_3.gcode", "; plate 3 gcode\n")
        file_path.write_bytes(buf.getvalue())

        mock_archive = MagicMock()
        mock_archive.id = 999
        mock_archive.printer_id = None
        mock_archive.filename = "Cube.gcode.3mf"
        mock_archive.print_name = "Cube"
        mock_archive.status = "archived"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.core.websocket.ws_manager.send_archive_created",
                new_callable=AsyncMock,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        # Three queue items, one per plate, with the correct plate_id and
        # consecutive positions starting at MAX(position)+1 = 1.
        assert len(added_items) == 3, f"Expected 3 queue items for 3-plate Send All, got {len(added_items)}"
        plate_ids = [q.plate_id for q in added_items]
        assert plate_ids == [1, 2, 3], f"plate_ids should preserve slice_info order, got {plate_ids}"
        positions = [q.position for q in added_items]
        assert positions == [1, 2, 3], f"positions should be consecutive, got {positions}"
        archive_ids = {q.archive_id for q in added_items}
        assert archive_ids == {999}, f"All queue items must share the single backing archive, got {archive_ids}"
        # auto_dispatch=False on the VP → every item is manual_start.
        assert all(q.manual_start for q in added_items)

    @pytest.mark.asyncio
    async def test_add_to_print_queue_captures_nozzle_mapping(self, tmp_path):
        """#1780: BambuStudio's project_file for H2C rack-swap (O1C2) sends
        per-filament physical nozzle position IDs in `nozzle_mapping`. VP
        intake must store it as a JSON string on the queue item so the
        dispatcher can replay it. Without this the H2C firmware falls back
        to "last matching nozzle" auto-pick and ignores the user's slicer
        choice.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=42,
            name="H2CRack",
            mode="queue",
            model="O1C2",
            access_code="12345678",
            serial_suffix="391800042",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        # Pre-populate as if BS's project_file arrived. Wire shape matches
        # BambuStudio's PrintJob params: nozzle_mapping = 32-entry array of
        # per-filament physical nozzle position IDs (verified via H2C wire
        # capture). The slicer-side `nozzles_info` field that the original
        # #1780 attempt also looked for was never actually sent — it has
        # been dropped from the capture path entirely.
        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "nozzle_mapping": [16, -1, -1, 1, -1, -1, -1, -1],
            },
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        item = added_items[0]
        assert item.nozzle_mapping is not None
        assert _json.loads(item.nozzle_mapping) == [16, -1, -1, 1, -1, -1, -1, -1]

    @pytest.mark.asyncio
    async def test_add_to_print_queue_no_nozzle_mapping_when_slicer_omits(self, tmp_path):
        """#1780: every model other than O1C2 sends no nozzle_mapping — the
        queue item must carry NULL, not an empty list. NULL is what the
        dispatch layer keys off of to skip the injection entirely on non-
        rack-swap printers.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=43,
            name="NotH2C",
            mode="queue",
            model="C11",
            access_code="12345678",
            serial_suffix="391800043",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        # X1C-style slicer command — no nozzle fields.
        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "timelapse": False, "bed_leveling": True},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        item = added_items[0]
        assert item.nozzle_mapping is None

    @pytest.mark.asyncio
    async def test_add_to_print_queue_ignores_ams_mapping_when_toggle_off(self, tmp_path):
        """With the per-VP `save_ams_mapping` toggle off, the slicer's
        AMS-slot pick must be ignored entirely — neither stamped on the queue
        item nor persisted to the archive.

        Taking the slicer's pick makes `_ensure_ams_mapping` return early, so
        `_compute_ams_mapping_for_printer` never runs — and that's where
        `prefer_lowest_filament`, its AMS-filament-backup gate (#1766) and the
        inventory-remain overrides live. Honouring it unconditionally would
        retire all of that for every existing queue-mode VP on upgrade, so it
        is opt-in like every other queue-mode behaviour toggle (#2700 review).
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=44,
            name="AMSMappingOff",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800044",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=False,
            target_printer_id=7,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "ams_mapping": [4, -1, 12, -1]},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ) as mock_archive_print,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].ams_mapping is None
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping"] is None

    @pytest.mark.asyncio
    async def test_add_to_print_queue_persists_ams_mapping_to_archive_when_toggle_on(self, tmp_path):
        """With the per-VP `save_ams_mapping` toggle on, the slicer's AMS
        pick must also be forwarded to `ArchiveService.archive_print` so it
        gets promoted to `archive.extra_data.slicer_ams_mapping` and a later
        reprint can reuse the exact physical spool.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=45,
            name="AMSMappingOn",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800045",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            target_printer_id=7,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "ams_mapping": [4, -1, 12, -1]},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ) as mock_archive_print,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert _json.loads(added_items[0].ams_mapping) == [4, -1, 12, -1]

        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping"] == [4, -1, 12, -1]
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping_printer_id"] == 7

    @pytest.mark.asyncio
    async def test_add_to_print_queue_ignores_ams_mapping_for_model_based_vp(self, tmp_path):
        """A model-based VP (`target_printer_id=None`, dispatched later by the
        scheduler to whichever printer matches) has no MQTT bridge to a real
        printer, so the slicer has no live AMS layout to resolve tray IDs
        against. Whatever it sends must be ignored on both the queue item and
        the archive — trusting it would dispatch the eventually-chosen
        printer onto a tray resolved against nothing (#2700 review).
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=451,
            name="AMSMappingModelBased",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800451",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            target_printer_id=None,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "ams_mapping": [4, -1, 12, -1]},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ) as mock_archive_print,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].ams_mapping is None
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping"] is None
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping_printer_id"] is None

    @pytest.mark.asyncio
    async def test_add_to_print_queue_ignores_unresolved_ams_mapping_sentinel(self, tmp_path):
        """#2589: an all -1 `ams_mapping` means the slicer's own race lost —
        every slot unresolved. Must be dropped (queue item AND archive),
        never trusted over a fresh scheduler-computed mapping.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=46,
            name="AMSMappingSentinel",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800046",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            target_printer_id=7,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "ams_mapping": [-1, -1, -1, -1]},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ) as mock_archive_print,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].ams_mapping is None
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping"] is None

    @pytest.mark.asyncio
    async def test_add_to_print_queue_force_color_match_keeps_mapping_off_the_queue_item(self, tmp_path):
        """`Force color match` and `Save AMS mapping` both on: the archive
        still gets the slicer's pick (that's what the toggle promises for
        later reprints), but THIS dispatch does not.

        Force color match's only effect on a fixed-printer item is via the
        per-slot `filament_overrides`, which are consumed inside
        `_compute_ams_mapping_for_printer` — the exact function a stored
        mapping makes `_ensure_ams_mapping` skip. Leaving the mapping on the
        item would silently retire the user's explicit strictness (#2700
        review).
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=47,
            name="AMSMappingVsForceColor",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800047",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            queue_force_color_match=True,
            target_printer_id=7,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "ams_mapping": [4, -1, 12, -1]},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ) as mock_archive_print,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].ams_mapping is None
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping"] == [4, -1, 12, -1]
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping_printer_id"] == 7

    @pytest.mark.asyncio
    async def test_add_to_print_queue_drops_ams_mapping_too_short_for_the_plate(self, tmp_path, monkeypatch):
        """A slicer mapping shorter than the plate's highest slot id can't
        address that plate's own slots, and `_ensure_ams_mapping` would keep it
        anyway (it only rejects an all-unresolved mapping). Drop it for that
        plate so the scheduler computes one from live AMS state instead.

        The mapping is indexed by the 3MF's file-global slot ids, so the plate
        that fits keeps it — only the one that doesn't falls back (#2700
        review).
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()
        mock_db.execute = AsyncMock()
        mock_db.execute.return_value.scalar.return_value = None
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=48,
            name="AMSMappingShort",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800048",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            target_printer_id=7,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        monkeypatch.setattr(inst, "_extract_plate_ids", lambda _p: [1, 2])

        # Plate 1 prints slots 1-2 (fits the 2-entry mapping); plate 2 also
        # prints slot 4, which the mapping cannot address.
        def fake_reqs(_path, plate_id):
            if plate_id == 1:
                return [
                    {"slot_id": 1, "type": "PLA", "color": "#FF0000"},
                    {"slot_id": 2, "type": "PLA", "color": "#00FF00"},
                ]
            return [
                {"slot_id": 1, "type": "PLA", "color": "#FF0000"},
                {"slot_id": 4, "type": "PLA", "color": "#0000FF"},
            ]

        monkeypatch.setattr(
            "backend.app.services.filament_requirements.extract_filament_requirements",
            fake_reqs,
        )

        await inst.on_print_command(
            file_path.name,
            {"command": "project_file", "ams_mapping": [4, 12]},
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ) as mock_archive_print,
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 2
        assert _json.loads(added_items[0].ams_mapping) == [4, 12]
        assert added_items[1].ams_mapping is None

        # The archive keeps the mapping either way — it is the file's, not one
        # plate's, and a reprint of the plate that fits still wants it.
        assert mock_archive_print.await_args.kwargs["slicer_ams_mapping"] == [4, 12]

    @pytest.mark.asyncio
    async def test_add_to_print_queue_nozzle_pick_replicated_across_plates(self, tmp_path, monkeypatch):
        """#1780 × #1697/#1188: a multi-plate Send All from BS must stamp the
        same nozzle_mapping on every plate's queue item, not only the first.
        Mirrors the per-plate stamping for gcode_injection,
        filament_overrides, etc.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()
        mock_db.execute = AsyncMock()
        mock_db.execute.return_value.scalar.return_value = None
        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=44,
            name="H2CMultiPlate",
            mode="queue",
            model="O1C2",
            access_code="12345678",
            serial_suffix="391800044",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        # Force 3 plates so the queue loop runs three times.
        monkeypatch.setattr(inst, "_extract_plate_ids", lambda _p: [1, 2, 3])

        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "nozzle_mapping": [16, 0],
            },
        )

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 3
        for item in added_items:
            assert _json.loads(item.nozzle_mapping) == [16, 0]

    @pytest.mark.asyncio
    async def test_on_print_command_late_mqtt_retroactively_stamps_queue_item(self, tmp_path):
        """#1780 round 3: Bambu Studio's MQTT project_file can arrive AFTER
        `_add_to_print_queue` already gave up waiting (observed at 2.085 s
        on H2C wireless setups). The queue item was committed with settings
        defaults; the slicer's nozzle_mapping + workflow flags must be
        patched onto it when MQTT lands, otherwise the H2C firmware falls
        back to auto-pick.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(
            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 100 + len(added_items)))[0]
        )

        async def _flush():
            # added_items[-1].id was set by `add`; nothing else to do.
            return None

        mock_db.flush = AsyncMock(side_effect=_flush)
        mock_db.commit = AsyncMock()

        # First execute() call (the position-max SELECT inside _add_to_print_queue)
        # returns None; second (the eligible-pending SELECT in
        # _restamp_recent_queue_item) returns the committed queue id; third
        # (the UPDATE) is fire-and-forget.
        position_max_result = MagicMock()
        position_max_result.scalar = MagicMock(return_value=None)
        select_pending_result = MagicMock()
        select_pending_result.all = MagicMock(return_value=[(101,)])
        update_result = MagicMock()
        mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=99,
            name="LateMQTT",
            mode="queue",
            model="O1C2",
            access_code="12345678",
            serial_suffix="391800099",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        # MQTT server presence enables the wait_for path; we don't actually
        # use any methods on it.
        inst._mqtt = MagicMock()

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        # 1. _add_to_print_queue runs WITHOUT a prior on_print_command —
        #    the wait_for times out (settings-default fallback) and the
        #    queue item is committed.
        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            # Shorten the wait so the test isn't slow.
            patch(
                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
                0.05,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].nozzle_mapping is None  # MQTT was never received
        assert file_path.name in inst._recent_queue_items

        # 2. MQTT project_file arrives AFTER the wait expired — must
        #    retroactively patch the queue item.
        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "file": file_path.name,
                "nozzle_mapping": [16, -1, -1, 1],
                "timelapse": True,
                "bed_leveling": False,
            },
        )

        # The UPDATE call is the third execute. Inspect its values.
        update_call = mock_db.execute.await_args_list[2]
        update_stmt = update_call.args[0]
        compiled = update_stmt.compile(compile_kwargs={"literal_binds": False})
        params = dict(compiled.params)
        assert _json.loads(params["nozzle_mapping"]) == [16, -1, -1, 1]
        assert params["timelapse"] is True
        assert params["bed_levelling"] == "off"  # MQTT bed_leveling → column bed_levelling (tri-state)
        # Recent-queue tracking dict is cleared after the patch.
        assert file_path.name not in inst._recent_queue_items

    @pytest.mark.asyncio
    async def test_on_print_command_late_mqtt_retroactively_stamps_archive_ams_mapping(self, tmp_path):
        """Same #1780-round-3 race as the test above, but for `ams_mapping`
        specifically: the archive was already created (with no
        `slicer_ams_mapping`, since the slicer's pick hadn't arrived yet)
        before this late MQTT lands. With the toggle on and a fixed target
        printer, the archive must be retroactively patched too — otherwise a
        reprint from this archive would never see the badge or the Mapping
        button, even though the queue item itself dispatches correctly.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(
            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 100 + len(added_items)))[0]
        )
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()

        mock_archive_row = MagicMock()
        mock_archive_row.id = 55
        mock_archive_row.extra_data = None

        position_max_result = MagicMock()
        position_max_result.scalar = MagicMock(return_value=None)
        select_pending_result = MagicMock()
        # (queue_item_id, archive_id) — matches the id set by db.add above.
        select_pending_result.all = MagicMock(return_value=[(101, 55)])
        update_result = MagicMock()
        select_archives_result = MagicMock()
        select_archives_result.scalars = MagicMock(
            return_value=MagicMock(all=MagicMock(return_value=[mock_archive_row]))
        )
        mock_db.execute = AsyncMock(
            side_effect=[position_max_result, select_pending_result, update_result, select_archives_result]
        )

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=98,
            name="LateMQTTArchivePatch",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800098",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            target_printer_id=7,
        )
        inst._mqtt = MagicMock()

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 55
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
                0.05,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        assert len(added_items) == 1
        assert added_items[0].ams_mapping is None  # MQTT was never received in time

        # MQTT project_file arrives late, carrying the slicer's AMS pick.
        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "file": file_path.name,
                "ams_mapping": [4, -1, 12, -1],
            },
        )

        # 3rd execute() is the queue-item UPDATE; 4th is the archive SELECT.
        update_call = mock_db.execute.await_args_list[2]
        compiled = update_call.args[0].compile(compile_kwargs={"literal_binds": False})
        assert _json.loads(dict(compiled.params)["ams_mapping"]) == [4, -1, 12, -1]

        assert mock_archive_row.extra_data == {
            "slicer_ams_mapping": {"mapping": [4, -1, 12, -1], "printer_id": 7},
        }

    @pytest.mark.asyncio
    async def test_on_print_command_late_mqtt_skips_archive_patch_for_model_based_vp(self, tmp_path):
        """The archive patch above must not fire for a model-based VP
        (`target_printer_id=None`) either — same rationale as the immediate
        path: no fixed printer means no live AMS layout to have resolved the
        late mapping against.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(
            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 200 + len(added_items)))[0]
        )
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()

        position_max_result = MagicMock()
        position_max_result.scalar = MagicMock(return_value=None)
        select_pending_result = MagicMock()
        select_pending_result.all = MagicMock(return_value=[(201, 65)])
        update_result = MagicMock()
        # No 4th execute() expected — the archive patch must be skipped
        # entirely, so only 3 calls should ever happen.
        mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=97,
            name="LateMQTTModelBased",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800097",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            target_printer_id=None,
        )
        inst._mqtt = MagicMock()

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 65
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
                0.05,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "file": file_path.name,
                "ams_mapping": [4, -1, 12, -1],
                # An unrelated field so `patch` isn't empty and the UPDATE
                # actually runs — isolates the assertion to "ams_mapping was
                # excluded" rather than "nothing happened at all".
                "timelapse": True,
            },
        )

        # 3rd execute() is the queue-item UPDATE — ams_mapping must be absent
        # from it. No 4th execute() (the archive SELECT) should follow.
        assert mock_db.execute.await_count == 3
        update_call = mock_db.execute.await_args_list[2]
        compiled = update_call.args[0].compile(compile_kwargs={"literal_binds": False})
        assert "ams_mapping" not in dict(compiled.params)
        assert dict(compiled.params)["timelapse"] is True

    @pytest.mark.asyncio
    async def test_on_print_command_late_mqtt_force_color_match_archives_but_does_not_stamp(self, tmp_path):
        """The late path splits the same way the immediate one does when
        `Force color match` is on: the archive still records the slicer's pick
        for later reprints, but it must not reach the queue item, or the
        overrides the user asked for would never be applied (#2700 review).
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(
            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 300 + len(added_items)))[0]
        )
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()

        mock_archive_row = MagicMock()
        mock_archive_row.id = 75
        mock_archive_row.extra_data = None

        position_max_result = MagicMock()
        position_max_result.scalar = MagicMock(return_value=None)
        select_pending_result = MagicMock()
        select_pending_result.all = MagicMock(return_value=[(301, 75)])
        update_result = MagicMock()
        select_archives_result = MagicMock()
        select_archives_result.scalars = MagicMock(
            return_value=MagicMock(all=MagicMock(return_value=[mock_archive_row]))
        )
        mock_db.execute = AsyncMock(
            side_effect=[position_max_result, select_pending_result, update_result, select_archives_result]
        )

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=96,
            name="LateMQTTForceColor",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800096",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=True,
            queue_force_color_match=True,
            target_printer_id=7,
        )
        inst._mqtt = MagicMock()

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 75
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
                0.05,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "file": file_path.name,
                "ams_mapping": [4, -1, 12, -1],
                # Keeps the UPDATE non-empty so the assertion below is "the
                # mapping was excluded", not "nothing ran".
                "timelapse": True,
            },
        )

        update_call = mock_db.execute.await_args_list[2]
        compiled = update_call.args[0].compile(compile_kwargs={"literal_binds": False})
        assert "ams_mapping" not in dict(compiled.params)
        assert dict(compiled.params)["timelapse"] is True

        assert mock_archive_row.extra_data == {
            "slicer_ams_mapping": {"mapping": [4, -1, 12, -1], "printer_id": 7},
        }

    @pytest.mark.asyncio
    async def test_on_print_command_late_mqtt_ignores_ams_mapping_when_toggle_off(self, tmp_path):
        """Toggle off: the late path must not stamp the queue item either, or
        the opt-in would leak in through the #1780 race window.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(
            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 400 + len(added_items)))[0]
        )
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()

        position_max_result = MagicMock()
        position_max_result.scalar = MagicMock(return_value=None)
        select_pending_result = MagicMock()
        select_pending_result.all = MagicMock(return_value=[(401, 85)])
        update_result = MagicMock()
        # No 4th execute(): with nothing to save there is no archive SELECT.
        mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=95,
            name="LateMQTTToggleOff",
            mode="queue",
            model="X1C",
            access_code="12345678",
            serial_suffix="391800095",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
            save_ams_mapping=False,
            target_printer_id=7,
        )
        inst._mqtt = MagicMock()

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        mock_archive = MagicMock()
        mock_archive.id = 85
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
                0.05,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        await inst.on_print_command(
            file_path.name,
            {
                "command": "project_file",
                "file": file_path.name,
                "ams_mapping": [4, -1, 12, -1],
                "timelapse": True,
            },
        )

        assert mock_db.execute.await_count == 3
        update_call = mock_db.execute.await_args_list[2]
        compiled = update_call.args[0].compile(compile_kwargs={"literal_binds": False})
        assert "ams_mapping" not in dict(compiled.params)
        assert dict(compiled.params)["timelapse"] is True

    @pytest.mark.asyncio
    async def test_add_to_print_queue_catches_mqtt_stashed_post_wait_timeout(self, tmp_path):
        """The actual race-window scenario: wait_for times out, then MQTT
        arrives and stashes options AFTER the wait but BEFORE the
        post-commit re-check. The post-commit pop must catch it.
        """
        import json as _json

        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        added_items: list = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(
            side_effect=lambda item: (added_items.append(item), setattr(item, "id", 300 + len(added_items)))[0]
        )
        mock_db.flush = AsyncMock()
        mock_db.commit = AsyncMock()

        position_max_result = MagicMock()
        position_max_result.scalar = MagicMock(return_value=None)
        select_pending_result = MagicMock()
        select_pending_result.all = MagicMock(return_value=[(301,)])
        update_result = MagicMock()
        mock_db.execute = AsyncMock(side_effect=[position_max_result, select_pending_result, update_result])

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=96,
            name="RaceCommitYield",
            mode="queue",
            model="O1C2",
            access_code="12345678",
            serial_suffix="391800096",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        inst._mqtt = MagicMock()

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")

        # Stash MQTT data on the FIRST commit (simulating MQTT arrival
        # during _add_to_print_queue's commit yield); _restamp also calls
        # db.commit later, so we one-shot the side effect.
        commit_calls = {"n": 0}

        async def _delayed_stash(*_args, **_kwargs):
            commit_calls["n"] += 1
            if commit_calls["n"] == 1:
                inst._slicer_print_options[file_path.name] = {
                    "command": "project_file",
                    "file": file_path.name,
                    "nozzle_mapping": [0, 16, -1, -1],
                    "timelapse": False,
                }
            return None

        mock_db.commit = AsyncMock(side_effect=_delayed_stash)

        mock_archive = MagicMock()
        mock_archive.id = 1
        mock_archive.print_name = "test"

        with (
            patch(
                "backend.app.api.routes.settings.get_setting",
                new_callable=AsyncMock,
                return_value=None,
            ),
            patch(
                "backend.app.services.archive.ArchiveService.archive_print",
                new_callable=AsyncMock,
                return_value=mock_archive,
            ),
            patch(
                "backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT",
                0.05,
            ),
        ):
            await inst._add_to_print_queue(file_path, "192.168.1.100")

        # Queue item INSERTed with defaults (wait timed out, no slicer_opts).
        assert len(added_items) == 1
        # But the post-commit pop caught the late stash and applied the
        # slicer nozzle_mapping via _restamp's UPDATE.
        update_call = mock_db.execute.await_args_list[2]
        update_stmt = update_call.args[0]
        compiled = update_stmt.compile(compile_kwargs={"literal_binds": False})
        params = dict(compiled.params)
        assert _json.loads(params["nozzle_mapping"]) == [0, 16, -1, -1]
        assert params["timelapse"] is False
        # _recent_queue_items entry was consumed by the post-commit
        # _restamp call.
        assert file_path.name not in inst._recent_queue_items
        # And the stash is empty.
        assert file_path.name not in inst._slicer_print_options

    @pytest.mark.asyncio
    async def test_on_print_command_late_mqtt_skips_already_dispatched_item(self, tmp_path):
        """Once the scheduler has picked the queue item up (status != pending),
        the retroactive patch is a no-op — racing the dispatcher would be
        unsafe.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        mock_db = AsyncMock()
        # The eligible-pending SELECT returns nothing — item is no longer pending.
        empty_result = MagicMock()
        empty_result.all = MagicMock(return_value=[])
        mock_db.execute = AsyncMock(return_value=empty_result)
        mock_db.commit = AsyncMock()

        mock_session_factory = MagicMock()
        mock_session_ctx = AsyncMock()
        mock_session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        mock_session_ctx.__aexit__ = AsyncMock(return_value=False)
        mock_session_factory.return_value = mock_session_ctx

        inst = VirtualPrinterInstance(
            vp_id=98,
            name="LateMQTTDispatched",
            mode="queue",
            model="O1C2",
            access_code="12345678",
            serial_suffix="391800098",
            base_dir=tmp_path,
            session_factory=mock_session_factory,
        )
        inst._mqtt = MagicMock()
        # Pre-seed the recent-queue dict — pretend _add_to_print_queue just
        # committed item id 42.
        inst._recent_queue_items["test.3mf"] = ([42], 1_000_000.0)
        # Drive _restamp via on_print_command on the late-MQTT path.
        with patch("backend.app.services.virtual_printer.manager.time.monotonic", return_value=1_000_001.0):
            await inst.on_print_command(
                "test.3mf",
                {
                    "command": "project_file",
                    "file": "test.3mf",
                    "nozzle_mapping": [16, -1],
                },
            )
        # No UPDATE was issued — only the eligibility SELECT ran.
        assert mock_db.execute.await_count == 1
        mock_db.commit.assert_not_awaited()
        assert "test.3mf" not in inst._recent_queue_items


class TestExtractSlicerAmsMappingJson:
    """Unit tests for `_extract_slicer_ams_mapping_json`, the pure-function
    parser that pulls the slicer's own live-resolved AMS-slot pick out of a
    captured project_file MQTT payload (see docstring in manager.py)."""

    def _extract(self, data):
        from backend.app.services.virtual_printer.manager import _extract_slicer_ams_mapping_json

        return _extract_slicer_ams_mapping_json(data, "[test]")

    def test_missing_field_returns_none(self):
        assert self._extract({}) is None

    def test_valid_int_list_returns_json(self):
        import json as _json

        result = self._extract({"ams_mapping": [4, -1, 12, -1]})
        assert result is not None
        assert _json.loads(result) == [4, -1, 12, -1]

    def test_stringified_json_is_parsed(self):
        import json as _json

        result = self._extract({"ams_mapping": "[0, 1, 2]"})
        assert result is not None
        assert _json.loads(result) == [0, 1, 2]

    def test_unparseable_string_returns_none(self):
        assert self._extract({"ams_mapping": "not json"}) is None

    def test_non_list_value_returns_none(self):
        assert self._extract({"ams_mapping": 42}) is None

    def test_empty_list_returns_none(self):
        assert self._extract({"ams_mapping": []}) is None

    def test_non_int_entries_return_none(self):
        assert self._extract({"ams_mapping": [1, "two", 3]}) is None

    def test_2589_all_unresolved_sentinel_returns_none(self):
        """#2589: every slot -1 means the slicer's own resolution race
        lost — never trust this over a fresh live computation."""
        assert self._extract({"ams_mapping": [-1, -1, -1, -1]}) is None

    def test_partially_resolved_mapping_is_kept(self):
        import json as _json

        # Only some slots resolved is still meaningful — keep it.
        result = self._extract({"ams_mapping": [-1, 4, -1, -1]})
        assert result is not None
        assert _json.loads(result) == [-1, 4, -1, -1]


class TestVirtualPrinterManager:
    """Tests for VirtualPrinterManager orchestrator."""

    @pytest.fixture
    def manager(self):
        """Create a VirtualPrinterManager instance."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterManager

        return VirtualPrinterManager()

    def test_manager_starts_empty(self, manager):
        """Verify manager starts with no instances."""
        assert len(manager._instances) == 0
        assert manager.is_enabled is False

    def test_manager_get_status_empty(self, manager):
        """Verify get_status returns disabled state when no instances."""
        status = manager.get_status()
        assert status["enabled"] is False
        assert status["running"] is False
        assert status["mode"] == "archive"

    def test_manager_is_enabled_with_instance(self, manager, tmp_path):
        """Verify is_enabled is True when instances exist."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="Test",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        manager._instances[1] = inst
        assert manager.is_enabled is True

    @pytest.mark.asyncio
    async def test_manager_remove_instance_server(self, manager, tmp_path):
        """Verify remove_instance stops and removes a server-mode instance."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="Test",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        await manager.remove_instance(1)

        assert 1 not in manager._instances
        inst.stop_server.assert_called_once()

    @pytest.mark.asyncio
    async def test_manager_remove_instance_proxy(self, manager, tmp_path):
        """Verify remove_instance stops proxy-mode instance."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=2,
            name="Proxy",
            mode="proxy",
            model="C11",
            access_code="",
            serial_suffix="391800002",
            target_printer_ip="192.168.1.100",
            base_dir=tmp_path,
        )
        inst.stop_proxy = AsyncMock()
        manager._instances[2] = inst

        await manager.remove_instance(2)

        assert 2 not in manager._instances
        inst.stop_proxy.assert_called_once()

    def test_manager_get_status_with_instance(self, manager, tmp_path):
        """Verify legacy get_status returns first instance data."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="Bambuddy",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        mock_task = MagicMock(done=MagicMock(return_value=False))
        inst._tasks = [mock_task]
        inst._pending_files = {"file1.3mf": Path("/tmp/file1.3mf")}  # nosec B108
        manager._instances[1] = inst

        status = manager.get_status()
        assert status["enabled"] is True
        assert status["running"] is True
        assert status["mode"] == "archive"
        assert status["name"] == "Bambuddy"
        assert status["serial"] == "01S00A391800001"
        assert status["model"] == "C11"
        assert status["model_name"] == "P1P"
        assert status["pending_files"] == 1

    def test_manager_get_all_status(self, manager, tmp_path):
        """Verify get_all_status returns status for all instances."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        for i in range(1, 3):
            inst = VirtualPrinterInstance(
                vp_id=i,
                name=f"VP{i}",
                mode="archive",
                model="C11",
                access_code="12345678",
                serial_suffix=f"39180000{i}",
                base_dir=tmp_path,
            )
            manager._instances[i] = inst

        statuses = manager.get_all_status()
        assert len(statuses) == 2
        assert statuses[0]["name"] == "VP1"
        assert statuses[1]["name"] == "VP2"

    @pytest.mark.asyncio
    async def test_manager_stop_all(self, manager, tmp_path):
        """Verify stop_all removes all instances."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        for i in range(1, 3):
            inst = VirtualPrinterInstance(
                vp_id=i,
                name=f"VP{i}",
                mode="archive",
                model="C11",
                access_code="12345678",
                serial_suffix=f"39180000{i}",
                base_dir=tmp_path,
            )
            inst.stop_server = AsyncMock()
            manager._instances[i] = inst

        await manager.stop_all()
        assert len(manager._instances) == 0

    # ========================================================================
    # Tests for sync_from_db config change detection
    # ========================================================================

    def _make_db_vp(self, **overrides):
        """Create a mock VirtualPrinter DB object."""
        defaults = {
            "id": 1,
            "name": "TestVP",
            "enabled": True,
            "mode": "archive",
            "model": "C11",
            "access_code": "12345678",
            "serial_suffix": "391800001",
            "bind_ip": "",
            "remote_interface_ip": "",
            "target_printer_id": None,
            "auto_dispatch": True,
            "tailscale_disabled": True,  # Opt-in default (#1070 UX fix)
            "queue_force_color_match": False,  # default — must be explicit so MagicMock truthiness doesn't trip the change detector
            "save_ams_mapping": False,  # same reason as above
            "gcode_injection": False,  # same reason as above
            "position": 0,
        }
        defaults.update(overrides)
        vp = MagicMock()
        for k, v in defaults.items():
            setattr(vp, k, v)
        return vp

    def _setup_sync_mocks(self, manager, enabled_vps, tmp_path):
        """Wire up session_factory mock for sync_from_db."""
        mock_result = MagicMock()
        mock_result.scalars.return_value.all.return_value = enabled_vps

        mock_db = AsyncMock()
        mock_db.execute = AsyncMock(return_value=mock_result)
        mock_db.__aenter__ = AsyncMock(return_value=mock_db)
        mock_db.__aexit__ = AsyncMock(return_value=False)

        manager._session_factory = MagicMock(return_value=mock_db)
        manager._base_dir = tmp_path

    @pytest.mark.asyncio
    async def test_sync_from_db_restarts_on_mode_change(self, manager, tmp_path):
        """Verify sync_from_db restarts VP when mode changes."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        # DB says mode changed to "review"
        db_vp = self._make_db_vp(mode="review")
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            # Patch VirtualPrinterInstance to prevent actual start
            with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
                mock_new = MagicMock()
                mock_new.start_server = AsyncMock()
                MockInst.return_value = mock_new

                await manager.sync_from_db()

            mock_remove.assert_called_once_with(1)

    @pytest.mark.asyncio
    async def test_sync_from_db_restarts_on_access_code_change(self, manager, tmp_path):
        """Verify sync_from_db restarts VP when access_code changes."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        db_vp = self._make_db_vp(access_code="newcode99")
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
                mock_new = MagicMock()
                mock_new.start_server = AsyncMock()
                MockInst.return_value = mock_new

                await manager.sync_from_db()

            mock_remove.assert_called_once_with(1)

    @pytest.mark.asyncio
    async def test_sync_from_db_skips_unchanged_instance(self, manager, tmp_path):
        """Verify sync_from_db does NOT restart when config is identical."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        manager._instances[1] = inst

        # DB matches running config exactly
        db_vp = self._make_db_vp()
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            await manager.sync_from_db()

            mock_remove.assert_not_called()

    @pytest.mark.asyncio
    async def test_sync_from_db_restarts_on_bind_ip_change(self, manager, tmp_path):
        """Verify sync_from_db restarts VP when bind_ip changes."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            bind_ip="192.168.1.10",
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        db_vp = self._make_db_vp(bind_ip="192.168.1.20")
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
                mock_new = MagicMock()
                mock_new.start_server = AsyncMock()
                MockInst.return_value = mock_new

                await manager.sync_from_db()

            mock_remove.assert_called_once_with(1)

    @pytest.mark.asyncio
    async def test_sync_from_db_restarts_on_model_change(self, manager, tmp_path):
        """Verify sync_from_db restarts VP when model changes."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        db_vp = self._make_db_vp(model="C12")
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
                mock_new = MagicMock()
                mock_new.start_server = AsyncMock()
                MockInst.return_value = mock_new

                await manager.sync_from_db()

            mock_remove.assert_called_once_with(1)

    @pytest.mark.asyncio
    async def test_sync_from_db_does_not_restart_on_tailscale_toggle(self, manager, tmp_path):
        """Flipping tailscale_disabled is purely informational — must NOT trigger a restart.

        Cert provisioning was removed; the toggle only governs whether the VP card surfaces
        the host's Tailscale IP/FQDN to the user. No service needs to reload, so changing
        it through sync_from_db should leave any running instance untouched.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            tailscale_disabled=False,
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        db_vp = self._make_db_vp(tailscale_disabled=True)
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            await manager.sync_from_db()

        mock_remove.assert_not_called()

    @pytest.mark.asyncio
    async def test_sync_from_db_restarts_on_gcode_injection_toggle(self, manager, tmp_path):
        """Toggling gcode_injection in the DB must restart the running instance.

        Without this, the in-memory ``self.gcode_injection`` keeps its old value
        and ``_add_to_print_queue`` stamps the stale flag on every new queue
        item — so disabling injection in the UI silently has no effect until
        the process restarts.
        """
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=1,
            name="TestVP",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800001",
            gcode_injection=True,
            base_dir=tmp_path,
        )
        inst.stop_server = AsyncMock()
        manager._instances[1] = inst

        db_vp = self._make_db_vp(gcode_injection=False)
        self._setup_sync_mocks(manager, [db_vp], tmp_path)

        with patch.object(manager, "remove_instance", new_callable=AsyncMock) as mock_remove:
            with patch("backend.app.services.virtual_printer.manager.VirtualPrinterInstance") as MockInst:
                mock_new = MagicMock()
                mock_new.start_server = AsyncMock()
                MockInst.return_value = mock_new

                await manager.sync_from_db()

            mock_remove.assert_called_once_with(1)


class TestFTPSession:
    """Tests for FTP session handling."""

    @pytest.fixture
    def mock_reader(self):
        """Create a mock StreamReader."""
        reader = AsyncMock()
        return reader

    @pytest.fixture
    def mock_writer(self):
        """Create a mock StreamWriter."""
        writer = MagicMock()
        writer.get_extra_info = MagicMock(return_value=("192.168.1.100", 12345))
        writer.write = MagicMock()
        writer.drain = AsyncMock()
        writer.close = MagicMock()
        writer.wait_closed = AsyncMock()
        writer.is_closing = MagicMock(return_value=False)
        return writer

    @pytest.fixture
    def ssl_context(self):
        """Create a mock SSL context."""
        return MagicMock()

    @pytest.fixture
    def session(self, mock_reader, mock_writer, ssl_context, tmp_path):
        """Create an FTPSession instance."""
        from backend.app.services.virtual_printer.ftp_server import FTPSession

        return FTPSession(
            reader=mock_reader,
            writer=mock_writer,
            upload_dir=tmp_path,
            access_code="12345678",
            ssl_context=ssl_context,
            on_file_received=None,
        )

    # ========================================================================
    # Tests for authentication
    # ========================================================================

    @pytest.mark.asyncio
    async def test_user_command_accepts_bblp(self, session):
        """Verify USER command accepts bblp user."""
        await session.cmd_USER("bblp")

        assert session.username == "bblp"

    @pytest.mark.asyncio
    async def test_pass_command_authenticates(self, session):
        """Verify PASS command authenticates with correct code."""
        session.username = "bblp"

        await session.cmd_PASS("12345678")

        assert session.authenticated is True

    @pytest.mark.asyncio
    async def test_pass_command_rejects_wrong_code(self, session):
        """Verify PASS command rejects wrong access code."""
        session.username = "bblp"

        await session.cmd_PASS("wrongcode")

        assert session.authenticated is False

    # ========================================================================
    # Tests for FTP commands
    # ========================================================================

    @pytest.mark.asyncio
    async def test_syst_command(self, session):
        """Verify SYST returns UNIX type."""
        await session.cmd_SYST("")

        session.writer.write.assert_called()
        call_args = session.writer.write.call_args[0][0].decode()
        assert "215" in call_args
        assert "UNIX" in call_args

    @pytest.mark.asyncio
    async def test_pwd_command_requires_auth(self, session):
        """Verify PWD requires authentication."""
        session.authenticated = False

        await session.cmd_PWD("")

        call_args = session.writer.write.call_args[0][0].decode()
        assert "530" in call_args

    @pytest.mark.asyncio
    async def test_pwd_command_when_authenticated(self, session):
        """Verify PWD returns root directory when authenticated."""
        session.authenticated = True

        await session.cmd_PWD("")

        call_args = session.writer.write.call_args[0][0].decode()
        assert "257" in call_args

    @pytest.mark.asyncio
    async def test_type_command_sets_binary(self, session):
        """Verify TYPE I sets binary mode."""
        session.authenticated = True

        await session.cmd_TYPE("I")

        assert session.transfer_type == "I"

    @pytest.mark.asyncio
    async def test_pbsz_command(self, session):
        """Verify PBSZ returns success."""
        await session.cmd_PBSZ("0")

        call_args = session.writer.write.call_args[0][0].decode()
        assert "200" in call_args

    @pytest.mark.asyncio
    async def test_prot_command_accepts_p(self, session):
        """Verify PROT P is accepted."""
        await session.cmd_PROT("P")

        call_args = session.writer.write.call_args[0][0].decode()
        assert "200" in call_args

    @pytest.mark.asyncio
    async def test_quit_command(self, session):
        """Verify QUIT sends goodbye and raises CancelledError."""
        with pytest.raises(asyncio.CancelledError):
            await session.cmd_QUIT("")


class TestSSDPServer:
    """Tests for Virtual Printer SSDP server."""

    @pytest.fixture
    def ssdp_server(self):
        """Create a VirtualPrinterSSDPServer instance."""
        from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer

        return VirtualPrinterSSDPServer(
            serial="TEST123",
            name="TestPrinter",
            model="BL-P001",
        )

    # ========================================================================
    # Tests for SSDP response
    # ========================================================================

    def test_build_notify_message(self, ssdp_server):
        """Verify NOTIFY packet contains required headers."""
        # Set a known IP for testing
        ssdp_server._local_ip = "192.168.1.100"

        message = ssdp_server._build_notify_message()

        assert b"NOTIFY" in message
        assert b"DevName.bambu.com: TestPrinter" in message
        assert b"USN: TEST123" in message

    def test_build_response_message(self, ssdp_server):
        """Verify response packet contains required headers."""
        # Set a known IP for testing
        ssdp_server._local_ip = "192.168.1.100"

        message = ssdp_server._build_response_message()

        assert b"HTTP/1.1 200 OK" in message
        assert b"DevName.bambu.com: TestPrinter" in message
        assert b"USN: TEST123" in message

    def test_ssdp_server_uses_correct_model(self, ssdp_server):
        """Verify SSDP server uses the provided model."""
        ssdp_server._local_ip = "192.168.1.100"

        message = ssdp_server._build_notify_message()

        assert b"DevModel.bambu.com: BL-P001" in message

    # ========================================================================
    # Tests for advertise_ip parameter
    # ========================================================================

    def test_advertise_ip_sets_local_ip(self):
        """Verify advertise_ip overrides auto-detection."""
        from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer

        server = VirtualPrinterSSDPServer(
            serial="TEST123",
            name="TestPrinter",
            model="BL-P001",
            advertise_ip="10.0.0.50",
        )

        assert server._local_ip == "10.0.0.50"

    def test_advertise_ip_empty_string_uses_auto_detect(self):
        """Verify empty advertise_ip falls back to auto-detection."""
        from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer

        server = VirtualPrinterSSDPServer(
            serial="TEST123",
            name="TestPrinter",
            model="BL-P001",
            advertise_ip="",
        )

        assert server._local_ip is None

    def test_advertise_ip_in_notify_message(self):
        """Verify NOTIFY message uses the advertise_ip."""
        from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer

        server = VirtualPrinterSSDPServer(
            serial="TEST123",
            name="TestPrinter",
            model="BL-P001",
            advertise_ip="10.0.0.50",
        )

        message = server._build_notify_message()

        assert b"Location: 10.0.0.50" in message

    def test_advertise_ip_in_response_message(self):
        """Verify M-SEARCH response uses the advertise_ip."""
        from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer

        server = VirtualPrinterSSDPServer(
            serial="TEST123",
            name="TestPrinter",
            model="BL-P001",
            advertise_ip="10.0.0.50",
        )

        message = server._build_response_message()

        assert b"Location: 10.0.0.50" in message

    def test_default_no_advertise_ip(self):
        """Verify default constructor has None local_ip (auto-detect)."""
        from backend.app.services.virtual_printer.ssdp_server import VirtualPrinterSSDPServer

        server = VirtualPrinterSSDPServer()

        assert server._local_ip is None


class TestCertificateService:
    """Tests for TLS certificate generation."""

    @pytest.fixture
    def cert_service(self, tmp_path):
        """Create a CertificateService instance."""
        from backend.app.services.virtual_printer.certificate import CertificateService

        return CertificateService(cert_dir=tmp_path, serial="TEST123")

    def test_generate_certificates(self, cert_service, tmp_path):
        """Verify certificates are generated correctly."""
        cert_path, key_path = cert_service.generate_certificates()

        assert cert_path.exists()
        assert key_path.exists()

        # Verify certificate content
        cert_content = cert_path.read_text()
        assert "BEGIN CERTIFICATE" in cert_content

        key_content = key_path.read_text()
        assert "BEGIN" in key_content and "KEY" in key_content

    def test_certificates_reused_if_exist(self, cert_service):
        """Verify existing certificates are reused."""
        # First generation
        cert_path1, key_path1 = cert_service.generate_certificates()
        mtime1 = cert_path1.stat().st_mtime

        # Second call should reuse (via ensure_certificates)
        cert_path2, key_path2 = cert_service.ensure_certificates()
        mtime2 = cert_path2.stat().st_mtime

        assert mtime1 == mtime2  # File wasn't regenerated

    def test_delete_certificates(self, cert_service):
        """Verify certificates can be deleted."""
        cert_service.generate_certificates()

        assert cert_service.cert_path.exists()
        assert cert_service.key_path.exists()

        cert_service.delete_certificates()

        assert not cert_service.cert_path.exists()
        assert not cert_service.key_path.exists()

    def test_ensure_creates_if_not_exist(self, cert_service):
        """Verify ensure_certificates generates if not existing."""
        assert not cert_service.cert_path.exists()

        cert_path, key_path = cert_service.ensure_certificates()

        assert cert_path.exists()
        assert key_path.exists()


class TestBindServer:
    """Tests for BindServer (port 3002 bind/detect protocol)."""

    @pytest.fixture
    def bind_server(self):
        """Create a BindServer instance."""
        from backend.app.services.virtual_printer.bind_server import BindServer

        return BindServer(
            serial="09400A391800001",
            model="O1D",
            name="Bambuddy",
        )

    def test_build_frame(self, bind_server):
        """Verify frame building produces correct format."""
        payload = {"login": {"command": "detect"}}
        frame = bind_server._build_frame(payload)

        # Header: 0xA5A5
        assert frame[:2] == b"\xa5\xa5"
        # Trailer: 0xA7A7
        assert frame[-2:] == b"\xa7\xa7"
        # Length field is total message size (LE uint16)
        import struct

        total_len = struct.unpack_from("<H", frame, 2)[0]
        assert total_len == len(frame)
        # JSON payload is between header and trailer
        import json

        json_bytes = frame[4:-2]
        parsed = json.loads(json_bytes)
        assert parsed == payload

    def test_parse_frame_valid(self, bind_server):
        """Verify valid frame parsing extracts JSON correctly."""
        import json
        import struct

        payload = {"login": {"command": "detect", "sequence_id": "20000"}}
        json_bytes = json.dumps(payload, separators=(",", ":")).encode()
        total_len = 4 + len(json_bytes) + 2
        frame = b"\xa5\xa5" + struct.pack("<H", total_len) + json_bytes + b"\xa7\xa7"

        result = bind_server._parse_frame(frame)

        assert result is not None
        assert result["login"]["command"] == "detect"
        assert result["login"]["sequence_id"] == "20000"

    def test_parse_frame_invalid_header(self, bind_server):
        """Verify invalid header returns None."""
        result = bind_server._parse_frame(b"\xbb\xbb\x06\x00{}\xa7\xa7")
        assert result is None

    def test_parse_frame_invalid_trailer(self, bind_server):
        """Verify invalid trailer returns None."""
        result = bind_server._parse_frame(b"\xa5\xa5\x06\x00{}\xbb\xbb")
        assert result is None

    def test_parse_frame_too_short(self, bind_server):
        """Verify short data returns None."""
        result = bind_server._parse_frame(b"\xa5\xa5\x00")
        assert result is None

    def test_parse_frame_invalid_json(self, bind_server):
        """Verify invalid JSON returns None."""
        import struct

        bad_json = b"not json"
        total_len = 4 + len(bad_json) + 2
        frame = b"\xa5\xa5" + struct.pack("<H", total_len) + bad_json + b"\xa7\xa7"
        result = bind_server._parse_frame(frame)
        assert result is None

    def test_build_frame_roundtrip(self, bind_server):
        """Verify build_frame output can be parsed back."""
        payload = {
            "login": {
                "bind": "free",
                "command": "detect",
                "connect": "lan",
                "dev_cap": 1,
                "id": "09400A391800001",
                "model": "O1D",
                "name": "Bambuddy",
                "sequence_id": 3021,
                "version": "01.00.00.00",
            }
        }
        frame = bind_server._build_frame(payload)
        parsed = bind_server._parse_frame(frame)

        assert parsed is not None
        assert parsed["login"]["id"] == "09400A391800001"
        assert parsed["login"]["model"] == "O1D"
        assert parsed["login"]["name"] == "Bambuddy"
        assert parsed["login"]["bind"] == "free"

    def test_bind_server_stores_config(self, bind_server):
        """Verify bind server stores serial, model, name."""
        assert bind_server.serial == "09400A391800001"
        assert bind_server.model == "O1D"
        assert bind_server.name == "Bambuddy"
        assert bind_server.version == "01.00.00.00"

    def test_bind_server_custom_version(self):
        """Verify custom firmware version is stored."""
        from backend.app.services.virtual_printer.bind_server import BindServer

        server = BindServer(
            serial="TEST123",
            model="C13",
            name="Test",
            version="02.03.04.05",
        )
        assert server.version == "02.03.04.05"

    def test_bind_ports_constant(self):
        """Verify BIND_PORTS includes both 3000 and 3002 for slicer compatibility."""
        from backend.app.services.virtual_printer.bind_server import BIND_PORTS

        assert 3000 in BIND_PORTS
        assert 3002 in BIND_PORTS

    def test_bind_server_initializes_empty_servers_list(self, bind_server):
        """Verify bind server starts with empty servers list."""
        assert bind_server._servers == []
        assert bind_server._running is False


class TestSlicerProxyManager:
    """Tests for SlicerProxyManager (proxy mode)."""

    @pytest.fixture
    def proxy_manager(self, tmp_path):
        """Create a SlicerProxyManager instance."""
        from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager

        # Create dummy cert files
        cert_path = tmp_path / "cert.pem"
        key_path = tmp_path / "key.pem"
        cert_path.write_text("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----")
        # Split string to avoid pre-commit hook false positive on test data
        key_path.write_text("-----BEGIN " + "PRIVATE KEY-----\ntest\n-----END " + "PRIVATE KEY-----")

        return SlicerProxyManager(
            target_host="192.168.1.100",
            cert_path=cert_path,
            key_path=key_path,
        )

    def test_proxy_manager_initializes_ports(self, proxy_manager):
        """Verify proxy manager has correct port constants."""
        # FTP proxy uses privileged port 990 to match what Bambu Studio expects
        assert proxy_manager.LOCAL_FTP_PORT == 990
        assert proxy_manager.LOCAL_MQTT_PORT == 8883
        assert proxy_manager.PRINTER_FTP_PORT == 990
        assert proxy_manager.PRINTER_MQTT_PORT == 8883
        assert proxy_manager.PRINTER_FILE_TRANSFER_PORT == 6000
        assert proxy_manager.PRINTER_RTSP_PORT == 322
        # Auxiliary ports: undocumented proprietary ports for A1/P1S etc.
        assert proxy_manager.PRINTER_AUX_PORTS == [2024, 2025, 2026]
        # Bind ports: both 3000 and 3002 for slicer compatibility
        assert proxy_manager.PRINTER_BIND_PORTS == [3000, 3002]
        # FTP data port range for transparent EPSV proxying
        assert proxy_manager.FTP_DATA_PORT_MIN == 50000
        assert proxy_manager.FTP_DATA_PORT_MAX == 50100

    def test_proxy_manager_stores_target_host(self, proxy_manager):
        """Verify proxy manager stores target host."""
        assert proxy_manager.target_host == "192.168.1.100"

    def test_get_status_before_start(self, proxy_manager):
        """Verify get_status returns zeros before start."""
        status = proxy_manager.get_status()

        assert status["running"] is False
        assert status["ftp_connections"] == 0
        assert status["mqtt_connections"] == 0

    @pytest.mark.asyncio
    async def test_proxy_start_creates_transparent_proxies(self, tmp_path):
        """Verify start() uses TCPProxy for FTP/FileTransfer/RTSP and TLSProxy only for MQTT.

        The transparent proxy architecture preserves end-to-end TLS between
        slicer and printer for all protocols except MQTT, which must be
        TLS-terminated to rewrite the printer's IP in MQTT payloads.
        """
        from unittest.mock import patch

        from backend.app.services.virtual_printer.tcp_proxy import (
            SlicerProxyManager,
            TCPProxy,
            TLSProxy,
        )

        cert_path = tmp_path / "cert.pem"
        key_path = tmp_path / "key.pem"
        cert_path.write_text("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----")
        key_path.write_text("-----BEGIN " + "PRIVATE KEY-----\ntest\n-----END " + "PRIVATE KEY-----")

        mgr = SlicerProxyManager(
            target_host="192.168.1.100",
            cert_path=cert_path,
            key_path=key_path,
            bind_address="10.0.0.1",
        )

        # Mock asyncio.create_task and asyncio.gather to prevent actual
        # server start. Close every coroutine handed to gather — otherwise
        # the ~110 run_with_logging() coros built inside start() are
        # garbage-collected unfinalized and surface later as
        # PytestUnraisableExceptionWarning at random in other tests.
        async def _close_pending(*coros, **_):
            for c in coros:
                if asyncio.iscoroutine(c):
                    c.close()

        with (
            patch("asyncio.create_task") as mock_create_task,
            patch("asyncio.gather", side_effect=_close_pending),
            patch.object(SlicerProxyManager, "_log_activity"),
        ):
            mock_create_task.return_value = MagicMock()
            await mgr.start()

        # FTP, FileTransfer, RTSP should be TCPProxy (transparent)
        assert isinstance(mgr._ftp_proxy, TCPProxy), "FTP should be TCPProxy (transparent)"
        assert isinstance(mgr._file_transfer_proxy, TCPProxy), "FileTransfer should be TCPProxy"
        assert isinstance(mgr._rtsp_proxy, TCPProxy), "RTSP should be TCPProxy"

        # MQTT should be TLSProxy (TLS-terminated for IP rewriting)
        assert isinstance(mgr._mqtt_proxy, TLSProxy), "MQTT should be TLSProxy (TLS-terminated)"

        # Auxiliary ports (2024-2026) should be TCPProxy (transparent)
        assert len(mgr._aux_proxies) == 3, "Should have 3 aux port proxies"
        for ap in mgr._aux_proxies:
            assert isinstance(ap, TCPProxy), "Aux proxies should be TCPProxy"
        assert mgr._aux_proxies[0].listen_port == 2024
        assert mgr._aux_proxies[0].target_port == 2024
        assert mgr._aux_proxies[2].listen_port == 2026

        # FTP data ports should be pre-created as TCPProxy instances
        assert len(mgr._ftp_data_proxies) == 101  # 50000-50100 inclusive
        for dp in mgr._ftp_data_proxies:
            assert isinstance(dp, TCPProxy), "FTP data proxies should be TCPProxy"

        # Verify FTP data proxies target the same port on the printer
        first_dp = mgr._ftp_data_proxies[0]
        assert first_dp.listen_port == 50000
        assert first_dp.target_port == 50000
        assert first_dp.target_host == "192.168.1.100"

        last_dp = mgr._ftp_data_proxies[-1]
        assert last_dp.listen_port == 50100
        assert last_dp.target_port == 50100

    def test_proxy_manager_mqtt_has_ip_rewriting(self, tmp_path):
        """Verify MQTT proxy is configured with IP rewriting when bind_address is set."""
        from backend.app.services.virtual_printer.tcp_proxy import SlicerProxyManager

        cert_path = tmp_path / "cert.pem"
        key_path = tmp_path / "key.pem"
        cert_path.write_text("-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----")
        key_path.write_text("-----BEGIN " + "PRIVATE KEY-----\ntest\n-----END " + "PRIVATE KEY-----")

        mgr = SlicerProxyManager(
            target_host="192.168.1.100",
            cert_path=cert_path,
            key_path=key_path,
            bind_address="10.0.0.1",
        )

        # Before start, proxies are None — verify constructor stores rewrite config
        assert mgr.bind_address == "10.0.0.1"
        assert mgr.target_host == "192.168.1.100"


class TestSSDPProxy:
    """Tests for SSDPProxy (cross-network SSDP relay)."""

    @pytest.fixture
    def ssdp_proxy(self):
        """Create an SSDPProxy instance."""
        from backend.app.services.virtual_printer.ssdp_server import SSDPProxy

        return SSDPProxy(
            local_interface_ip="192.168.1.100",
            remote_interface_ip="10.0.0.100",
            target_printer_ip="192.168.1.50",
        )

    def test_ssdp_proxy_stores_interface_ips(self, ssdp_proxy):
        """Verify SSDPProxy stores interface IPs correctly."""
        assert ssdp_proxy.local_interface_ip == "192.168.1.100"
        assert ssdp_proxy.remote_interface_ip == "10.0.0.100"
        assert ssdp_proxy.target_printer_ip == "192.168.1.50"

    def test_rewrite_ssdp_location(self, ssdp_proxy):
        """Verify SSDP Location header is rewritten to remote interface IP."""
        original_packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: TestPrinter\r\n\r\n"

        rewritten = ssdp_proxy._rewrite_ssdp(original_packet)

        # Location should be changed to remote interface IP
        assert b"Location: 10.0.0.100" in rewritten
        assert b"Location: 192.168.1.50" not in rewritten
        # Other headers should be preserved
        assert b"DevName.bambu.com: TestPrinter" in rewritten

    def test_rewrite_ssdp_location_case_insensitive(self, ssdp_proxy):
        """Verify SSDP Location rewrite is case insensitive."""
        original_packet = b"NOTIFY * HTTP/1.1\r\nlocation: 192.168.1.50\r\n\r\n"

        rewritten = ssdp_proxy._rewrite_ssdp(original_packet)

        assert b"10.0.0.100" in rewritten

    def test_rewrite_ssdp_location_no_match(self, ssdp_proxy):
        """Verify packet without Location header is returned unchanged."""
        original_packet = b"NOTIFY * HTTP/1.1\r\nDevName.bambu.com: Test\r\n\r\n"

        rewritten = ssdp_proxy._rewrite_ssdp(original_packet)

        # No Location header, but _rewrite_ssdp logs a warning and returns as-is
        assert b"DevName.bambu.com: Test" in rewritten

    def test_parse_ssdp_message(self, ssdp_proxy):
        """Verify SSDP message parsing extracts headers."""
        packet = (
            b"NOTIFY * HTTP/1.1\r\n"
            b"Location: 192.168.1.50\r\n"
            b"DevName.bambu.com: TestPrinter\r\n"
            b"DevModel.bambu.com: BL-P001\r\n"
            b"\r\n"
        )

        headers = ssdp_proxy._parse_ssdp_message(packet)

        assert headers["location"] == "192.168.1.50"
        assert headers["devname.bambu.com"] == "TestPrinter"
        assert headers["devmodel.bambu.com"] == "BL-P001"


class TestVirtualPrinterManagerDirectories:
    """Tests for VirtualPrinterManager directory management."""

    def test_ensure_base_directories_creates_subdirs(self, tmp_path):
        """Verify _ensure_base_directories creates required base directories."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterManager

        manager = VirtualPrinterManager()
        manager._base_dir = tmp_path / "virtual_printer"
        manager._ensure_base_directories()

        assert (tmp_path / "virtual_printer").exists()
        assert (tmp_path / "virtual_printer" / "uploads").exists()
        assert (tmp_path / "virtual_printer" / "certs").exists()

    def test_ensure_base_directories_handles_permission_error(self, tmp_path, caplog):
        """Verify _ensure_base_directories logs error on permission failure."""
        import logging

        from backend.app.services.virtual_printer.manager import VirtualPrinterManager

        manager = VirtualPrinterManager()
        vp_dir = tmp_path / "virtual_printer"
        manager._base_dir = vp_dir

        original_mkdir = type(vp_dir).mkdir

        def mock_mkdir(self, *args, **kwargs):
            if "virtual_printer" in str(self):
                raise PermissionError("Permission denied")
            return original_mkdir(self, *args, **kwargs)

        with caplog.at_level(logging.ERROR), patch.object(type(vp_dir), "mkdir", mock_mkdir):
            manager._ensure_base_directories()
            assert "Permission denied" in caplog.text

    def test_instance_creates_per_vp_directories(self, tmp_path):
        """Verify VirtualPrinterInstance creates per-VP upload and cert dirs."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        VirtualPrinterInstance(
            vp_id=42,
            name="Test",
            mode="archive",
            model="C11",
            access_code="12345678",
            serial_suffix="391800042",
            base_dir=tmp_path,
        )

        assert (tmp_path / "uploads" / "42").exists()
        assert (tmp_path / "uploads" / "42" / "cache").exists()
        assert (tmp_path / "certs" / "42").exists()


class TestVirtualPrinterCameraPassthrough:
    """Tests for the non-proxy VP camera pass-through port selection (#1868).

    The camera port must follow the TARGET printer's model, not the VP's
    spoofed model: RTSP models (X1/X2/H2/P2S) use 322, chamber-image models
    (A1/P1) use 6000. Before the fix, the manager hardcoded 322, so P1S /
    A1 targets got a 322 listener with no upstream and BambuStudio /
    OrcaSlicer Liveview failed with error `[2:-10061]` (the reporter's
    symptom in #1868).
    """

    @staticmethod
    def _patch_start_server(monkeypatch):
        """Patch every non-camera constructor start_server touches so the
        test only exercises the camera-port branch. Returns the list that
        ``TCPProxy`` calls get captured into."""
        from backend.app.services.virtual_printer import manager as vp_manager

        tcp_calls: list[dict] = []

        class FakeTCPProxy:
            def __init__(self, **kwargs):
                tcp_calls.append(kwargs)
                self.kwargs = kwargs

            async def start(self):
                return None

            async def stop(self):
                return None

        # Camera pass-through is the only TCPProxy start_server constructs.
        monkeypatch.setattr(vp_manager, "TCPProxy", FakeTCPProxy)

        # No-op every other service constructor + resolve_cert_and_advertise.
        # start_server() awaits `.ready.wait()` on FTP/MQTT/Bind/SSDP after
        # spawning them, so each fake instance must carry an already-set
        # asyncio.Event as `.ready` — a plain MagicMock returns another
        # MagicMock for `.wait()`, which `asyncio.gather` then rejects with
        # `TypeError: An asyncio.Future, a coroutine or an awaitable is
        # required`.
        def _service_factory():
            def make(*args, **kwargs):
                inst = MagicMock()
                inst.start = AsyncMock(return_value=None)
                inst.stop = AsyncMock(return_value=None)
                ready = asyncio.Event()
                ready.set()
                inst.ready = ready
                return inst

            return make

        for name in (
            "VirtualPrinterFTPServer",
            "SimpleMQTTServer",
            "MQTTBridge",
            "BindServer",
            "VirtualPrinterSSDPServer",
            "SSDPProxy",
        ):
            monkeypatch.setattr(vp_manager, name, _service_factory())

        return tcp_calls

    async def _run_start_server(
        self,
        tmp_path,
        monkeypatch,
        *,
        target_model: str,
        target_ip: str = "192.168.1.100",
    ) -> list[dict]:
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        tcp_calls = self._patch_start_server(monkeypatch)

        instance = VirtualPrinterInstance(
            vp_id=99,
            name="CamTest",
            mode="archive",
            model="BL-P001",  # VP's spoofed identity — irrelevant here
            access_code="12345678",
            serial_suffix="391800099",
            target_printer_id=7,
            base_dir=tmp_path,
        )

        # printer_manager stub returns a client with the model + ip we want.
        client = MagicMock()
        client.ip_address = target_ip
        client.model = target_model
        printer_manager = MagicMock()
        printer_manager.get_client.return_value = client
        instance._printer_manager = printer_manager

        # Cert / advertise resolution — start_server calls this early. Patch
        # to a fixed tuple so no filesystem I/O is required.
        monkeypatch.setattr(
            instance,
            "_resolve_cert_and_advertise",
            lambda: (Path("/tmp/cert.pem"), Path("/tmp/key.pem"), "192.168.1.1"),  # nosec B108
        )

        try:
            await instance.start_server()
        finally:
            for task in instance._tasks:
                task.cancel()
            await asyncio.gather(*instance._tasks, return_exceptions=True)

        return tcp_calls

    @pytest.mark.asyncio
    async def test_rtsp_model_p2s_opens_port_322(self, tmp_path, monkeypatch):
        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="P2S")
        assert any(c["listen_port"] == 322 and c["target_port"] == 322 for c in calls), (
            f"Expected 322 pass-through for RTSP model P2S, got {calls}"
        )

    @pytest.mark.asyncio
    async def test_chamber_image_model_p1s_opens_port_6000(self, tmp_path, monkeypatch):
        """#1868 regression guard: P1S target must expose 6000, not 322."""
        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="P1S")
        assert any(c["listen_port"] == 6000 and c["target_port"] == 6000 for c in calls), (
            f"Expected 6000 pass-through for chamber-image model P1S (#1868), got {calls}"
        )
        assert not any(c["listen_port"] == 322 for c in calls), f"P1S should NOT get a 322 listener, got {calls}"

    @pytest.mark.asyncio
    async def test_chamber_image_model_a1_opens_port_6000(self, tmp_path, monkeypatch):
        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="A1")
        assert any(c["listen_port"] == 6000 and c["target_port"] == 6000 for c in calls)

    @pytest.mark.asyncio
    async def test_rtsp_model_x1c_opens_port_322(self, tmp_path, monkeypatch):
        calls = await self._run_start_server(tmp_path, monkeypatch, target_model="X1C")
        assert any(c["listen_port"] == 322 and c["target_port"] == 322 for c in calls)


class TestVirtualPrinterInstanceProxyMode:
    """Tests for VirtualPrinterInstance proxy mode."""

    @pytest.fixture
    def proxy_instance(self, tmp_path):
        """Create a proxy-mode VirtualPrinterInstance."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        return VirtualPrinterInstance(
            vp_id=10,
            name="ProxyTest",
            mode="proxy",
            model="C11",
            access_code="",
            serial_suffix="391800010",
            target_printer_ip="192.168.1.100",
            target_printer_serial="01P00A000000001",
            base_dir=tmp_path,
        )

    def test_proxy_instance_properties(self, proxy_instance):
        """Verify proxy instance stores config correctly."""
        assert proxy_instance.is_proxy is True
        assert proxy_instance.mode == "proxy"
        assert proxy_instance.target_printer_ip == "192.168.1.100"
        assert proxy_instance.target_printer_serial == "01P00A000000001"

    def test_proxy_instance_does_not_require_access_code(self, proxy_instance):
        """Verify proxy mode can have empty access code."""
        assert proxy_instance.access_code == ""

    def test_get_status_proxy_includes_proxy_fields(self, proxy_instance):
        """Verify get_status includes proxy fields when proxy is active."""
        mock_proxy = MagicMock()
        mock_proxy.get_status.return_value = {
            "running": True,
            "ftp_port": 990,
            "mqtt_port": 8883,
            "ftp_connections": 1,
            "mqtt_connections": 2,
            "target_host": "192.168.1.100",
        }
        proxy_instance._proxy = mock_proxy

        status = proxy_instance.get_status()
        assert "proxy" in status
        assert status["proxy"]["ftp_port"] == 990
        assert status["proxy"]["mqtt_connections"] == 2

    def test_proxy_instance_stores_remote_interface(self, tmp_path):
        """Verify proxy instance stores remote_interface_ip."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=11,
            name="Proxy2",
            mode="proxy",
            model="C11",
            access_code="",
            serial_suffix="391800011",
            target_printer_ip="192.168.1.100",
            remote_interface_ip="10.0.0.50",
            base_dir=tmp_path,
        )
        assert inst.remote_interface_ip == "10.0.0.50"


class TestVirtualPrinterInstanceIPOverride:
    """Tests for remote_interface_ip and bind_ip on VirtualPrinterInstance."""

    @pytest.fixture
    def instance_with_remote_ip(self, tmp_path):
        """Create an instance with remote_interface_ip set."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        return VirtualPrinterInstance(
            vp_id=20,
            name="IPTest",
            mode="archive",
            model="BL-P001",
            access_code="12345678",
            serial_suffix="391800020",
            bind_ip="192.168.1.50",
            remote_interface_ip="10.0.0.50",
            base_dir=tmp_path,
        )

    def test_instance_stores_bind_ip(self, instance_with_remote_ip):
        """Verify bind_ip is stored."""
        assert instance_with_remote_ip.bind_ip == "192.168.1.50"

    def test_instance_stores_remote_interface_ip(self, instance_with_remote_ip):
        """Verify remote_interface_ip is stored."""
        assert instance_with_remote_ip.remote_interface_ip == "10.0.0.50"

    def test_generate_certificates_includes_remote_and_bind_ip(self, instance_with_remote_ip):
        """Verify generate_certificates passes remote_interface_ip and bind_ip as SANs."""
        with (
            patch.object(instance_with_remote_ip._cert_service, "delete_printer_certificate"),
            patch.object(
                instance_with_remote_ip._cert_service,
                "generate_certificates",
                return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")),  # nosec B108
            ) as mock_gen,
        ):
            instance_with_remote_ip.generate_certificates()
            mock_gen.assert_called_once_with(additional_ips=["10.0.0.50", "192.168.1.50"])

    def test_generate_certificates_no_remote_ip(self, tmp_path):
        """Verify generate_certificates passes only bind_ip when no remote_interface_ip."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=21,
            name="NoRemote",
            mode="archive",
            model="BL-P001",
            access_code="12345678",
            serial_suffix="391800021",
            bind_ip="192.168.1.50",
            base_dir=tmp_path,
        )

        with (
            patch.object(inst._cert_service, "delete_printer_certificate"),
            patch.object(
                inst._cert_service,
                "generate_certificates",
                return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")),  # nosec B108
            ) as mock_gen,
        ):
            inst.generate_certificates()
            mock_gen.assert_called_once_with(additional_ips=["192.168.1.50"])

    def test_generate_certificates_no_ips(self, tmp_path):
        """Verify generate_certificates passes None when no IPs configured."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=22,
            name="NoIPs",
            mode="archive",
            model="BL-P001",
            access_code="12345678",
            serial_suffix="391800022",
            base_dir=tmp_path,
        )

        with (
            patch.object(inst._cert_service, "delete_printer_certificate"),
            patch.object(
                inst._cert_service,
                "generate_certificates",
                return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")),  # nosec B108
            ) as mock_gen,
        ):
            inst.generate_certificates()
            mock_gen.assert_called_once_with(additional_ips=None)


class TestBindServer:
    """Tests for the BindServer (port 3002 bind/detect protocol)."""

    @pytest.fixture
    def bind_server(self):
        """Create a BindServer instance."""
        from backend.app.services.virtual_printer.bind_server import BindServer

        return BindServer(
            serial="01S00C000000001",
            model="BL-P001",
            name="Bambuddy",
        )

    def test_build_frame(self, bind_server):
        """Verify frame format: 0xA5A5 + len(u16le) + JSON + 0xA7A7."""
        payload = {"login": {"command": "detect"}}
        frame = bind_server._build_frame(payload)

        assert frame[:2] == b"\xa5\xa5"
        assert frame[-2:] == b"\xa7\xa7"

        # Length field is total message size
        import struct

        total_len = struct.unpack_from("<H", frame, 2)[0]
        assert total_len == len(frame)

        # JSON payload is between header and trailer
        import json

        json_bytes = frame[4:-2]
        parsed = json.loads(json_bytes)
        assert parsed == payload

    def test_parse_frame_valid(self, bind_server):
        """Verify valid frame parsing."""
        frame = bind_server._build_frame({"login": {"command": "detect", "sequence_id": "20000"}})
        result = bind_server._parse_frame(frame)

        assert result is not None
        assert result["login"]["command"] == "detect"
        assert result["login"]["sequence_id"] == "20000"

    def test_parse_frame_invalid_header(self, bind_server):
        """Verify invalid header returns None."""
        frame = b"\xb5\xb5\x10\x00" + b'{"login":{}}' + b"\xa7\xa7"
        assert bind_server._parse_frame(frame) is None

    def test_parse_frame_invalid_trailer(self, bind_server):
        """Verify invalid trailer returns None."""
        frame = b"\xa5\xa5\x10\x00" + b'{"login":{}}' + b"\xb7\xb7"
        assert bind_server._parse_frame(frame) is None

    def test_parse_frame_too_short(self, bind_server):
        """Verify short data returns None."""
        assert bind_server._parse_frame(b"\xa5\xa5\x00") is None
        assert bind_server._parse_frame(b"") is None

    def test_parse_frame_invalid_json(self, bind_server):
        """Verify invalid JSON returns None."""
        import struct

        bad_json = b"not json"
        total_len = 4 + len(bad_json) + 2
        frame = b"\xa5\xa5" + struct.pack("<H", total_len) + bad_json + b"\xa7\xa7"
        assert bind_server._parse_frame(frame) is None

    def test_build_frame_roundtrip(self, bind_server):
        """Verify build then parse roundtrip."""
        original = {"login": {"bind": "free", "command": "detect", "id": "01S00C000000001"}}
        frame = bind_server._build_frame(original)
        parsed = bind_server._parse_frame(frame)
        assert parsed == original

    def test_bind_server_stores_config(self, bind_server):
        """Verify config is stored correctly."""
        assert bind_server.serial == "01S00C000000001"
        assert bind_server.model == "BL-P001"
        assert bind_server.name == "Bambuddy"
        assert bind_server.version == "01.00.00.00"

    def test_bind_server_custom_version(self):
        """Verify custom firmware version is stored."""
        from backend.app.services.virtual_printer.bind_server import BindServer

        server = BindServer(
            serial="01S00C000000001",
            model="BL-P001",
            name="Bambuddy",
            version="01.09.00.10",
        )
        assert server.version == "01.09.00.10"

    def test_bind_ports_includes_both(self):
        """Verify BIND_PORTS includes both 3000 and 3002 for slicer compatibility."""
        from backend.app.services.virtual_printer.bind_server import BIND_PORTS

        assert 3000 in BIND_PORTS
        assert 3002 in BIND_PORTS

    def test_bind_server_initializes_empty_servers_list(self, bind_server):
        """Verify bind server starts with empty servers list."""
        assert bind_server._servers == []
        assert bind_server._running is False

    @pytest.mark.asyncio
    async def test_start_server_creates_bind_server(self, tmp_path):
        """Verify start_server creates BindServer with correct params."""
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        inst = VirtualPrinterInstance(
            vp_id=99,
            name="Bambuddy",
            mode="archive",
            model="BL-P001",
            access_code="12345678",
            serial_suffix="391800099",
            bind_ip="192.168.1.50",
            base_dir=tmp_path,
        )

        # Each mocked child service exposes a real asyncio.Event for the
        # readiness barrier added in start_server (set on instantiation so
        # the barrier returns immediately in tests).
        ready_event = asyncio.Event()
        ready_event.set()

        def with_ready(*_args, **_kwargs):
            child = MagicMock()
            child.ready = ready_event
            return child

        with (
            patch(
                "backend.app.services.virtual_printer.manager.VirtualPrinterSSDPServer",
                side_effect=with_ready,
            ),
            patch(
                "backend.app.services.virtual_printer.manager.VirtualPrinterFTPServer",
                side_effect=with_ready,
            ),
            patch(
                "backend.app.services.virtual_printer.manager.SimpleMQTTServer",
                side_effect=with_ready,
            ),
            patch(
                "backend.app.services.virtual_printer.manager.BindServer",
                side_effect=with_ready,
            ) as mock_bind_cls,
            patch.object(inst._cert_service, "delete_printer_certificate"),
            patch.object(
                inst._cert_service,
                "generate_certificates",
                return_value=(Path("/tmp/cert.pem"), Path("/tmp/key.pem")),  # nosec B108
            ),
        ):
            await inst.start_server()

            mock_bind_cls.assert_called_once_with(
                serial=inst.serial,
                model="BL-P001",
                name="Bambuddy",
                bind_address="192.168.1.50",
                cert_path=Path("/tmp/cert.pem"),  # nosec B108
                key_path=Path("/tmp/key.pem"),  # nosec B108
            )


class TestResolveModelCodes:
    """Tests for model code resolution (display name → SSDP code)."""

    def test_display_name_to_model_code_maps_all_models(self):
        """Verify reverse mapping covers all VIRTUAL_PRINTER_MODELS entries."""
        from backend.app.services.virtual_printer.manager import DISPLAY_NAME_TO_MODEL_CODE, VIRTUAL_PRINTER_MODELS

        for _code, display_name in VIRTUAL_PRINTER_MODELS.items():
            assert display_name in DISPLAY_NAME_TO_MODEL_CODE
            # For non-duplicate display names, should map back to a valid code
            assert DISPLAY_NAME_TO_MODEL_CODE[display_name] in VIRTUAL_PRINTER_MODELS

    def test_resolve_printer_model_with_ssdp_code(self):
        """SSDP codes pass through unchanged."""
        from backend.app.api.routes.virtual_printers import _resolve_printer_model

        assert _resolve_printer_model("BL-P001") == "BL-P001"
        assert _resolve_printer_model("O1D") == "O1D"
        assert _resolve_printer_model("N2S") == "N2S"

    def test_resolve_printer_model_with_display_name(self):
        """Display names resolve to SSDP codes."""
        from backend.app.api.routes.virtual_printers import _resolve_printer_model

        assert _resolve_printer_model("X1C") == "BL-P001"
        assert _resolve_printer_model("H2D") == "O1D"
        assert _resolve_printer_model("A1") == "N2S"
        assert _resolve_printer_model("P1S") == "C12"

    def test_resolve_printer_model_with_none_or_unknown(self):
        """None and unknown values return None."""
        from backend.app.api.routes.virtual_printers import _resolve_printer_model

        assert _resolve_printer_model(None) is None
        assert _resolve_printer_model("UnknownModel") is None


class TestMqttIpRewrite:
    """Tests for TLSProxy._rewrite_mqtt_ip() MQTT packet IP rewriting."""

    @staticmethod
    def _build_mqtt_publish(topic: str, payload: bytes) -> bytes:
        """Build a minimal MQTT PUBLISH packet."""
        # PUBLISH fixed header: type 3, no flags
        topic_bytes = topic.encode("utf-8")
        # Variable header: topic length (2 bytes) + topic
        var_header = len(topic_bytes).to_bytes(2, "big") + topic_bytes
        body = var_header + payload

        # Encode remaining length
        remaining = len(body)
        header = bytearray([0x30])  # PUBLISH, QoS 0
        while True:
            encoded_byte = remaining % 128
            remaining //= 128
            if remaining > 0:
                encoded_byte |= 0x80
            header.append(encoded_byte)
            if remaining == 0:
                break

        return bytes(header) + body

    @staticmethod
    def _build_mqtt_pingreq() -> bytes:
        """Build an MQTT PINGREQ packet (2 bytes, no payload)."""
        return b"\xc0\x00"

    def test_rewrite_ip_in_publish(self):
        """IP string in PUBLISH payload is rewritten."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        payload = b'{"rtsp_url":"rtsps://192.168.1.100:322/live"}'
        packet = self._build_mqtt_publish("device/status", payload)

        result, buf = TLSProxy._rewrite_mqtt_ip(packet, b"192.168.1.100", b"10.0.0.1", bytearray())

        assert b"10.0.0.1" in result
        assert b"192.168.1.100" not in result

    def test_no_rewrite_when_ip_absent(self):
        """Packets without the target IP are passed through unchanged."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        payload = b'{"status":"idle"}'
        packet = self._build_mqtt_publish("device/status", payload)

        result, buf = TLSProxy._rewrite_mqtt_ip(packet, b"192.168.1.100", b"10.0.0.1", bytearray())

        assert result == packet

    def test_non_publish_packets_unchanged(self):
        """Non-PUBLISH packets (e.g. PINGREQ) are never rewritten."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        pingreq = self._build_mqtt_pingreq()
        result, buf = TLSProxy._rewrite_mqtt_ip(pingreq, b"192.168.1.100", b"10.0.0.1", bytearray())

        assert result == pingreq

    def test_rewrite_preserves_packet_framing(self):
        """Rewritten packet has valid MQTT remaining length."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        # Use IPs of different lengths to test length re-encoding
        old_ip = b"192.168.255.133"  # 15 bytes
        new_ip = b"10.0.0.1"  # 8 bytes

        payload = b'{"ip":"192.168.255.133"}'
        packet = self._build_mqtt_publish("device/status", payload)

        result, buf = TLSProxy._rewrite_mqtt_ip(packet, old_ip, new_ip, bytearray())

        # Parse the result to verify framing
        assert result[0] == 0x30  # PUBLISH header byte
        # Decode remaining length
        pos = 1
        remaining = 0
        multiplier = 1
        while True:
            b = result[pos]
            pos += 1
            remaining += (b & 0x7F) * multiplier
            multiplier *= 128
            if (b & 0x80) == 0:
                break

        # Remaining length should match actual data
        assert pos + remaining == len(result)
        assert new_ip in result

    def test_incomplete_packet_buffered(self):
        """Incomplete packet at end of chunk is buffered for next call."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        payload = b'{"ip":"192.168.1.100"}'
        packet = self._build_mqtt_publish("device/status", payload)

        # Split packet in the middle
        half = len(packet) // 2
        chunk1 = packet[:half]
        chunk2 = packet[half:]

        result1, buf = TLSProxy._rewrite_mqtt_ip(chunk1, b"192.168.1.100", b"10.0.0.1", bytearray())
        # First chunk should be buffered (incomplete packet)
        assert len(buf) > 0

        result2, buf = TLSProxy._rewrite_mqtt_ip(chunk2, b"192.168.1.100", b"10.0.0.1", buf)
        # Second chunk completes the packet, IP should be rewritten
        combined = result1 + result2
        assert b"10.0.0.1" in combined
        assert b"192.168.1.100" not in combined

    def test_multiple_packets_in_one_chunk(self):
        """Multiple MQTT packets in a single chunk are all processed."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        payload1 = b'{"ip":"192.168.1.100"}'
        payload2 = b'{"other":"data"}'
        packet1 = self._build_mqtt_publish("topic1", payload1)
        packet2 = self._build_mqtt_publish("topic2", payload2)

        combined = packet1 + packet2
        result, buf = TLSProxy._rewrite_mqtt_ip(combined, b"192.168.1.100", b"10.0.0.1", bytearray())

        assert b"10.0.0.1" in result
        assert b"192.168.1.100" not in result
        # Second packet should still be present
        assert b"other" in result

    def test_extra_replacements(self):
        """Extra replacement pairs (e.g. integer IP) are also applied."""
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        payload = b'{"net":{"info":[{"ip":2248124608}]}}'
        packet = self._build_mqtt_publish("device/status", payload)

        result, buf = TLSProxy._rewrite_mqtt_ip(
            packet,
            b"NOMATCH",
            b"NOREPLACE",
            bytearray(),
            extra_replacements=[(b"2248124608", b"285190336")],
        )

        assert b"285190336" in result
        assert b"2248124608" not in result


class TestIpToLeIntBytes:
    """Tests for TLSProxy._ip_to_le_int_bytes() integer IP conversion."""

    def test_converts_ip_to_le_int(self):
        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        assert TLSProxy._ip_to_le_int_bytes("192.168.255.133") == b"2248124608"
        assert TLSProxy._ip_to_le_int_bytes("192.168.255.16") == b"285190336"
        assert TLSProxy._ip_to_le_int_bytes("10.0.0.1") == b"16777226"

    def test_roundtrip(self):
        """Verify the integer converts back to the correct IP."""
        import struct

        from backend.app.services.virtual_printer.tcp_proxy import TLSProxy

        for ip in ["192.168.1.1", "10.0.0.1", "172.16.0.100", "192.168.255.133"]:
            le_int = int(TLSProxy._ip_to_le_int_bytes(ip))
            parts = ip.split(".")
            expected = struct.unpack("<I", bytes(int(p) for p in parts))[0]
            assert le_int == expected


class TestSSDPProxyName:
    """Tests for SSDPProxy VP name rewriting."""

    @pytest.fixture
    def ssdp_proxy_with_name(self):
        from backend.app.services.virtual_printer.ssdp_server import SSDPProxy

        return SSDPProxy(
            local_interface_ip="192.168.1.100",
            remote_interface_ip="10.0.0.100",
            target_printer_ip="192.168.1.50",
            name="H2D-1 Proxy",
        )

    @pytest.fixture
    def ssdp_proxy_without_name(self):
        from backend.app.services.virtual_printer.ssdp_server import SSDPProxy

        return SSDPProxy(
            local_interface_ip="192.168.1.100",
            remote_interface_ip="10.0.0.100",
            target_printer_ip="192.168.1.50",
        )

    def test_rewrite_uses_configured_name(self, ssdp_proxy_with_name):
        """When name is set, DevName is replaced entirely."""
        packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: RealPrinter\r\nDevBind.bambu.com: cloud\r\n\r\n"
        rewritten = ssdp_proxy_with_name._rewrite_ssdp(packet)

        assert b"DevName.bambu.com: H2D-1 Proxy" in rewritten
        assert b"RealPrinter" not in rewritten

    def test_rewrite_appends_proxy_without_name(self, ssdp_proxy_without_name):
        """When no name is set, ' - Proxy' is appended to the real name."""
        packet = b"NOTIFY * HTTP/1.1\r\nLocation: 192.168.1.50\r\nDevName.bambu.com: RealPrinter\r\nDevBind.bambu.com: cloud\r\n\r\n"
        rewritten = ssdp_proxy_without_name._rewrite_ssdp(packet)

        assert b"DevName.bambu.com: RealPrinter - Proxy" in rewritten


class TestVPProjectFileStashKey:
    """Regression: `on_print_command` MUST stash slicer options under the
    FTP filename (`data["file"]`, with extension), NOT under `filename`
    (the slicer's `subtask_name`, bare).

    #1780 root cause (real bundle, 2026-06-21): BambuStudio sends
    `subtask_name = "Model_Name"` (bare) and `file = "Model_Name.gcode.3mf"`
    (with extension). `_add_to_print_queue` looks up the stash under
    `file_path.name` from the FTP receive side, which always has the
    extension. If the stash uses `subtask_name`, lookup misses → every
    captured slicer field (bed_leveling, flow_cali, vibration_cali,
    layer_inspect, timelapse, nozzle_mapping) silently falls back to
    settings defaults on every Bambu Studio "Send" upload.

    `filename` (subtask_name) must still flow to `_schedule_finish_release`
    untouched — push_status echoes it back as gcode_file / subtask_name and
    the slicer matches against its own local subtask_name there. So
    `on_print_command` keeps `filename` for state-feedback but derives the
    stash key from `data["file"]`.
    """

    @pytest.fixture
    def instance(self, tmp_path):
        from backend.app.services.virtual_printer.manager import VirtualPrinterInstance

        return VirtualPrinterInstance(
            vp_id=99,
            name="StashKeyTest",
            mode="queue",
            model="O1C2",
            access_code="12345678",
            serial_suffix="999999999",
            base_dir=tmp_path,
        )

    @pytest.mark.asyncio
    async def test_stash_key_uses_file_field_not_subtask_name(self, instance):
        """BambuStudio's real wire shape: `subtask_name` ≠ `file`.
        on_print_command must stash under `data["file"]` so the FTP-side
        `_add_to_print_queue` lookup matches.
        """
        # mqtt_server.py:_handle_publish hands the bare subtask_name as
        # `filename` and the full print_data body as `data`. The FTP filename
        # lives in `data["file"]`.
        await instance.on_print_command(
            "Filament_Track_Switch_Holder",  # subtask_name (bare)
            {
                "command": "project_file",
                "subtask_name": "Filament_Track_Switch_Holder",
                "file": "Filament_Track_Switch_Holder.gcode.3mf",
                "nozzle_mapping": [16, -1, -1, 1],
            },
        )

        # Stash MUST be under the FTP filename, not the bare subtask_name.
        # `_add_to_print_queue` does `_slicer_print_options.pop(file_path.name, None)`
        # where file_path.name == "Filament_Track_Switch_Holder.gcode.3mf".
        assert "Filament_Track_Switch_Holder.gcode.3mf" in instance._slicer_print_options
        assert "Filament_Track_Switch_Holder" not in instance._slicer_print_options
        # Body must carry nozzle_mapping verbatim.
        stashed = instance._slicer_print_options["Filament_Track_Switch_Holder.gcode.3mf"]
        assert stashed["nozzle_mapping"] == [16, -1, -1, 1]

    @pytest.mark.asyncio
    async def test_stash_key_falls_back_to_filename_when_file_absent(self, instance):
        """Defensive fallback: a slicer that omits the `file` field entirely
        (legacy / non-3MF) must fall back to `filename` (subtask_name), not
        leave the stash unkeyed."""
        await instance.on_print_command(
            "BareName",
            {
                "command": "project_file",
                "subtask_name": "BareName",
                # no "file" field
            },
        )

        assert "BareName" in instance._slicer_print_options

    @pytest.mark.asyncio
    async def test_stash_key_signals_event_under_file_key(self, instance):
        """`_add_to_print_queue` registers a wait-event under `file_path.name`
        when the slicer's project_file arrives late. on_print_command must
        signal THAT event (keyed by the FTP filename), not one keyed by
        subtask_name — else the waiter times out even though the stash is
        present and addressable."""
        import asyncio

        ftp_filename = "Filament_Track_Switch_Holder.gcode.3mf"
        event = asyncio.Event()
        instance._slicer_print_options_events[ftp_filename] = event

        await instance.on_print_command(
            "Filament_Track_Switch_Holder",  # bare subtask_name
            {
                "command": "project_file",
                "subtask_name": "Filament_Track_Switch_Holder",
                "file": ftp_filename,
            },
        )

        # Event keyed by FTP filename must fire even though on_print_command
        # was called with the bare subtask_name.
        assert event.is_set()