mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
maziggy
72044e3a53
When a print targets a single plate from a multi-plate 3MF, both the
internal Filament Inventory tracker and the Spoolman-mode tracker parsed
the 3MF without a plate filter and summed every plate's filament — so a
single lid print debited the spool the entire file's grey + black totals.
The 3MF parser already supports plate_id (queue pre-flight uses it at
print_queue.py:254/:286). Plumbed it through both dispatch paths:
Queue path:
- PrintSession gains a plate_id field; on_print_start queries the
printer's currently-printing queue row and records queue_item.plate_id
onto the session.
- _track_from_3mf accepts plate_id and passes it to the extractor.
- store_print_data moves its existing queue-item lookup above the
extract and uses queue_item.plate_id as the plate filter.
Direct-Print path (reprintArchive / printLibraryFile — never goes
through the queue):
- _print_plate_ids dict added in main.py, parallel to _print_ams_mappings.
- register_expected_print accepts plate_id and stores it; the 2 sites in
background_dispatch.py and the 1 site in print_scheduler.py now pass
it (resolve was already happening, just needed reordering before the
register call so the value is available).
- Expected-print promotion in main.py injects _print_plate_ids[archive_id]
into the session, guarded so a queue capture wins over the dict.
- _get_start_plate_id helper feeds plate_id into all 3
_store_spoolman_print_data call sites; spoolman_tracking.store_print_data
takes the caller value first, falls back to queue_item.plate_id.
PrintArchive.filament_used_grams stays file-level summed by design
(#1593's contract — the archive describes the file, not the run); only
the per-run usage attribution becomes plate-aware. Single-plate direct
prints resolve to plate_id=1 → plate 1 = whole file, identical to the
prior no-filter behaviour.
2339 lines
89 KiB
Python
2339 lines
89 KiB
Python
"""Unit tests for usage_tracker.py — 3MF-primary filament tracking.
|
|
|
|
Tests the unified tracking logic: 3MF slicer estimates as primary path,
|
|
AMS remain% delta as fallback, per-layer gcode for partial prints,
|
|
slot-to-tray mapping resolution, and notification variable formatting.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.usage_tracker import (
|
|
PrintSession,
|
|
_active_sessions,
|
|
_decode_mqtt_mapping,
|
|
_find_3mf_by_filename,
|
|
_match_slots_by_color,
|
|
_track_from_3mf,
|
|
on_print_complete,
|
|
on_print_start,
|
|
)
|
|
|
|
|
|
def _make_spool(spool_id=1, label_weight=1000, weight_used=0, tag_uid=None, tray_uuid=None):
|
|
"""Create a mock Spool object."""
|
|
spool = MagicMock()
|
|
spool.id = spool_id
|
|
spool.label_weight = label_weight
|
|
spool.weight_used = weight_used
|
|
spool.tag_uid = tag_uid
|
|
spool.tray_uuid = tray_uuid
|
|
spool.last_used = None
|
|
spool.cost_per_kg = None
|
|
spool.material = "PLA"
|
|
return spool
|
|
|
|
|
|
def _make_assignment(spool_id=1, printer_id=1, ams_id=0, tray_id=0):
|
|
"""Create a mock SpoolAssignment object."""
|
|
assignment = MagicMock()
|
|
assignment.spool_id = spool_id
|
|
assignment.printer_id = printer_id
|
|
assignment.ams_id = ams_id
|
|
assignment.tray_id = tray_id
|
|
return assignment
|
|
|
|
|
|
def _make_archive(archive_id=1, file_path="archives/1/test.3mf", extra_data=None):
|
|
"""Create a mock PrintArchive object."""
|
|
archive = MagicMock()
|
|
archive.id = archive_id
|
|
archive.file_path = file_path
|
|
archive.extra_data = extra_data
|
|
return archive
|
|
|
|
|
|
def _make_queue_item(ams_mapping=None, status="printing", plate_id=None):
|
|
"""Create a mock PrintQueueItem object."""
|
|
item = MagicMock()
|
|
item.ams_mapping = ams_mapping
|
|
item.status = status
|
|
item.plate_id = plate_id
|
|
return item
|
|
|
|
|
|
def _mock_db_execute(*return_values):
|
|
"""Create a mock db with execute() that returns values in sequence."""
|
|
db = AsyncMock()
|
|
results = []
|
|
for val in return_values:
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = val
|
|
results.append(result)
|
|
db.execute = AsyncMock(side_effect=results)
|
|
return db
|
|
|
|
|
|
def _mock_db_sequential(responses):
|
|
"""Create mock db that returns responses in order."""
|
|
db = AsyncMock()
|
|
call_count = [0]
|
|
|
|
async def mock_execute(*args, **kwargs):
|
|
idx = call_count[0]
|
|
call_count[0] += 1
|
|
result = MagicMock()
|
|
if idx < len(responses):
|
|
result.scalar_one_or_none.return_value = responses[idx]
|
|
else:
|
|
result.scalar_one_or_none.return_value = None
|
|
# For cost aggregation queries that use .scalar() instead of .scalar_one_or_none()
|
|
result.scalar.return_value = None
|
|
return result
|
|
|
|
db.execute = mock_execute
|
|
return db
|
|
|
|
|
|
class TestOnPrintStart:
|
|
"""Tests for on_print_start()."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_sessions(self):
|
|
_active_sessions.clear()
|
|
yield
|
|
_active_sessions.clear()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_remain_data(self):
|
|
"""Captures AMS remain% at print start."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}, {"id": 1, "remain": 50}]}]},
|
|
tray_now=5,
|
|
)
|
|
|
|
await on_print_start(1, {"subtask_name": "Benchy"}, printer_manager)
|
|
|
|
assert 1 in _active_sessions
|
|
session = _active_sessions[1]
|
|
assert session.print_name == "Benchy"
|
|
assert session.tray_remain_start == {(0, 0): 80, (0, 1): 50}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_tray_now_at_start(self):
|
|
"""Captures tray_now at print start for later use in usage tracking."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
|
|
tray_now=9,
|
|
)
|
|
|
|
await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
|
|
|
|
assert _active_sessions[1].tray_now_at_start == 9
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_now_at_start_255_when_unloaded(self):
|
|
"""Captures tray_now=255 when printer has no filament loaded at start."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
|
|
tray_now=255,
|
|
)
|
|
|
|
await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
|
|
|
|
assert _active_sessions[1].tray_now_at_start == 255
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_session_without_remain(self):
|
|
"""Creates session even without valid remain data (for 3MF tracking)."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": -1}]}]},
|
|
tray_now=255,
|
|
)
|
|
|
|
await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
|
|
|
|
assert 1 in _active_sessions
|
|
assert _active_sessions[1].tray_remain_start == {}
|
|
|
|
|
|
class TestOnPrintComplete:
|
|
"""Tests for on_print_complete() — path ordering and interaction."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_sessions(self):
|
|
_active_sessions.clear()
|
|
yield
|
|
_active_sessions.clear()
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_get_setting(self):
|
|
with patch(
|
|
"backend.app.api.routes.settings.get_setting",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
):
|
|
yield
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bl_spool_uses_3mf(self):
|
|
"""BL spool (with tag_uid) is tracked via 3MF, not just AMS delta."""
|
|
spool = _make_spool(spool_id=1, tag_uid="AABB1122", label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, printer_id=1, ams_id=0, tray_id=0)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# Setup: session with AMS remain data
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="Benchy",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
)
|
|
|
|
# Mock printer state: tray_now=0 (AMS0-T0), single filament
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]},
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=0,
|
|
)
|
|
|
|
# db returns: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 15.0, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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 on_print_complete(
|
|
printer_id=1,
|
|
data={"status": "completed"},
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
archive_id=10,
|
|
)
|
|
|
|
# 3MF path should handle it (BL guard removed)
|
|
assert len(results) >= 1
|
|
assert results[0]["spool_id"] == 1
|
|
assert results[0]["weight_used"] == 15.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ams_delta_fallback_no_archive(self):
|
|
"""AMS delta tracks consumption when archive_id is None."""
|
|
spool = _make_spool(spool_id=2, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=2)
|
|
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="Test",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
)
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]},
|
|
tray_now=0,
|
|
last_loaded_tray=-1,
|
|
)
|
|
|
|
# Pad 2 Nones for _find_3mf_by_filename DB queries (library + archive search),
|
|
# then assignment and spool for the AMS fallback path
|
|
db = _mock_db_sequential([None, None, assignment, spool])
|
|
|
|
results = await on_print_complete(
|
|
printer_id=1,
|
|
data={"status": "completed"},
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
archive_id=None,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 2
|
|
# 10% of 1000g = 100g
|
|
assert results[0]["weight_used"] == 100.0
|
|
assert results[0]["percent_used"] == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_double_tracking(self):
|
|
"""When 3MF handles a tray, AMS delta skips it."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
_active_sessions[1] = PrintSession(
|
|
printer_id=1,
|
|
print_name="Benchy",
|
|
started_at=datetime.now(timezone.utc),
|
|
tray_remain_start={(0, 0): 80},
|
|
)
|
|
|
|
# tray_now=0 matches the single filament slot
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 70}]}]},
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=0,
|
|
)
|
|
|
|
# db returns: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 15.0, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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 on_print_complete(
|
|
printer_id=1,
|
|
data={"status": "completed"},
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
archive_id=10,
|
|
)
|
|
|
|
# Only 1 result (3MF), NOT 2 (3MF + AMS delta)
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 15.0
|
|
|
|
|
|
class TestTrackFrom3mf:
|
|
"""Tests for _track_from_3mf() — per-layer, linear scaling, and slot mapping."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefers_live_assignment_when_reassigned_mid_print(self):
|
|
"""If tray assignment changed during print, track usage on the new spool."""
|
|
spool_old = _make_spool(spool_id=1, label_weight=1000)
|
|
spool_new = _make_spool(spool_id=2, label_weight=1000)
|
|
archive = _make_archive(archive_id=80)
|
|
|
|
live_assignment = _make_assignment(spool_id=2, ams_id=0, tray_id=0)
|
|
started_at = datetime.now(timezone.utc)
|
|
live_assignment.created_at = started_at + timedelta(seconds=5)
|
|
|
|
# db: archive, queue_item(None), live assignment lookup, spool_new lookup
|
|
db = _mock_db_sequential([archive, None, live_assignment, spool_new])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=0,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=80,
|
|
status="completed",
|
|
print_name="MidPrintReassign",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
spool_assignments={(0, 0): spool_old.id},
|
|
print_started_at=started_at,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == spool_new.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_keeps_snapshot_when_live_assignment_predates_print(self):
|
|
"""If live assignment predates print start, preserve snapshot spool mapping."""
|
|
spool_old = _make_spool(spool_id=1, label_weight=1000)
|
|
archive = _make_archive(archive_id=81)
|
|
|
|
live_assignment = _make_assignment(spool_id=2, ams_id=0, tray_id=0)
|
|
started_at = datetime.now(timezone.utc)
|
|
live_assignment.created_at = started_at - timedelta(seconds=5)
|
|
|
|
# db: archive, queue_item(None), live assignment lookup, spool_old lookup
|
|
db = _mock_db_sequential([archive, None, live_assignment, spool_old])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=0,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=81,
|
|
status="completed",
|
|
print_name="SnapshotPreserved",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
spool_assignments={(0, 0): spool_old.id},
|
|
print_started_at=started_at,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == spool_old.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_linear_fallback_for_partial_print(self):
|
|
"""Falls back to linear scaling when gcode layer data unavailable."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=50,
|
|
layer_num=25,
|
|
tray_now=0,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 20.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value=None, # No layer data available
|
|
),
|
|
):
|
|
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=10,
|
|
status="failed",
|
|
print_name="Benchy",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
# 50% of 20g = 10g
|
|
assert results[0]["weight_used"] == 10.0
|
|
# Tray should be marked as handled
|
|
assert (0, 0) in handled_trays
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_per_layer_partial_print(self):
|
|
"""Failed print at layer N uses gcode cumulative data."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=50,
|
|
layer_num=25,
|
|
tray_now=0,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 20.0, "type": "PLA", "color": ""}]
|
|
# Per-layer data: at layer 25, filament 0 used 5000mm
|
|
layer_data = {10: {0: 2000.0}, 25: {0: 5000.0}, 50: {0: 10000.0}}
|
|
filament_props = {1: {"density": 1.24, "diameter": 1.75}}
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value=layer_data,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.get_cumulative_usage_at_layer",
|
|
return_value={0: 5000.0},
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_properties_from_3mf",
|
|
return_value=filament_props,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.mm_to_grams",
|
|
return_value=12.0, # 5000mm at 1.75mm/1.24g/cm3
|
|
),
|
|
):
|
|
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=10,
|
|
status="failed",
|
|
print_name="Benchy",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
# Should use per-layer grams (12.0g), not linear scale (10.0g)
|
|
assert results[0]["weight_used"] == 12.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_completed_print_uses_full_weight(self):
|
|
"""Completed print uses full 3MF weight (scale=1.0)."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=0,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 20.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=10,
|
|
status="completed",
|
|
print_name="Benchy",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 20.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_now_override_for_single_filament(self):
|
|
"""Single-filament non-queue print uses tray_now instead of slot_id mapping."""
|
|
# Spool 2 is at AMS1-T3 (global_tray_id=7)
|
|
spool = _make_spool(spool_id=2, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=2, ams_id=1, tray_id=3)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
# tray_now=7 = (ams_id=1, tray_id=3), the ACTUAL tray used
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=7,
|
|
)
|
|
|
|
# 3MF has slot_id=12 (would default-map to ams_id=2, tray_id=3 — WRONG)
|
|
filament_usage = [{"slot_id": 12, "used_g": 10.6, "type": "PLA", "color": "#FF0000"}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=10,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 2
|
|
assert results[0]["ams_id"] == 1
|
|
assert results[0]["tray_id"] == 3
|
|
assert results[0]["weight_used"] == 10.6
|
|
assert (1, 3) in handled_trays
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_queue_ams_mapping_overrides_default(self):
|
|
"""Queue item ams_mapping overrides default slot_id mapping."""
|
|
# Spool at AMS1-T3 (global_tray_id=7)
|
|
spool = _make_spool(spool_id=5, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=5, ams_id=1, tray_id=3)
|
|
archive = _make_archive(archive_id=20)
|
|
# Queue item maps slot 1 → global tray 7 (ams_id=1, tray_id=3)
|
|
queue_item = _make_queue_item(ams_mapping="[7, -1, -1, -1]")
|
|
|
|
# db: archive, queue_item, assignment, spool
|
|
db = _mock_db_sequential([archive, queue_item, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=7,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 25.0, "type": "PETG", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=20,
|
|
status="completed",
|
|
print_name="Queue Print",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 5
|
|
assert results[0]["ams_id"] == 1
|
|
assert results[0]["tray_id"] == 3
|
|
assert results[0]["weight_used"] == 25.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_filament_uses_queue_mapping(self):
|
|
"""Multi-filament queue prints use ams_mapping for each slot."""
|
|
spool_a = _make_spool(spool_id=1, label_weight=1000)
|
|
spool_b = _make_spool(spool_id=2, label_weight=1000)
|
|
assign_a = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
assign_b = _make_assignment(spool_id=2, ams_id=1, tray_id=2)
|
|
archive = _make_archive(archive_id=30)
|
|
# slot 1 → tray 0 (AMS0-T0), slot 2 → tray 6 (AMS1-T2)
|
|
queue_item = _make_queue_item(ams_mapping="[0, 6]")
|
|
|
|
# db: archive, queue_item, assign_a, spool_a, assign_b, spool_b
|
|
db = _mock_db_sequential([archive, queue_item, assign_a, spool_a, assign_b, spool_b])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=6,
|
|
)
|
|
|
|
filament_usage = [
|
|
{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""},
|
|
{"slot_id": 2, "used_g": 5.0, "type": "PETG", "color": ""},
|
|
]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=30,
|
|
status="completed",
|
|
print_name="Multi",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 2
|
|
assert results[0]["spool_id"] == 1
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 0
|
|
assert results[0]["weight_used"] == 10.0
|
|
assert results[1]["spool_id"] == 2
|
|
assert results[1]["ams_id"] == 1
|
|
assert results[1]["tray_id"] == 2
|
|
assert results[1]["weight_used"] == 5.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_tray_now_override_for_multi_filament(self):
|
|
"""Multi-filament non-queue prints fall back to default mapping, not tray_now."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# db: archive, queue_item(None), assignment, spool (2nd slot has no assignment)
|
|
db = _mock_db_sequential([archive, None, assignment, spool, None])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=4, # tray_now won't be used
|
|
)
|
|
|
|
# Two filament slots with usage
|
|
filament_usage = [
|
|
{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""},
|
|
{"slot_id": 2, "used_g": 5.0, "type": "PETG", "color": ""},
|
|
]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=10,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
# Should use default mapping (slot 1 → tray 0, slot 2 → tray 1)
|
|
assert len(results) == 1 # Only slot 1 has assignment
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stored_ams_mapping_overrides_all(self):
|
|
"""Stored ams_mapping from print command takes priority over queue and tray_now."""
|
|
# Spool at AMS2-T1 (global_tray_id=9)
|
|
spool = _make_spool(spool_id=10, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=10, ams_id=2, tray_id=1)
|
|
archive = _make_archive(archive_id=50)
|
|
|
|
# db: archive, assignment, spool (no queue lookup when ams_mapping provided)
|
|
db = _mock_db_sequential([archive, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=0, # Different from mapped tray — should be ignored
|
|
last_loaded_tray=0,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 2, "used_g": 1.57, "type": "PLA", "color": "#FFFFFF"}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
mock_settings.base_dir = MagicMock()
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
mock_settings.base_dir.__truediv__ = MagicMock(return_value=mock_path)
|
|
|
|
# ams_mapping: slot 2 (index 1) -> tray 9 (AMS2-T1)
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=50,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
ams_mapping=[-1, 9],
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 10
|
|
assert results[0]["ams_id"] == 2
|
|
assert results[0]["tray_id"] == 1
|
|
assert results[0]["weight_used"] == 1.6 # rounded
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_last_loaded_tray_fallback(self):
|
|
"""Falls back to last_loaded_tray when tray_now_at_start and current tray_now are both 255."""
|
|
# Spool at AMS2-T1 (global_tray_id=9)
|
|
spool = _make_spool(spool_id=11, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=11, ams_id=2, tray_id=1)
|
|
archive = _make_archive(archive_id=60)
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
# H2D scenario: tray_now=255 at completion, but last_loaded_tray=9
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=255,
|
|
last_loaded_tray=9,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 6, "used_g": 1.52, "type": "PLA", "color": "#7CC4D5"}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=60,
|
|
status="completed",
|
|
print_name="Cube",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=255, # H2D: 255 at start too
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 11
|
|
assert results[0]["ams_id"] == 2
|
|
assert results[0]["tray_id"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_now_at_start_preferred_over_last_loaded(self):
|
|
"""tray_now_at_start is used before last_loaded_tray fallback."""
|
|
spool = _make_spool(spool_id=3, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=3, ams_id=1, tray_id=1)
|
|
archive = _make_archive(archive_id=70)
|
|
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
# tray_now_at_start=5 (valid), last_loaded_tray=9 (different) — should use 5
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=255,
|
|
last_loaded_tray=9,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 5.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=70,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=5, # AMS1-T1
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["ams_id"] == 1
|
|
assert results[0]["tray_id"] == 1
|
|
|
|
|
|
class TestTrayChangeSplit:
|
|
"""Tests for mid-print tray switch weight splitting in _track_from_3mf()."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_switch_splits_weight_with_gcode(self):
|
|
"""Two-tray runout: weight split using per-layer gcode data."""
|
|
spool_a = _make_spool(spool_id=10, label_weight=1000)
|
|
spool_b = _make_spool(spool_id=20, label_weight=1000)
|
|
assign_a = _make_assignment(spool_id=10, ams_id=0, tray_id=1)
|
|
assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=0)
|
|
archive = _make_archive(archive_id=100)
|
|
|
|
# db: archive, queue_item(None), then for each segment: assignment, spool
|
|
db = _mock_db_sequential([archive, None, assign_a, spool_a, assign_b, spool_b])
|
|
|
|
# Tray change log: started on tray 1, switched to tray 0 at layer 60
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=0,
|
|
last_loaded_tray=0,
|
|
total_layers=100,
|
|
tray_change_log=[(1, 0), (0, 60)],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 30.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value={30: {0: 3000.0}, 60: {0: 6000.0}, 100: {0: 10000.0}},
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.get_cumulative_usage_at_layer",
|
|
side_effect=lambda data, layer: {0: {0: 0.0, 60: 6000.0, 100: 10000.0}.get(layer, 0.0)},
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_properties_from_3mf",
|
|
return_value={1: {"density": 1.24, "diameter": 1.75}},
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.mm_to_grams",
|
|
side_effect=lambda mm, d, dens: round(mm * 0.003, 1), # Simple conversion
|
|
),
|
|
):
|
|
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=100,
|
|
status="completed",
|
|
print_name="Runout Test",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
# Two results: one per tray segment
|
|
assert len(results) == 2
|
|
# First segment: tray 1 (AMS0-T1), layers 0→60
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 1
|
|
assert results[0]["spool_id"] == 10
|
|
assert results[0]["weight_used"] == 18.0 # 6000mm * 0.003
|
|
# Second segment: tray 0 (AMS0-T0), layers 60→end = 30.0 - 18.0 = 12.0
|
|
assert results[1]["ams_id"] == 0
|
|
assert results[1]["tray_id"] == 0
|
|
assert results[1]["spool_id"] == 20
|
|
assert results[1]["weight_used"] == 12.0
|
|
# Both trays handled
|
|
assert (0, 1) in handled_trays
|
|
assert (0, 0) in handled_trays
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_switch_linear_fallback(self):
|
|
"""Two-tray runout without per-layer gcode: linear split by layer ratio."""
|
|
spool_a = _make_spool(spool_id=10, label_weight=1000)
|
|
spool_b = _make_spool(spool_id=20, label_weight=1000)
|
|
assign_a = _make_assignment(spool_id=10, ams_id=0, tray_id=2)
|
|
assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=1)
|
|
archive = _make_archive(archive_id=101)
|
|
|
|
db = _mock_db_sequential([archive, None, assign_a, spool_a, assign_b, spool_b])
|
|
|
|
# Tray 2 from layer 0, switched to tray 1 at layer 40 (of 100 total)
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=1,
|
|
last_loaded_tray=1,
|
|
total_layers=100,
|
|
tray_change_log=[(2, 0), (1, 40)],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 50.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value=None, # No per-layer gcode available
|
|
),
|
|
):
|
|
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=101,
|
|
status="completed",
|
|
print_name="Linear Fallback",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 2
|
|
# Linear split: tray 2 for 40/100 layers = 20g
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 2
|
|
assert results[0]["weight_used"] == 20.0
|
|
# Last segment gets remainder: 50 - 20 = 30g
|
|
assert results[1]["ams_id"] == 0
|
|
assert results[1]["tray_id"] == 1
|
|
assert results[1]["weight_used"] == 30.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_switch_overrides_print_cmd_mapping(self):
|
|
"""tray_change_log evidence overrides slot_to_tray captured at print start.
|
|
|
|
Regression for #957: when AMS auto-falls-back from an empty spool to a
|
|
same-material sibling, the print_cmd's mapping (which named the
|
|
original tray) is stale by the time the print finishes. Before this
|
|
fix, the splitting branch was gated on ``not slot_to_tray`` so the
|
|
slicer mapping was preferred even when the printer actually fed from
|
|
a different tray — Path 1 credited the original tray with the full
|
|
3MF estimate and Path 2 layered the AMS-fallback delta on top, so
|
|
spool consumption double-counted (e.g. 78 g print credited as 78 g
|
|
+ 60 g = 138 g). This test pins the new behavior: when
|
|
tray_change_log has > 1 entries, splitting takes over regardless of
|
|
whether ams_mapping was provided.
|
|
"""
|
|
spool_a = _make_spool(spool_id=10, label_weight=1000)
|
|
spool_b = _make_spool(spool_id=20, label_weight=1000)
|
|
assign_a = _make_assignment(spool_id=10, ams_id=0, tray_id=0)
|
|
assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=1)
|
|
archive = _make_archive(archive_id=200)
|
|
|
|
# No queue_item placeholder: passing ams_mapping bypasses the queue lookup
|
|
# at usage_tracker.py:816 (`if not slot_to_tray and archive_id`).
|
|
db = _mock_db_sequential([archive, assign_a, spool_a, assign_b, spool_b])
|
|
|
|
# Slicer mapping says slot 1 -> tray 0; printer actually swapped to tray 1 at layer 30
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=1,
|
|
last_loaded_tray=1,
|
|
total_layers=100,
|
|
tray_change_log=[(0, 0), (1, 30)],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 78.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value=None, # No per-layer data — exercises linear fallback
|
|
),
|
|
):
|
|
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=200,
|
|
status="completed",
|
|
print_name="Runout w/ slicer mapping",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
# ams_mapping captured at print start — slicer told us slot 0 -> tray 0
|
|
# (1-based slot_id=1 -> 0-based slot index 0).
|
|
ams_mapping=[0],
|
|
)
|
|
|
|
# Splitting branch ran despite ams_mapping being set: two segments,
|
|
# one per tray, total weight matches the 3MF estimate (no double-count).
|
|
assert len(results) == 2
|
|
total = sum(r["weight_used"] for r in results)
|
|
assert total == pytest.approx(78.0, abs=0.1)
|
|
# Both trays now in handled_trays so Path 2 (remain%-delta) skips them.
|
|
assert (0, 0) in handled_trays
|
|
assert (0, 1) in handled_trays
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_tray_change_uses_normal_path(self):
|
|
"""Single-entry tray_change_log falls through to normal tray_now_at_start logic."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=2)
|
|
archive = _make_archive(archive_id=102)
|
|
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
# Only one entry = no switch, should use normal path
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=2,
|
|
last_loaded_tray=2,
|
|
total_layers=100,
|
|
tray_change_log=[(2, 0)],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 15.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=102,
|
|
status="completed",
|
|
print_name="No Switch",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=2,
|
|
)
|
|
|
|
# Normal path: single result, full weight
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 15.0
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_tray_change_log_uses_normal_path(self):
|
|
"""Empty tray_change_log (e.g. server restart) falls through to existing logic."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
archive = _make_archive(archive_id=103)
|
|
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
# Empty log (server restarted mid-print)
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=0,
|
|
last_loaded_tray=0,
|
|
total_layers=100,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=103,
|
|
status="completed",
|
|
print_name="Restart Recovery",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=0,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 10.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_switch_segment_no_spool(self):
|
|
"""Segment with no spool assignment is skipped; other segments still tracked."""
|
|
spool_b = _make_spool(spool_id=20, label_weight=1000)
|
|
assign_b = _make_assignment(spool_id=20, ams_id=0, tray_id=3)
|
|
archive = _make_archive(archive_id=104)
|
|
|
|
# db: archive, queue_item(None), 1st segment: no assignment, 2nd segment: assignment, spool
|
|
db = _mock_db_sequential([archive, None, None, assign_b, spool_b])
|
|
|
|
# Tray 5 (no spool) from layer 0, switched to tray 3 at layer 50
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=3,
|
|
last_loaded_tray=3,
|
|
total_layers=100,
|
|
tray_change_log=[(5, 0), (3, 50)],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 40.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value=None, # No per-layer data
|
|
),
|
|
):
|
|
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=104,
|
|
status="completed",
|
|
print_name="Missing Spool",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
# Only the second segment (tray 3) tracked; first segment (tray 5) skipped
|
|
assert len(results) == 1
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 3
|
|
assert results[0]["spool_id"] == 20
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tray_switch_three_segments(self):
|
|
"""Three-segment switch (rare): A→B→C split by linear fallback."""
|
|
spool_a = _make_spool(spool_id=1, label_weight=1000)
|
|
spool_b = _make_spool(spool_id=2, label_weight=1000)
|
|
spool_c = _make_spool(spool_id=3, label_weight=1000)
|
|
assign_a = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
assign_b = _make_assignment(spool_id=2, ams_id=0, tray_id=1)
|
|
assign_c = _make_assignment(spool_id=3, ams_id=0, tray_id=2)
|
|
archive = _make_archive(archive_id=105)
|
|
|
|
db = _mock_db_sequential(
|
|
[
|
|
archive,
|
|
None,
|
|
assign_a,
|
|
spool_a,
|
|
assign_b,
|
|
spool_b,
|
|
assign_c,
|
|
spool_c,
|
|
]
|
|
)
|
|
|
|
# 3 segments: tray 0 (0-30), tray 1 (30-70), tray 2 (70-end)
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
progress=100,
|
|
layer_num=100,
|
|
tray_now=2,
|
|
last_loaded_tray=2,
|
|
total_layers=100,
|
|
tray_change_log=[(0, 0), (1, 30), (2, 70)],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 100.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf",
|
|
return_value=None,
|
|
),
|
|
):
|
|
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=105,
|
|
status="completed",
|
|
print_name="Triple Switch",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 3
|
|
# Tray 0: 30/100 * 100g = 30g
|
|
assert results[0]["weight_used"] == 30.0
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 0
|
|
# Tray 1: 40/100 * 100g = 40g
|
|
assert results[1]["weight_used"] == 40.0
|
|
assert results[1]["ams_id"] == 0
|
|
assert results[1]["tray_id"] == 1
|
|
# Tray 2: remainder = 100 - 30 - 40 = 30g
|
|
assert results[2]["weight_used"] == 30.0
|
|
assert results[2]["ams_id"] == 0
|
|
assert results[2]["tray_id"] == 2
|
|
|
|
|
|
class TestDecodeMqttMapping:
|
|
"""Tests for _decode_mqtt_mapping() — snow-encoded MQTT mapping to global tray IDs."""
|
|
|
|
def test_none_input(self):
|
|
assert _decode_mqtt_mapping(None) is None
|
|
|
|
def test_empty_list(self):
|
|
assert _decode_mqtt_mapping([]) is None
|
|
|
|
def test_all_unmapped(self):
|
|
"""All 65535 values → None (no valid mappings)."""
|
|
assert _decode_mqtt_mapping([65535, 65535, 65535]) is None
|
|
|
|
def test_single_ams_slots(self):
|
|
"""AMS 0 slots: snow values 0-3 → global tray IDs 0-3."""
|
|
assert _decode_mqtt_mapping([0, 1, 2, 3]) == [0, 1, 2, 3]
|
|
|
|
def test_multi_ams_slots(self):
|
|
"""AMS 1 (hw_id=1): snow 256=AMS1-T0, 257=AMS1-T1 → global 4, 5."""
|
|
assert _decode_mqtt_mapping([256, 257]) == [4, 5]
|
|
|
|
def test_ams_ht_slot(self):
|
|
"""AMS-HT (hw_id=128): snow 32768 → global 128."""
|
|
assert _decode_mqtt_mapping([32768]) == [128]
|
|
|
|
def test_external_spool(self):
|
|
"""External spool: ams_hw_id=254, slot=0 → global 254."""
|
|
# snow = 254 * 256 + 0 = 65024
|
|
assert _decode_mqtt_mapping([65024]) == [254]
|
|
|
|
def test_mixed_with_unmapped(self):
|
|
"""Mix of valid and unmapped (65535) values."""
|
|
result = _decode_mqtt_mapping([1, 65535, 0])
|
|
assert result == [1, -1, 0]
|
|
|
|
def test_h2c_real_mapping(self):
|
|
"""Real H2C mapping from MQTT logs: [1, 0, 65535*4, 32768]."""
|
|
mapping = [1, 0, 65535, 65535, 65535, 65535, 32768]
|
|
result = _decode_mqtt_mapping(mapping)
|
|
assert result == [1, 0, -1, -1, -1, -1, 128]
|
|
|
|
def test_non_int_values_treated_as_unmapped(self):
|
|
"""Non-integer values in the mapping are treated as unmapped."""
|
|
assert _decode_mqtt_mapping(["foo", 0]) == [-1, 0]
|
|
|
|
|
|
class TestMatchSlotsByColor:
|
|
"""Tests for _match_slots_by_color() — color-based filament slot to AMS tray matching."""
|
|
|
|
def _ams(self, trays):
|
|
"""Build AMS data from list of (ams_id, tray_id, color_hex, tray_type) tuples."""
|
|
units: dict[int, list] = {}
|
|
for ams_id, tray_id, color, tray_type in trays:
|
|
units.setdefault(ams_id, []).append({"id": tray_id, "tray_color": color, "tray_type": tray_type})
|
|
return [{"id": aid, "tray": t} for aid, t in units.items()]
|
|
|
|
def _usage(self, slots):
|
|
"""Build filament_usage from list of (slot_id, color_hex) tuples."""
|
|
return [{"slot_id": sid, "used_g": 10.0, "type": "PLA", "color": color} for sid, color in slots]
|
|
|
|
def test_none_inputs(self):
|
|
assert _match_slots_by_color(None, None) is None
|
|
assert _match_slots_by_color([], None) is None
|
|
assert _match_slots_by_color(None, {"ams": []}) is None
|
|
|
|
def test_empty_ams(self):
|
|
usage = self._usage([(1, "#FF0000")])
|
|
assert _match_slots_by_color(usage, {"ams": []}) is None
|
|
|
|
def test_single_slot_single_tray(self):
|
|
"""One 3MF slot matches one AMS tray by color."""
|
|
ams = self._ams([(0, 0, "FF0000FF", "PLA")])
|
|
usage = self._usage([(1, "#FF0000")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [0]
|
|
|
|
def test_a1_mini_three_colors(self):
|
|
"""A1 Mini: 3 slots match 3 distinct AMS trays."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "FF0000FF", "PLA"), # Red
|
|
(0, 1, "00FF00FF", "PLA"), # Green
|
|
(0, 2, "0000FFFF", "PLA"), # Blue
|
|
]
|
|
)
|
|
usage = self._usage([(1, "#FF0000"), (2, "#00FF00"), (3, "#0000FF")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [0, 1, 2]
|
|
|
|
def test_dual_ams_p2s_like(self):
|
|
"""P2S with dual AMS: slots from second AMS unit."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "AAAAAAFF", "PLA"),
|
|
(0, 1, "BBBBBBFF", "PLA"),
|
|
(1, 0, "CC0000FF", "PETG"), # global_id=4
|
|
(1, 1, "00CC00FF", "PETG"), # global_id=5
|
|
]
|
|
)
|
|
usage = self._usage([(1, "#CC0000"), (2, "#00CC00")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [4, 5]
|
|
|
|
def test_ams_ht_global_id(self):
|
|
"""AMS-HT (ams_id >= 128) uses raw ams_id as global tray ID."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "FF0000FF", "PLA"),
|
|
(128, 0, "0000FFFF", "PLA"), # AMS-HT → global_id=128
|
|
]
|
|
)
|
|
usage = self._usage([(1, "#FF0000"), (2, "#0000FF")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [0, 128]
|
|
|
|
def test_ambiguous_same_color_returns_none(self):
|
|
"""Two trays with the same color → ambiguous → None."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "FF0000FF", "PLA"),
|
|
(0, 1, "FF0000FF", "PLA"), # Same red
|
|
]
|
|
)
|
|
usage = self._usage([(1, "#FF0000")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) is None
|
|
|
|
def test_no_matching_color_returns_none(self):
|
|
"""3MF slot color not found in any AMS tray → None."""
|
|
ams = self._ams([(0, 0, "00FF00FF", "PLA")])
|
|
usage = self._usage([(1, "#FF0000")]) # Red, but AMS has green
|
|
assert _match_slots_by_color(usage, {"ams": ams}) is None
|
|
|
|
def test_color_normalization_strips_alpha(self):
|
|
"""AMS colors (RRGGBBAA) and 3MF colors (#RRGGBB) match after normalization."""
|
|
ams = self._ams([(0, 0, "AABBCC80", "PLA")]) # 8-char with alpha
|
|
usage = self._usage([(1, "#AABBCC")]) # 6-char with #
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [0]
|
|
|
|
def test_case_insensitive(self):
|
|
"""Color matching is case-insensitive."""
|
|
ams = self._ams([(0, 0, "aaBBccFF", "PLA")])
|
|
usage = self._usage([(1, "#AAbbCC")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [0]
|
|
|
|
def test_empty_tray_color_skipped(self):
|
|
"""Trays with empty color are skipped (not matched)."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "", "PLA"),
|
|
(0, 1, "FF0000FF", "PLA"),
|
|
]
|
|
)
|
|
usage = self._usage([(1, "#FF0000")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [1]
|
|
|
|
def test_empty_tray_type_skipped(self):
|
|
"""Trays with empty tray_type are skipped (unloaded slot)."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "FF0000FF", ""), # Empty slot
|
|
(0, 1, "FF0000FF", "PLA"), # Loaded slot
|
|
]
|
|
)
|
|
usage = self._usage([(1, "#FF0000")])
|
|
assert _match_slots_by_color(usage, {"ams": ams}) == [1]
|
|
|
|
def test_short_slot_color_returns_none(self):
|
|
"""3MF slot with color < 6 chars → can't match → None."""
|
|
ams = self._ams([(0, 0, "FF0000FF", "PLA")])
|
|
usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": "#FFF"}]
|
|
assert _match_slots_by_color(usage, {"ams": ams}) is None
|
|
|
|
def test_slot_id_zero_skipped(self):
|
|
"""Slots with slot_id=0 are skipped."""
|
|
ams = self._ams([(0, 0, "FF0000FF", "PLA")])
|
|
usage = [{"slot_id": 0, "used_g": 10.0, "type": "PLA", "color": "#FF0000"}]
|
|
assert _match_slots_by_color(usage, {"ams": ams}) is None
|
|
|
|
def test_ams_data_as_list(self):
|
|
"""Handles ams_raw as a plain list (some printer models)."""
|
|
ams_list = [{"id": 0, "tray": [{"id": 0, "tray_color": "FF0000FF", "tray_type": "PLA"}]}]
|
|
usage = self._usage([(1, "#FF0000")])
|
|
assert _match_slots_by_color(usage, ams_list) == [0]
|
|
|
|
def test_same_color_two_trays_disambiguated_by_usage(self):
|
|
"""Two trays same color, two slots same color → unique assignment via used_trays tracking."""
|
|
ams = self._ams(
|
|
[
|
|
(0, 0, "FF0000FF", "PLA"),
|
|
(0, 1, "FF0000FF", "PLA"),
|
|
]
|
|
)
|
|
# Two slots both wanting red — first gets tray 0, second gets tray 1? No.
|
|
# When first slot takes the only available, second has 1 left → should work
|
|
usage = self._usage([(1, "#FF0000"), (2, "#FF0000")])
|
|
# First slot: candidates=[0,1], available=[0,1], len!=1 → None
|
|
assert _match_slots_by_color(usage, {"ams": ams}) is None
|
|
|
|
def test_dict_wrapper_with_ams_key(self):
|
|
"""Standard dict format with 'ams' key."""
|
|
ams_data = {"ams": [{"id": 0, "tray": [{"id": 0, "tray_color": "00FF00FF", "tray_type": "PLA"}]}]}
|
|
usage = self._usage([(1, "#00FF00")])
|
|
assert _match_slots_by_color(usage, ams_data) == [0]
|
|
|
|
|
|
class TestMqttMappingIntegration:
|
|
"""Integration tests: MQTT mapping field used in _track_from_3mf."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_h2c_multi_filament_uses_mqtt_mapping(self):
|
|
"""H2C: 3 filaments resolved via MQTT mapping field (no ams_mapping, no queue)."""
|
|
# AMS0-T1 (White PLA), AMS0-T0 (Black PLA), AMS128-T0 (Red PLA)
|
|
spool_white = _make_spool(spool_id=1, label_weight=1000)
|
|
spool_black = _make_spool(spool_id=2, label_weight=1000)
|
|
spool_red = _make_spool(spool_id=3, label_weight=1000)
|
|
assign_white = _make_assignment(spool_id=1, ams_id=0, tray_id=1)
|
|
assign_black = _make_assignment(spool_id=2, ams_id=0, tray_id=0)
|
|
assign_red = _make_assignment(spool_id=3, ams_id=128, tray_id=0)
|
|
archive = _make_archive(archive_id=12)
|
|
|
|
# db: archive, then 3 pairs of (assignment, spool)
|
|
# No queue lookup because MQTT mapping is found first
|
|
db = _mock_db_sequential(
|
|
[
|
|
archive,
|
|
assign_white,
|
|
spool_white,
|
|
assign_black,
|
|
spool_black,
|
|
assign_red,
|
|
spool_red,
|
|
]
|
|
)
|
|
|
|
# MQTT mapping: slot0→AMS0-T1(1), slot1→AMS0-T0(0), slots2-5→unmapped, slot6→AMS128-T0(32768)
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"mapping": [1, 0, 65535, 65535, 65535, 65535, 32768]},
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=255,
|
|
)
|
|
|
|
# 3MF slots 1, 2, 7 (1-based) → indices 0, 1, 6 in mapping
|
|
filament_usage = [
|
|
{"slot_id": 1, "used_g": 21.16, "type": "PLA", "color": "#FFFFFF"},
|
|
{"slot_id": 2, "used_g": 24.22, "type": "PLA", "color": "#000000"},
|
|
{"slot_id": 7, "used_g": 18.47, "type": "PLA", "color": "#F72323"},
|
|
]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=12,
|
|
status="completed",
|
|
print_name="Cube + Cube + Cube",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 3
|
|
|
|
# slot_id=1 → mapping[0]=1 → AMS0-T1 (White PLA)
|
|
assert results[0]["spool_id"] == 1
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 1
|
|
assert results[0]["weight_used"] == 21.2
|
|
|
|
# slot_id=2 → mapping[1]=0 → AMS0-T0 (Black PLA)
|
|
assert results[1]["spool_id"] == 2
|
|
assert results[1]["ams_id"] == 0
|
|
assert results[1]["tray_id"] == 0
|
|
assert results[1]["weight_used"] == 24.2
|
|
|
|
# slot_id=7 → mapping[6]=32768 → AMS128-T0 (Red PLA)
|
|
assert results[2]["spool_id"] == 3
|
|
assert results[2]["ams_id"] == 128
|
|
assert results[2]["tray_id"] == 0
|
|
assert results[2]["weight_used"] == 18.5
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_print_cmd_mapping_takes_priority_over_mqtt(self):
|
|
"""ams_mapping from print command is used even when MQTT mapping exists."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=2)
|
|
archive = _make_archive(archive_id=10)
|
|
|
|
# db: archive, assignment, spool (no queue lookup when ams_mapping provided)
|
|
db = _mock_db_sequential([archive, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"mapping": [0, 65535]}, # MQTT says slot 0 → AMS0-T0
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=255,
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=10,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
ams_mapping=[2], # Print cmd says slot 0 → AMS0-T2 (overrides MQTT)
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["ams_id"] == 0
|
|
assert results[0]["tray_id"] == 2 # From print_cmd mapping, not MQTT
|
|
|
|
|
|
class TestPositionBasedFallbackEmptyAmsSlot:
|
|
"""Position-based mapping fallback (#1607): when no explicit mapping is
|
|
available, the slicer's Nth filament must map to the Nth *loaded* AMS tray
|
|
(skipping empty slots), not the Nth physical slot position. BambuStudio /
|
|
OrcaSlicer compact their filament-assignment UI by hiding unloaded AMS
|
|
slots, so the 3MF slot list is dense even when the AMS itself has gaps."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_external_routed_correctly_when_ams_has_empty_middle_slot(self):
|
|
"""Reporter's scenario: AMS trays 0-2 loaded, tray 3 empty, external
|
|
loaded. Slicer emits 4 filaments — slot 4 = external. Without the fix
|
|
the position-based fallback maps slot 4 to the empty AMS tray 3
|
|
(since `available_trays = [0, 1, 2, 3, 254]`) and external usage is
|
|
silently dropped because no spool is assigned to AMS0-T3.
|
|
After the fix, empty AMS slots are filtered (tray_type is empty) so
|
|
`available_trays = [0, 1, 2, 254]` and slot 4 correctly resolves to
|
|
the external (global tray 254 → AMS255-T0)."""
|
|
# Spool fed via external (vt_tray 254 → AMS255-T0)
|
|
spool = _make_spool(spool_id=42, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=42, ams_id=255, tray_id=0)
|
|
archive = _make_archive(archive_id=70)
|
|
|
|
# db: archive, queue_item(None), assignment, spool
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
# AMS reports 4 physical tray slots but slot 3 has no spool (empty
|
|
# tray_type); external spool is loaded in vt_tray.
|
|
# No `mapping` field on the state — forces fallback through path 5.
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={
|
|
"ams": [
|
|
{
|
|
"id": 0,
|
|
"tray": [
|
|
{"id": 0, "tray_type": "PLA"},
|
|
{"id": 1, "tray_type": "PETG"},
|
|
{"id": 2, "tray_type": "ABS"},
|
|
{"id": 3, "tray_type": ""}, # empty slot
|
|
],
|
|
}
|
|
],
|
|
"vt_tray": [{"id": 254, "tray_type": "PLA"}],
|
|
},
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=254,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
# 3MF has 4 dense filament slots — slot 4 is the external. Only slot 4
|
|
# has weight (other slots came from AMS spools handled separately).
|
|
filament_usage = [{"slot_id": 4, "used_g": 12.3, "type": "PLA", "color": "#00AABB"}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=70,
|
|
status="completed",
|
|
print_name="External + AMS print",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
# The external spool was charged, NOT the empty AMS slot.
|
|
assert results[0]["spool_id"] == 42
|
|
assert results[0]["ams_id"] == 255
|
|
assert results[0]["tray_id"] == 0
|
|
assert results[0]["weight_used"] == 12.3
|
|
assert (255, 0) in handled_trays
|
|
# Critical assertion: AMS0-T3 (the empty slot) was NOT charged.
|
|
assert (0, 3) not in handled_trays
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dense_ams_unchanged_no_empty_slots(self):
|
|
"""Sanity check: when every AMS slot is loaded, the position-based
|
|
fallback still works for the slicer's external = last slot case."""
|
|
spool = _make_spool(spool_id=99, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=99, ams_id=255, tray_id=0)
|
|
archive = _make_archive(archive_id=71)
|
|
|
|
db = _mock_db_sequential([archive, None, assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={
|
|
"ams": [
|
|
{
|
|
"id": 0,
|
|
"tray": [
|
|
{"id": 0, "tray_type": "PLA"},
|
|
{"id": 1, "tray_type": "PETG"},
|
|
{"id": 2, "tray_type": "ABS"},
|
|
{"id": 3, "tray_type": "TPU"},
|
|
],
|
|
}
|
|
],
|
|
"vt_tray": [{"id": 254, "tray_type": "PLA"}],
|
|
},
|
|
progress=100,
|
|
layer_num=50,
|
|
tray_now=254,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
# 5 filaments, slot 5 = external. available_trays = [0,1,2,3,254] →
|
|
# slot_id=5 → available_trays[4] = 254.
|
|
filament_usage = [{"slot_id": 5, "used_g": 7.5, "type": "PLA", "color": ""}]
|
|
handled_trays: set[tuple[int, int]] = set()
|
|
|
|
with (
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
):
|
|
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=71,
|
|
status="completed",
|
|
print_name="Dense AMS + external",
|
|
handled_trays=handled_trays,
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 99
|
|
assert results[0]["ams_id"] == 255
|
|
assert results[0]["tray_id"] == 0
|
|
|
|
|
|
class TestNotificationVariables:
|
|
"""Tests for filament_details formatting in notifications."""
|
|
|
|
def test_filament_details_single_slot(self):
|
|
"""Single slot produces 'PLA: 15.2g' format."""
|
|
slots = [{"type": "PLA", "used_g": 15.2, "slot_id": 1, "color": "#FF0000"}]
|
|
parts = []
|
|
for slot in slots:
|
|
ftype = slot.get("type", "Unknown") or "Unknown"
|
|
used = slot.get("used_g", 0)
|
|
parts.append(f"{ftype}: {used:.1f}g")
|
|
result = " | ".join(parts)
|
|
assert result == "PLA: 15.2g"
|
|
|
|
def test_filament_details_multi_slot(self):
|
|
"""Multiple slots produce 'PLA: 10.0g | PETG: 5.0g' format."""
|
|
slots = [
|
|
{"type": "PLA", "used_g": 10.0, "slot_id": 1, "color": ""},
|
|
{"type": "PETG", "used_g": 5.0, "slot_id": 2, "color": ""},
|
|
]
|
|
parts = []
|
|
for slot in slots:
|
|
ftype = slot.get("type", "Unknown") or "Unknown"
|
|
used = slot.get("used_g", 0)
|
|
parts.append(f"{ftype}: {used:.1f}g")
|
|
result = " | ".join(parts)
|
|
assert result == "PLA: 10.0g | PETG: 5.0g"
|
|
|
|
def test_filament_details_empty_type(self):
|
|
"""Empty type defaults to 'Unknown'."""
|
|
slots = [{"type": "", "used_g": 5.0, "slot_id": 1, "color": ""}]
|
|
parts = []
|
|
for slot in slots:
|
|
ftype = slot.get("type", "Unknown") or "Unknown"
|
|
used = slot.get("used_g", 0)
|
|
parts.append(f"{ftype}: {used:.1f}g")
|
|
result = " | ".join(parts)
|
|
assert result == "Unknown: 5.0g"
|
|
|
|
def test_filament_grams_scaled_for_partial(self):
|
|
"""filament_grams is scaled by progress for partial prints."""
|
|
filament_used_grams = 20.0
|
|
progress = 50
|
|
scale = max(0.0, min(progress / 100.0, 1.0))
|
|
scaled = round(filament_used_grams * scale, 1)
|
|
assert scaled == 10.0
|
|
|
|
def test_filament_grams_zero_progress(self):
|
|
"""Progress=0 at cancellation gives 0.0g."""
|
|
filament_used_grams = 20.0
|
|
progress = 0
|
|
scale = max(0.0, min(progress / 100.0, 1.0))
|
|
scaled = round(filament_used_grams * scale, 1)
|
|
assert scaled == 0.0
|
|
|
|
def test_slot_scaling_for_partial(self):
|
|
"""Per-slot usage is scaled linearly for partial prints."""
|
|
slots = [
|
|
{"type": "PLA", "used_g": 20.0, "slot_id": 1, "color": ""},
|
|
{"type": "PETG", "used_g": 10.0, "slot_id": 2, "color": ""},
|
|
]
|
|
progress = 30
|
|
scale = max(0.0, min(progress / 100.0, 1.0))
|
|
scaled_slots = [{**s, "used_g": round(s["used_g"] * scale, 1)} for s in slots]
|
|
assert scaled_slots[0]["used_g"] == 6.0
|
|
assert scaled_slots[1]["used_g"] == 3.0
|
|
|
|
|
|
class TestOnPrintStartAmsMapping:
|
|
"""Tests for ams_mapping capture in on_print_start()."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_sessions(self):
|
|
_active_sessions.clear()
|
|
yield
|
|
_active_sessions.clear()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_ams_mapping_from_data(self):
|
|
"""on_print_start captures ams_mapping from the data dict into the session."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
|
|
tray_now=0,
|
|
)
|
|
|
|
await on_print_start(1, {"subtask_name": "Test", "ams_mapping": [3, -1, -1, 2]}, printer_manager)
|
|
|
|
assert _active_sessions[1].ams_mapping == [3, -1, -1, 2]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ams_mapping_none_when_not_in_data(self):
|
|
"""Session ams_mapping is None when data dict has no ams_mapping."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
|
|
tray_now=0,
|
|
)
|
|
|
|
await on_print_start(1, {"subtask_name": "Test"}, printer_manager)
|
|
|
|
assert _active_sessions[1].ams_mapping is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_captures_queue_plate_id(self):
|
|
"""on_print_start records the queue item's plate_id onto the session (#1697)."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
|
|
tray_now=0,
|
|
)
|
|
|
|
queue_item = _make_queue_item(plate_id=2)
|
|
# on_print_start now executes: SpoolAssignment lookup, then PrintQueueItem lookup.
|
|
db = AsyncMock()
|
|
assignment_result = MagicMock()
|
|
assignment_result.scalars.return_value.all.return_value = []
|
|
queue_result = MagicMock()
|
|
queue_result.scalars.return_value.first.return_value = queue_item
|
|
db.execute = AsyncMock(side_effect=[assignment_result, queue_result])
|
|
|
|
await on_print_start(1, {"subtask_name": "Test"}, printer_manager, db=db)
|
|
|
|
assert _active_sessions[1].plate_id == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plate_id_none_when_no_queue_item(self):
|
|
"""Direct/library prints with no queue item leave session.plate_id = None."""
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]},
|
|
tray_now=0,
|
|
)
|
|
|
|
db = AsyncMock()
|
|
assignment_result = MagicMock()
|
|
assignment_result.scalars.return_value.all.return_value = []
|
|
queue_result = MagicMock()
|
|
queue_result.scalars.return_value.first.return_value = None
|
|
db.execute = AsyncMock(side_effect=[assignment_result, queue_result])
|
|
|
|
await on_print_start(1, {"subtask_name": "Test"}, printer_manager, db=db)
|
|
|
|
assert _active_sessions[1].plate_id is None
|
|
|
|
|
|
class TestFindThreemfByFilename:
|
|
"""Tests for _find_3mf_by_filename() — library/archive search without archive_id."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_finds_library_file(self):
|
|
"""Finds a 3MF from library files matching filename."""
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
lib_file = MagicMock()
|
|
lib_file.file_path = "library/BMCU-BADGE.3mf"
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [lib_file]
|
|
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(return_value=mock_result)
|
|
|
|
base_dir = MagicMock(spec=Path)
|
|
candidate = MagicMock(spec=Path)
|
|
candidate.exists.return_value = True
|
|
candidate.suffix = ".3mf"
|
|
base_dir.__truediv__ = MagicMock(return_value=candidate)
|
|
|
|
result = await _find_3mf_by_filename(1, "BMCU-BADGE.3mf", db, base_dir)
|
|
|
|
assert result == candidate
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_none_for_empty_filename(self):
|
|
"""Returns None when filename is empty or just extensions."""
|
|
db = AsyncMock()
|
|
base_dir = MagicMock()
|
|
|
|
result = await _find_3mf_by_filename(1, ".3mf", db, base_dir)
|
|
assert result is None
|
|
|
|
result = await _find_3mf_by_filename(1, "", db, base_dir)
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_falls_through_to_archive_search(self):
|
|
"""Falls back to previous archives when library search returns no results."""
|
|
from pathlib import Path
|
|
|
|
# Library returns nothing
|
|
empty_result = MagicMock()
|
|
empty_result.scalars.return_value.all.return_value = []
|
|
|
|
# Archive returns a match
|
|
archive = MagicMock()
|
|
archive.id = 35
|
|
archive.file_path = "archives/35/BMCU-BADGE.3mf"
|
|
archive_result = MagicMock()
|
|
archive_result.scalars.return_value.all.return_value = [archive]
|
|
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(side_effect=[empty_result, archive_result])
|
|
|
|
base_dir = MagicMock(spec=Path)
|
|
candidate = MagicMock(spec=Path)
|
|
candidate.exists.return_value = True
|
|
candidate.suffix = ".3mf"
|
|
base_dir.__truediv__ = MagicMock(return_value=candidate)
|
|
|
|
result = await _find_3mf_by_filename(1, "BMCU-BADGE.3mf", db, base_dir)
|
|
|
|
assert result == candidate
|
|
assert db.execute.call_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_none_when_nothing_found(self):
|
|
"""Returns None when neither library nor archives have a matching 3MF."""
|
|
empty_result = MagicMock()
|
|
empty_result.scalars.return_value.all.return_value = []
|
|
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(return_value=empty_result)
|
|
|
|
base_dir = MagicMock()
|
|
|
|
result = await _find_3mf_by_filename(1, "nonexistent.3mf", db, base_dir)
|
|
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strips_path_and_extensions(self):
|
|
"""Correctly strips path components and extensions for search."""
|
|
empty_result = MagicMock()
|
|
empty_result.scalars.return_value.all.return_value = []
|
|
|
|
db = AsyncMock()
|
|
db.execute = AsyncMock(return_value=empty_result)
|
|
|
|
base_dir = MagicMock()
|
|
|
|
# Should search for "BMCU-BADGE" base name even with path and .gcode.3mf
|
|
await _find_3mf_by_filename(1, "/sdcard/BMCU-BADGE.gcode.3mf", db, base_dir)
|
|
|
|
# Verify the execute was called (search was attempted with stripped name)
|
|
assert db.execute.call_count == 2 # library + archive search
|
|
|
|
|
|
class TestTrackFrom3mfWithPreresolvedPath:
|
|
"""Tests for _track_from_3mf() with threemf_path (no archive needed)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uses_preresolved_path_without_archive(self):
|
|
"""When threemf_path is provided with archive_id=None, uses the path directly."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=3)
|
|
|
|
# DB: 1st call = assignment lookup (live), 2nd = spool lookup
|
|
db = _mock_db_sequential([assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": []}]},
|
|
tray_now=255,
|
|
last_loaded_tray=3,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 5.0, "type": "PETG", "color": "#FFFFFF"}]
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
):
|
|
mock_settings.base_dir = MagicMock()
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=None,
|
|
status="completed",
|
|
print_name="BMCU-BADGE",
|
|
handled_trays=set(),
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
ams_mapping=[3, -1, -1, -1],
|
|
threemf_path=mock_path,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["spool_id"] == 1
|
|
assert results[0]["weight_used"] == 5.0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skips_queue_lookup_without_archive_id(self):
|
|
"""When archive_id is None, queue item lookup is skipped."""
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
|
|
db = _mock_db_sequential([assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": []}]},
|
|
tray_now=0,
|
|
last_loaded_tray=0,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
filament_usage = [{"slot_id": 1, "used_g": 2.0, "type": "PLA", "color": "#FF0000"}]
|
|
|
|
with (
|
|
patch(
|
|
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
|
|
return_value=filament_usage,
|
|
),
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
):
|
|
mock_settings.base_dir = MagicMock()
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
|
|
# Should NOT fail even though there's no archive_id for queue lookup
|
|
results = await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=None,
|
|
status="completed",
|
|
print_name="Test",
|
|
handled_trays=set(),
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=0,
|
|
threemf_path=mock_path,
|
|
)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["weight_used"] == 2.0
|
|
|
|
|
|
class TestTrackFrom3mfPlateId:
|
|
"""plate_id must propagate from PrintSession through _track_from_3mf to the
|
|
3MF parser, so multi-plate files dispatched for one plate only count that
|
|
plate's filament (#1697)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_passes_plate_id_to_3mf_extract(self):
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
|
|
db = _mock_db_sequential([assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": []}]},
|
|
tray_now=0,
|
|
last_loaded_tray=0,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
extract_mock = MagicMock(return_value=[{"slot_id": 1, "used_g": 190.0, "type": "PETG", "color": "#888888"}])
|
|
|
|
with (
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", extract_mock),
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
):
|
|
mock_settings.base_dir = MagicMock()
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
|
|
await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=None,
|
|
status="completed",
|
|
print_name="GridfinityLid",
|
|
handled_trays=set(),
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=0,
|
|
threemf_path=mock_path,
|
|
plate_id=2,
|
|
)
|
|
|
|
# plate_id=2 passed positionally as second arg
|
|
assert extract_mock.call_count == 1
|
|
assert extract_mock.call_args.args[1] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plate_id_none_for_non_queue_print(self):
|
|
spool = _make_spool(spool_id=1, label_weight=1000)
|
|
assignment = _make_assignment(spool_id=1, ams_id=0, tray_id=0)
|
|
|
|
db = _mock_db_sequential([assignment, spool])
|
|
|
|
printer_manager = MagicMock()
|
|
printer_manager.get_status.return_value = SimpleNamespace(
|
|
raw_data={"ams": [{"id": 0, "tray": []}]},
|
|
tray_now=0,
|
|
last_loaded_tray=0,
|
|
tray_change_log=[],
|
|
)
|
|
|
|
extract_mock = MagicMock(return_value=[{"slot_id": 1, "used_g": 5.0, "type": "PLA", "color": "#FF0000"}])
|
|
|
|
with (
|
|
patch("backend.app.utils.threemf_tools.extract_filament_usage_from_3mf", extract_mock),
|
|
patch("backend.app.core.config.settings") as mock_settings,
|
|
):
|
|
mock_settings.base_dir = MagicMock()
|
|
mock_path = MagicMock()
|
|
mock_path.exists.return_value = True
|
|
|
|
# No plate_id kwarg — direct/library Print flow.
|
|
await _track_from_3mf(
|
|
printer_id=1,
|
|
archive_id=None,
|
|
status="completed",
|
|
print_name="DirectPrint",
|
|
handled_trays=set(),
|
|
printer_manager=printer_manager,
|
|
db=db,
|
|
tray_now_at_start=0,
|
|
threemf_path=mock_path,
|
|
)
|
|
|
|
assert extract_mock.call_args.args[1] is None
|