mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
454457a0af
Everything the completion path needs to split a print's filament across the trays it fed from lived only in memory: the dispatched plate and slot-to-tray mapping, the spool-assignment snapshot, and the tray-change log. A print that outlived a restart lost all of it and fell back to what the printer reports at completion -- which, with AMS Filament Backup on, is the substitute tray. The whole print was charged to the spool that only finished it while the spool that ran dry was charged nothing. Persist that context in a new active_print_sessions row, append tray changes as they happen, and restore both the session and the printer's tray-change log at restart recovery. Seed the log from the current tray when there is nothing to restore, since last_loaded_tray advances even when no change is logged. Rank the queue item's stored ams_mapping above the printer's live mapping field, which is what backup rewrites. Recover plate_id from the archive or queue item, and give extract_layer_filament_usage_from_3mf a plate_id instead of taking the first .gcode member -- a Bambu Studio export stores plate 2 first, so per-layer figures were measured against the wrong plate for both inventory backends. Stop auto-unlinking a spool assignment when its slot reports empty during a running print. At a runout the spool is still in the AMS, and dropping the link leaves the completion path nothing to charge. Capture the print-start context for both inventory backends. Spoolman's own durable row (#1820) carries its plate-scoped figures and dispatched mapping but not the tray-change log, and its slot assignments -- the way. Registration in _active_sessions stays gated, since on_ams_change reads it to decide whether to skip the remain%-based weight sync (#880).
1651 lines
71 KiB
Python
1651 lines
71 KiB
Python
"""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
|