bambuddy/backend/tests/integration/test_inventory_assign.py

"""Integration tests for inventory spool assignment — tray_info_idx resolution.

Tests that the spool's own slicer_filament (including PFUS* cloud-synced
custom presets) takes priority, with slot reuse and generic fallback as
lower-priority fallbacks.
"""

from unittest.mock import MagicMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.models.spool import Spool


@pytest.fixture
async def spool_factory(db_session: AsyncSession):
    """Factory to create test spools."""
    _counter = [0]

    async def _create_spool(**kwargs):
        _counter[0] += 1
        defaults = {
            "material": "PLA",
            "subtype": "Basic",
            "brand": "Devil Design",
            "color_name": "Red",
            "rgba": "FF0000FF",
            "label_weight": 1000,
            "weight_used": 0,
            "slicer_filament": "PFUS9ac902733670a9",
        }
        defaults.update(kwargs)
        spool = Spool(**defaults)
        db_session.add(spool)
        await db_session.commit()
        await db_session.refresh(spool)
        return spool

    return _create_spool


def _make_mock_status(ams_data=None, vt_tray=None, nozzles=None, ams_extruder_map=None):
    """Build a mock printer status with optional AMS/nozzle data."""
    status = MagicMock()
    raw = {}
    if ams_data is not None:
        raw["ams"] = {"ams": ams_data}
    if vt_tray is not None:
        raw["vt_tray"] = vt_tray
    status.raw_data = raw
    status.nozzles = nozzles or [MagicMock(nozzle_diameter="0.4")]
    status.ams_extruder_map = ams_extruder_map
    return status


