bambuddy/backend/tests/unit/test_scheduler_external_spool_nozzle_2771.py

"""Regression tests for external-spool nozzle routing on AMS-less printers (#2771).

A fleet of X2Ds with no AMS, printing from external spools, could not be sent a
job with "Any X2D": the print uploaded, then the firmware rejected it with
0700_8012 "Failed to get AMS mapping table" and the item failed after three
dispatch attempts. Sending the same file to a named printer worked, because that
path carries a mapping the *frontend* resolved and the scheduler's matcher never
runs.

Cause: ``_build_loaded_filaments`` derived dual-nozzle status from
``ams_extruder_map``, which is built from AMS info bits — a dual-nozzle printer
with zero AMS units reports an empty map. Every external spool then got
``extruder_id=None``, and the nozzle-aware hard filter in
``_match_filaments_to_slots`` rejected it because ``None`` equals neither 0 nor
1. Nothing matched, the mapping came back all -1 and was cleared to None, and the
print command went out as ``use_ams: true`` with no mapping table at all.

This is the backend half of #1257, which fixed the identical logic in
``useFilamentMapping.ts`` and left this copy behind; the first two tests below
mirror its frontend regression tests.

The second half covers the guard that keeps a genuinely unmappable job from
being uploaded at all, since without an AMS there is no "load another spool and
press Resume" recovery for the firmware error to lead to.
"""

import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from backend.app.services.print_scheduler import (
    PrintScheduler,
    _unmatched_filament_message,
)

# Two external feeds, as an X2D/H2D reports them: 254 is Ext-L (deputy/left,
# extruder 1) and 255 is Ext-R (main/right, extruder 0).
DUAL_EXTERNAL = [
    {"id": "254", "tray_type": "PETG", "tray_color": "000000FF", "tray_info_idx": "GFG00"},
    {"id": "255", "tray_type": "PLA", "tray_color": "FFFFFFFF", "tray_info_idx": "GFA00"},
]

REAL_NOZZLES = [
    SimpleNamespace(nozzle_diameter="0.4"),
    SimpleNamespace(nozzle_diameter="0.4"),
]
# The state seeds `nozzles` with two empty NozzleInfo stubs even on single-nozzle
# printers, so the second entry's presence proves nothing — only a diameter does.
STUB_NOZZLES = [
    SimpleNamespace(nozzle_diameter="0.4"),
    SimpleNamespace(nozzle_diameter=""),
]


def _status(raw_data, nozzles=None):
    return SimpleNamespace(raw_data=raw_data, nozzles=nozzles)


@pytest.fixture
def scheduler():
    return PrintScheduler()


class TestExternalSpoolExtruderRouting:
    """``_build_loaded_filaments`` must route external spools without an AMS."""

    def test_dual_nozzle_without_ams_routes_both_external_feeds(self, scheduler):
        """The X2D case from the report: no AMS, so ams_extruder_map is empty."""
        loaded = scheduler._build_loaded_filaments(
            _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
        )

        by_tray = {f["global_tray_id"]: f for f in loaded}
        assert by_tray[254]["extruder_id"] == 1  # Ext-L -> left
        assert by_tray[255]["extruder_id"] == 0  # Ext-R -> right

    def test_single_nozzle_stub_does_not_fabricate_an_extruder(self, scheduler):
        """A P1S/A1/X1C must keep extruder_id=None, matching pre-fix behaviour.

        Sibling regression to the fix: `nozzles` always has two entries, so
        inferring dual-nozzle from its length would hand every single-nozzle
        printer's external spool a nozzle it does not have.
        """
        loaded = scheduler._build_loaded_filaments(
            _status(
                {"ams": [], "ams_extruder_map": {}, "vt_tray": [DUAL_EXTERNAL[0]]},
                STUB_NOZZLES,
            )
        )

        assert len(loaded) == 1
        assert loaded[0]["extruder_id"] is None

    def test_two_external_feeds_alone_imply_dual_nozzle(self, scheduler):
        """Fallback signal: only dual-nozzle hardware exposes two external feeds.

        Kept for firmware revisions that report the feeds but not the nozzle
        diameters — here `nozzles` is absent entirely.
        """
        loaded = scheduler._build_loaded_filaments(
            _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL})
        )

        assert {f["extruder_id"] for f in loaded} == {0, 1}

    def test_populated_ams_extruder_map_still_implies_dual_nozzle(self, scheduler):
        """The original signal keeps working when there IS an AMS."""
        loaded = scheduler._build_loaded_filaments(
            _status(
                {"ams": [], "ams_extruder_map": {"0": 1}, "vt_tray": [DUAL_EXTERNAL[0]]},
                STUB_NOZZLES,
            )
        )

        assert loaded[0]["extruder_id"] == 1

    def test_mapping_resolves_for_the_nozzle_the_spool_feeds(self, scheduler):
        """End to end: the matcher now finds the external spool, as it did for
        the working named-printer dispatch (which sent ams_mapping [254])."""
        loaded = scheduler._build_loaded_filaments(
            _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
        )
        req = {"slot_id": 1, "type": "PETG", "color": "#000000", "tray_info_idx": "GFG00"}

        assert scheduler._match_filaments_to_slots([{**req, "nozzle_id": 1}], loaded) == [254]
        # Nothing PETG on the right nozzle — still correctly unmatched.
        assert scheduler._match_filaments_to_slots([{**req, "nozzle_id": 0}], loaded) == [-1]


