bambuddy/backend/tests/unit/services/test_spoolman_unmapped_external_3405.py

"""An X2D print from the right-hand external spool, sent without ams_mapping2 (#3405).

The X2D has two external holders and lists both in ``vt_tray``, loaded or not.
A print command that carries no ``ams_mapping2`` leaves the external spool as -1
in the flat mapping (#3166 resolves it only from ams_mapping2). Until #3166's
follow-up, Spoolman mode read that -1 as the first external holder present,
254, and charged the empty left holder; since then it charged nothing. The
built-in inventory drops a mapping that names no tray for any used slot and
lets the printer's ``tray_now`` decide. Spoolman mode now does the same, and
reads ``tray_now`` 255 as the right-hand holder on a printer that has one, as
the built-in inventory does.

Values are the reporter's ``/status``: 254 empty, 255 loaded with red PLA,
``tray_now`` 255.
"""

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

import pytest

from backend.app.services import spoolman_tracking
from backend.app.services.spoolman_tracking import _single_slot_tray_from_state
from backend.app.services.usage_tracker import has_second_external_holder, mapping_names_no_used_tray

X2D_VT_TRAY = [
    {"id": 254, "tray_type": "", "tray_color": "00000000", "tray_info_idx": ""},
    {"id": 255, "tray_type": "PLA", "tray_color": "F72323FF", "tray_info_idx": "GFA01"},
]
USAGE = [{"slot_id": 1, "used_g": 12.5, "type": "PLA", "color": "#F72323"}]
RIGHT_HOLDER_SPOOL = 77


def _x2d_state(tray_now=255, vt_tray=X2D_VT_TRAY, last_loaded_tray=-1, tray_change_log=(), **raw):
    return SimpleNamespace(
        raw_data={"vt_tray": vt_tray, **raw},
        layer_num=0,
        total_layers=0,
        tray_change_log=list(tray_change_log),
        tray_now=tray_now,
        last_loaded_tray=last_loaded_tray,
    )


class TestTheRules:
    def test_a_mapping_of_only_minus_one_names_no_used_tray(self):
        assert mapping_names_no_used_tray([-1], USAGE)
        assert not mapping_names_no_used_tray([255], USAGE)
        # Another slot is mapped: the -1 is a slot the print does not feed.
        two = [*USAGE, {"slot_id": 2, "used_g": 3.0}]
        assert not mapping_names_no_used_tray([-1, 5], two)
        assert not mapping_names_no_used_tray(None, USAGE)

    def test_255_is_a_tray_only_on_a_printer_with_two_holders(self):
        assert has_second_external_holder({"vt_tray": X2D_VT_TRAY})
        assert not has_second_external_holder({"vt_tray": [X2D_VT_TRAY[0]]})
        assert not has_second_external_holder(None)

    def test_tray_now_255_is_the_right_holder(self):
        assert _single_slot_tray_from_state(_x2d_state(), USAGE, 255) == (1, 255)

    @pytest.mark.parametrize("stale", [3, 254])
    def test_the_right_holder_beats_a_tray_an_earlier_print_loaded(self, stale):
        """A print from 255 never updates last_loaded_tray: it still names the
        AMS tray, or the left holder, an earlier print used."""
        state = _x2d_state(last_loaded_tray=stale)
        assert _single_slot_tray_from_state(state, USAGE, 255) == (1, 255)

    def test_a_tray_loaded_during_the_print_still_wins_over_255(self):
        state = _x2d_state(last_loaded_tray=3, tray_change_log=[(3, 2)])
        assert _single_slot_tray_from_state(state, USAGE, 255) == (1, 3)

    def test_last_loaded_254_is_no_survivor(self):
        """Same cap as the built-in inventory: 0-253 only."""
        state = _x2d_state(vt_tray=[X2D_VT_TRAY[0]], last_loaded_tray=254)
        assert _single_slot_tray_from_state(state, USAGE, 255) is None

    def test_tray_now_255_on_a_one_holder_printer_is_nothing(self):
        assert _single_slot_tray_from_state(_x2d_state(vt_tray=[X2D_VT_TRAY[0]]), USAGE, 255) is None


class _AsyncCtx:
    def __init__(self, db):
        self._db = db

    async def __aenter__(self):
        return self._db

    async def __aexit__(self, *exc):
        return False


def _slot_assignment(_printer_id, ams_id, tray_id):
    # The reporter's spool, assigned to the holder it sits on: 255 is (255, 1).
    return RIGHT_HOLDER_SPOOL if (ams_id, tray_id) == (255, 1) else None


