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

"""A VP with "Save AMS mapping" keeps the slicer's external-spool pick (#3237).

The slicer writes an external spool as -1 in the flat ``ams_mapping``, the same
as a filament with no tray, and names it only in ``ams_mapping2``. Saving the
flat list alone made dispatch send the external filament as unmapped, and the
printer stopped with 0700-8012 before the first layer.
"""

import json
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from backend.app.services.virtual_printer.manager import (
    VirtualPrinterInstance,
    _extract_slicer_ams_mapping_json,
)

EXT_LEFT = {"ams_id": 254, "slot_id": 0}
EXT_RIGHT = {"ams_id": 255, "slot_id": 0}
UNMAPPED = {"ams_id": 255, "slot_id": 255}

# The reporter's H2C job: left external spool plus AMS 0 slots 3 and 2.
H2C_PAYLOAD = {
    "command": "project_file",
    "ams_mapping": [-1, 3, 2],
    "ams_mapping2": [EXT_LEFT, {"ams_id": 0, "slot_id": 3}, {"ams_id": 0, "slot_id": 2}],
}


@pytest.fixture(autouse=True)
def _fake_file_manager_store():
    """Uploads go to the File Manager (library_store); its own tests cover it.

    Here it stands in with file id 1 and the default folder settings, so these
    tests see what the virtual printer does around it.
    """
    stored = MagicMock(id=1, filename="upload.3mf", folder_id=None)
    with (
        patch(
            "backend.app.services.virtual_printer.library_store.store_upload",
            AsyncMock(return_value=(stored, False)),
        ) as store,
        patch(
            "backend.app.services.virtual_printer.library_store.upload_settings",
            AsyncMock(return_value=(None, False)),
        ),
    ):
        yield store


def _extract(data, is_dual_nozzle):
    raw = _extract_slicer_ams_mapping_json(data, "[test]", is_dual_nozzle=is_dual_nozzle)
    return None if raw is None else json.loads(raw)


class TestExtract:
    def test_left_external_spool_on_dual_nozzle(self):
        assert _extract(H2C_PAYLOAD, is_dual_nozzle=True) == [254, 3, 2]

    def test_right_external_spool_on_dual_nozzle(self):
        data = {"ams_mapping": [-1, 3], "ams_mapping2": [EXT_RIGHT, {"ams_id": 0, "slot_id": 3}]}
        assert _extract(data, is_dual_nozzle=True) == [255, 3]

    def test_single_nozzle_external_spool_is_254(self):
        # Single-nozzle printers have one external spool; the wire calls it
        # 255, Bambuddy's global tray is 254.
        data = {"ams_mapping": [0, -1], "ams_mapping2": [{"ams_id": 0, "slot_id": 0}, EXT_RIGHT]}
        assert _extract(data, is_dual_nozzle=False) == [0, 254]

    def test_unused_filament_stays_unmapped(self):
        data = {"ams_mapping": [-1, 3], "ams_mapping2": [UNMAPPED, {"ams_id": 0, "slot_id": 3}]}
        assert _extract(data, is_dual_nozzle=True) == [-1, 3]

    def test_external_only_job_is_kept(self):
        # Every flat entry is -1, but this is a real pick, not the #2589
        # unresolved sentinel.
        assert _extract({"ams_mapping": [-1], "ams_mapping2": [EXT_LEFT]}, is_dual_nozzle=True) == [254]

    def test_unresolved_sentinel_still_dropped(self):
        data = {"ams_mapping": [-1, -1], "ams_mapping2": [UNMAPPED, UNMAPPED]}
        assert _extract(data, is_dual_nozzle=True) is None

    def test_stringified_mapping2_is_parsed(self):
        data = {"ams_mapping": "[-1, 3, 2]", "ams_mapping2": json.dumps(H2C_PAYLOAD["ams_mapping2"])}
        assert _extract(data, is_dual_nozzle=True) == [254, 3, 2]

    def test_missing_or_unusable_mapping2_keeps_flat_list(self):
        assert _extract({"ams_mapping": [-1, 3, 2]}, is_dual_nozzle=True) == [-1, 3, 2]
        assert _extract({"ams_mapping": [-1, 3, 2], "ams_mapping2": "not json"}, is_dual_nozzle=True) == [-1, 3, 2]
        # Length mismatch: positions can't be paired, leave it alone.
        assert _extract({"ams_mapping": [-1, 3, 2], "ams_mapping2": [EXT_LEFT]}, is_dual_nozzle=True) == [-1, 3, 2]


def _instance(tmp_path, *, model="O1C", printer_manager=None, session_factory=None):
    return VirtualPrinterInstance(
        vp_id=3237,
        name="ExtSpool",
        mode="queue",
        model=model,
        access_code="12345678",
        serial_suffix="391803237",
        base_dir=tmp_path,
        session_factory=session_factory,
        save_ams_mapping=True,
        target_printer_id=7,
        printer_manager=printer_manager,
    )


def _manager_with(client):
    manager = MagicMock()
    manager.get_client = MagicMock(return_value=client)
    return manager


