"""What the Spoolman AMS syncs write to the slot table, and announce (SM-C3, SSA-2, SSA-3). - The automatic pass read the slot table, then awaited Spoolman for every tray and wrote every synced slot back, unchanged ones included. A spool assigned by hand in between was overwritten with what the pass had read. It now writes only slots whose spool it changed, and only while they still hold what it read. - The Sync buttons in Settings wrote slots without announcing them, refreshing the slot's preset or clearing its grace hold, as the automatic pass does. - Both seeded a newly created spool's weight from the built-in inventory's slot assignments, which in Spoolman mode are left over from before the switch. """ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from httpx import AsyncClient from sqlalchemy import select, update from backend.app.models.spool import Spool from backend.app.models.spool_assignment import SpoolAssignment from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment from backend.app.services.spoolman import AMSTray from backend.tests.integration.test_spoolman_ams_sync_broadcast import ( _enable_spoolman, _run_ams_change, _slot_events, _tray, ) pytestmark = [pytest.mark.asyncio, pytest.mark.integration] async def _slots(db_session, printer_id) -> dict[tuple[int, int], int]: rows = ( await db_session.execute( select(SpoolmanSlotAssignment) .where(SpoolmanSlotAssignment.printer_id == printer_id) .execution_options(populate_existing=True) ) ).scalars() return {(row.ams_id, row.tray_id): row.spoolman_spool_id for row in rows} async def _printer_with_slot(db_session, printer_factory, spool_id: int = 7) -> int: await _enable_spoolman(db_session) printer = await printer_factory() db_session.add(SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=0, spoolman_spool_id=spool_id)) await db_session.commit() return printer.id def _assigning_by_hand_meanwhile(db_session, printer_id, slot, spool_id, returns): """sync_ams_tray, during which the user assigns *spool_id* to *slot*.""" async def sync(tray, *args, **kwargs): await db_session.execute( update(SpoolmanSlotAssignment) .where( SpoolmanSlotAssignment.printer_id == printer_id, SpoolmanSlotAssignment.ams_id == slot[0], SpoolmanSlotAssignment.tray_id == slot[1], ) .values(spoolman_spool_id=spool_id) ) await db_session.commit() return returns return AsyncMock(side_effect=sync) AMS = [{"id": 0, "tray": [{"id": 0}, {"id": 1}]}] @pytest.mark.parametrize("synced", [7, 8], ids=["unchanged", "changed"]) async def test_the_automatic_pass_keeps_a_spool_assigned_meanwhile( async_client: AsyncClient, db_session, printer_factory, synced ): printer_id = await _printer_with_slot(db_session, printer_factory) sync = _assigning_by_hand_meanwhile(db_session, printer_id, (0, 0), 9, {"id": synced}) broadcast, _client = await _run_ams_change(printer_id, AMS[:1], parsed={(0, 0): _tray(0, 0)}, sync=sync) assert await _slots(db_session, printer_id) == {(0, 0): 9} assert (0, 0) not in _slot_events(broadcast), "nothing the pass wrote changed" async def test_the_automatic_pass_announces_no_slot_it_left_as_it_was( async_client: AsyncClient, db_session, printer_factory ): printer_id = await _printer_with_slot(db_session, printer_factory) broadcast, _client = await _run_ams_change( printer_id, AMS[:1], parsed={(0, 0): _tray(0, 0)}, sync=AsyncMock(return_value={"id": 7}) ) assert await _slots(db_session, printer_id) == {(0, 0): 7} assert (0, 0) not in _slot_events(broadcast) async def test_the_automatic_pass_still_writes_and_announces_what_it_changed( async_client: AsyncClient, db_session, printer_factory ): printer_id = await _printer_with_slot(db_session, printer_factory) broadcast, _client = await _run_ams_change( printer_id, AMS[:1], parsed={(0, 0): _tray(0, 0)}, sync=AsyncMock(return_value={"id": 8}) ) assert await _slots(db_session, printer_id) == {(0, 0): 8} assert _slot_events(broadcast) == {(0, 0)} # ── The Sync buttons ───────────────────────────────────────────────────────── PATHS = ["/api/v1/spoolman/sync/{printer_id}", "/api/v1/spoolman/sync-all"] def _parsed_tray(ams_id, tray_data): return AMSTray( ams_id=ams_id, tray_id=int(tray_data.get("id", 0)), tray_type="PLA", tray_sub_brands="PLA Matte", tray_color="042F56FF", remain=-1, tag_uid="7757EF0100000100", tray_uuid="EB335968299543078925C71D83DA3864", tray_info_idx="GFA01", tray_weight=1000, ) async def _manual_sync(async_client, printer_id, path, synced: int, *, trays=None, sync=None): trays = trays or [{"id": 0, "tray_type": "PLA"}] state = SimpleNamespace(state="IDLE", raw_data={"ams": [{"id": 0, "tray": trays}]}) client = MagicMock() client.health_check = AsyncMock(return_value=True) client.get_spools = AsyncMock(return_value=[]) # A tray without a type reads empty. client.parse_ams_tray = MagicMock( side_effect=lambda ams_id, tray: _parsed_tray(ams_id, tray) if tray.get("tray_type") else None ) client.sync_ams_tray = sync or AsyncMock(return_value={"id": synced}) preset = AsyncMock() with ( patch("backend.app.api.routes.spoolman.get_spoolman_client", AsyncMock(return_value=client)), patch("backend.app.api.routes.spoolman.printer_manager") as pm, patch("backend.app.services.spoolman_slot_sync.ws_manager") as ws, patch("backend.app.services.slot_preset_writer.upsert_slot_preset_for_spoolman_spool", preset), patch("backend.app.services.slot_unlink_grace.forget_slot") as forget, ): ws.broadcast = AsyncMock() pm.get_status = MagicMock(return_value=state) pm.get_all_statuses = MagicMock(return_value={printer_id: state}) response = await async_client.post(path.format(printer_id=printer_id)) assert response.status_code == 200, response.text return client, ws.broadcast, preset, forget @pytest.mark.parametrize("path", PATHS) async def test_a_sync_button_announces_a_changed_slot(async_client: AsyncClient, db_session, printer_factory, path): printer_id = await _printer_with_slot(db_session, printer_factory) _client, broadcast, preset, forget = await _manual_sync(async_client, printer_id, path, synced=8) assert await _slots(db_session, printer_id) == {(0, 0): 8} assert _slot_events(broadcast) == {(0, 0)} assert preset.await_args.kwargs["spoolman_spool"] == {"id": 8} forget.assert_called_once_with(printer_id, 0, 0) @pytest.mark.parametrize("path", PATHS) async def test_a_sync_button_does_not_seed_a_spool_from_the_built_in_inventory( async_client: AsyncClient, db_session, printer_factory, path ): """A slot assignment left from before the switch to Spoolman: 880 g of 1000 used, nothing to do with the spool now in the slot.""" printer_id = await _printer_with_slot(db_session, printer_factory) stale = Spool(material="PLA", label_weight=1000, weight_used=880) db_session.add(stale) await db_session.flush() db_session.add(SpoolAssignment(spool_id=stale.id, printer_id=printer_id, ams_id=0, tray_id=0)) await db_session.commit() client, *_ = await _manual_sync(async_client, printer_id, path, synced=8) assert client.sync_ams_tray.await_args.kwargs.get("inventory_remaining") is None @pytest.mark.parametrize("path", PATHS) async def test_a_sync_button_keeps_a_slot_that_was_filled_meanwhile( async_client: AsyncClient, db_session, printer_factory, path ): """Slot 0 read empty, so the sync would unlink spool 7; by then it holds 9.""" printer_id = await _printer_with_slot(db_session, printer_factory) sync = _assigning_by_hand_meanwhile(db_session, printer_id, (0, 0), 9, {"id": 5}) await _manual_sync( async_client, printer_id, path, synced=5, trays=[{"id": 0}, {"id": 1, "tray_type": "PLA"}], sync=sync ) assert await _slots(db_session, printer_id) == {(0, 0): 9, (0, 1): 5} @pytest.mark.parametrize("path", PATHS) async def test_a_sync_button_still_unlinks_a_slot_that_reads_empty( async_client: AsyncClient, db_session, printer_factory, path ): printer_id = await _printer_with_slot(db_session, printer_factory) _client, broadcast, *_ = await _manual_sync(async_client, printer_id, path, synced=5, trays=[{"id": 0}]) assert await _slots(db_session, printer_id) == {} assert _slot_events(broadcast) == {(0, 0)} async def test_the_automatic_pass_keeps_a_spool_assigned_to_a_slot_it_read_empty( async_client: AsyncClient, db_session, printer_factory ): await _enable_spoolman(db_session) printer_id = (await printer_factory()).id async def sync(tray, *args, **kwargs): db_session.add(SpoolmanSlotAssignment(printer_id=printer_id, ams_id=0, tray_id=0, spoolman_spool_id=9)) await db_session.commit() return {"id": 8} await _run_ams_change(printer_id, AMS[:1], parsed={(0, 0): _tray(0, 0)}, sync=AsyncMock(side_effect=sync)) assert await _slots(db_session, printer_id) == {(0, 0): 9} async def test_without_the_slot_table_as_read_the_syncs_write_as_before(db_session, printer_factory): """Its load failed: nothing to compare against, so a swap and a removal are written as before rather than dropped.""" from backend.app.services.spoolman_slot_sync import write_slot_results printer_id = await _printer_with_slot(db_session, printer_factory, spool_id=10) db_session.add(SpoolmanSlotAssignment(printer_id=printer_id, ams_id=0, tray_id=1, spoolman_spool_id=11)) await db_session.commit() with patch("backend.app.services.spoolman_slot_sync.ws_manager") as ws: ws.broadcast = AsyncMock() changed = await write_slot_results(db_session, None, [(printer_id, 0, 0, 99)], [(printer_id, 0, 1)]) assert await _slots(db_session, printer_id) == {(0, 0): 99} assert sorted(changed) == [(printer_id, 0, 0), (printer_id, 0, 1)]