"""Integration tests for inventory spool assignment — tray_info_idx resolution. Tests that the spool's own slicer_filament (including PFUS* cloud-synced custom presets) takes priority, with slot reuse and generic fallback as lower-priority fallbacks. """ from unittest.mock import MagicMock, patch import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from backend.app.models.spool import Spool @pytest.fixture async def spool_factory(db_session: AsyncSession): """Factory to create test spools.""" _counter = [0] async def _create_spool(**kwargs): _counter[0] += 1 defaults = { "material": "PLA", "subtype": "Basic", "brand": "Devil Design", "color_name": "Red", "rgba": "FF0000FF", "label_weight": 1000, "weight_used": 0, "slicer_filament": "PFUS9ac902733670a9", } defaults.update(kwargs) spool = Spool(**defaults) db_session.add(spool) await db_session.commit() await db_session.refresh(spool) return spool return _create_spool def _make_mock_status(ams_data=None, vt_tray=None, nozzles=None, ams_extruder_map=None): """Build a mock printer status with optional AMS/nozzle data.""" status = MagicMock() raw = {} if ams_data is not None: raw["ams"] = {"ams": ams_data} if vt_tray is not None: raw["vt_tray"] = vt_tray status.raw_data = raw status.nozzles = nozzles or [MagicMock(nozzle_diameter="0.4")] status.ams_extruder_map = ams_extruder_map return status class TestAssignSpoolTrayInfoIdx: """Tests for tray_info_idx resolution during spool assignment.""" @pytest.mark.asyncio @pytest.mark.integration async def test_pfus_slicer_filament_used_directly(self, async_client: AsyncClient, printer_factory, spool_factory): """PFUS* cloud-synced custom preset IDs are sent to the printer.""" 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"] == "PFUS9ac902733670a9" @pytest.mark.asyncio @pytest.mark.integration async def test_spool_preset_takes_priority_over_slot( self, async_client: AsyncClient, printer_factory, spool_factory ): """Spool's own slicer_filament takes priority over slot's existing preset.""" 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 # Spool's own preset wins over slot's existing one assert call_kwargs.kwargs["tray_info_idx"] == "PFUS9ac902733670a9" @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 own slicer_filament is used regardless of what's on the slot.""" 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"] == "PFUS9ac902733670a9" @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_used_over_slot_pfus(self, async_client: AsyncClient, printer_factory, spool_factory): """Spool's own PFUS preset is used even when slot has a different PFUS.""" 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 # Spool's own preset wins assert call_kwargs.kwargs["tray_info_idx"] == "PFUS1111111111" @pytest.mark.asyncio @pytest.mark.integration async def test_generic_on_slot_not_reused_over_spool_preset( self, async_client: AsyncClient, printer_factory, spool_factory ): """Generic ID on slot (e.g. GFB99) must not override spool's own preset.""" 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 # Spool's preset wins — generic on slot must not be sticky assert call_kwargs.kwargs["tray_info_idx"] == "PFUScda4c46fc9031" @pytest.mark.asyncio @pytest.mark.integration async def test_no_preset_with_generic_on_slot_still_uses_generic( self, async_client: AsyncClient, printer_factory, spool_factory ): """Spool without preset + generic on slot → generic fallback (not slot reuse).""" printer = await printer_factory(name="P2S") spool = await spool_factory(slicer_filament=None, material="ABS") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot has generic ABS status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 1, "tray_info_idx": "GFB99", "tray_type": "ABS"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # Still gets generic, but via fallback — not via sticky reuse assert call_kwargs.kwargs["tray_info_idx"] == "GFB99" @pytest.mark.asyncio @pytest.mark.integration async def test_no_preset_reuses_specific_slot_preset( self, async_client: AsyncClient, printer_factory, spool_factory ): """Spool without preset + specific preset on slot → reuse slot's preset.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament=None, material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot has a specific Bambu PLA preset (not generic) status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 0, "tray_info_idx": "GFA05", "tray_type": "PLA"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # Slot's specific preset is reused when spool has no own preset assert call_kwargs.kwargs["tray_info_idx"] == "GFA05" class TestAssignSpoolPresetMapping: """Tests that assign_spool saves the slot preset mapping for correct UI display.""" @pytest.mark.asyncio @pytest.mark.integration async def test_preset_mapping_saved_with_slicer_filament_name( self, async_client: AsyncClient, printer_factory, spool_factory ): """Slot preset mapping uses slicer_filament_name (not material+subtype).""" printer = await printer_factory(name="X1C") spool = await spool_factory( slicer_filament="GFA05", slicer_filament_name="Bambu PLA Silk", material="PLA", subtype="Silk", brand="Bambu", ) mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 1, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1}, ) assert response.status_code == 200 # Verify via the slot presets API presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets") assert presets_resp.status_code == 200 presets = presets_resp.json() # Key is str(ams_id * 4 + tray_id) — ams 0, tray 1 → "1" assert "1" in presets # Must use slicer_filament_name, NOT "PLA Silk" from material+subtype assert presets["1"]["preset_name"] == "Bambu PLA Silk" assert presets["1"]["preset_id"] == "GFSA05" @pytest.mark.asyncio @pytest.mark.integration async def test_preset_mapping_overwrites_old_mapping( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """Assigning a new spool overwrites the old slot preset mapping.""" from backend.app.models.slot_preset import SlotPresetMapping printer = await printer_factory(name="X1C") # Pre-existing mapping (e.g. from previous manual configuration) old_mapping = SlotPresetMapping( printer_id=printer.id, ams_id=0, tray_id=2, preset_id="GFSA01", preset_name="Bambu PLA Matte", preset_source="cloud", ) db_session.add(old_mapping) await db_session.commit() # Assign a "Generic PLA Silk" spool to same slot spool = await spool_factory( slicer_filament="GFL96", slicer_filament_name="Generic PLA Silk", material="PLA", subtype="Silk", ) mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 2, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 2}, ) assert response.status_code == 200 # Verify via the slot presets API to avoid stale session cache presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets") assert presets_resp.status_code == 200 presets = presets_resp.json() # Key is str(ams_id * 4 + tray_id) — ams 0, tray 2 → "2" assert "2" in presets # Old "Bambu PLA Matte" must be overwritten assert presets["2"]["preset_name"] == "Generic PLA Silk" assert presets["2"]["preset_id"] == "GFSL96" @pytest.mark.asyncio @pytest.mark.integration async def test_preset_mapping_fallback_to_tray_sub_brands( self, async_client: AsyncClient, printer_factory, spool_factory ): """When slicer_filament_name is null, falls back to tray_sub_brands.""" from backend.app.models.slot_preset import SlotPresetMapping printer = await printer_factory(name="A1M") spool = await spool_factory( slicer_filament="GFL05", slicer_filament_name=None, material="PLA", subtype="Matte", brand="Overture", ) mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 # Verify via the slot presets API presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets") assert presets_resp.status_code == 200 presets = presets_resp.json() # Key is str(ams_id * 4 + tray_id) — ams 0, tray 0 → "0" assert "0" in presets # Falls back to tray_sub_brands ("Overture PLA Matte") assert presets["0"]["preset_name"] == "Overture PLA Matte" class TestAssignSpoolLiveCaliIdx: """P9-TEST-BE-3: assign_spool falls back to live tray cali_idx when no K-profile stored.""" @pytest.mark.asyncio @pytest.mark.integration async def test_no_kprofile_uses_live_cali_idx(self, async_client: AsyncClient, printer_factory, spool_factory): """When no KProfile row exists, live tray cali_idx is sent via extrusion_cali_sel.""" 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": 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() call_kwargs = mock_client.extrusion_cali_sel.call_args[1] assert call_kwargs["cali_idx"] == 42 @pytest.mark.asyncio @pytest.mark.integration async def test_no_kprofile_no_live_cali_idx_nothing_sent( self, async_client: AsyncClient, printer_factory, spool_factory ): """When tray has no cali_idx, extrusion_cali_sel is not called.""" 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_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_negative_live_cali_idx_not_sent(self, async_client: AsyncClient, printer_factory, spool_factory): """A negative live cali_idx (-1) is invalid and must not be sent.""" 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_not_called() class TestAssignSpoolEmptySlotPreConfig: """SpoolBuddy primary workflow: weigh-then-assign before the spool is in the AMS. Bambu firmware silently drops ams_filament_setting / extrusion_cali_sel for unloaded slots — there's no filament context for the cali_idx to attach to. The endpoint persists the SpoolAssignment row with an empty fingerprint_type (the "pending config" marker) and skips the MQTT publish; on_ams_change re-fires the full configuration when filament is later inserted. """ @pytest.mark.asyncio @pytest.mark.integration async def test_empty_slot_skips_mqtt_but_persists_assignment( self, async_client: AsyncClient, printer_factory, spool_factory ): """Assigning to an empty slot skips MQTT and returns pending_config=True.""" 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 # Slot found but empty (tray_type=""): the SpoolBuddy scenario 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 body = response.json() assert body["pending_config"] is True assert body["configured"] is False # Critical: no MQTT was published (firmware would drop it) mock_client.ams_set_filament_setting.assert_not_called() mock_client.extrusion_cali_sel.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_empty_slot_no_ams_data_skips_mqtt(self, async_client: AsyncClient, printer_factory, spool_factory): """No AMS data at all (printer offline, no telemetry yet) → still pre-config.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="GFL05", material="PLA") mock_client = MagicMock() # No AMS data — fingerprint_type stays None, treated as empty status = _make_mock_status(ams_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 assert response.json()["pending_config"] is True mock_client.ams_set_filament_setting.assert_not_called() @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" 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_falls_back_to_tray_type_empty( self, async_client: AsyncClient, printer_factory, spool_factory ): """Older firmware without state field + empty tray_type → pending.""" 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, "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 # Legacy fallback: empty tray_type + no state → treated as empty. mock_client.ams_set_filament_setting.assert_not_called() body = response.json() assert body["pending_config"] 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_state_never_eleven_firmware_with_empty_tray_marks_pending( self, async_client: AsyncClient, printer_factory, spool_factory ): """Same firmwares as above, but the slot is truly unconfigured (tray_type=''). Neither signal points to 'loaded', so this should still pending-config — the user has to configure or insert filament before MQTT can fire. Pins that the disjunction didn't accidentally flip empty slots into the loaded branch.""" 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": 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_not_called() body = response.json() assert body["pending_config"] 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