class TestUnmatchedFilamentMessage:
    """The message has to name the filament and, on dual-nozzle, the nozzle."""

    def test_names_type_colour_and_nozzle(self):
        message = _unmatched_filament_message(
            [{"slot_id": 1, "type": "PETG", "color": "#000000", "nozzle_id": 0}],
            [{"type": "PETG", "color": "#000000", "extruder_id": 1}],
        )

        assert "PETG #000000 (right nozzle)" in message
        assert "PETG #000000 (left nozzle)" in message

    def test_omits_nozzle_on_single_nozzle_printers(self):
        message = _unmatched_filament_message(
            [{"slot_id": 1, "type": "ABS", "color": "#FF0000"}],
            [{"type": "PLA", "color": "#000000"}],
        )

        assert "ABS #FF0000" in message
        assert "nozzle" not in message


class TestUnmappableWithoutAmsGuard:
    """``_ensure_ams_mapping`` reports only a positive, unrecoverable finding."""

    def _item(self, ams_mapping=None):
        item = MagicMock()
        item.id = 22
        item.printer_id = 4
        item.ams_mapping = ams_mapping
        item.filament_overrides = None
        return item

    async def _ensure(self, scheduler, computed, status):
        scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=computed)
        scheduler._get_filament_requirements = AsyncMock(
            return_value=[{"slot_id": 1, "type": "PETG", "color": "#000000", "nozzle_id": 0}]
        )
        with patch("backend.app.services.print_scheduler.printer_manager") as pm:
            pm.get_status.return_value = status
            return await scheduler._ensure_ams_mapping(AsyncMock(), 4, self._item())

    @pytest.mark.asyncio
    async def test_reports_when_nothing_matches_and_there_is_no_ams(self, scheduler):
        status = _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)

        message = await self._ensure(scheduler, [-1], status)

        assert message is not None
        assert "no AMS" in message

    @pytest.mark.asyncio
    async def test_silent_when_an_ams_is_attached(self, scheduler):
        """With an AMS the user can load a spool and press Resume, so the
        firmware's own error is worth reaching — behaviour is unchanged."""
        status = _status(
            {
                "ams": [{"id": 0, "tray": [{"id": 0, "tray_type": "PLA", "tray_color": "FF0000"}]}],
                "ams_extruder_map": {},
                "vt_tray": DUAL_EXTERNAL,
            },
            REAL_NOZZLES,
        )

        assert await self._ensure(scheduler, [-1], status) is None

    @pytest.mark.asyncio
    async def test_silent_when_the_ams_field_has_not_arrived_yet(self, scheduler):
        """Absence of an AMS report is not a report of no AMS.

        `raw_data["ams"]` appears only once an AMS push has been handled, so a
        missing key means a reconnect or a cold start — where a fully loaded
        AMS is briefly invisible and everything would look unmappable.
        """
        status = _status({"ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)

        assert await self._ensure(scheduler, [-1], status) is None

    @pytest.mark.asyncio
    async def test_silent_when_the_matcher_never_ran(self, scheduler):
        """A None mapping means no requirements parsed or nothing loaded — not
        evidence of a mismatch. Fail-safe: dispatch as before."""
        status = _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)

        assert await self._ensure(scheduler, None, status) is None

    @pytest.mark.asyncio
    async def test_silent_when_the_mapping_resolves(self, scheduler):
        status = _status({"ams": [], "ams_extruder_map": {}, "vt_tray": DUAL_EXTERNAL}, REAL_NOZZLES)
        item = self._item()
        scheduler._compute_ams_mapping_for_printer = AsyncMock(return_value=[254])

        with patch("backend.app.services.print_scheduler.printer_manager") as pm:
            pm.get_status.return_value = status
            assert await scheduler._ensure_ams_mapping(AsyncMock(), 4, item) is None

        assert json.loads(item.ams_mapping) == [254]

    @pytest.mark.asyncio
    async def test_silent_when_the_printer_status_is_gone(self, scheduler):
        assert await self._ensure(scheduler, [-1], None) is None


class TestFailUnmappableItem:
    """The guard fails the item instead of spending an upload on it."""

    @pytest.mark.asyncio
    async def test_marks_failed_with_the_message(self, scheduler):
        db = AsyncMock()
        item = MagicMock()
        item.id = 22
        item.created_by_id = 1

        with (
            patch("backend.app.services.print_scheduler.notification_service") as notify,
            patch("backend.app.services.print_scheduler.ws_manager"),
        ):
            notify.on_queue_job_failed = AsyncMock()
            scheduler._get_job_name = AsyncMock(return_value="Fidget")
            scheduler._get_printer = AsyncMock(return_value=SimpleNamespace(name="X2D-1"))

            await scheduler._fail_unmappable_item(db, item, 4, "needs PETG")

        assert item.status == "failed"
        assert item.error_message == "needs PETG"
        assert item.completed_at is not None
        db.commit.assert_awaited()
        notify.on_queue_job_failed.assert_awaited_once()