class TestAssignSpoolTrayInfoIdx:
    """Tests for tray_info_idx resolution during spool assignment."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_pfus_slicer_filament_falls_back_to_generic(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """PFUS* cloud setting_ids are rejected by the slicer as tray_info_idx, so the
        no-kp path falls back to the generic material id (PLA → GFL99). The K-profile
        realignment path translates PFUS → P-prefix when a stored kp exists; that's
        covered separately."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "", "tray_type": "PLA"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "GFL99"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_pfus_spool_reuses_valid_slot_preset(self, async_client: AsyncClient, printer_factory, spool_factory):
        """When the spool's PFUS gets discarded as slicer-invalid, the slot's existing
        valid P-prefix preset is reused if it matches the spool's material — preserves
        the printer's calibration context rather than resetting to generic."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Slot already configured by slicer with cloud-synced preset
        status = _make_mock_status(
            ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "P4d64437"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_spool_preset_used_even_if_different_material_on_slot(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Spool's material drives the fallback generic id. Slot's existing PLA preset
        is overridden because the spool is PETG → GFG99."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PETG")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Slot currently has PLA but spool is PETG
        status = _make_mock_status(
            ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "GFG99"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_gf_slicer_filament_kept(self, async_client: AsyncClient, printer_factory, spool_factory):
        """Standard GF* IDs from spool.slicer_filament are used directly."""
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "GFL05"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_empty_slicer_filament_uses_generic(self, async_client: AsyncClient, printer_factory, spool_factory):
        """Spool with no slicer_filament gets a generic ID from material type."""
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament=None, material="ABS")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "ABS"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "GFB99"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_spool_pfus_falls_back_to_generic_over_slot_pfus(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Both spool and slot have PFUS values — both rejected as tray_info_idx —
        falls back to generic material id (PLA → GFL99)."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFUS1111111111", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Slot has a PFUS* ID from some previous config
        status = _make_mock_status(
            ams_data=[{"id": 0, "tray": [{"id": 0, "tray_info_idx": "PFUS2222222222", "tray_type": "PLA"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "GFL99"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_generic_on_slot_falls_back_to_material_generic(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """When spool's PFUS is discarded and slot only has a generic ID, the result
        comes from the spool's material (ABS → GFB99) — not from the slot. Important
        because the generic-id check (`not in _generic_id_values`) prevents stale
        generic reuse and routes the decision through the material fallback."""
        printer = await printer_factory(name="P2S")
        spool = await spool_factory(slicer_filament="PFUScda4c46fc9031", material="ABS")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Slot stuck on generic ABS from a previous assignment
        status = _make_mock_status(
            ams_data=[{"id": 0, "tray": [{"id": 1, "tray_info_idx": "GFB99", "tray_type": "ABS"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "GFB99"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_preset_with_generic_on_slot_still_uses_generic(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Spool without preset + generic on slot → generic fallback (not slot reuse)."""
        printer = await printer_factory(name="P2S")
        spool = await spool_factory(slicer_filament=None, material="ABS")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Slot has generic ABS
        status = _make_mock_status(
            ams_data=[{"id": 0, "tray": [{"id": 1, "tray_info_idx": "GFB99", "tray_type": "ABS"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            # Still gets generic, but via fallback — not via sticky reuse
            assert call_kwargs.kwargs["tray_info_idx"] == "GFB99"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_preset_reuses_specific_slot_preset(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Spool without preset + specific preset on slot → reuse slot's preset."""
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament=None, material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Slot has a specific Bambu PLA preset (not generic)
        status = _make_mock_status(
            ams_data=[{"id": 0, "tray": [{"id": 0, "tray_info_idx": "GFA05", "tray_type": "PLA"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            # Slot's specific preset is reused when spool has no own preset
            assert call_kwargs.kwargs["tray_info_idx"] == "GFA05"


class TestAssignSpoolPresetMapping:
    """Tests that assign_spool saves the slot preset mapping for correct UI display."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_preset_mapping_saved_with_slicer_filament_name(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Slot preset mapping uses slicer_filament_name (not material+subtype)."""

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(
            slicer_filament="GFA05",
            slicer_filament_name="Bambu PLA Silk",
            material="PLA",
            subtype="Silk",
            brand="Bambu",
        )

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True
        status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 1, "tray_type": "PLA"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1},
            )

        assert response.status_code == 200

        # Verify via the slot presets API
        presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets")
        assert presets_resp.status_code == 200
        presets = presets_resp.json()
        # Key is str(ams_id * 4 + tray_id) — ams 0, tray 1 → "1"
        assert "1" in presets
        # Must use slicer_filament_name, NOT "PLA Silk" from material+subtype
        assert presets["1"]["preset_name"] == "Bambu PLA Silk"
        assert presets["1"]["preset_id"] == "GFSA05"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_preset_mapping_overwrites_old_mapping(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """Assigning a new spool overwrites the old slot preset mapping."""
        from backend.app.models.slot_preset import SlotPresetMapping

        printer = await printer_factory(name="X1C")

        # Pre-existing mapping (e.g. from previous manual configuration)
        old_mapping = SlotPresetMapping(
            printer_id=printer.id,
            ams_id=0,
            tray_id=2,
            preset_id="GFSA01",
            preset_name="Bambu PLA Matte",
            preset_source="cloud",
        )
        db_session.add(old_mapping)
        await db_session.commit()

        # Assign a "Generic PLA Silk" spool to same slot
        spool = await spool_factory(
            slicer_filament="GFL96",
            slicer_filament_name="Generic PLA Silk",
            material="PLA",
            subtype="Silk",
        )

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True
        status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 2, "tray_type": "PLA"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 2},
            )

        assert response.status_code == 200

        # Verify via the slot presets API to avoid stale session cache
        presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets")
        assert presets_resp.status_code == 200
        presets = presets_resp.json()
        # Key is str(ams_id * 4 + tray_id) — ams 0, tray 2 → "2"
        assert "2" in presets
        # Old "Bambu PLA Matte" must be overwritten
        assert presets["2"]["preset_name"] == "Generic PLA Silk"
        assert presets["2"]["preset_id"] == "GFSL96"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_preset_mapping_fallback_to_tray_sub_brands(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """When slicer_filament_name is null, falls back to tray_sub_brands."""
        from backend.app.models.slot_preset import SlotPresetMapping

        printer = await printer_factory(name="A1M")
        spool = await spool_factory(
            slicer_filament="GFL05",
            slicer_filament_name=None,
            material="PLA",
            subtype="Matte",
            brand="Overture",
        )

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True
        status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

        assert response.status_code == 200

        # Verify via the slot presets API
        presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets")
        assert presets_resp.status_code == 200
        presets = presets_resp.json()
        # Key is str(ams_id * 4 + tray_id) — ams 0, tray 0 → "0"
        assert "0" in presets
        # Falls back to tray_sub_brands ("Overture PLA Matte")
        assert presets["0"]["preset_name"] == "Overture PLA Matte"


class TestAssignSpoolLiveCaliIdx:
    """assign_spool always resets the slot to Default K when the spool has no stored K-profile."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_kprofile_resets_to_default_k(self, async_client: AsyncClient, printer_factory, spool_factory):
        """When no KProfile row exists, slot resets to cali_idx=-1 (Default K) regardless of live value."""
        printer = await printer_factory()
        spool = await spool_factory()

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True
        # Live cali_idx=42 belongs to whatever filament was previously calibrated
        # in this slot. Applying it to a different spool would use the wrong K
        # value, so the assign flow must override it with Default K (-1).
        tray_data = {
            "id": 1,
            "cali_idx": 42,
            "tray_color": "FF0000FF",
            "tray_type": "PLA",
            "tray_sub_brands": "PLA Basic",
            "tray_id_name": "GFL99",
        }
        status = _make_mock_status(ams_data=[{"id": 0, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1},
            )

        assert response.status_code == 200
        mock_client.extrusion_cali_sel.assert_called_once()
        assert mock_client.extrusion_cali_sel.call_args[1]["cali_idx"] == -1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_kprofile_no_live_cali_idx_sends_default(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """When tray has no cali_idx, extrusion_cali_sel is sent with cali_idx=-1 (Default)."""
        printer = await printer_factory()
        spool = await spool_factory()

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True
        tray_data = {
            "id": 0,
            "cali_idx": None,
            "tray_color": "FF0000FF",
            "tray_type": "PLA",
            "tray_sub_brands": "PLA Basic",
            "tray_id_name": "GFL99",
        }
        status = _make_mock_status(ams_data=[{"id": 0, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

        assert response.status_code == 200
        mock_client.extrusion_cali_sel.assert_called_once()
        assert mock_client.extrusion_cali_sel.call_args[1]["cali_idx"] == -1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_negative_live_cali_idx_sends_default(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """A negative live cali_idx (-1) falls through and is sent as Default (cali_idx=-1)."""
        printer = await printer_factory()
        spool = await spool_factory()

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True
        tray_data = {
            "id": 0,
            "cali_idx": -1,
            "tray_color": "FF0000FF",
            "tray_type": "PLA",
            "tray_sub_brands": "PLA Basic",
            "tray_id_name": "GFL99",
        }
        status = _make_mock_status(ams_data=[{"id": 0, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

        assert response.status_code == 200
        mock_client.extrusion_cali_sel.assert_called_once()
        assert mock_client.extrusion_cali_sel.call_args[1]["cali_idx"] == -1


class TestAssignSpoolEmptySlotPreConfig:
    """Assign path under ambiguous / explicit-empty AMS state.

    Updated for the #1322 follow-up: only the firmware's *explicit* empty
    signal (state ∈ {9, 10}) skips MQTT. Anything else — including the
    SpoolBuddy weigh-then-assign-before-insert case where state/tray_type
    can't tell us whether a spool is loaded — attempts MQTT. The deferred-
    config workflow still works because on_ams_change at main.py:1031-1054
    re-fires when an AMS push eventually reports the loaded slot.
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_empty_tray_type_without_state_still_fires_mqtt(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """tray_type='' with no state field: AMS can't tell us whether a
        spool is loaded. Trust the user's Assign click and fire MQTT —
        firmware accepts it when a spool is physically there, drops it
        silently otherwise (no harm)."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_type": ""}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        mock_client.ams_set_filament_setting.assert_called_once()
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_no_ams_data_with_no_client_marks_pending(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """No AMS data + no MQTT client (printer offline, no telemetry):
        publish can't happen, so configured=False and pending_config=True so
        on_ams_change replay picks it up when the printer comes online."""
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        # No AMS data — fingerprint_type stays None.
        status = _make_mock_status(ams_data=[])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = None  # Printer offline, no MQTT client.
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

        assert response.status_code == 200
        body = response.json()
        assert body["pending_config"] is True
        assert body["configured"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_loaded_slot_publishes_mqtt_immediately(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Loaded slot (tray_type non-empty) → MQTT fires + pending_config=False."""
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(
            ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_info_idx": "GFL05"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0},
            )

        assert response.status_code == 200
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True
        mock_client.ams_set_filament_setting.assert_called_once()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_on_ams_change_fires_config_when_pre_assigned_slot_loads(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """Pre-config replay: SpoolAssignment with empty fingerprint + slot now loaded → MQTT fires."""
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        # Pre-existing assignment with empty fingerprint (the SpoolBuddy state)
        pre_assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=2,
            tray_id=3,
            fingerprint_color=None,
            fingerprint_type=None,
        )
        db_session.add(pre_assignment)
        await db_session.commit()

        # Filament has now been physically inserted into the slot.
        # state=11 ("filament fed to extruder") is the load signal we trigger on.
        ams_data = [{"id": 2, "tray": [{"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "state": 11}]}]

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=ams_data)
        printer_info = MagicMock(name="H2D", serial_number="0948BB540200427")

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = printer_info
            mock_pm_main.get_status.return_value = status
            mock_pm_main.get_client.return_value = mock_client
            mock_pm_main.get_model.return_value = "H2D"
            mock_pm_inv.get_client.return_value = mock_client
            mock_pm_inv.get_status.return_value = status
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        # Full filament setting was published when the slot transitioned to loaded
        mock_client.ams_set_filament_setting.assert_called_once()
        call_kwargs = mock_client.ams_set_filament_setting.call_args.kwargs
        assert call_kwargs["ams_id"] == 2
        assert call_kwargs["tray_id"] == 3
        assert call_kwargs["tray_info_idx"] == "GFL05"

        # Fingerprint was updated so the next push doesn't re-fire
        await db_session.refresh(pre_assignment)
        assert pre_assignment.fingerprint_type == "PLA"
        assert pre_assignment.fingerprint_color == "FF0000FF"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_on_ams_change_does_not_refire_for_already_configured_slot(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """Once fingerprint_type is set, subsequent AMS pushes must not re-fire MQTT."""
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        # Assignment already configured (fingerprint stamped)
        configured_assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=0,
            tray_id=0,
            fingerprint_color="FF0000FF",
            fingerprint_type="PLA",
        )
        db_session.add(configured_assignment)
        await db_session.commit()

        ams_data = [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF", "state": 11}]}]

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=ams_data)
        printer_info = MagicMock(name="X1C", serial_number="00M00A391800004")

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = printer_info
            mock_pm_main.get_status.return_value = status
            mock_pm_main.get_client.return_value = mock_client
            mock_pm_main.get_model.return_value = "X1C"
            mock_pm_inv.get_client.return_value = mock_client
            mock_pm_inv.get_status.return_value = status
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        # Fingerprint was already set — re-fire path skipped
        mock_client.ams_set_filament_setting.assert_not_called()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_on_ams_change_fires_replay_when_tray_type_appears_without_state_11(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """A1 Mini / P1S firmware variant of the SpoolBuddy pre-config replay
        (#1322). The user pre-assigned via SpoolBuddy (fingerprint empty), then
        configured the slot manually in Bambu Studio so tray_type went from ''
        to 'PLA' — but state stays at 3 because these firmwares never set it
        to 11. With state-only detection the replay never fired."""
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="A1 mini")
        spool = await spool_factory(slicer_filament="GFL05", material="PLA")

        pre_assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=0,
            tray_id=3,
            fingerprint_color=None,
            fingerprint_type=None,
        )
        db_session.add(pre_assignment)
        await db_session.commit()

        # state=3 (never goes to 11 on A1 Mini BMCU 01.07.02.00) but tray_type
        # is now configured — the replay must fire on this transition too.
        ams_data = [
            {
                "id": 0,
                "tray": [{"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "state": 3, "tray_info_idx": "GFL05"}],
            }
        ]

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=ams_data)
        printer_info = MagicMock(name="A1 mini", serial_number="0309CA391800999")

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = printer_info
            mock_pm_main.get_status.return_value = status
            mock_pm_main.get_client.return_value = mock_client
            mock_pm_main.get_model.return_value = "A1 mini"
            mock_pm_inv.get_client.return_value = mock_client
            mock_pm_inv.get_status.return_value = status
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        # Replay fired despite state never being 11 — the disjunction picked
        # up tray_type going non-empty.
        mock_client.ams_set_filament_setting.assert_called_once()
        await db_session.refresh(pre_assignment)
        assert pre_assignment.fingerprint_type == "PLA"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_auto_unlink_broadcasts_assignment_change(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """#2575 follow-up: when on_ams_change auto-unlinks a stale external-spool
        assignment, it must broadcast spool_assignment_changed. Only the manual
        REST endpoints did, so open browsers kept rendering the unlinked spool —
        the reporter read that as "the fix didn't work" when the DB was correct."""
        from unittest.mock import AsyncMock

        from sqlalchemy import select

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament="GFU01", material="TPU")

        # TPU inventory spool assigned to the external slot (ams_id=255, tray 0)
        assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=255,
            tray_id=0,
            fingerprint_color="000000FF",
            fingerprint_type="TPU",
        )
        db_session.add(assignment)
        await db_session.commit()

        # The printer's external spool now reports ABS — the assignment is stale.
        vt_tray = [
            {
                "id": "254",
                "tray_type": "ABS",
                "tray_color": "000000FF",
                "tag_uid": "0000000000000000",
                "tray_uuid": "00000000000000000000000000000000",
            }
        ]
        status = _make_mock_status(ams_data=[], vt_tray=vt_tray)
        printer_info = MagicMock(name="X1C", serial_number="00M00A391800004")

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = printer_info
            mock_pm_main.get_status.return_value = status
            mock_pm_main.get_client.return_value = MagicMock()
            mock_pm_main.get_model.return_value = "X1C"
            mock_pm_inv.get_client.return_value = MagicMock()
            mock_pm_inv.get_status.return_value = status
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, [])

            # The stale TPU assignment on the now-ABS external slot was unlinked...
            gone = await db_session.execute(
                select(SpoolAssignment).where(
                    SpoolAssignment.printer_id == printer.id,
                    SpoolAssignment.ams_id == 255,
                    SpoolAssignment.tray_id == 0,
                )
            )
            assert gone.scalar_one_or_none() is None

            # ...and the frontend was told about it.
            change_events = [
                c.args[0]
                for c in mock_ws.broadcast.await_args_list
                if c.args and isinstance(c.args[0], dict) and c.args[0].get("type") == "spool_assignment_changed"
            ]
            assert change_events, "auto-unlink must broadcast spool_assignment_changed"
            assert change_events[0]["printer_id"] == printer.id
            assert change_events[0]["ams_id"] == 255
            assert change_events[0]["tray_id"] == 0


class TestAssignSpoolEmptyDetection:
    """Bambu firmware reports tray.state — 11=loaded, 9=empty, 10=spool present
    but filament not in feeder. The assign route must prefer that signal over
    tray_type for the empty-vs-loaded check, because a manual "Reset slot"
    clears tray_type to "" while leaving filament physically loaded — the
    legacy heuristic would route to the pending-config path and skip MQTT
    forever, since on_ams_change replay only fires on an empty→loaded
    transition that never comes when the slot is already loaded.
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_state_loaded_with_empty_tray_type_fires_mqtt(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Post-reset case: state=11 (loaded) but tray_type='' — MQTT must fire."""
        printer = await printer_factory()
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # Simulates the "reset slot" aftermath: filament physically loaded
        # (state=11) but tray_type/tray_color/tray_info_idx have been cleared.
        tray_data = {"id": 3, "state": 11, "tray_type": "", "tray_color": "", "tray_info_idx": ""}
        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        # MQTT must have fired — the bug was that legacy detection saw the
        # empty tray_type and skipped this entirely.
        mock_client.ams_set_filament_setting.assert_called_once()
        # Response must report configured=True, pending_config=False — the
        # slot is loaded, just had stale metadata cleared.
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_state_empty_skips_mqtt_and_marks_pending(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Genuinely empty slot: state=9 — MQTT skipped, pending_config=True."""
        printer = await printer_factory()
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True

        tray_data = {"id": 3, "state": 9, "tray_type": "", "tray_color": "", "tray_info_idx": ""}
        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        # SpoolBuddy weigh-then-assign workflow: firmware drops MQTT for
        # unloaded slots, so we don't bother sending it.
        mock_client.ams_set_filament_setting.assert_not_called()
        body = response.json()
        assert body["pending_config"] is True
        assert body["configured"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_state_missing_falls_back_to_tray_type_loaded(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Older firmware without state field: tray_type='PLA' → treated as loaded."""
        printer = await printer_factory()
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True

        # No 'state' key at all — older firmware behaviour.
        tray_data = {"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF"}
        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        # Legacy fallback: tray_type non-empty → treated as loaded → MQTT fires.
        mock_client.ams_set_filament_setting.assert_called_once()
        body = response.json()
        assert body["pending_config"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_state_missing_with_empty_tray_type_still_fires_mqtt(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """Older firmware without state field + empty tray_type still fires MQTT.

        The AMS doesn't tell us whether a spool is physically loaded in this
        case (no state, no tray_type), so the assign click is the user's
        assertion that a spool is there. Firmware silently drops the push on
        a truly empty slot — no harm done, and on_ams_change replay handles
        the deferred-config case (#1322 follow-up).
        """
        printer = await printer_factory()
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        tray_data = {"id": 3, "tray_type": "", "tray_color": ""}
        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        mock_client.ams_set_filament_setting.assert_called_once()
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_state_never_eleven_firmware_with_loaded_tray_fires_mqtt(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """A1 Mini BMCU 01.07.02.00 and P1S Standard AMS 00.00.06.75 always
        report tray.state=3, never 11 — even for fully-loaded configured slots.
        A state-only check classified those as empty and skipped MQTT (#1322).
        With the disjunctive check, tray_type='PLA' alone is enough to fire."""
        printer = await printer_factory()
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        # state=3, tray_type non-empty — A1 Mini / P1S configured slot.
        tray_data = {"id": 3, "state": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "tray_info_idx": "GFL99"}
        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        mock_client.ams_set_filament_setting.assert_called_once()
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_post_reset_slot_with_state_3_still_fires_mqtt(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """A1 Mini BMCU / P1S Standard AMS post-"Reset Slot" with spool still
        inserted: state=3, tray_type="". The AMS gives us no signal to tell
        this apart from a truly-empty slot. We trust the user's Assign click
        and fire MQTT — firmware accepts the push because a spool is
        physically there (#1322 follow-up by @RosdasHH).

        Replaces the previous "marks_pending" assertion which was the bug:
        that gate created a deadlock because the AMS would never report a
        state change (nothing physically changed), so on_ams_change replay
        never re-fired the deferred config either.
        """
        printer = await printer_factory()
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        tray_data = {"id": 3, "state": 3, "tray_type": "", "tray_color": "00000000", "tray_info_idx": ""}
        status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

        assert response.status_code == 200
        mock_client.ams_set_filament_setting.assert_called_once()
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_external_slot_state_loaded_with_empty_tray_type_fires_mqtt(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """External (vt_tray) slot post-reset: same fix applies for ams_id=255."""
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True

        # External slot tray_id=0 → vt_tray id=254. state=11 (loaded), tray_type
        # cleared by reset.
        vt_data = [{"id": 254, "state": 11, "tray_type": "", "tray_color": "", "tray_info_idx": ""}]
        status = _make_mock_status(ams_data=[], vt_tray=vt_data)

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 255, "tray_id": 0},
            )

        assert response.status_code == 200
        mock_client.ams_set_filament_setting.assert_called_once()
        body = response.json()
        assert body["pending_config"] is False
        assert body["configured"] is True


class TestAssignSpoolPfcnCloudPreset:
    """Assign path for PFCN-prefix cloud presets (#1648).

    PFCN is a third Bambu cloud preset shape alongside PFUS (cloud user-created)
    and GFS (Bambu official) — used for cloud-shared / partner-uploaded
    presets like Polymaker's "(Custom)" Bambu Lab H2D variants. Before #1648
    the assign path skipped the cloud-detail lookup and left the raw PFCN
    string in tray_info_idx, which the printer's calibration table can't
    resolve. ConfigureAmsSlotModal rescued each assignment by doing the lookup
    itself, making "Configure" feel like a mandatory follow-up step.
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_pfcn_falls_back_to_generic_when_cloud_unavailable(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """When cloud auth isn't available (e.g. user not logged into Bambu Cloud),
        the raw PFCN must be discarded as slicer-invalid and the slot configures
        with the spool's generic material id (PLA → GFL99). Pre-fix behaviour
        was to leak the raw PFCN, which the slicer can't resolve."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFCN80e80c1f79db85", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "", "tray_type": "PLA"}]}])

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            # PFCN never leaks into tray_info_idx — must resolve to the
            # generic-material fallback when cloud lookup couldn't.
            assert call_kwargs.kwargs["tray_info_idx"] == "GFL99"
            assert not call_kwargs.kwargs["tray_info_idx"].startswith("PFCN")

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_pfcn_spool_reuses_valid_slot_preset(self, async_client: AsyncClient, printer_factory, spool_factory):
        """Symmetry with the PFUS case: when the spool's PFCN is discarded as
        slicer-invalid, the slot's existing valid P-prefix preset is reused
        if material matches — preserves calibration context instead of
        resetting to generic."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFCN80e80c1f79db85", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(
            ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA"}]}]
        )

        with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            assert call_kwargs.kwargs["tray_info_idx"] == "P4d64437"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_pfcn_resolves_to_filament_id_via_cloud_lookup(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """When the user is authenticated against Bambu Cloud, the PFCN setting_id
        triggers the same cloud-detail lookup as PFUS / GFS — extracts the real
        filament_id from `detail["filament_id"]` and ships that as
        tray_info_idx. This is the happy path the Configure modal already had
        but the assign path didn't, #1648."""
        printer = await printer_factory(name="H2D")
        spool = await spool_factory(slicer_filament="PFCN80e80c1f79db85", material="PLA")

        mock_client = MagicMock()
        mock_client.ams_set_filament_setting.return_value = True
        mock_client.extrusion_cali_sel.return_value = True

        status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "", "tray_type": "PLA"}]}])

        # Cloud responds with a real filament_id for the PFCN preset — exactly
        # what the Configure modal already exploits.
        mock_cloud = MagicMock()
        mock_cloud.is_authenticated = True

        async def fake_get_detail(setting_id):
            assert setting_id == "PFCN80e80c1f79db85"
            return {"filament_id": "GFL05", "name": "Polymaker PLA Matte"}

        async def fake_close():
            return None

        mock_cloud.get_setting_detail = fake_get_detail
        mock_cloud.close = fake_close

        async def fake_build_cloud(_db, _user):
            return mock_cloud

        with (
            patch("backend.app.services.printer_manager.printer_manager") as mock_pm,
            patch("backend.app.api.routes.cloud.build_authenticated_cloud", new=fake_build_cloud),
        ):
            mock_pm.get_client.return_value = mock_client
            mock_pm.get_status.return_value = status

            response = await async_client.post(
                "/api/v1/inventory/assignments",
                json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3},
            )

            assert response.status_code == 200
            call_kwargs = mock_client.ams_set_filament_setting.call_args
            # tray_info_idx is the resolved cloud filament_id; setting_id is the
            # original PFCN (which the slicer needs separately).
            assert call_kwargs.kwargs["tray_info_idx"] == "GFL05"
            assert call_kwargs.kwargs["setting_id"] == "PFCN80e80c1f79db85"


def _make_printing_status(ams_data, state="RUNNING"):
    """Printer status carrying an explicit gcode state for the runout guard."""
    status = _make_mock_status(ams_data=ams_data)
    status.state = state
    return status


class TestAutoUnlinkDuringRunout:
    """A slot that reports empty mid-print is a filament runout, not a spool
    swap — the spool is still in the AMS, just consumed.

    Unlinking there erased the only record of which spool fed the print, so the
    completion path had nothing to charge the runout segment to. With AMS
    filament backup that is the normal course of events, not an edge case.
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cleared_tray_data_keeps_the_assignment_while_printing(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="H2D")
        spool = await spool_factory(material="ABS", rgba="616777FF")

        assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=0,
            tray_id=2,
            fingerprint_color="616777FF",
            fingerprint_type="ABS",
        )
        db_session.add(assignment)
        await db_session.commit()

        # The firmware clears colour and type when it unloads a spool it just
        # emptied (state 26 = "not loaded").
        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
            mock_pm_main.get_status.return_value = _make_printing_status(ams_data)
            mock_pm_main.get_model.return_value = "H2D"
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        # on_ams_change committed through its own session — drop this one's
        # identity map so the assertion reads the database, not a cached row.
        db_session.expunge_all()
        remaining = await db_session.get(SpoolAssignment, assignment.id)
        assert remaining is not None, "runout must not unlink the spool that fed the print"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_cleared_tray_data_still_unlinks_when_idle(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """Off the print, an emptied slot really does mean the spool is gone."""
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="H2D")
        spool = await spool_factory(material="ABS", rgba="616777FF")

        assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=0,
            tray_id=2,
            fingerprint_color="616777FF",
            fingerprint_type="ABS",
        )
        db_session.add(assignment)
        await db_session.commit()
        assignment_id = assignment.id

        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
            mock_pm_main.get_status.return_value = _make_printing_status(ams_data, state="IDLE")
            mock_pm_main.get_model.return_value = "H2D"
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        db_session.expunge_all()
        assert await db_session.get(SpoolAssignment, assignment_id) is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_genuinely_different_filament_still_unlinks_while_printing(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """The guard is for blank tray data only — a real swap must still
        reconcile, or the wrong spool gets charged."""
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="H2D")
        spool = await spool_factory(material="ABS", rgba="616777FF")

        assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=0,
            tray_id=2,
            fingerprint_color="616777FF",
            fingerprint_type="ABS",
        )
        db_session.add(assignment)
        await db_session.commit()
        assignment_id = assignment.id

        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "PETG", "tray_color": "6EE53CFF", "state": 11}]}]

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
            mock_pm_main.get_status.return_value = _make_printing_status(ams_data)
            mock_pm_main.get_model.return_value = "H2D"
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        db_session.expunge_all()
        assert await db_session.get(SpoolAssignment, assignment_id) is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_slot_missing_from_ams_data_keeps_the_assignment_while_printing(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="H2D")
        spool = await spool_factory(material="ABS", rgba="616777FF")

        assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=0,
            tray_id=2,
            fingerprint_color="616777FF",
            fingerprint_type="ABS",
        )
        db_session.add(assignment)
        await db_session.commit()
        assignment_id = assignment.id

        # Tray 2 dropped out of the payload entirely.
        ams_data = [{"id": 0, "tray": [{"id": 0, "tray_type": "ABS", "tray_color": "FFFFFFFF", "state": 11}]}]

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
            mock_pm_main.get_status.return_value = _make_printing_status(ams_data)
            mock_pm_main.get_model.return_value = "H2D"
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer.id, ams_data)

        db_session.expunge_all()
        assert await db_session.get(SpoolAssignment, assignment_id) is not None


class TestSpoolmanSlotAssignmentDuringRunout:
    """`spoolman_slot_assignments` is how a tag-less spool assigned through the
    Bambuddy UI is resolved at completion (#1459). Deleting the row when a slot
    empties mid-print loses the runout segment's usage — the same failure the
    internal inventory's auto-unlink had, so it needs the same guard."""

    async def _enable_spoolman(self, db_session):
        from backend.app.models.settings import Settings

        for key, value in (
            ("spoolman_enabled", "true"),
            ("spoolman_sync_mode", "auto"),
            ("spoolman_url", "http://spoolman.test"),
        ):
            db_session.add(Settings(key=key, value=value))
        await db_session.commit()

    async def _run(self, printer_id, state):
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change

        # A tray the firmware has cleared: parse_ams_tray returns None, which
        # is what marks the slot empty for the cleanup pass.
        ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}]

        spoolman_client = MagicMock()
        spoolman_client.health_check = AsyncMock(return_value=True)
        spoolman_client.get_spools = AsyncMock(return_value=[])
        spoolman_client.sync_ams_tray = AsyncMock(return_value=None)
        # None is what marks the slot empty for the cleanup pass.
        spoolman_client.parse_ams_tray.return_value = None

        with (
            patch("backend.app.main.printer_manager") as mock_pm_main,
            patch("backend.app.main.mqtt_relay") as mock_relay,
            patch("backend.app.main.ws_manager") as mock_ws,
            patch("backend.app.main.get_spoolman_client", new=AsyncMock(return_value=spoolman_client)),
        ):
            mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427")
            mock_pm_main.get_status.return_value = state
            mock_pm_main.get_model.return_value = "H2D"
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer_id, ams_data)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_slot_row_survives_a_runout(
        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
    ):
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment

        await self._enable_spoolman(db_session)
        printer = await printer_factory(name="H2D")
        row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41)
        db_session.add(row)
        await db_session.commit()
        row_id = row.id

        await self._run(printer.id, _make_printing_status(None))

        db_session.expunge_all()
        assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_slot_row_is_still_cleaned_up_when_idle(
        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
    ):
        """Proves the guard is what saved the row above, not an unreachable
        code path."""
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment

        await self._enable_spoolman(db_session)
        printer = await printer_factory(name="H2D")
        row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41)
        db_session.add(row)
        await db_session.commit()
        row_id = row.id

        await self._run(printer.id, _make_printing_status(None, state="IDLE"))

        db_session.expunge_all()
        assert await db_session.get(SpoolmanSlotAssignment, row_id) is None