"""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, timezone from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import pytest from backend.app.services.usage_tracker import ( PrintSession, _active_sessions, _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 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"): """Create a mock PrintQueueItem object.""" item = MagicMock() item.ams_mapping = ams_mapping item.status = status 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 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}]}]} ) 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_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}]}]} ) 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.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}]}]}, ) # db returns assignment then spool db = _mock_db_sequential([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_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 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