bambuddy/backend/tests/unit/test_ams_mapping_per_printer_2799.py

"""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