|
|
@@ -0,0 +1,803 @@
|
|
|
+"""Filament attribution has to survive a restart mid-print.
|
|
|
+
|
|
|
+A 14-hour print that spans a Bambuddy restart used to lose everything the
|
|
|
+completion path needs: the plate (so the 3MF parser summed every plate of a
|
|
|
+multi-plate file), the dispatched slot-to-tray mapping (so it fell back to the
|
|
|
+live MQTT ``mapping`` field, which AMS filament backup rewrites to the
|
|
|
+substitute tray), the spool-assignment snapshot, and the tray-change log that
|
|
|
+splits weight across a runout. The whole print was then charged to whichever
|
|
|
+spool happened to finish it, while the spool that actually ran dry was charged
|
|
|
+nothing.
|
|
|
+
|
|
|
+These tests cover the durable ``active_print_sessions`` row that fixes that,
|
|
|
+plus the plate and mapping fallbacks the completion path now applies.
|
|
|
+"""
|
|
|
+
|
|
|
+from datetime import datetime, timezone
|
|
|
+from types import SimpleNamespace
|
|
|
+from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+from sqlalchemy import select
|
|
|
+
|
|
|
+from backend.app.models.active_print_session import ActivePrintSession
|
|
|
+from backend.app.models.printer import Printer
|
|
|
+from backend.app.services.usage_tracker import (
|
|
|
+ PrintSession,
|
|
|
+ _active_sessions,
|
|
|
+ _track_from_3mf,
|
|
|
+ clear_persisted_session,
|
|
|
+ get_persisted_print_name,
|
|
|
+ on_print_complete,
|
|
|
+ persist_session,
|
|
|
+ record_tray_change,
|
|
|
+ restore_session,
|
|
|
+)
|
|
|
+
|
|
|
+
|
|
|
+def _make_spool(spool_id=1, label_weight=1000, weight_used=0):
|
|
|
+ spool = MagicMock()
|
|
|
+ spool.id = spool_id
|
|
|
+ spool.label_weight = label_weight
|
|
|
+ spool.weight_used = weight_used
|
|
|
+ spool.tag_uid = None
|
|
|
+ spool.tray_uuid = None
|
|
|
+ spool.last_used = None
|
|
|
+ spool.cost_per_kg = None
|
|
|
+ spool.material = "ABS"
|
|
|
+ spool.rgba = "616777FF"
|
|
|
+ return spool
|
|
|
+
|
|
|
+
|
|
|
+def _make_assignment(spool_id=1, ams_id=0, tray_id=0):
|
|
|
+ assignment = MagicMock()
|
|
|
+ assignment.spool_id = spool_id
|
|
|
+ assignment.printer_id = 1
|
|
|
+ assignment.ams_id = ams_id
|
|
|
+ assignment.tray_id = tray_id
|
|
|
+ assignment.created_at = None
|
|
|
+ return assignment
|
|
|
+
|
|
|
+
|
|
|
+def _make_archive(archive_id=1, plate_id=None, file_path="archives/1/multi_plate.3mf"):
|
|
|
+ archive = MagicMock()
|
|
|
+ archive.id = archive_id
|
|
|
+ archive.file_path = file_path
|
|
|
+ archive.plate_id = plate_id
|
|
|
+ archive.extra_data = None
|
|
|
+ return archive
|
|
|
+
|
|
|
+
|
|
|
+def _make_queue_item(item_id=629, ams_mapping=None, plate_id=None):
|
|
|
+ item = MagicMock()
|
|
|
+ item.id = item_id
|
|
|
+ item.ams_mapping = ams_mapping
|
|
|
+ item.plate_id = plate_id
|
|
|
+ item.status = "printing"
|
|
|
+ return item
|
|
|
+
|
|
|
+
|
|
|
+def _mock_db_sequential(responses):
|
|
|
+ """Mock db whose execute() yields the given rows in order."""
|
|
|
+ db = AsyncMock()
|
|
|
+ call_count = [0]
|
|
|
+
|
|
|
+ async def mock_execute(*args, **kwargs):
|
|
|
+ idx = call_count[0]
|
|
|
+ call_count[0] += 1
|
|
|
+ result = MagicMock()
|
|
|
+ value = responses[idx] if idx < len(responses) else None
|
|
|
+ result.scalar_one_or_none.return_value = value
|
|
|
+ result.scalars.return_value.first.return_value = value
|
|
|
+ result.scalar.return_value = None
|
|
|
+ return result
|
|
|
+
|
|
|
+ db.execute = mock_execute
|
|
|
+ return db
|
|
|
+
|
|
|
+
|
|
|
+def _patched_3mf(filament_usage, capture=None):
|
|
|
+ """Patch the 3MF extract, optionally recording the plate_id it was given."""
|
|
|
+
|
|
|
+ def _extract(path, plate_id=None):
|
|
|
+ if capture is not None:
|
|
|
+ capture.append(plate_id)
|
|
|
+ return filament_usage
|
|
|
+
|
|
|
+ return patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", side_effect=_extract)
|
|
|
+
|
|
|
+
|
|
|
+def _settings_patch():
|
|
|
+ mock_settings = patch("backend.app.core.config.settings")
|
|
|
+ return mock_settings
|
|
|
+
|
|
|
+
|
|
|
+class TestPersistedSessionRoundTrip:
|
|
|
+ """The row is the only thing that outlives the process."""
|
|
|
+
|
|
|
+ @pytest.fixture(autouse=True)
|
|
|
+ def _clear_sessions(self):
|
|
|
+ _active_sessions.clear()
|
|
|
+ yield
|
|
|
+ _active_sessions.clear()
|
|
|
+
|
|
|
+ @pytest.fixture
|
|
|
+ async def printer(self, db_session):
|
|
|
+ row = Printer(name="H2D-1", ip_address="192.168.0.10", access_code="1234", serial_number="TESTSERIAL")
|
|
|
+ db_session.add(row)
|
|
|
+ await db_session.commit()
|
|
|
+ return row
|
|
|
+
|
|
|
+ def _session(self, printer_id):
|
|
|
+ return PrintSession(
|
|
|
+ printer_id=printer_id,
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
|
|
|
+ tray_remain_start={(0, 2): 84, (0, 3): 100},
|
|
|
+ tray_now_at_start=2,
|
|
|
+ spool_assignments={(0, 2): 69, (0, 3): 68},
|
|
|
+ ams_mapping=[2],
|
|
|
+ plate_id=1,
|
|
|
+ )
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_restore_rebuilds_the_session_and_returns_the_tray_log(self, db_session, printer):
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0)])
|
|
|
+ _active_sessions.clear()
|
|
|
+
|
|
|
+ log = await restore_session(db_session, printer.id)
|
|
|
+
|
|
|
+ assert log == [[2, 0]]
|
|
|
+ restored = _active_sessions[printer.id]
|
|
|
+ assert restored.plate_id == 1
|
|
|
+ assert restored.ams_mapping == [2]
|
|
|
+ assert restored.tray_now_at_start == 2
|
|
|
+ # Tuple keys survive the JSON round trip — the completion path indexes
|
|
|
+ # the snapshot by (ams_id, tray_id).
|
|
|
+ assert restored.spool_assignments == {(0, 2): 69, (0, 3): 68}
|
|
|
+ assert restored.tray_remain_start == {(0, 2): 84, (0, 3): 100}
|
|
|
+ assert restored.started_at.tzinfo is not None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_tray_changes_accumulate_in_order(self, db_session, printer):
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0)])
|
|
|
+
|
|
|
+ # The runout sequence from the reported print: A3 empties, the AMS
|
|
|
+ # parks, then filament backup brings A4 in.
|
|
|
+ await record_tray_change(db_session, printer.id, 254, 670)
|
|
|
+ await record_tray_change(db_session, printer.id, 3, 675)
|
|
|
+
|
|
|
+ assert await restore_session(db_session, printer.id) == [[2, 0], [254, 670], [3, 675]]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_tray_change_without_a_session_is_a_noop(self, db_session, printer):
|
|
|
+ await record_tray_change(db_session, printer.id, 3, 675)
|
|
|
+
|
|
|
+ row = await db_session.get(ActivePrintSession, printer.id)
|
|
|
+ assert row is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_print_start_overwrites_a_row_left_by_a_missed_completion(self, db_session, printer):
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0), (3, 675)])
|
|
|
+
|
|
|
+ second = self._session(printer.id)
|
|
|
+ second.print_name = "Cover"
|
|
|
+ second.plate_id = 2
|
|
|
+ second.ams_mapping = [5]
|
|
|
+ second.spool_assignments = {(1, 0): 60}
|
|
|
+ await persist_session(db_session, second, [(5, 0)])
|
|
|
+
|
|
|
+ rows = (await db_session.execute(select(ActivePrintSession))).scalars().all()
|
|
|
+ assert len(rows) == 1
|
|
|
+ log = await restore_session(db_session, printer.id)
|
|
|
+ assert log == [[5, 0]]
|
|
|
+ assert _active_sessions[printer.id].plate_id == 2
|
|
|
+ assert _active_sessions[printer.id].spool_assignments == {(1, 0): 60}
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_clear_removes_the_row(self, db_session, printer):
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0)])
|
|
|
+
|
|
|
+ await clear_persisted_session(db_session, printer.id)
|
|
|
+
|
|
|
+ assert await restore_session(db_session, printer.id) is None
|
|
|
+ assert await get_persisted_print_name(db_session, printer.id) is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_clear_is_safe_without_a_row(self, db_session, printer):
|
|
|
+ await clear_persisted_session(db_session, printer.id)
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_print_name_is_readable_for_the_identity_check(self, db_session, printer):
|
|
|
+ await persist_session(db_session, self._session(printer.id), None)
|
|
|
+
|
|
|
+ assert await get_persisted_print_name(db_session, printer.id) == "AMS_Rack"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_completion_falls_back_to_the_persisted_row(self, db_session, printer):
|
|
|
+ """No in-memory session (the restart case): the plate, the mapping and
|
|
|
+ the assignment snapshot must still reach the 3MF path."""
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0), (3, 675)])
|
|
|
+ _active_sessions.clear()
|
|
|
+
|
|
|
+ captured = {}
|
|
|
+
|
|
|
+ async def _fake_track(*args, **kwargs):
|
|
|
+ captured.update(kwargs)
|
|
|
+ return []
|
|
|
+
|
|
|
+ with (
|
|
|
+ patch("backend.app.api.routes.settings.get_setting", new_callable=AsyncMock, return_value=None),
|
|
|
+ patch("backend.app.services.usage_tracker._track_from_3mf", side_effect=_fake_track),
|
|
|
+ ):
|
|
|
+ await on_print_complete(
|
|
|
+ printer.id,
|
|
|
+ {"status": "completed", "subtask_name": "AMS_Rack"},
|
|
|
+ MagicMock(),
|
|
|
+ db_session,
|
|
|
+ archive_id=312,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert captured["plate_id"] == 1
|
|
|
+ assert captured["ams_mapping"] == [2]
|
|
|
+ assert captured["tray_now_at_start"] == 2
|
|
|
+ assert captured["spool_assignments"] == {(0, 2): 69, (0, 3): 68}
|
|
|
+
|
|
|
+
|
|
|
+class TestPlateIdRecovery:
|
|
|
+ """Without the plate, the 3MF parser sums every plate in the file and the
|
|
|
+ whole multi-plate total lands on one spool."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_archive_plate_id_is_used_when_the_session_is_gone(self):
|
|
|
+ archive = _make_archive(archive_id=312, plate_id=1)
|
|
|
+ spool = _make_spool(spool_id=68)
|
|
|
+ assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
|
|
|
+ db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
+ seen_plate_ids: list = []
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=809,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=809,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ _patched_3mf([{"slot_id": 1, "used_g": 1122.44, "type": "ABS", "color": "#808080"}], seen_plate_ids),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=312,
|
|
|
+ status="completed",
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=None,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert seen_plate_ids == [1]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_queue_item_plate_id_is_used_when_the_archive_has_none(self):
|
|
|
+ archive = _make_archive(archive_id=312, plate_id=None)
|
|
|
+ queue_item = _make_queue_item(plate_id=2)
|
|
|
+ spool = _make_spool(spool_id=68)
|
|
|
+ assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
|
|
|
+ # db: archive, the single queue lookup (plate + mapping share it),
|
|
|
+ # then assignment and spool
|
|
|
+ db = _mock_db_sequential([archive, queue_item, assignment, spool])
|
|
|
+ seen_plate_ids: list = []
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=361,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=361,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ _patched_3mf([{"slot_id": 1, "used_g": 318.82, "type": "ABS", "color": "#808080"}], seen_plate_ids),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=312,
|
|
|
+ status="completed",
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=None,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert seen_plate_ids == [2]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_caller_plate_id_wins_over_the_database(self):
|
|
|
+ archive = _make_archive(archive_id=312, plate_id=1)
|
|
|
+ spool = _make_spool(spool_id=68)
|
|
|
+ assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
|
|
|
+ db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
+ seen_plate_ids: list = []
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=361,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=361,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ _patched_3mf([{"slot_id": 1, "used_g": 318.82, "type": "ABS", "color": "#808080"}], seen_plate_ids),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=312,
|
|
|
+ status="completed",
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=2,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert seen_plate_ids == [2]
|
|
|
+
|
|
|
+
|
|
|
+class TestMappingPriority:
|
|
|
+ """AMS filament backup rewrites the printer's live ``mapping`` field to the
|
|
|
+ substitute tray. Read at completion it names the tray that finished the
|
|
|
+ print, not the one the slicer assigned."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_queue_mapping_beats_the_live_mqtt_mapping(self):
|
|
|
+ archive = _make_archive(archive_id=312, plate_id=1)
|
|
|
+ # Dispatched against AMS0-T2 (global tray 2); the printer now reports
|
|
|
+ # tray 3 because backup swapped in the neighbouring spool.
|
|
|
+ queue_item = _make_queue_item(ams_mapping="[2]")
|
|
|
+ spool_69 = _make_spool(spool_id=69)
|
|
|
+ assign_69 = _make_assignment(spool_id=69, ams_id=0, tray_id=2)
|
|
|
+ db = _mock_db_sequential([archive, queue_item, assign_69, spool_69])
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=809,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=809,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ _patched_3mf([{"slot_id": 1, "used_g": 1122.44, "type": "ABS", "color": "#808080"}]),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ results = await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=312,
|
|
|
+ status="completed",
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=1,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert len(results) == 1
|
|
|
+ assert results[0]["spool_id"] == 69
|
|
|
+ assert (results[0]["ams_id"], results[0]["tray_id"]) == (0, 2)
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_mqtt_mapping_still_used_for_a_direct_print(self):
|
|
|
+ """No queue item — the live field is the only mapping there is."""
|
|
|
+ archive = _make_archive(archive_id=400, plate_id=1)
|
|
|
+ spool = _make_spool(spool_id=68)
|
|
|
+ assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
|
|
|
+ db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=100,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=100,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ _patched_3mf([{"slot_id": 1, "used_g": 50.0, "type": "ABS", "color": "#808080"}]),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ results = await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=400,
|
|
|
+ status="completed",
|
|
|
+ print_name="Cover",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=1,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert len(results) == 1
|
|
|
+ assert (results[0]["ams_id"], results[0]["tray_id"]) == (0, 3)
|
|
|
+
|
|
|
+
|
|
|
+class TestRestoreOnRestartRecovery:
|
|
|
+ """``on_print_running_observed`` is the only hook that fires when Bambuddy
|
|
|
+ comes up mid-print — the #1304 guard suppresses ``on_print_start``."""
|
|
|
+
|
|
|
+ @pytest.fixture(autouse=True)
|
|
|
+ def _clear_sessions(self):
|
|
|
+ _active_sessions.clear()
|
|
|
+ yield
|
|
|
+ _active_sessions.clear()
|
|
|
+
|
|
|
+ @pytest.fixture
|
|
|
+ async def printer(self, db_session):
|
|
|
+ row = Printer(name="H2D-1", ip_address="192.168.0.10", access_code="1234", serial_number="TESTSERIAL")
|
|
|
+ db_session.add(row)
|
|
|
+ await db_session.commit()
|
|
|
+ return row
|
|
|
+
|
|
|
+ def _state(self, **overrides):
|
|
|
+ state = SimpleNamespace(
|
|
|
+ subtask_name="AMS_Rack",
|
|
|
+ tray_change_log=[],
|
|
|
+ tray_now=3,
|
|
|
+ layer_num=700,
|
|
|
+ last_loaded_tray=-1,
|
|
|
+ )
|
|
|
+ for key, value in overrides.items():
|
|
|
+ setattr(state, key, value)
|
|
|
+ return state
|
|
|
+
|
|
|
+ def _session(self, printer_id, print_name="AMS_Rack"):
|
|
|
+ return PrintSession(
|
|
|
+ printer_id=printer_id,
|
|
|
+ print_name=print_name,
|
|
|
+ started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
|
|
|
+ tray_now_at_start=2,
|
|
|
+ spool_assignments={(0, 2): 69},
|
|
|
+ ams_mapping=[2],
|
|
|
+ plate_id=1,
|
|
|
+ )
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_persisted_log_comes_back_onto_the_printer_state(self, db_session, printer):
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0)])
|
|
|
+ _active_sessions.clear()
|
|
|
+ state = self._state()
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
|
|
|
+
|
|
|
+ assert state.tray_change_log == [(2, 0)]
|
|
|
+ assert _active_sessions[printer.id].plate_id == 1
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_entries_seen_by_this_process_are_kept_after_the_persisted_ones(self, db_session, printer):
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ await persist_session(db_session, self._session(printer.id), [(2, 0)])
|
|
|
+ state = self._state(tray_change_log=[(3, 675)])
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
|
|
|
+
|
|
|
+ assert state.tray_change_log == [(2, 0), (3, 675)]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_no_persisted_row_seeds_from_the_tray_feeding_now(self, db_session, printer):
|
|
|
+ """A print that started before this build still gets its remaining
|
|
|
+ segment attributed to the right spool."""
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ state = self._state()
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
|
|
|
+
|
|
|
+ assert state.tray_change_log == [(3, 700)]
|
|
|
+ assert state.last_loaded_tray == 3
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unloaded_tray_seeds_nothing(self, db_session, printer):
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ state = self._state(tray_now=255)
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
|
|
|
+
|
|
|
+ assert state.tray_change_log == []
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_row_from_a_different_print_is_discarded(self, db_session, printer):
|
|
|
+ """A completion Bambuddy never saw leaves a row behind; it must not
|
|
|
+ attach itself to whatever is running now."""
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ await persist_session(db_session, self._session(printer.id, print_name="Old_Print"), [(2, 0)])
|
|
|
+ _active_sessions.clear()
|
|
|
+ state = self._state(subtask_name="AMS_Rack")
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
|
|
|
+
|
|
|
+ assert printer.id not in _active_sessions
|
|
|
+ assert await restore_session(db_session, printer.id) is None
|
|
|
+ # Still seeded, so the rest of the running print stays attributable.
|
|
|
+ assert state.tray_change_log == [(3, 700)]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_an_unloaded_tray_does_not_clobber_last_loaded_tray(self, db_session, printer):
|
|
|
+ """``last_loaded_tray`` is the fallback that survives the end-of-print
|
|
|
+ retract to 255; writing 255 into it would defeat its whole purpose."""
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ state = self._state(tray_now=255, last_loaded_tray=2)
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, state, db_session, MagicMock())
|
|
|
+
|
|
|
+ assert state.last_loaded_tray == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_failure_is_swallowed_so_the_caller_keeps_going(self, db_session, printer):
|
|
|
+ """The caller still has to capture its timelapse baseline before the
|
|
|
+ printer uploads the in-flight recording — there is no second chance."""
|
|
|
+ from backend.app.main import _restore_usage_tracking_session
|
|
|
+
|
|
|
+ broken = SimpleNamespace() # no subtask_name, no tray fields at all
|
|
|
+
|
|
|
+ await _restore_usage_tracking_session(printer.id, broken, db_session, MagicMock())
|
|
|
+
|
|
|
+
|
|
|
+class TestPlateNotInTheFile:
|
|
|
+ """A recovered plate has to be treated as a hint, not a filter that can
|
|
|
+ silently zero out a print's usage."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_falls_back_to_the_whole_file_when_the_plate_is_absent(self):
|
|
|
+ """The archive's own 3MF can be gone, with a same-named library file
|
|
|
+ substituted that was sliced with different plates."""
|
|
|
+ archive = _make_archive(archive_id=312, plate_id=7)
|
|
|
+ spool = _make_spool(spool_id=68)
|
|
|
+ assignment = _make_assignment(spool_id=68, ams_id=0, tray_id=3)
|
|
|
+ db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
+ calls: list = []
|
|
|
+
|
|
|
+ def _extract(path, plate_id=None):
|
|
|
+ calls.append(plate_id)
|
|
|
+ return [] if plate_id is not None else [{"slot_id": 1, "used_g": 12.0, "type": "ABS", "color": ""}]
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=10,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=10,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", side_effect=_extract),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ results = await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=312,
|
|
|
+ status="completed",
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=None,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert calls == [7, None]
|
|
|
+ assert len(results) == 1
|
|
|
+ assert results[0]["weight_used"] == 12.0
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_file_with_no_usage_at_all_still_records_nothing(self):
|
|
|
+ archive = _make_archive(archive_id=312, plate_id=1)
|
|
|
+ db = _mock_db_sequential([archive, None])
|
|
|
+
|
|
|
+ printer_manager = MagicMock()
|
|
|
+ printer_manager.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={"mapping": [3]},
|
|
|
+ progress=100,
|
|
|
+ layer_num=10,
|
|
|
+ tray_now=255,
|
|
|
+ tray_change_log=[],
|
|
|
+ total_layers=10,
|
|
|
+ )
|
|
|
+
|
|
|
+ with (
|
|
|
+ _settings_patch() as mock_settings,
|
|
|
+ patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", return_value=[]),
|
|
|
+ ):
|
|
|
+ mock_settings.base_dir = MagicMock()
|
|
|
+ mock_path = MagicMock()
|
|
|
+ mock_path.exists.return_value = True
|
|
|
+ mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
+
|
|
|
+ results = await _track_from_3mf(
|
|
|
+ printer_id=1,
|
|
|
+ archive_id=312,
|
|
|
+ status="completed",
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ handled_trays=set(),
|
|
|
+ printer_manager=printer_manager,
|
|
|
+ db=db,
|
|
|
+ plate_id=1,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert results == []
|
|
|
+
|
|
|
+
|
|
|
+class TestSpoolmanParity:
|
|
|
+ """Both inventory backends need the same restart protection.
|
|
|
+
|
|
|
+ Spoolman's own durable row (#1820) already carries its plate-scoped 3MF
|
|
|
+ figures and the mapping it was dispatched with, but not the tray-change
|
|
|
+ log — and that log is the only record of which spool fed which layers when
|
|
|
+ AMS Filament Backup swaps trays. Capturing it for one backend only would
|
|
|
+ leave Spoolman users with the bug this fixes for everyone else.
|
|
|
+ """
|
|
|
+
|
|
|
+ @pytest.fixture(autouse=True)
|
|
|
+ def _clear_sessions(self):
|
|
|
+ _active_sessions.clear()
|
|
|
+ yield
|
|
|
+ _active_sessions.clear()
|
|
|
+
|
|
|
+ @pytest.fixture
|
|
|
+ async def printer(self, db_session):
|
|
|
+ row = Printer(name="H2D-1", ip_address="192.168.0.10", access_code="1234", serial_number="TESTSERIAL")
|
|
|
+ db_session.add(row)
|
|
|
+ await db_session.commit()
|
|
|
+ return row
|
|
|
+
|
|
|
+ def _printer_manager(self):
|
|
|
+ pm = MagicMock()
|
|
|
+ pm.get_status.return_value = SimpleNamespace(
|
|
|
+ raw_data={
|
|
|
+ "ams": {"ams": [{"id": 0, "tray": [{"id": 2, "remain": 84, "tray_type": "ABS"}]}]},
|
|
|
+ "vt_tray": [],
|
|
|
+ "mapping": [2],
|
|
|
+ },
|
|
|
+ tray_now=2,
|
|
|
+ last_loaded_tray=2,
|
|
|
+ tray_change_log=[(2, 0)],
|
|
|
+ )
|
|
|
+ return pm
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_row_is_written_with_spoolman_enabled(self, db_session, printer):
|
|
|
+ from backend.app.services.usage_tracker import on_print_start
|
|
|
+
|
|
|
+ await on_print_start(
|
|
|
+ printer.id,
|
|
|
+ {"subtask_name": "AMS_Rack", "ams_mapping": [2]},
|
|
|
+ self._printer_manager(),
|
|
|
+ db=db_session,
|
|
|
+ spoolman_owns_usage=True,
|
|
|
+ )
|
|
|
+
|
|
|
+ row = await db_session.get(ActivePrintSession, printer.id)
|
|
|
+ assert row is not None
|
|
|
+ assert row.print_name == "AMS_Rack"
|
|
|
+ assert row.tray_change_log == [[2, 0]]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_spoolman_does_not_get_an_in_memory_session(self, db_session, printer):
|
|
|
+ """``_active_sessions`` doubles as on_ams_change's "skip the remain%
|
|
|
+ weight sync, the internal tracker will deduct precisely" flag (#880).
|
|
|
+ A session the internal tracker will never complete would suppress a
|
|
|
+ sync Spoolman users still need."""
|
|
|
+ from backend.app.services.usage_tracker import on_print_start
|
|
|
+
|
|
|
+ await on_print_start(
|
|
|
+ printer.id,
|
|
|
+ {"subtask_name": "AMS_Rack", "ams_mapping": [2]},
|
|
|
+ self._printer_manager(),
|
|
|
+ db=db_session,
|
|
|
+ spoolman_owns_usage=True,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert printer.id not in _active_sessions
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_internal_tracker_still_gets_one(self, db_session, printer):
|
|
|
+ from backend.app.services.usage_tracker import on_print_start
|
|
|
+
|
|
|
+ await on_print_start(
|
|
|
+ printer.id,
|
|
|
+ {"subtask_name": "AMS_Rack", "ams_mapping": [2]},
|
|
|
+ self._printer_manager(),
|
|
|
+ db=db_session,
|
|
|
+ spoolman_owns_usage=False,
|
|
|
+ )
|
|
|
+
|
|
|
+ assert _active_sessions[printer.id].ams_mapping == [2]
|
|
|
+ assert await db_session.get(ActivePrintSession, printer.id) is not None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_restore_can_return_the_log_without_publishing_a_session(self, db_session, printer):
|
|
|
+ await persist_session(
|
|
|
+ db_session,
|
|
|
+ PrintSession(
|
|
|
+ printer_id=printer.id,
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
|
|
|
+ ),
|
|
|
+ [(2, 0), (3, 675)],
|
|
|
+ )
|
|
|
+ _active_sessions.clear()
|
|
|
+
|
|
|
+ log = await restore_session(db_session, printer.id, register_active=False)
|
|
|
+
|
|
|
+ assert log == [[2, 0], [3, 675]]
|
|
|
+ assert printer.id not in _active_sessions
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_discard_clears_both_halves(self, db_session, printer):
|
|
|
+ from backend.app.services.usage_tracker import discard_session
|
|
|
+
|
|
|
+ session = PrintSession(
|
|
|
+ printer_id=printer.id,
|
|
|
+ print_name="AMS_Rack",
|
|
|
+ started_at=datetime(2026, 8, 11, 9, 25, 6, tzinfo=timezone.utc),
|
|
|
+ )
|
|
|
+ _active_sessions[printer.id] = session
|
|
|
+ await persist_session(db_session, session, [(2, 0)])
|
|
|
+
|
|
|
+ await discard_session(db_session, printer.id)
|
|
|
+
|
|
|
+ assert printer.id not in _active_sessions
|
|
|
+ assert await db_session.get(ActivePrintSession, printer.id) is None
|
maziggy