class TestTargetIsDualNozzle:
    def test_live_detection_wins(self, tmp_path):
        client = MagicMock(_is_dual_nozzle=True, model="Some future model")
        assert _instance(tmp_path, model="BL-P001", printer_manager=_manager_with(client))._target_is_dual_nozzle()

    def test_client_model(self, tmp_path):
        client = MagicMock(_is_dual_nozzle=False, model="H2C")
        assert _instance(tmp_path, model="BL-P001", printer_manager=_manager_with(client))._target_is_dual_nozzle()

    def test_single_nozzle_client(self, tmp_path):
        client = MagicMock(_is_dual_nozzle=False, model="X1C")
        assert not _instance(tmp_path, printer_manager=_manager_with(client))._target_is_dual_nozzle()

    def test_falls_back_to_vp_model_without_client(self, tmp_path):
        assert _instance(tmp_path, model="O1C", printer_manager=_manager_with(None))._target_is_dual_nozzle()
        assert not _instance(tmp_path, model="BL-P001")._target_is_dual_nozzle()


@pytest.mark.asyncio
async def test_queue_item_and_archive_keep_left_external_spool(tmp_path):
    added_items = []
    mock_db = AsyncMock()
    mock_db.add = MagicMock(side_effect=added_items.append)
    mock_db.commit = AsyncMock()
    session_ctx = AsyncMock()
    session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
    session_ctx.__aexit__ = AsyncMock(return_value=False)
    session_factory = MagicMock(return_value=session_ctx)

    client = MagicMock(_is_dual_nozzle=True, model="H2C")
    inst = _instance(tmp_path, printer_manager=_manager_with(client), session_factory=session_factory)

    file_path = tmp_path / "test.3mf"
    file_path.write_bytes(b"fake3mf")
    await inst.on_print_command(file_path.name, dict(H2C_PAYLOAD))

    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 len(added_items) == 1
    assert json.loads(added_items[0].ams_mapping) == [254, 3, 2]
    # Kept on the item for the archive its print makes.
    assert json.loads(added_items[0].slicer_ams_mapping)["mapping"] == [254, 3, 2]


# An H3C job: AMS 3H slot 1, then the external inputs of hotends 0, 1 and 5.
H3C_PAYLOAD = {
    "command": "project_file",
    "ams_mapping": [28, -1, -1, -1],
    "ams_mapping2": [
        {"ams_id": 32, "slot_id": 0},
        {"ams_id": 255, "slot_id": 0},
        {"ams_id": 254, "slot_id": 0},
        {"ams_id": 250, "slot_id": 0},
    ],
}


class TestRackPrinterExternalInputs:
    """An H3C has one external input per hotend (247-255); each keeps its own id."""

    def test_rack_layout_keeps_each_input(self):
        from backend.app.utils.tray_ids import TrayLayout

        raw = _extract_slicer_ams_mapping_json(H3C_PAYLOAD, "[test]", layout=TrayLayout.ams3h())
        assert json.loads(raw) == [28, 255, 254, 250]

    def test_target_h3c_layout_from_its_model(self, tmp_path):
        manager = _manager_with(None)
        manager.get_status = MagicMock(return_value=None)
        manager.get_model = MagicMock(return_value="H3C")
        assert _instance(tmp_path, model="N10", printer_manager=manager)._target_tray_layout().rack

    def test_any_available_h3c_uses_the_vp_model(self, tmp_path):
        inst = _instance(tmp_path, model="N10")
        inst.target_printer_id = None
        assert inst._target_tray_layout().rack

    def test_other_targets_are_not_rack(self, tmp_path):
        manager = _manager_with(None)
        manager.get_status = MagicMock(return_value=None)
        manager.get_model = MagicMock(return_value="H2C")
        assert not _instance(tmp_path, printer_manager=manager)._target_tray_layout().rack
        inst = _instance(tmp_path, model="BL-P001")
        inst.target_printer_id = None
        assert not inst._target_tray_layout().rack

    @pytest.mark.asyncio
    async def test_queue_item_keeps_h3c_external_inputs(self, tmp_path):
        added_items = []
        mock_db = AsyncMock()
        mock_db.add = MagicMock(side_effect=added_items.append)
        mock_db.commit = AsyncMock()
        session_ctx = AsyncMock()
        session_ctx.__aenter__ = AsyncMock(return_value=mock_db)
        session_ctx.__aexit__ = AsyncMock(return_value=False)
        session_factory = MagicMock(return_value=session_ctx)

        manager = _manager_with(MagicMock(_is_dual_nozzle=False, model="N10"))
        manager.get_status = MagicMock(return_value=None)
        manager.get_model = MagicMock(return_value="N10")
        inst = _instance(tmp_path, model="N10", printer_manager=manager, session_factory=session_factory)

        file_path = tmp_path / "test.3mf"
        file_path.write_bytes(b"fake3mf")
        await inst.on_print_command(file_path.name, dict(H3C_PAYLOAD))
        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 len(added_items) == 1
        assert json.loads(added_items[0].ams_mapping) == [28, 255, 254, 250]