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 AsyncMock, 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"

    @pytest.mark.asyncio
    @pytest.mark.integration
    @pytest.mark.parametrize("slot_generic", ["GFL96", "GFL95"])
    async def test_no_preset_keeps_a_silk_or_high_speed_generic_on_the_slot(
        self, async_client: AsyncClient, printer_factory, spool_factory, slot_generic
    ):
        """PLA Silk and PLA High Speed joined the generic table (#3273), but the
        reuse check kept its old set: a slot set to one of them still keeps it
        for a preset-less PLA spool rather than dropping to Generic PLA."""
        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

        status = _make_mock_status(
            ams_data=[{"id": 0, "tray": [{"id": 0, "tray_info_idx": slot_generic, "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"] == slot_generic


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"
        # The filament id the slot was given goes with it, so the slot card can
        # tell a later re-configuration from elsewhere (#3216).
        assert presets["1"]["tray_info_idx"] == "GFA05"

    @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",
            tray_info_idx="GFA01",
        )
        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"
        # ...and so must the filament id recorded with it (#3216).
        assert presets["2"]["tray_info_idx"] == "GFL96"

    @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 TestAssignSpoolPresenceBit:
    """#3084: the slot the firmware says is full, and the cache says is empty.

    ``apply_tray_exist_bits`` stamps ``state = 9`` on every slot whose
    ``tray_exist_bits`` bit is 0 and annotates ``exists`` on every slot it
    looks at — but when the bit comes back it only refreshes ``exists`` and
    leaves the 9 where it was. Swap a Bambu spool for a non-RFID one and the
    slot sits at ``exists=True, state=9`` until something configures it.

    Reported on an H2D/H2C AMS-HT: remove the Bambu spool (bits ``f``), insert
    a third-party one 9 seconds later (bits ``1000f``), then Assign Spool 28
    seconds after that — and no ``ams_filament_setting`` was published at all.
    Configure worked, because it publishes unconditionally.
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_bit_overrules_a_stale_empty_state(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """exists=True with a leftover state=9 — MQTT must fire."""
        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

        tray_data = {"id": 0, "state": 9, "exists": True, "tray_type": "", "tray_color": "", "tray_info_idx": ""}
        status = _make_mock_status(ams_data=[{"id": 128, "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": 128, "tray_id": 0},
            )

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

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_empty_bit_does_not_start_suppressing_pushes(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """The bit overrules the 9 and nothing else.

        Reading it the other way too would be tidier — skip the push firmware
        is going to drop — but it also means a slot that silently stops
        configuring on whichever AMS variant we compute the bit position
        wrong for. The saving is one MQTT message; the failure is the bug
        this commit is fixing, inverted. So a state that does not say "empty"
        still publishes, exactly as it did before.
        """
        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

        tray_data = {"id": 3, "state": 11, "exists": False, "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()
        assert response.json()["pending_config"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_pre_assign_workflow_still_skips_a_genuinely_empty_slot(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """SpoolBuddy weighs a spool and assigns it before it goes in. Bit
        clear and state 9 agree that the slot is empty, so the push is still
        deferred to the replay — unchanged."""
        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

        tray_data = {"id": 3, "state": 9, "exists": False, "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_not_called()
        body = response.json()
        assert body["configured"] is False
        assert body["pending_config"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_unannotated_tray_still_reads_the_state(
        self, async_client: AsyncClient, printer_factory, spool_factory
    ):
        """No presence bit in the payload → the 9/10 heuristic still decides.

        The external spool's ``vt_tray`` has no bit in the mask, and neither do
        the hand-built payloads every other test in this file uses.
        """
        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

        status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "state": 9, "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_not_called()
        assert response.json()["pending_config"] is True

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deferred_config_fires_for_a_spool_the_ams_cannot_name(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """The pre-assign workflow's half of the same bug.

        A non-RFID spool inserted into a pre-assigned slot brings no
        ``tray_type`` with it, and the stale 9 kept the replay's "loaded" test
        false, so the deferred configuration never fired for it either. The
        presence bit is the only thing in the payload that changed.
        """
        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_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()

        ams_data = [{"id": 2, "tray": [{"id": 3, "state": 9, "exists": True, "tray_type": "", "tray_color": ""}]}]

        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)

        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 = MagicMock(name="H2D", serial_number="0948BB540200427")
            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)

        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"

        # The assignment is still there — the pass that fires the config is the
        # same pass that deletes stale ones (#3100).
        db_session.expunge_all()
        assert await db_session.get(SpoolAssignment, pre_assignment.id) is not None


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 --
        once it has stayed empty for the grace period (#3186)."""
        from unittest.mock import AsyncMock

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

        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()

            clock = [1000.0]
            with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
                await on_ams_change(printer.id, ams_data)
                db_session.expunge_all()
                assert await db_session.get(SpoolAssignment, assignment_id) is not None

                clock[0] += slot_unlink_grace.GRACE_SECONDS
                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 TestAutoUnlinkOccupiedSlot:
    """#3100: six saved assignments deleted across three X1 Carbons.

    Each one had an explicit earlier assignment and then an ``Auto-unlink ...
    fingerprint mismatch``, and the reporter recovered the mappings from logs
    because they were gone from the inventory API, not merely hidden. Two
    shapes, one cause: a slot the presence bit calls occupied while the tray
    reports nothing about what is in it.

    ``cur=/ fp=BCBCBCFF/PLA spool=8A8F92FF/PLA`` — an established assignment,
    a blank idle report, and the row deleted. The spool never went anywhere;
    the AMS simply had nothing to say about a filament it cannot read.
    """

    @staticmethod
    async def _run(printer_id, ams_data, status):
        from unittest.mock import AsyncMock

        from backend.app.main import on_ams_change

        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="X1C", serial_number="0948BB540200427")
            mock_pm_main.get_status.return_value = status
            mock_pm_main.get_model.return_value = "X1C"
            mock_pm_main.get_client.return_value = None
            mock_relay.on_ams_change = AsyncMock()
            mock_ws.send_printer_status = AsyncMock()
            mock_ws.broadcast = AsyncMock()

            await on_ams_change(printer_id, ams_data)

    @staticmethod
    async def _assignment(db_session, printer, spool):
        from backend.app.models.spool_assignment import SpoolAssignment

        assignment = SpoolAssignment(
            spool_id=spool.id,
            printer_id=printer.id,
            ams_id=1,
            tray_id=1,
            fingerprint_color="BCBCBCFF",
            fingerprint_type="PLA",
        )
        db_session.add(assignment)
        await db_session.commit()
        return assignment.id

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_blank_report_from_an_occupied_slot_keeps_the_assignment(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
        assignment_id = await self._assignment(db_session, printer, spool)

        ams_data = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
        await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))

        db_session.expunge_all()
        assert await db_session.get(SpoolAssignment, assignment_id) is not None, (
            "a spool the AMS cannot identify is not a spool that was removed"
        )

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_blank_report_from_an_empty_slot_still_unlinks(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """The guard reads the bit, not the blankness — take the spool out and
        the assignment still goes, which is what makes the test above a
        distinction rather than a blanket reprieve."""
        from backend.app.models.spool_assignment import SpoolAssignment
        from backend.app.services import slot_unlink_grace

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
        assignment_id = await self._assignment(db_session, printer, spool)

        ams_data = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))
            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))

        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_in_an_occupied_slot_still_unlinks(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """The guard is for a blank report only. A slot that names a filament
        which is not the assigned spool is a swap, bit set or not."""
        from backend.app.models.spool_assignment import SpoolAssignment

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
        assignment_id = await self._assignment(db_session, printer, spool)

        ams_data = [
            {"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "PETG", "tray_color": "00FF00FF", "state": 11}]}
        ]
        await self._run(printer.id, ams_data, _make_printing_status(ams_data, state="IDLE"))

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


class TestAutoUnlinkHeldForGrace:
    """#3186: one push is not proof a spool was taken out.

    An idle X1 Carbon on X1Plus sent a push that cleared a whole AMS unit --
    presence bits off, colour and type blank -- and four saved assignments were
    deleted in the same instant (``Auto-unlinked 4 stale spool assignments``).
    The spools never moved. A removal is now held for the grace period and
    unlinked only if the slot is still empty then.
    """

    @staticmethod
    async def _assign_unit(db_session, printer, spool):
        from backend.app.models.spool_assignment import SpoolAssignment

        ids = []
        for tray_id in range(4):
            a = SpoolAssignment(
                spool_id=spool.id,
                printer_id=printer.id,
                ams_id=1,
                tray_id=tray_id,
                fingerprint_color="858585FF",
                fingerprint_type="PLA",
            )
            db_session.add(a)
            await db_session.commit()
            ids.append(a.id)
        return ids

    @staticmethod
    def _unit(blank: bool):
        return [
            {
                "id": 1,
                "tray": [
                    {"id": t, "exists": False, "tray_type": "", "tray_color": "", "state": 9}
                    if blank
                    else {"id": t, "exists": True, "tray_type": "PLA", "tray_color": "858585FF", "state": 11}
                    for t in range(4)
                ],
            }
        ]

    @staticmethod
    async def _surviving(db_session, ids):
        from backend.app.models.spool_assignment import SpoolAssignment

        db_session.expunge_all()
        return [i for i in ids if await db_session.get(SpoolAssignment, i) is not None]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_whole_unit_blanking_for_one_push_keeps_every_assignment(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        from backend.app.services import slot_unlink_grace

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="000000FF")
        ids = await self._assign_unit(db_session, printer, spool)

        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            blank, back = self._unit(blank=True), self._unit(blank=False)
            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
            assert await self._surviving(db_session, ids) == ids

            clock[0] += 30
            await TestAutoUnlinkOccupiedSlot._run(printer.id, back, _make_printing_status(back, state="IDLE"))

            # The recovery dropped the holds, so a blank long after the first
            # one starts a fresh clock instead of confirming the old one.
            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))

        assert await self._surviving(db_session, ids) == ids

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_unit_missing_from_one_push_keeps_its_assignments(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="000000FF")
        ids = await self._assign_unit(db_session, printer, spool)

        # AMS 1 dropped out of the payload entirely.
        other_unit = [{"id": 0, "tray": [{"id": 0, "exists": True, "tray_type": "PLA", "tray_color": "FFFFFFFF"}]}]
        await TestAutoUnlinkOccupiedSlot._run(printer.id, other_unit, _make_printing_status(other_unit, state="IDLE"))

        assert await self._surviving(db_session, ids) == ids

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_slot_that_stays_empty_is_unlinked_by_the_recheck(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """No second push has to arrive: a spool taken out and left out never
        changes the AMS hash again, so the held removal re-checks itself."""
        from backend.app.main import _recheck_held_unlinks
        from backend.app.services import slot_unlink_grace

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="000000FF")
        ids = await self._assign_unit(db_session, printer, spool)
        blank = self._unit(blank=True)

        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
            assert printer.id in slot_unlink_grace._recheck_tasks, "a held removal must schedule its re-check"
            assert await self._surviving(db_session, ids) == ids

            clock[0] += slot_unlink_grace.GRACE_SECONDS
            with (
                patch("backend.app.main.printer_manager") as mock_pm_main,
                patch("backend.app.main.ws_manager") as mock_ws,
            ):
                mock_pm_main.get_status.return_value = _make_printing_status(blank, state="IDLE")
                mock_ws.broadcast = AsyncMock()
                await _recheck_held_unlinks(printer.id)

        assert await self._surviving(db_session, ids) == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_the_recheck_leaves_a_slot_that_came_back(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        from backend.app.main import _recheck_held_unlinks
        from backend.app.services import slot_unlink_grace

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="000000FF")
        ids = await self._assign_unit(db_session, printer, spool)
        blank, back = self._unit(blank=True), self._unit(blank=False)

        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            await TestAutoUnlinkOccupiedSlot._run(printer.id, blank, _make_printing_status(blank, state="IDLE"))
            clock[0] += slot_unlink_grace.GRACE_SECONDS
            with (
                patch("backend.app.main.printer_manager") as mock_pm_main,
                patch("backend.app.main.ws_manager") as mock_ws,
            ):
                mock_pm_main.get_status.return_value = _make_printing_status(back, state="IDLE")
                mock_ws.broadcast = AsyncMock()
                await _recheck_held_unlinks(printer.id)

        assert await self._surviving(db_session, ids) == ids


