|
|
@@ -15,6 +15,7 @@ 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,
|
|
|
@@ -254,8 +255,9 @@ class TestOnPrintComplete:
|
|
|
last_loaded_tray=-1,
|
|
|
)
|
|
|
|
|
|
- # db returns assignment then spool
|
|
|
- db = _mock_db_sequential([assignment, spool])
|
|
|
+ # 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,
|
|
|
@@ -1739,3 +1741,237 @@ class TestNotificationVariables:
|
|
|
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
|
|
|
+
|
|
|
+
|
|
|
+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
|
maziggy