def _client():
    client = AsyncMock()
    client.find_spool_by_tag = AsyncMock(return_value=None)
    client.get_spool = AsyncMock(
        return_value={"id": RIGHT_HOLDER_SPOOL, "filament": {"color_hex": "F72323", "material": "PLA"}}
    )
    client.use_spool = AsyncMock()
    return client


def _patches(client, state, db):
    pm = MagicMock()
    pm.get_status.return_value = state
    return (
        patch.object(spoolman_tracking, "async_session", lambda: _AsyncCtx(db)),
        patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
        patch.object(spoolman_tracking, "_get_spoolman_client_with_fallback", AsyncMock(return_value=client)),
        patch.object(spoolman_tracking, "_get_printer_serial", AsyncMock(return_value="SER")),
        patch.object(
            spoolman_tracking, "_resolve_spool_id_via_slot_assignment", AsyncMock(side_effect=_slot_assignment)
        ),
        patch("backend.app.services.printer_manager.printer_manager", pm),
    )


def _tracking(slot_to_tray):
    return SimpleNamespace(
        filament_usage=list(USAGE),
        ams_trays={
            "254": {"tray_uuid": "", "tag_uid": "", "tray_type": ""},
            "255": {"tray_uuid": "", "tag_uid": "", "tray_type": "PLA"},
        },
        slot_to_tray=slot_to_tray,
        tray_remain_start=None,
        layer_usage=None,
        filament_properties=None,
        tray_now_at_start=255,
        archive_id=12,
    )


class TestReportUsage:
    @staticmethod
    async def _run(slot_to_tray, state=None):
        archive = SimpleNamespace(filament_color="#F72323", filament_type="PLA")
        rows = iter([_tracking(slot_to_tray)])

        def _next_row(*_args, **_kwargs):
            result = MagicMock()
            result.scalar_one_or_none.return_value = next(rows, archive)
            return result

        db = AsyncMock()
        db.execute = AsyncMock(side_effect=_next_row)
        db.delete = AsyncMock()
        db.commit = AsyncMock()
        client = _client()
        p = _patches(client, state or _x2d_state(), db)
        with p[0], p[1], p[2], p[3], p[4], p[5]:
            await spoolman_tracking.report_usage(printer_id=1, archive_id=12)
        return client

    @pytest.mark.asyncio
    async def test_the_loaded_right_holder_is_charged(self):
        client = await self._run([-1])

        client.use_spool.assert_awaited_once_with(RIGHT_HOLDER_SPOOL, 12.5)

    @pytest.mark.asyncio
    async def test_a_same_coloured_ams_tray_is_not_charged_instead(self):
        """The colour match knows no external holder; the built-in inventory
        never runs it for a mapping that named no tray."""
        ams = [{"id": 0, "tray": [{"id": 0, "tray_color": "F72323FF", "tray_type": "PLA"}]}]
        client = await self._run([-1], state=_x2d_state(ams=ams))

        client.use_spool.assert_awaited_once_with(RIGHT_HOLDER_SPOOL, 12.5)

    @pytest.mark.asyncio
    async def test_a_resolved_mapping_is_unchanged(self):
        """With ams_mapping2 the captured mapping already says 255 (#3166)."""
        client = await self._run([255])

        client.use_spool.assert_awaited_once_with(RIGHT_HOLDER_SPOOL, 12.5)


class TestPartialUsage:
    @pytest.mark.asyncio
    async def test_an_aborted_print_charges_the_right_holder_too(self):
        tracking = _tracking([-1])
        tracking.layer_usage = {}
        # A same-coloured AMS tray must not take it either, as on completion.
        ams = [{"id": 0, "tray": [{"id": 0, "tray_color": "F72323FF", "tray_type": "PLA"}]}]
        state = _x2d_state(ams=ams)
        state.layer_num, state.total_layers = 50, 100
        report = AsyncMock(return_value=1)
        p = _patches(_client(), state, AsyncMock())
        with (
            p[0],
            p[1],
            p[2],
            p[3],
            p[4],
            p[5],
            patch.object(spoolman_tracking, "_report_spool_usage_for_slots", report),
        ):
            await spoolman_tracking._report_partial_usage(1, tracking, last_progress=50)

        report.assert_awaited()
        slot_to_tray = report.await_args.args[3]
        assert spoolman_tracking._resolve_global_tray_id(1, slot_to_tray, {254: {}, 255: {}}) == 255