class TestAutoUnlinkSwapDuringGrace:
    """A non-RFID spool swapped out and another put in within the grace period.

    The new spool reports occupied-but-blank, which on its own keeps an
    assignment (#3100). After an *empty* report it means the spool changed,
    so the held removal has to keep running rather than be cancelled -- or the
    old spool would stay assigned to the new one indefinitely.
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_swap_still_unlinks_once_the_grace_runs_out(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        from backend.app.models.spool_assignment import SpoolAssignment
        from backend.app.services import slot_unlink_grace

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
        assignment_id = await TestAutoUnlinkOccupiedSlot._assignment(db_session, printer, spool)

        taken_out = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
        new_spool = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            await TestAutoUnlinkOccupiedSlot._run(printer.id, taken_out, _make_printing_status(taken_out, state="IDLE"))
            clock[0] += 20
            await TestAutoUnlinkOccupiedSlot._run(printer.id, new_spool, _make_printing_status(new_spool, state="IDLE"))
            db_session.expunge_all()
            assert await db_session.get(SpoolAssignment, assignment_id) is not None

            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await TestAutoUnlinkOccupiedSlot._run(printer.id, new_spool, _make_printing_status(new_spool, state="IDLE"))

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

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_new_assignment_to_the_slot_starts_clean(
        self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession
    ):
        """Assigning a spool to the slot after the swap -- even the same spool
        again -- makes a new row, and the old clock does not carry over to it."""
        from backend.app.models.spool_assignment import SpoolAssignment
        from backend.app.services import slot_unlink_grace

        printer = await printer_factory(name="X1C")
        spool = await spool_factory(material="PLA", rgba="8A8F92FF")
        old_id = await TestAutoUnlinkOccupiedSlot._assignment(db_session, printer, spool)

        taken_out = [{"id": 1, "tray": [{"id": 1, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
        back_in = [{"id": 1, "tray": [{"id": 1, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            await TestAutoUnlinkOccupiedSlot._run(printer.id, taken_out, _make_printing_status(taken_out, state="IDLE"))

            # The user puts the spool back and assigns it again through the
            # endpoint, which replaces the row -- on SQLite quite possibly under
            # the same id, so only dropping the hold keeps the old clock off it.
            hold_key = ("inventory", 1, 1, spool.id)
            assert slot_unlink_grace.is_held(printer.id, hold_key)
            with patch("backend.app.services.printer_manager.printer_manager") as mock_pm:
                mock_pm.get_client.return_value = None
                mock_pm.get_status.return_value = _make_mock_status(ams_data=back_in)
                response = await async_client.post(
                    "/api/v1/inventory/assignments",
                    json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 1, "tray_id": 1},
                )
            assert response.status_code == 200
            assert not slot_unlink_grace.is_held(printer.id, hold_key), "an assignment must drop the old hold"
            new_id = response.json()["id"]
            assert await db_session.get(SpoolAssignment, old_id) is None or new_id == old_id

            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await TestAutoUnlinkOccupiedSlot._run(printer.id, back_in, _make_printing_status(back_in, state="IDLE"))

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


class TestSpoolmanSlotUnlinkHeldForGrace:
    """Spoolman mode's half of #3186: the slot row waits out the same grace."""

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

        await TestSpoolmanSlotAssignmentDuringRunout()._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

        blank = [{"id": 0, "tray": [{"id": 2, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
        clock = [1000.0]
        with (
            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_ws.broadcast = AsyncMock()
            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)
            db_session.expunge_all()
            assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None

            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)

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

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

        await TestSpoolmanSlotAssignmentDuringRunout()._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

        taken_out = [{"id": 0, "tray": [{"id": 2, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
        new_spool = [{"id": 0, "tray": [{"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
        clock = [1000.0]
        with (
            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_ws.broadcast = AsyncMock()
            await _expire_spoolman_empty_slots(printer.id, taken_out, printing_now=False)
            clock[0] += 20
            await _expire_spoolman_empty_slots(printer.id, new_spool, printing_now=False)
            db_session.expunge_all()
            assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None

            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await _expire_spoolman_empty_slots(printer.id, new_spool, printing_now=False)

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

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_occupied_unreadable_slot_with_no_hold_keeps_its_row(
        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
    ):
        """#3100 still holds for the re-check: never reported empty, never unlinked."""
        from backend.app.main import _expire_spoolman_empty_slots
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
        from backend.app.services import slot_unlink_grace

        await TestSpoolmanSlotAssignmentDuringRunout()._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

        unreadable = [{"id": 0, "tray": [{"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9}]}]
        clock = [1000.0]
        with (
            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_ws.broadcast = AsyncMock()
            for _ in range(3):
                await _expire_spoolman_empty_slots(printer.id, unreadable, printing_now=False)
                clock[0] += slot_unlink_grace.GRACE_SECONDS

        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_recheck_keeps_a_row_whose_slot_reads_again(
        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
    ):
        from backend.app.main import _expire_spoolman_empty_slots
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
        from backend.app.services import slot_unlink_grace

        await TestSpoolmanSlotAssignmentDuringRunout()._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

        blank = [{"id": 0, "tray": [{"id": 2, "exists": False, "tray_type": "", "tray_color": "", "state": 9}]}]
        back = [{"id": 0, "tray": [{"id": 2, "exists": True, "tray_type": "PLA", "tray_color": "858585FF"}]}]
        clock = [1000.0]
        with (
            patch.object(slot_unlink_grace, "_now", lambda: clock[0]),
            patch("backend.app.main.ws_manager") as mock_ws,
        ):
            mock_ws.broadcast = AsyncMock()
            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)
            clock[0] += 30
            await _expire_spoolman_empty_slots(printer.id, back, printing_now=False)
            clock[0] += slot_unlink_grace.GRACE_SECONDS
            await _expire_spoolman_empty_slots(printer.id, blank, printing_now=False)

        db_session.expunge_all()
        assert await db_session.get(SpoolmanSlotAssignment, row_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, tray=None):
        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": [tray or {"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_an_occupied_slot_the_ams_cannot_read_keeps_its_row(
        self, async_client: AsyncClient, printer_factory, db_session: AsyncSession
    ):
        """Spoolman mode's half of #3100.

        parse_ams_tray calls a tray with no type empty, and a tag-less spool
        has none until something configures it — so the row assigned through
        the UI was deleted by the first idle push after the spool went in.
        Firmware's presence bit is the same answer the built-in inventory
        uses, so the two modes stay in step.
        """
        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"),
            tray={"id": 2, "exists": True, "tray_type": "", "tray_color": "", "state": 9},
        )

        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 -- once the slot has stayed empty for the grace period."""
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
        from backend.app.services import slot_unlink_grace

        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

        clock = [1000.0]
        with patch.object(slot_unlink_grace, "_now", lambda: clock[0]):
            await self._run(printer.id, _make_printing_status(None, state="IDLE"))
            db_session.expunge_all()
            assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None

            clock[0] += slot_unlink_grace.GRACE_SECONDS
            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