|
|
@@ -0,0 +1,858 @@
|
|
|
+"""Regression tests for the cross-printer AMS mapping bug (#2799).
|
|
|
+
|
|
|
+One model queued to three P2S at once produced three queue items carrying a
|
|
|
+byte-identical ``ams_mapping``. That mapping was correct for the printer it was
|
|
|
+computed against and wrong for the other two, whose AMS held the same spools in
|
|
|
+different slots — so two of the three printed the lettering in ASA where the 3MF
|
|
|
+strictly requires PETG. An explicit ``ams_mapping`` bypasses the firmware's own
|
|
|
+type check, so nothing downstream caught it.
|
|
|
+
|
|
|
+The reproduction below is the real incident: the 3MF needs light-blue PETG in
|
|
|
+slot 1 and dark-blue PETG in slot 3; P2S-4 holds them in trays 2 and 1, while
|
|
|
+P2S-5 holds them in trays 0 and 2 with ASA in tray 1. ``[2, -1, 1]`` is right on
|
|
|
+the first and feeds ASA into slot 3 on the second.
|
|
|
+"""
|
|
|
+
|
|
|
+import json
|
|
|
+from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.services.print_scheduler import PrintScheduler
|
|
|
+
|
|
|
+# 3MF requirements: slot 1 light-blue PETG (body), slot 3 dark-blue PETG (text).
|
|
|
+REQUIREMENTS = [
|
|
|
+ {"slot_id": 1, "type": "PETG", "color": "#76D9F4", "tray_info_idx": "GFG99"},
|
|
|
+ {"slot_id": 3, "type": "PETG", "color": "#2850E0", "tray_info_idx": "GFG99"},
|
|
|
+]
|
|
|
+
|
|
|
+# AMS layouts as reported over MQTT, same spools in a different slot order.
|
|
|
+P2S_4_TRAYS = [
|
|
|
+ {"global_tray_id": 0, "type": "ASA", "color": "161616FF"},
|
|
|
+ {"global_tray_id": 1, "type": "PETG", "color": "2850E0FF"},
|
|
|
+ {"global_tray_id": 2, "type": "PETG", "color": "76D9F4FF"},
|
|
|
+ {"global_tray_id": 3, "type": "ASA", "color": "898989FF"},
|
|
|
+]
|
|
|
+P2S_5_TRAYS = [
|
|
|
+ {"global_tray_id": 0, "type": "PETG", "color": "76D9F4FF"},
|
|
|
+ {"global_tray_id": 1, "type": "ASA", "color": "161616FF"},
|
|
|
+ {"global_tray_id": 2, "type": "PETG", "color": "2850E0FF"},
|
|
|
+ {"global_tray_id": 3, "type": "ASA", "color": "898989FF"},
|
|
|
+]
|
|
|
+
|
|
|
+# The mapping the print dialog stamped onto all three items.
|
|
|
+SHARED_MAPPING = [2, -1, 1]
|
|
|
+
|
|
|
+
|
|
|
+def _item(**overrides):
|
|
|
+ item = MagicMock()
|
|
|
+ item.id = 276
|
|
|
+ item.printer_id = 8
|
|
|
+ item.ams_mapping = json.dumps(SHARED_MAPPING)
|
|
|
+ item.skip_filament_check = False
|
|
|
+ item.filament_overrides = None
|
|
|
+ item.required_filament_types = None
|
|
|
+ item.manual_start = False
|
|
|
+ item.filament_short = False
|
|
|
+ for key, value in overrides.items():
|
|
|
+ setattr(item, key, value)
|
|
|
+ return item
|
|
|
+
|
|
|
+
|
|
|
+def _scheduler(trays, requirements=REQUIREMENTS):
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ scheduler._build_loaded_filaments = MagicMock(return_value=list(trays))
|
|
|
+ scheduler._get_filament_requirements = AsyncMock(
|
|
|
+ return_value=[dict(r) for r in requirements] if requirements else None
|
|
|
+ )
|
|
|
+ return scheduler
|
|
|
+
|
|
|
+
|
|
|
+class TestStoredMappingConflict:
|
|
|
+ """The stored mapping is re-checked against the printer about to run it."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_mapping_fits_the_printer_it_was_computed_for(self):
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), SHARED_MAPPING)
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_same_mapping_on_a_differently_loaded_printer_conflicts(self):
|
|
|
+ """The incident: slot 3 would have printed in ASA."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 8, _item(), SHARED_MAPPING)
|
|
|
+ assert conflict is not None
|
|
|
+ assert "slot 3" in conflict
|
|
|
+ assert "ASA" in conflict
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_tray_not_loaded_conflicts(self):
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS[:2]) # trays 2 and 3 removed
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), SHARED_MAPPING)
|
|
|
+ assert conflict is not None
|
|
|
+ assert "does not have loaded" in conflict
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_mapping_too_short_for_the_plate_conflicts(self):
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), [2])
|
|
|
+ assert conflict is not None
|
|
|
+ assert "slot 3" in conflict
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unreported_external_feed_is_not_a_conflict(self):
|
|
|
+ """An external spool we have not heard about is absence of evidence."""
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), [254, -1, 255])
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_external_feed_is_type_checked_like_any_tray(self):
|
|
|
+ """Two printers with different filament in the external feed is the same
|
|
|
+ failure this check exists for, so >=254 is not a free pass."""
|
|
|
+ trays = P2S_4_TRAYS + [{"global_tray_id": 254, "type": "ASA", "color": "161616FF"}]
|
|
|
+ scheduler = _scheduler(trays)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), [254, -1, 1])
|
|
|
+ assert conflict is not None
|
|
|
+ assert "ASA" in conflict
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_an_external_feed_does_not_excuse_the_rest_of_the_mapping(self):
|
|
|
+ """An unrecognised >=254 entry is absence of evidence about that slot
|
|
|
+ only. Bailing out of the whole pass on it let a foreign tray later in
|
|
|
+ the mapping through unexamined."""
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), [254, -1, 7])
|
|
|
+ assert conflict is not None
|
|
|
+ assert "slot 3" in conflict
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unresolved_required_slot_is_not_a_conflict(self):
|
|
|
+ """-1 says the matcher had nothing, not that the mapping is foreign.
|
|
|
+ Recomputing on it would discard the slots the user did resolve by hand;
|
|
|
+ _block_on_unmatched_filament owns this case instead."""
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), [2, -1, -1])
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_foreign_tray_is_caught_without_reading_the_3mf(self):
|
|
|
+ """The cheap pass runs on live status alone, so a foreign mapping is
|
|
|
+ rejected without opening the 3MF."""
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 9, _item(), [7, -1, 1])
|
|
|
+ assert conflict is not None
|
|
|
+ assert "does not have loaded" in conflict
|
|
|
+ scheduler._get_filament_requirements.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_no_status_is_not_a_conflict(self):
|
|
|
+ """Absence of evidence never triggers a recompute."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = None
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 8, _item(), SHARED_MAPPING)
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_no_trays_reported_yet_is_not_a_conflict(self):
|
|
|
+ scheduler = _scheduler([])
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 8, _item(), SHARED_MAPPING)
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unreadable_requirements_is_not_a_conflict(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS, requirements=None)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 8, _item(), SHARED_MAPPING)
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_mapping_that_is_not_a_list_is_not_a_conflict(self):
|
|
|
+ """A corrupted row is not evidence about the printer, and the guard that
|
|
|
+ keeps it from being indexed was the one thing here without a test."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 8, _item(), {"1": 2})
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_boolean_is_not_read_as_tray_one(self):
|
|
|
+ """`true` is an int in Python, so it would otherwise be judged against
|
|
|
+ whatever tray 1 holds — here ASA, i.e. a conflict reported for a slot
|
|
|
+ nothing is actually known about."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ conflict = await scheduler._stored_mapping_conflict(AsyncMock(), 8, _item(), [True, -1, 2])
|
|
|
+ assert conflict is None
|
|
|
+
|
|
|
+
|
|
|
+class TestEnsureAmsMappingRevalidates:
|
|
|
+ """``_ensure_ams_mapping`` acts on the conflict verdict."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_conflicting_mapping_is_recomputed(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[0, -1, 2])
|
|
|
+ item = _item()
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, 2]
|
|
|
+ scheduler._compute_ams_mapping_for_printer.assert_awaited_once()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_fitting_mapping_is_left_untouched(self):
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[0, -1, 2])
|
|
|
+ item = _item(printer_id=9)
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 9, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == SHARED_MAPPING
|
|
|
+ scheduler._compute_ams_mapping_for_printer.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_print_anyway_keeps_the_users_mapping(self):
|
|
|
+ """skip_filament_check is the user overriding exactly this judgement."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[0, -1, 2])
|
|
|
+ item = _item(skip_filament_check=True)
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == SHARED_MAPPING
|
|
|
+ scheduler._compute_ams_mapping_for_printer.assert_not_awaited()
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_failed_recompute_does_not_fall_back_to_the_bad_mapping(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=None)
|
|
|
+ scheduler._unmappable_without_ams_message = AsyncMock(return_value=None)
|
|
|
+ item = _item()
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+
|
|
|
+ assert item.ams_mapping is None
|
|
|
+
|
|
|
+
|
|
|
+class TestBlockOnUnmatchedFilament:
|
|
|
+ """A slot the plate prints with no tray on this printer holds the job."""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unresolved_required_slot_holds(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(db, item)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+ assert item.manual_start is True
|
|
|
+ # The reason names the filament, not a machine token, because it renders
|
|
|
+ # on the queue row.
|
|
|
+ assert item.waiting_reason.startswith("Needs ")
|
|
|
+ assert "PETG" in item.waiting_reason
|
|
|
+ assert "#2850E0" in item.waiting_reason
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_padding_for_a_slot_this_plate_skips_does_not_hold(self):
|
|
|
+ """-1 at slot 2, which this plate never prints, is not a missing filament.
|
|
|
+ This is why the hold intersects with the requirement list rather than
|
|
|
+ reading the mapping array on its own."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, 2]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is False
|
|
|
+ assert item.manual_start is False
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_mapping_too_short_for_the_plate_holds(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=json.dumps([0]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_print_anyway_dispatches(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]), skip_filament_check=True)
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is False
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_mapping_that_resolved_nothing_at_all_holds(self):
|
|
|
+ """The complete miss used to fall out of this gate and into the ugly path.
|
|
|
+
|
|
|
+ `_ensure_ams_mapping` clears a rejected mapping the recompute could not
|
|
|
+ replace, and bailing on `not item.ams_mapping` then dispatched the item
|
|
|
+ with no mapping at all: several megabytes uploaded, 0700_8012 from the
|
|
|
+ firmware, retries burned. Live status reporting loaded trays while the
|
|
|
+ matcher resolved none of them is a positive finding, so it holds.
|
|
|
+ """
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=None)
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+ assert item.manual_start is True
|
|
|
+ assert item.waiting_reason.startswith("Needs ")
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_cleared_mapping_on_a_printer_with_no_trays_reported_dispatches(self):
|
|
|
+ """#2589 is the same shape and must stay untouched: there the mapping is
|
|
|
+ bogus because the AMS was not known yet, and an empty loaded list is
|
|
|
+ exactly that ignorance. Holding on it would hold on absence of
|
|
|
+ evidence."""
|
|
|
+ scheduler = _scheduler([])
|
|
|
+ item = _item(ams_mapping=None)
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is False
|
|
|
+ assert item.manual_start is False
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_cleared_mapping_with_no_status_dispatches(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=None)
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = None
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is False
|
|
|
+ assert item.manual_start is False
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_boolean_tray_id_holds_rather_than_reading_as_tray_one(self):
|
|
|
+ """`true` is an int in Python. A hand-written payload carrying it would
|
|
|
+ otherwise be read as tray 1 and dispatched as whatever sits there."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=json.dumps([True, -1, 2]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+ assert "#76D9F4" in item.waiting_reason
|
|
|
+
|
|
|
+
|
|
|
+class TestHoldOnTheModelBasedPath:
|
|
|
+ """A held "any P2S" job must not stay pinned to the P2S that held it.
|
|
|
+
|
|
|
+ The model-based path commits `item.printer_id` before the gates run, so a
|
|
|
+ hold would otherwise leave an unpinned job pinned to the one printer that
|
|
|
+ could not run it, and the next press of the start button retries the same
|
|
|
+ dead end. The printer-assigned path keeps its printer: that one was the
|
|
|
+ user's choice, not the scheduler's.
|
|
|
+ """
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_model_based_hold_releases_the_printer_and_its_mapping(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item, release_assignment=True)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+ assert item.printer_id is None
|
|
|
+ # The mapping goes with the assignment: its tray IDs were resolved
|
|
|
+ # against the printer just released, so leaving it would hold the job
|
|
|
+ # again on the next printer even when that one has the filament.
|
|
|
+ assert item.ams_mapping is None
|
|
|
+ assert item.waiting_reason.startswith("Needs ")
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_notification_still_names_the_printer_that_held_it(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ await scheduler._block_on_unmatched_filament(AsyncMock(), item, release_assignment=True)
|
|
|
+
|
|
|
+ # Looked up before the release, or it would be looked up as None.
|
|
|
+ assert scheduler._get_printer.await_args.args[1] == 8
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_printer_assigned_hold_keeps_both(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+ assert item.printer_id == 8
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, -1]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_unreadable_requirements_dispatches(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS, requirements=None)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is False
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_nozzle_bound_filament_on_the_wrong_side_holds(self):
|
|
|
+ """The H2D case the type-only scan waved through: PETG is loaded, but on
|
|
|
+ the other nozzle's AMS, so the nozzle-aware matcher left the slot at -1.
|
|
|
+ Reading the mapping inherits that restriction for free."""
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="H2D"))
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert blocked is True
|
|
|
+
|
|
|
+
|
|
|
+class TestModelBasedVirtualPrinterMapping:
|
|
|
+ """A model-based Virtual Printer stamps a mapping nothing can attribute.
|
|
|
+
|
|
|
+ `virtual_printer/manager.py` writes the slicer's own AMS pick onto the queue
|
|
|
+ item it creates, on the line below `printer_id=self.target_printer_id` —
|
|
|
+ which is None for an "Any P2S" VP. So the tray IDs the slicer resolved are
|
|
|
+ carried by an item the scheduler will hand to whichever printer of that
|
|
|
+ model happens to be free, and the row itself records no printer they could
|
|
|
+ be checked against. Same failure as the print dialog's, arriving from a
|
|
|
+ different direction, and reachable whenever the per-VP `save_ams_mapping`
|
|
|
+ opt-in is on without `queue_force_color_match`.
|
|
|
+
|
|
|
+ The item has no `printer_id` until the model-based path assigns one, so
|
|
|
+ these go through `_ensure_ams_mapping`'s `printer_id` argument rather than
|
|
|
+ `item.printer_id` — which is how that path calls it.
|
|
|
+ """
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_slicer_mapping_is_rechecked_against_the_printer_chosen_for_it(self):
|
|
|
+ scheduler = _scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[0, -1, 2])
|
|
|
+ # Fresh from the VP: no printer_id, mapping resolved by the slicer.
|
|
|
+ item = _item(printer_id=None)
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 10, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, 2]
|
|
|
+ # Judged against the printer the scheduler picked, not the row's own
|
|
|
+ # (absent) one.
|
|
|
+ scheduler._compute_ams_mapping_for_printer.assert_awaited_once()
|
|
|
+ assert scheduler._compute_ams_mapping_for_printer.await_args.args[1] == 10
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_slicer_mapping_survives_on_a_printer_it_happens_to_fit(self):
|
|
|
+ """Not every VP mapping is wrong — one that fits the chosen printer is
|
|
|
+ kept, so the slicer's deliberate slot pick is not thrown away."""
|
|
|
+ scheduler = _scheduler(P2S_4_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[0, -1, 2])
|
|
|
+ item = _item(printer_id=None)
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 9, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == SHARED_MAPPING
|
|
|
+ scheduler._compute_ams_mapping_for_printer.assert_not_awaited()
|
|
|
+
|
|
|
+
|
|
|
+class TestFilamentRequirementMemo:
|
|
|
+ """The two gates parse the same 3MF per dispatch attempt (#2799 review).
|
|
|
+
|
|
|
+ A one-file fan-out to many idle printers repeats that for every one of them
|
|
|
+ inside a single pass, each time a DB lookup plus a zip open and XML parse.
|
|
|
+ The memo is scoped to the pass so a file re-sliced between ticks is re-read.
|
|
|
+ """
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _db_returning(archive):
|
|
|
+ result = MagicMock()
|
|
|
+ result.scalar_one_or_none.return_value = archive
|
|
|
+ db = AsyncMock()
|
|
|
+ db.execute = AsyncMock(return_value=result)
|
|
|
+ return db
|
|
|
+
|
|
|
+ @pytest.fixture
|
|
|
+ def parsed(self, tmp_path, monkeypatch):
|
|
|
+ from backend.app.services import print_scheduler as ps
|
|
|
+
|
|
|
+ (tmp_path / "plate.3mf").write_bytes(b"")
|
|
|
+ monkeypatch.setattr(ps.settings, "base_dir", tmp_path)
|
|
|
+
|
|
|
+ calls = []
|
|
|
+
|
|
|
+ def fake_extract(path, plate_id=None):
|
|
|
+ calls.append((path, plate_id))
|
|
|
+ return [{"slot_id": 1, "type": "PETG", "color": "#76D9F4", "tray_info_idx": ""}]
|
|
|
+
|
|
|
+ monkeypatch.setattr(
|
|
|
+ "backend.app.services.filament_requirements.extract_filament_requirements",
|
|
|
+ fake_extract,
|
|
|
+ )
|
|
|
+ return calls
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_same_plate_is_parsed_once_per_pass(self, parsed):
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ db = self._db_returning(MagicMock(file_path="plate.3mf"))
|
|
|
+ item = _item(archive_id=731, library_file_id=None, plate_id=22)
|
|
|
+
|
|
|
+ await scheduler._get_filament_requirements(db, item)
|
|
|
+ await scheduler._get_filament_requirements(db, item)
|
|
|
+
|
|
|
+ assert len(parsed) == 1
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_each_caller_gets_its_own_copy(self, parsed):
|
|
|
+ """`_apply_filament_overrides` rewrites these dicts in place, so handing
|
|
|
+ out the cached list would leak one item's overrides into the next."""
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ db = self._db_returning(MagicMock(file_path="plate.3mf"))
|
|
|
+ item = _item(archive_id=731, library_file_id=None, plate_id=22)
|
|
|
+
|
|
|
+ first = await scheduler._get_filament_requirements(db, item)
|
|
|
+ first[0]["type"] = "ASA" # an override applied by one caller
|
|
|
+
|
|
|
+ second = await scheduler._get_filament_requirements(db, item)
|
|
|
+ assert second[0]["type"] == "PETG"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_different_plate_is_parsed_separately(self, parsed):
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ db = self._db_returning(MagicMock(file_path="plate.3mf"))
|
|
|
+
|
|
|
+ await scheduler._get_filament_requirements(db, _item(archive_id=731, library_file_id=None, plate_id=22))
|
|
|
+ await scheduler._get_filament_requirements(db, _item(archive_id=731, library_file_id=None, plate_id=23))
|
|
|
+
|
|
|
+ assert len(parsed) == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_memo_does_not_outlive_the_pass(self, parsed):
|
|
|
+ """check_queue clears it at the top of every pass."""
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ db = self._db_returning(MagicMock(file_path="plate.3mf"))
|
|
|
+ item = _item(archive_id=731, library_file_id=None, plate_id=22)
|
|
|
+
|
|
|
+ await scheduler._get_filament_requirements(db, item)
|
|
|
+ scheduler._filament_req_memo.clear()
|
|
|
+ await scheduler._get_filament_requirements(db, item)
|
|
|
+
|
|
|
+ assert len(parsed) == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_an_unreadable_file_is_remembered_too(self, parsed):
|
|
|
+ """The miss is worth caching as well — otherwise a missing file costs a
|
|
|
+ DB lookup per gate per attempt for nothing."""
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ db = self._db_returning(None)
|
|
|
+ item = _item(archive_id=999, library_file_id=None, plate_id=1)
|
|
|
+
|
|
|
+ assert await scheduler._get_filament_requirements(db, item) is None
|
|
|
+ assert await scheduler._get_filament_requirements(db, item) is None
|
|
|
+ assert db.execute.await_count == 1
|
|
|
+
|
|
|
+
|
|
|
+class TestTheHoldReasonSurvivesTheNextPass:
|
|
|
+ """The reason is the hold's whole explanation, so a later pass must keep it.
|
|
|
+
|
|
|
+ Since #3074 the scheduler clears the reason of every staged item on each
|
|
|
+ pass, because a reason left from before staging would otherwise stand
|
|
|
+ forever. The unmatched-filament hold writes its reason at the moment it
|
|
|
+ stages the item, so clearing it would leave a job that waits for a start
|
|
|
+ press with nothing on the row saying which spool to load.
|
|
|
+ """
|
|
|
+
|
|
|
+ @pytest.fixture
|
|
|
+ async def session_maker(self):
|
|
|
+ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
+
|
|
|
+ import backend.app.models # noqa: F401 - populate Base.metadata
|
|
|
+ from backend.app.core.database import Base
|
|
|
+
|
|
|
+ engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
|
|
+ async with engine.begin() as conn:
|
|
|
+ await conn.run_sync(Base.metadata.create_all)
|
|
|
+ try:
|
|
|
+ yield async_sessionmaker(engine, expire_on_commit=False)
|
|
|
+ finally:
|
|
|
+ await engine.dispose()
|
|
|
+
|
|
|
+ async def _staged_item(self, session_maker, reason):
|
|
|
+ from backend.app.models.print_queue import PrintQueueItem
|
|
|
+ from backend.app.models.printer import Printer
|
|
|
+
|
|
|
+ async with session_maker() as db:
|
|
|
+ printer = Printer(
|
|
|
+ name="P2S-5", serial_number="SERIAL", ip_address="10.0.0.1", access_code="code", model="P2S"
|
|
|
+ )
|
|
|
+ db.add(printer)
|
|
|
+ await db.flush()
|
|
|
+ item = PrintQueueItem(
|
|
|
+ printer_id=printer.id, status="pending", position=0, manual_start=True, waiting_reason=reason
|
|
|
+ )
|
|
|
+ db.add(item)
|
|
|
+ await db.commit()
|
|
|
+ return item.id
|
|
|
+
|
|
|
+ async def _reason_after_a_pass(self, session_maker, item_id):
|
|
|
+ from backend.app.models.print_queue import PrintQueueItem
|
|
|
+
|
|
|
+ scheduler = PrintScheduler()
|
|
|
+ with (
|
|
|
+ patch("backend.app.services.print_scheduler.async_session", session_maker),
|
|
|
+ patch("backend.app.services.print_scheduler.printer_manager.is_connected", MagicMock(return_value=True)),
|
|
|
+ patch("backend.app.services.print_scheduler.printer_manager.get_status", MagicMock(return_value=None)),
|
|
|
+ patch.object(scheduler, "_is_printer_idle", MagicMock(return_value=True)),
|
|
|
+ patch.object(scheduler, "_check_auto_drying", AsyncMock()),
|
|
|
+ ):
|
|
|
+ await scheduler.check_queue()
|
|
|
+ async with session_maker() as db:
|
|
|
+ return (await db.get(PrintQueueItem, item_id)).waiting_reason
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_the_hold_reason_stays_on_a_staged_item(self, session_maker):
|
|
|
+ item_id = await self._staged_item(session_maker, "Needs PETG #2850E0")
|
|
|
+
|
|
|
+ assert await self._reason_after_a_pass(session_maker, item_id) == "Needs PETG #2850E0"
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_any_other_reason_on_a_staged_item_is_still_cleared(self, session_maker):
|
|
|
+ """#3074's rule for everything else is unchanged."""
|
|
|
+ item_id = await self._staged_item(session_maker, "Busy: P2S-5")
|
|
|
+
|
|
|
+ assert await self._reason_after_a_pass(session_maker, item_id) is None
|
|
|
+
|
|
|
+
|
|
|
+def _matching_scheduler(trays):
|
|
|
+ """A scheduler whose real matcher runs against ``trays``."""
|
|
|
+ scheduler = _scheduler(trays)
|
|
|
+ scheduler._get_bool_setting = AsyncMock(return_value=False)
|
|
|
+ scheduler._get_job_name = AsyncMock(return_value="body1")
|
|
|
+ scheduler._get_printer = AsyncMock(return_value=MagicMock(model="P2S"))
|
|
|
+ return scheduler
|
|
|
+
|
|
|
+
|
|
|
+class TestUnresolvedSlotsAreMatchedAgain:
|
|
|
+ """A fitting mapping keeps its resolved slots and has only its gaps matched again.
|
|
|
+
|
|
|
+ The hold keeps a pinned item's mapping, and an unresolved slot is not a
|
|
|
+ conflict, so without this nothing would look at the gap again: loading the
|
|
|
+ missing spool and pressing Start would hold the job again, every time.
|
|
|
+ """
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_spool_loaded_after_the_hold_is_found(self):
|
|
|
+ # Held with slot 3 unresolved; dark-blue PETG has since gone into tray 2.
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(db, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, 2]
|
|
|
+ assert blocked is False
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_tray_the_user_picked_is_kept_and_not_handed_out_twice(self):
|
|
|
+ """The user put slot 1 on tray 2 by hand. Slot 3's best match is that same
|
|
|
+ tray, but it is taken, so slot 3 gets the other PETG instead of sharing."""
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=json.dumps([2, -1, -1]))
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == [2, -1, 0]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_still_missing_leaves_the_mapping_and_holds(self):
|
|
|
+ trays = [
|
|
|
+ {"global_tray_id": 0, "type": "PETG", "color": "76D9F4FF"},
|
|
|
+ {"global_tray_id": 1, "type": "ASA", "color": "161616FF"},
|
|
|
+ ]
|
|
|
+ scheduler = _matching_scheduler(trays)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+ blocked = await scheduler._block_on_unmatched_filament(db, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, -1]
|
|
|
+ assert blocked is True
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_print_anyway_leaves_the_gap_alone(self):
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]), skip_filament_check=True)
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, -1]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_complete_mapping_is_not_matched_again(self):
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[0, -1, 2])
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, 2]))
|
|
|
+ db = AsyncMock()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ await scheduler._ensure_ams_mapping(db, 8, item)
|
|
|
+
|
|
|
+ scheduler._compute_ams_mapping_for_printer.assert_not_awaited()
|
|
|
+
|
|
|
+
|
|
|
+class TestMissingFilamentForStart:
|
|
|
+ """The Start button's question: would this item be held again right now?"""
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_names_a_filament_that_is_still_not_loaded(self):
|
|
|
+ trays = [
|
|
|
+ {"global_tray_id": 0, "type": "PETG", "color": "76D9F4FF"},
|
|
|
+ {"global_tray_id": 1, "type": "ASA", "color": "161616FF"},
|
|
|
+ ]
|
|
|
+ scheduler = _matching_scheduler(trays)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ missing = await scheduler.missing_filament_for_start(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert missing is not None and len(missing) == 1
|
|
|
+ assert "PETG" in missing[0] and "#2850E0" in missing[0]
|
|
|
+ # Asking is not acting: the stored mapping is the scheduler's to change.
|
|
|
+ assert json.loads(item.ams_mapping) == [0, -1, -1]
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_nothing_missing_once_the_spool_is_loaded(self):
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=json.dumps([0, -1, -1]))
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ assert await scheduler.missing_filament_for_start(AsyncMock(), item) is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_a_mapping_from_another_printer_is_judged_by_a_fresh_match(self):
|
|
|
+ """The #2799 mapping on P2S-5 conflicts, so Start judges a fresh match,
|
|
|
+ which finds both filaments."""
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item()
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ assert await scheduler.missing_filament_for_start(AsyncMock(), item) is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_no_mapping_and_nothing_matching_is_missing(self):
|
|
|
+ trays = [{"global_tray_id": 0, "type": "ASA", "color": "161616FF"}]
|
|
|
+ scheduler = _matching_scheduler(trays)
|
|
|
+ item = _item(ams_mapping=None)
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ missing = await scheduler.missing_filament_for_start(AsyncMock(), item)
|
|
|
+
|
|
|
+ assert missing is not None and len(missing) == 2
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "overrides",
|
|
|
+ [
|
|
|
+ {"skip_filament_check": True},
|
|
|
+ # A model-based item picks its printer at dispatch.
|
|
|
+ {"printer_id": None},
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_does_not_judge_what_dispatch_would_not(self, overrides):
|
|
|
+ trays = [{"global_tray_id": 0, "type": "ASA", "color": "161616FF"}]
|
|
|
+ scheduler = _matching_scheduler(trays)
|
|
|
+ item = _item(ams_mapping=None, **overrides)
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = MagicMock()
|
|
|
+ assert await scheduler.missing_filament_for_start(AsyncMock(), item) is None
|
|
|
+
|
|
|
+ @pytest.mark.asyncio
|
|
|
+ async def test_no_status_is_not_a_finding(self):
|
|
|
+ scheduler = _matching_scheduler(P2S_5_TRAYS)
|
|
|
+ item = _item(ams_mapping=None)
|
|
|
+
|
|
|
+ with patch("backend.app.services.print_scheduler.printer_manager") as pm:
|
|
|
+ pm.get_status.return_value = None
|
|
|
+ assert await scheduler.missing_filament_for_start(AsyncMock(), item) is None
|
Martin Grolmus