"""Integration tests for inventory spool assignment — tray_info_idx resolution. Tests that the spool's own slicer_filament (including PFUS* cloud-synced custom presets) takes priority, with slot reuse and generic fallback as lower-priority fallbacks. """ from unittest.mock import MagicMock, patch import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from backend.app.models.spool import Spool @pytest.fixture async def spool_factory(db_session: AsyncSession): """Factory to create test spools.""" _counter = [0] async def _create_spool(**kwargs): _counter[0] += 1 defaults = { "material": "PLA", "subtype": "Basic", "brand": "Devil Design", "color_name": "Red", "rgba": "FF0000FF", "label_weight": 1000, "weight_used": 0, "slicer_filament": "PFUS9ac902733670a9", } defaults.update(kwargs) spool = Spool(**defaults) db_session.add(spool) await db_session.commit() await db_session.refresh(spool) return spool return _create_spool def _make_mock_status(ams_data=None, vt_tray=None, nozzles=None, ams_extruder_map=None): """Build a mock printer status with optional AMS/nozzle data.""" status = MagicMock() raw = {} if ams_data is not None: raw["ams"] = {"ams": ams_data} if vt_tray is not None: raw["vt_tray"] = vt_tray status.raw_data = raw status.nozzles = nozzles or [MagicMock(nozzle_diameter="0.4")] status.ams_extruder_map = ams_extruder_map return status class TestAssignSpoolTrayInfoIdx: """Tests for tray_info_idx resolution during spool assignment.""" @pytest.mark.asyncio @pytest.mark.integration async def test_pfus_slicer_filament_falls_back_to_generic( self, async_client: AsyncClient, printer_factory, spool_factory ): """PFUS* cloud setting_ids are rejected by the slicer as tray_info_idx, so the no-kp path falls back to the generic material id (PLA → GFL99). The K-profile realignment path translates PFUS → P-prefix when a stored kp exists; that's covered separately.""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "", "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFL99" @pytest.mark.asyncio @pytest.mark.integration async def test_pfus_spool_reuses_valid_slot_preset(self, async_client: AsyncClient, printer_factory, spool_factory): """When the spool's PFUS gets discarded as slicer-invalid, the slot's existing valid P-prefix preset is reused if it matches the spool's material — preserves the printer's calibration context rather than resetting to generic.""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot already configured by slicer with cloud-synced preset status = _make_mock_status( ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "P4d64437" @pytest.mark.asyncio @pytest.mark.integration async def test_spool_preset_used_even_if_different_material_on_slot( self, async_client: AsyncClient, printer_factory, spool_factory ): """Spool's material drives the fallback generic id. Slot's existing PLA preset is overridden because the spool is PETG → GFG99.""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PETG") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot currently has PLA but spool is PETG status = _make_mock_status( ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFG99" @pytest.mark.asyncio @pytest.mark.integration async def test_gf_slicer_filament_kept(self, async_client: AsyncClient, printer_factory, spool_factory): """Standard GF* IDs from spool.slicer_filament are used directly.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="GFL05", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFL05" @pytest.mark.asyncio @pytest.mark.integration async def test_empty_slicer_filament_uses_generic(self, async_client: AsyncClient, printer_factory, spool_factory): """Spool with no slicer_filament gets a generic ID from material type.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament=None, material="ABS") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "ABS"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFB99" @pytest.mark.asyncio @pytest.mark.integration async def test_spool_pfus_falls_back_to_generic_over_slot_pfus( self, async_client: AsyncClient, printer_factory, spool_factory ): """Both spool and slot have PFUS values — both rejected as tray_info_idx — falls back to generic material id (PLA → GFL99).""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFUS1111111111", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot has a PFUS* ID from some previous config status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 0, "tray_info_idx": "PFUS2222222222", "tray_type": "PLA"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFL99" @pytest.mark.asyncio @pytest.mark.integration async def test_generic_on_slot_falls_back_to_material_generic( self, async_client: AsyncClient, printer_factory, spool_factory ): """When spool's PFUS is discarded and slot only has a generic ID, the result comes from the spool's material (ABS → GFB99) — not from the slot. Important because the generic-id check (`not in _generic_id_values`) prevents stale generic reuse and routes the decision through the material fallback.""" printer = await printer_factory(name="P2S") spool = await spool_factory(slicer_filament="PFUScda4c46fc9031", material="ABS") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot stuck on generic ABS from a previous assignment status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 1, "tray_info_idx": "GFB99", "tray_type": "ABS"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "GFB99" @pytest.mark.asyncio @pytest.mark.integration async def test_no_preset_with_generic_on_slot_still_uses_generic( self, async_client: AsyncClient, printer_factory, spool_factory ): """Spool without preset + generic on slot → generic fallback (not slot reuse).""" printer = await printer_factory(name="P2S") spool = await spool_factory(slicer_filament=None, material="ABS") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot has generic ABS status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 1, "tray_info_idx": "GFB99", "tray_type": "ABS"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # Still gets generic, but via fallback — not via sticky reuse assert call_kwargs.kwargs["tray_info_idx"] == "GFB99" @pytest.mark.asyncio @pytest.mark.integration async def test_no_preset_reuses_specific_slot_preset( self, async_client: AsyncClient, printer_factory, spool_factory ): """Spool without preset + specific preset on slot → reuse slot's preset.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament=None, material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Slot has a specific Bambu PLA preset (not generic) status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 0, "tray_info_idx": "GFA05", "tray_type": "PLA"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # Slot's specific preset is reused when spool has no own preset assert call_kwargs.kwargs["tray_info_idx"] == "GFA05" class TestAssignSpoolPresetMapping: """Tests that assign_spool saves the slot preset mapping for correct UI display.""" @pytest.mark.asyncio @pytest.mark.integration async def test_preset_mapping_saved_with_slicer_filament_name( self, async_client: AsyncClient, printer_factory, spool_factory ): """Slot preset mapping uses slicer_filament_name (not material+subtype).""" printer = await printer_factory(name="X1C") spool = await spool_factory( slicer_filament="GFA05", slicer_filament_name="Bambu PLA Silk", material="PLA", subtype="Silk", brand="Bambu", ) mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 1, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1}, ) assert response.status_code == 200 # Verify via the slot presets API presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets") assert presets_resp.status_code == 200 presets = presets_resp.json() # Key is str(ams_id * 4 + tray_id) — ams 0, tray 1 → "1" assert "1" in presets # Must use slicer_filament_name, NOT "PLA Silk" from material+subtype assert presets["1"]["preset_name"] == "Bambu PLA Silk" assert presets["1"]["preset_id"] == "GFSA05" @pytest.mark.asyncio @pytest.mark.integration async def test_preset_mapping_overwrites_old_mapping( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """Assigning a new spool overwrites the old slot preset mapping.""" from backend.app.models.slot_preset import SlotPresetMapping printer = await printer_factory(name="X1C") # Pre-existing mapping (e.g. from previous manual configuration) old_mapping = SlotPresetMapping( printer_id=printer.id, ams_id=0, tray_id=2, preset_id="GFSA01", preset_name="Bambu PLA Matte", preset_source="cloud", ) db_session.add(old_mapping) await db_session.commit() # Assign a "Generic PLA Silk" spool to same slot spool = await spool_factory( slicer_filament="GFL96", slicer_filament_name="Generic PLA Silk", material="PLA", subtype="Silk", ) mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 2, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 2}, ) assert response.status_code == 200 # Verify via the slot presets API to avoid stale session cache presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets") assert presets_resp.status_code == 200 presets = presets_resp.json() # Key is str(ams_id * 4 + tray_id) — ams 0, tray 2 → "2" assert "2" in presets # Old "Bambu PLA Matte" must be overwritten assert presets["2"]["preset_name"] == "Generic PLA Silk" assert presets["2"]["preset_id"] == "GFSL96" @pytest.mark.asyncio @pytest.mark.integration async def test_preset_mapping_fallback_to_tray_sub_brands( self, async_client: AsyncClient, printer_factory, spool_factory ): """When slicer_filament_name is null, falls back to tray_sub_brands.""" from backend.app.models.slot_preset import SlotPresetMapping printer = await printer_factory(name="A1M") spool = await spool_factory( slicer_filament="GFL05", slicer_filament_name=None, material="PLA", subtype="Matte", brand="Overture", ) mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 # Verify via the slot presets API presets_resp = await async_client.get(f"/api/v1/printers/{printer.id}/slot-presets") assert presets_resp.status_code == 200 presets = presets_resp.json() # Key is str(ams_id * 4 + tray_id) — ams 0, tray 0 → "0" assert "0" in presets # Falls back to tray_sub_brands ("Overture PLA Matte") assert presets["0"]["preset_name"] == "Overture PLA Matte" class TestAssignSpoolLiveCaliIdx: """assign_spool always resets the slot to Default K when the spool has no stored K-profile.""" @pytest.mark.asyncio @pytest.mark.integration async def test_no_kprofile_resets_to_default_k(self, async_client: AsyncClient, printer_factory, spool_factory): """When no KProfile row exists, slot resets to cali_idx=-1 (Default K) regardless of live value.""" printer = await printer_factory() spool = await spool_factory() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Live cali_idx=42 belongs to whatever filament was previously calibrated # in this slot. Applying it to a different spool would use the wrong K # value, so the assign flow must override it with Default K (-1). tray_data = { "id": 1, "cali_idx": 42, "tray_color": "FF0000FF", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_id_name": "GFL99", } status = _make_mock_status(ams_data=[{"id": 0, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 1}, ) assert response.status_code == 200 mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args[1]["cali_idx"] == -1 @pytest.mark.asyncio @pytest.mark.integration async def test_no_kprofile_no_live_cali_idx_sends_default( self, async_client: AsyncClient, printer_factory, spool_factory ): """When tray has no cali_idx, extrusion_cali_sel is sent with cali_idx=-1 (Default).""" printer = await printer_factory() spool = await spool_factory() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True tray_data = { "id": 0, "cali_idx": None, "tray_color": "FF0000FF", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_id_name": "GFL99", } status = _make_mock_status(ams_data=[{"id": 0, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args[1]["cali_idx"] == -1 @pytest.mark.asyncio @pytest.mark.integration async def test_negative_live_cali_idx_sends_default( self, async_client: AsyncClient, printer_factory, spool_factory ): """A negative live cali_idx (-1) falls through and is sent as Default (cali_idx=-1).""" printer = await printer_factory() spool = await spool_factory() mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True tray_data = { "id": 0, "cali_idx": -1, "tray_color": "FF0000FF", "tray_type": "PLA", "tray_sub_brands": "PLA Basic", "tray_id_name": "GFL99", } status = _make_mock_status(ams_data=[{"id": 0, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 mock_client.extrusion_cali_sel.assert_called_once() assert mock_client.extrusion_cali_sel.call_args[1]["cali_idx"] == -1 class TestAssignSpoolEmptySlotPreConfig: """Assign path under ambiguous / explicit-empty AMS state. Updated for the #1322 follow-up: only the firmware's *explicit* empty signal (state ∈ {9, 10}) skips MQTT. Anything else — including the SpoolBuddy weigh-then-assign-before-insert case where state/tray_type can't tell us whether a spool is loaded — attempts MQTT. The deferred- config workflow still works because on_ams_change at main.py:1031-1054 re-fires when an AMS push eventually reports the loaded slot. """ @pytest.mark.asyncio @pytest.mark.integration async def test_empty_tray_type_without_state_still_fires_mqtt( self, async_client: AsyncClient, printer_factory, spool_factory ): """tray_type='' with no state field: AMS can't tell us whether a spool is loaded. Trust the user's Assign click and fire MQTT — firmware accepts it when a spool is physically there, drops it silently otherwise (no harm).""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="GFL05", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_type": ""}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 mock_client.ams_set_filament_setting.assert_called_once() body = response.json() assert body["pending_config"] is False assert body["configured"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_no_ams_data_with_no_client_marks_pending( self, async_client: AsyncClient, printer_factory, spool_factory ): """No AMS data + no MQTT client (printer offline, no telemetry): publish can't happen, so configured=False and pending_config=True so on_ams_change replay picks it up when the printer comes online.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="GFL05", material="PLA") # No AMS data — fingerprint_type stays None. status = _make_mock_status(ams_data=[]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = None # Printer offline, no MQTT client. mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 body = response.json() assert body["pending_config"] is True assert body["configured"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_loaded_slot_publishes_mqtt_immediately( self, async_client: AsyncClient, printer_factory, spool_factory ): """Loaded slot (tray_type non-empty) → MQTT fires + pending_config=False.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="GFL05", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status( ams_data=[{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_info_idx": "GFL05"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 0, "tray_id": 0}, ) assert response.status_code == 200 body = response.json() assert body["pending_config"] is False assert body["configured"] is True mock_client.ams_set_filament_setting.assert_called_once() @pytest.mark.asyncio @pytest.mark.integration async def test_on_ams_change_fires_config_when_pre_assigned_slot_loads( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """Pre-config replay: SpoolAssignment with empty fingerprint + slot now loaded → MQTT fires.""" from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="GFL05", material="PLA") # Pre-existing assignment with empty fingerprint (the SpoolBuddy state) pre_assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=2, tray_id=3, fingerprint_color=None, fingerprint_type=None, ) db_session.add(pre_assignment) await db_session.commit() # Filament has now been physically inserted into the slot. # state=11 ("filament fed to extruder") is the load signal we trigger on. ams_data = [{"id": 2, "tray": [{"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "state": 11}]}] mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=ams_data) printer_info = MagicMock(name="H2D", serial_number="0948BB540200427") with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = printer_info mock_pm_main.get_status.return_value = status mock_pm_main.get_client.return_value = mock_client mock_pm_main.get_model.return_value = "H2D" mock_pm_inv.get_client.return_value = mock_client mock_pm_inv.get_status.return_value = status mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) # Full filament setting was published when the slot transitioned to loaded mock_client.ams_set_filament_setting.assert_called_once() call_kwargs = mock_client.ams_set_filament_setting.call_args.kwargs assert call_kwargs["ams_id"] == 2 assert call_kwargs["tray_id"] == 3 assert call_kwargs["tray_info_idx"] == "GFL05" # Fingerprint was updated so the next push doesn't re-fire await db_session.refresh(pre_assignment) assert pre_assignment.fingerprint_type == "PLA" assert pre_assignment.fingerprint_color == "FF0000FF" @pytest.mark.asyncio @pytest.mark.integration async def test_on_ams_change_does_not_refire_for_already_configured_slot( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """Once fingerprint_type is set, subsequent AMS pushes must not re-fire MQTT.""" from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="GFL05", material="PLA") # Assignment already configured (fingerprint stamped) configured_assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=0, fingerprint_color="FF0000FF", fingerprint_type="PLA", ) db_session.add(configured_assignment) await db_session.commit() ams_data = [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000FF", "state": 11}]}] mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=ams_data) printer_info = MagicMock(name="X1C", serial_number="00M00A391800004") with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = printer_info mock_pm_main.get_status.return_value = status mock_pm_main.get_client.return_value = mock_client mock_pm_main.get_model.return_value = "X1C" mock_pm_inv.get_client.return_value = mock_client mock_pm_inv.get_status.return_value = status mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) # Fingerprint was already set — re-fire path skipped mock_client.ams_set_filament_setting.assert_not_called() @pytest.mark.asyncio @pytest.mark.integration async def test_on_ams_change_fires_replay_when_tray_type_appears_without_state_11( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """A1 Mini / P1S firmware variant of the SpoolBuddy pre-config replay (#1322). The user pre-assigned via SpoolBuddy (fingerprint empty), then configured the slot manually in Bambu Studio so tray_type went from '' to 'PLA' — but state stays at 3 because these firmwares never set it to 11. With state-only detection the replay never fired.""" from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="A1 mini") spool = await spool_factory(slicer_filament="GFL05", material="PLA") pre_assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=3, fingerprint_color=None, fingerprint_type=None, ) db_session.add(pre_assignment) await db_session.commit() # state=3 (never goes to 11 on A1 Mini BMCU 01.07.02.00) but tray_type # is now configured — the replay must fire on this transition too. ams_data = [ { "id": 0, "tray": [{"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "state": 3, "tray_info_idx": "GFL05"}], } ] mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=ams_data) printer_info = MagicMock(name="A1 mini", serial_number="0309CA391800999") with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = printer_info mock_pm_main.get_status.return_value = status mock_pm_main.get_client.return_value = mock_client mock_pm_main.get_model.return_value = "A1 mini" mock_pm_inv.get_client.return_value = mock_client mock_pm_inv.get_status.return_value = status mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) # Replay fired despite state never being 11 — the disjunction picked # up tray_type going non-empty. mock_client.ams_set_filament_setting.assert_called_once() await db_session.refresh(pre_assignment) assert pre_assignment.fingerprint_type == "PLA" @pytest.mark.asyncio @pytest.mark.integration async def test_auto_unlink_broadcasts_assignment_change( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """#2575 follow-up: when on_ams_change auto-unlinks a stale external-spool assignment, it must broadcast spool_assignment_changed. Only the manual REST endpoints did, so open browsers kept rendering the unlinked spool — the reporter read that as "the fix didn't work" when the DB was correct.""" from unittest.mock import AsyncMock from sqlalchemy import select from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="GFU01", material="TPU") # TPU inventory spool assigned to the external slot (ams_id=255, tray 0) assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=255, tray_id=0, fingerprint_color="000000FF", fingerprint_type="TPU", ) db_session.add(assignment) await db_session.commit() # The printer's external spool now reports ABS — the assignment is stale. vt_tray = [ { "id": "254", "tray_type": "ABS", "tray_color": "000000FF", "tag_uid": "0000000000000000", "tray_uuid": "00000000000000000000000000000000", } ] status = _make_mock_status(ams_data=[], vt_tray=vt_tray) printer_info = MagicMock(name="X1C", serial_number="00M00A391800004") with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.services.printer_manager.printer_manager") as mock_pm_inv, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = printer_info mock_pm_main.get_status.return_value = status mock_pm_main.get_client.return_value = MagicMock() mock_pm_main.get_model.return_value = "X1C" mock_pm_inv.get_client.return_value = MagicMock() mock_pm_inv.get_status.return_value = status mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, []) # The stale TPU assignment on the now-ABS external slot was unlinked... gone = await db_session.execute( select(SpoolAssignment).where( SpoolAssignment.printer_id == printer.id, SpoolAssignment.ams_id == 255, SpoolAssignment.tray_id == 0, ) ) assert gone.scalar_one_or_none() is None # ...and the frontend was told about it. change_events = [ c.args[0] for c in mock_ws.broadcast.await_args_list if c.args and isinstance(c.args[0], dict) and c.args[0].get("type") == "spool_assignment_changed" ] assert change_events, "auto-unlink must broadcast spool_assignment_changed" assert change_events[0]["printer_id"] == printer.id assert change_events[0]["ams_id"] == 255 assert change_events[0]["tray_id"] == 0 class TestAssignSpoolEmptyDetection: """Bambu firmware reports tray.state — 11=loaded, 9=empty, 10=spool present but filament not in feeder. The assign route must prefer that signal over tray_type for the empty-vs-loaded check, because a manual "Reset slot" clears tray_type to "" while leaving filament physically loaded — the legacy heuristic would route to the pending-config path and skip MQTT forever, since on_ams_change replay only fires on an empty→loaded transition that never comes when the slot is already loaded. """ @pytest.mark.asyncio @pytest.mark.integration async def test_state_loaded_with_empty_tray_type_fires_mqtt( self, async_client: AsyncClient, printer_factory, spool_factory ): """Post-reset case: state=11 (loaded) but tray_type='' — MQTT must fire.""" printer = await printer_factory() spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # Simulates the "reset slot" aftermath: filament physically loaded # (state=11) but tray_type/tray_color/tray_info_idx have been cleared. tray_data = {"id": 3, "state": 11, "tray_type": "", "tray_color": "", "tray_info_idx": ""} status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 # MQTT must have fired — the bug was that legacy detection saw the # empty tray_type and skipped this entirely. mock_client.ams_set_filament_setting.assert_called_once() # Response must report configured=True, pending_config=False — the # slot is loaded, just had stale metadata cleared. body = response.json() assert body["pending_config"] is False assert body["configured"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_state_empty_skips_mqtt_and_marks_pending( self, async_client: AsyncClient, printer_factory, spool_factory ): """Genuinely empty slot: state=9 — MQTT skipped, pending_config=True.""" printer = await printer_factory() spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True tray_data = {"id": 3, "state": 9, "tray_type": "", "tray_color": "", "tray_info_idx": ""} status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 # SpoolBuddy weigh-then-assign workflow: firmware drops MQTT for # unloaded slots, so we don't bother sending it. mock_client.ams_set_filament_setting.assert_not_called() body = response.json() assert body["pending_config"] is True assert body["configured"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_state_missing_falls_back_to_tray_type_loaded( self, async_client: AsyncClient, printer_factory, spool_factory ): """Older firmware without state field: tray_type='PLA' → treated as loaded.""" printer = await printer_factory() spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True # No 'state' key at all — older firmware behaviour. tray_data = {"id": 3, "tray_type": "PLA", "tray_color": "FF0000FF"} status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 # Legacy fallback: tray_type non-empty → treated as loaded → MQTT fires. mock_client.ams_set_filament_setting.assert_called_once() body = response.json() assert body["pending_config"] is False @pytest.mark.asyncio @pytest.mark.integration async def test_state_missing_with_empty_tray_type_still_fires_mqtt( self, async_client: AsyncClient, printer_factory, spool_factory ): """Older firmware without state field + empty tray_type still fires MQTT. The AMS doesn't tell us whether a spool is physically loaded in this case (no state, no tray_type), so the assign click is the user's assertion that a spool is there. Firmware silently drops the push on a truly empty slot — no harm done, and on_ams_change replay handles the deferred-config case (#1322 follow-up). """ printer = await printer_factory() spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True tray_data = {"id": 3, "tray_type": "", "tray_color": ""} status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 mock_client.ams_set_filament_setting.assert_called_once() body = response.json() assert body["pending_config"] is False assert body["configured"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_state_never_eleven_firmware_with_loaded_tray_fires_mqtt( self, async_client: AsyncClient, printer_factory, spool_factory ): """A1 Mini BMCU 01.07.02.00 and P1S Standard AMS 00.00.06.75 always report tray.state=3, never 11 — even for fully-loaded configured slots. A state-only check classified those as empty and skipped MQTT (#1322). With the disjunctive check, tray_type='PLA' alone is enough to fire.""" printer = await printer_factory() spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True # state=3, tray_type non-empty — A1 Mini / P1S configured slot. tray_data = {"id": 3, "state": 3, "tray_type": "PLA", "tray_color": "FF0000FF", "tray_info_idx": "GFL99"} status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 mock_client.ams_set_filament_setting.assert_called_once() body = response.json() assert body["pending_config"] is False assert body["configured"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_post_reset_slot_with_state_3_still_fires_mqtt( self, async_client: AsyncClient, printer_factory, spool_factory ): """A1 Mini BMCU / P1S Standard AMS post-"Reset Slot" with spool still inserted: state=3, tray_type="". The AMS gives us no signal to tell this apart from a truly-empty slot. We trust the user's Assign click and fire MQTT — firmware accepts the push because a spool is physically there (#1322 follow-up by @RosdasHH). Replaces the previous "marks_pending" assertion which was the bug: that gate created a deadlock because the AMS would never report a state change (nothing physically changed), so on_ams_change replay never re-fired the deferred config either. """ printer = await printer_factory() spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True tray_data = {"id": 3, "state": 3, "tray_type": "", "tray_color": "00000000", "tray_info_idx": ""} status = _make_mock_status(ams_data=[{"id": 2, "tray": [tray_data]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 mock_client.ams_set_filament_setting.assert_called_once() body = response.json() assert body["pending_config"] is False assert body["configured"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_external_slot_state_loaded_with_empty_tray_type_fires_mqtt( self, async_client: AsyncClient, printer_factory, spool_factory ): """External (vt_tray) slot post-reset: same fix applies for ams_id=255.""" printer = await printer_factory(name="X1C") spool = await spool_factory(slicer_filament="PFUS9ac902733670a9", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True # External slot tray_id=0 → vt_tray id=254. state=11 (loaded), tray_type # cleared by reset. vt_data = [{"id": 254, "state": 11, "tray_type": "", "tray_color": "", "tray_info_idx": ""}] status = _make_mock_status(ams_data=[], vt_tray=vt_data) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 255, "tray_id": 0}, ) assert response.status_code == 200 mock_client.ams_set_filament_setting.assert_called_once() body = response.json() assert body["pending_config"] is False assert body["configured"] is True class TestAssignSpoolPfcnCloudPreset: """Assign path for PFCN-prefix cloud presets (#1648). PFCN is a third Bambu cloud preset shape alongside PFUS (cloud user-created) and GFS (Bambu official) — used for cloud-shared / partner-uploaded presets like Polymaker's "(Custom)" Bambu Lab H2D variants. Before #1648 the assign path skipped the cloud-detail lookup and left the raw PFCN string in tray_info_idx, which the printer's calibration table can't resolve. ConfigureAmsSlotModal rescued each assignment by doing the lookup itself, making "Configure" feel like a mandatory follow-up step. """ @pytest.mark.asyncio @pytest.mark.integration async def test_pfcn_falls_back_to_generic_when_cloud_unavailable( self, async_client: AsyncClient, printer_factory, spool_factory ): """When cloud auth isn't available (e.g. user not logged into Bambu Cloud), the raw PFCN must be discarded as slicer-invalid and the slot configures with the spool's generic material id (PLA → GFL99). Pre-fix behaviour was to leak the raw PFCN, which the slicer can't resolve.""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFCN80e80c1f79db85", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "", "tray_type": "PLA"}]}]) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # PFCN never leaks into tray_info_idx — must resolve to the # generic-material fallback when cloud lookup couldn't. assert call_kwargs.kwargs["tray_info_idx"] == "GFL99" assert not call_kwargs.kwargs["tray_info_idx"].startswith("PFCN") @pytest.mark.asyncio @pytest.mark.integration async def test_pfcn_spool_reuses_valid_slot_preset(self, async_client: AsyncClient, printer_factory, spool_factory): """Symmetry with the PFUS case: when the spool's PFCN is discarded as slicer-invalid, the slot's existing valid P-prefix preset is reused if material matches — preserves calibration context instead of resetting to generic.""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFCN80e80c1f79db85", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status( ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "P4d64437", "tray_type": "PLA"}]}] ) with patch("backend.app.services.printer_manager.printer_manager") as mock_pm: mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args assert call_kwargs.kwargs["tray_info_idx"] == "P4d64437" @pytest.mark.asyncio @pytest.mark.integration async def test_pfcn_resolves_to_filament_id_via_cloud_lookup( self, async_client: AsyncClient, printer_factory, spool_factory ): """When the user is authenticated against Bambu Cloud, the PFCN setting_id triggers the same cloud-detail lookup as PFUS / GFS — extracts the real filament_id from `detail["filament_id"]` and ships that as tray_info_idx. This is the happy path the Configure modal already had but the assign path didn't, #1648.""" printer = await printer_factory(name="H2D") spool = await spool_factory(slicer_filament="PFCN80e80c1f79db85", material="PLA") mock_client = MagicMock() mock_client.ams_set_filament_setting.return_value = True mock_client.extrusion_cali_sel.return_value = True status = _make_mock_status(ams_data=[{"id": 2, "tray": [{"id": 3, "tray_info_idx": "", "tray_type": "PLA"}]}]) # Cloud responds with a real filament_id for the PFCN preset — exactly # what the Configure modal already exploits. mock_cloud = MagicMock() mock_cloud.is_authenticated = True async def fake_get_detail(setting_id): assert setting_id == "PFCN80e80c1f79db85" return {"filament_id": "GFL05", "name": "Polymaker PLA Matte"} async def fake_close(): return None mock_cloud.get_setting_detail = fake_get_detail mock_cloud.close = fake_close async def fake_build_cloud(_db, _user): return mock_cloud with ( patch("backend.app.services.printer_manager.printer_manager") as mock_pm, patch("backend.app.api.routes.cloud.build_authenticated_cloud", new=fake_build_cloud), ): mock_pm.get_client.return_value = mock_client mock_pm.get_status.return_value = status response = await async_client.post( "/api/v1/inventory/assignments", json={"spool_id": spool.id, "printer_id": printer.id, "ams_id": 2, "tray_id": 3}, ) assert response.status_code == 200 call_kwargs = mock_client.ams_set_filament_setting.call_args # tray_info_idx is the resolved cloud filament_id; setting_id is the # original PFCN (which the slicer needs separately). assert call_kwargs.kwargs["tray_info_idx"] == "GFL05" assert call_kwargs.kwargs["setting_id"] == "PFCN80e80c1f79db85" def _make_printing_status(ams_data, state="RUNNING"): """Printer status carrying an explicit gcode state for the runout guard.""" status = _make_mock_status(ams_data=ams_data) status.state = state return status class TestAutoUnlinkDuringRunout: """A slot that reports empty mid-print is a filament runout, not a spool swap — the spool is still in the AMS, just consumed. Unlinking there erased the only record of which spool fed the print, so the completion path had nothing to charge the runout segment to. With AMS filament backup that is the normal course of events, not an edge case. """ @pytest.mark.asyncio @pytest.mark.integration async def test_cleared_tray_data_keeps_the_assignment_while_printing( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="H2D") spool = await spool_factory(material="ABS", rgba="616777FF") assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, fingerprint_color="616777FF", fingerprint_type="ABS", ) db_session.add(assignment) await db_session.commit() # The firmware clears colour and type when it unloads a spool it just # emptied (state 26 = "not loaded"). ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}] with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427") mock_pm_main.get_status.return_value = _make_printing_status(ams_data) mock_pm_main.get_model.return_value = "H2D" mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) # on_ams_change committed through its own session — drop this one's # identity map so the assertion reads the database, not a cached row. db_session.expunge_all() remaining = await db_session.get(SpoolAssignment, assignment.id) assert remaining is not None, "runout must not unlink the spool that fed the print" @pytest.mark.asyncio @pytest.mark.integration async def test_cleared_tray_data_still_unlinks_when_idle( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """Off the print, an emptied slot really does mean the spool is gone.""" from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="H2D") spool = await spool_factory(material="ABS", rgba="616777FF") assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, fingerprint_color="616777FF", fingerprint_type="ABS", ) db_session.add(assignment) await db_session.commit() assignment_id = assignment.id ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}] with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427") mock_pm_main.get_status.return_value = _make_printing_status(ams_data, state="IDLE") mock_pm_main.get_model.return_value = "H2D" mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) db_session.expunge_all() assert await db_session.get(SpoolAssignment, assignment_id) is None @pytest.mark.asyncio @pytest.mark.integration async def test_a_genuinely_different_filament_still_unlinks_while_printing( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): """The guard is for blank tray data only — a real swap must still reconcile, or the wrong spool gets charged.""" from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="H2D") spool = await spool_factory(material="ABS", rgba="616777FF") assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, fingerprint_color="616777FF", fingerprint_type="ABS", ) db_session.add(assignment) await db_session.commit() assignment_id = assignment.id ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "PETG", "tray_color": "6EE53CFF", "state": 11}]}] with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427") mock_pm_main.get_status.return_value = _make_printing_status(ams_data) mock_pm_main.get_model.return_value = "H2D" mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) db_session.expunge_all() assert await db_session.get(SpoolAssignment, assignment_id) is None @pytest.mark.asyncio @pytest.mark.integration async def test_slot_missing_from_ams_data_keeps_the_assignment_while_printing( self, async_client: AsyncClient, printer_factory, spool_factory, db_session: AsyncSession ): from unittest.mock import AsyncMock from backend.app.main import on_ams_change from backend.app.models.spool_assignment import SpoolAssignment printer = await printer_factory(name="H2D") spool = await spool_factory(material="ABS", rgba="616777FF") assignment = SpoolAssignment( spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=2, fingerprint_color="616777FF", fingerprint_type="ABS", ) db_session.add(assignment) await db_session.commit() assignment_id = assignment.id # Tray 2 dropped out of the payload entirely. ams_data = [{"id": 0, "tray": [{"id": 0, "tray_type": "ABS", "tray_color": "FFFFFFFF", "state": 11}]}] with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, ): mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427") mock_pm_main.get_status.return_value = _make_printing_status(ams_data) mock_pm_main.get_model.return_value = "H2D" mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer.id, ams_data) db_session.expunge_all() assert await db_session.get(SpoolAssignment, assignment_id) is not None class TestSpoolmanSlotAssignmentDuringRunout: """`spoolman_slot_assignments` is how a tag-less spool assigned through the Bambuddy UI is resolved at completion (#1459). Deleting the row when a slot empties mid-print loses the runout segment's usage — the same failure the internal inventory's auto-unlink had, so it needs the same guard.""" async def _enable_spoolman(self, db_session): from backend.app.models.settings import Settings for key, value in ( ("spoolman_enabled", "true"), ("spoolman_sync_mode", "auto"), ("spoolman_url", "http://spoolman.test"), ): db_session.add(Settings(key=key, value=value)) await db_session.commit() async def _run(self, printer_id, state): from unittest.mock import AsyncMock from backend.app.main import on_ams_change # A tray the firmware has cleared: parse_ams_tray returns None, which # is what marks the slot empty for the cleanup pass. ams_data = [{"id": 0, "tray": [{"id": 2, "tray_type": "", "tray_color": "", "state": 26}]}] spoolman_client = MagicMock() spoolman_client.health_check = AsyncMock(return_value=True) spoolman_client.get_spools = AsyncMock(return_value=[]) spoolman_client.sync_ams_tray = AsyncMock(return_value=None) # None is what marks the slot empty for the cleanup pass. spoolman_client.parse_ams_tray.return_value = None with ( patch("backend.app.main.printer_manager") as mock_pm_main, patch("backend.app.main.mqtt_relay") as mock_relay, patch("backend.app.main.ws_manager") as mock_ws, patch("backend.app.main.get_spoolman_client", new=AsyncMock(return_value=spoolman_client)), ): mock_pm_main.get_printer.return_value = MagicMock(name="H2D", serial_number="0948BB540200427") mock_pm_main.get_status.return_value = state mock_pm_main.get_model.return_value = "H2D" mock_relay.on_ams_change = AsyncMock() mock_ws.send_printer_status = AsyncMock() mock_ws.broadcast = AsyncMock() await on_ams_change(printer_id, ams_data) @pytest.mark.asyncio @pytest.mark.integration async def test_the_slot_row_survives_a_runout( self, async_client: AsyncClient, printer_factory, db_session: AsyncSession ): from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment await self._enable_spoolman(db_session) printer = await printer_factory(name="H2D") row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41) db_session.add(row) await db_session.commit() row_id = row.id await self._run(printer.id, _make_printing_status(None)) db_session.expunge_all() assert await db_session.get(SpoolmanSlotAssignment, row_id) is not None @pytest.mark.asyncio @pytest.mark.integration async def test_the_slot_row_is_still_cleaned_up_when_idle( self, async_client: AsyncClient, printer_factory, db_session: AsyncSession ): """Proves the guard is what saved the row above, not an unreachable code path.""" from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment await self._enable_spoolman(db_session) printer = await printer_factory(name="H2D") row = SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=2, spoolman_spool_id=41) db_session.add(row) await db_session.commit() row_id = row.id await self._run(printer.id, _make_printing_status(None, state="IDLE")) db_session.expunge_all() assert await db_session.get(SpoolmanSlotAssignment, row_id) is None