bambuddy/backend/tests/unit/test_rfid_refresh_ack_3206.py

"""RFID refresh reports what the printer said, and falls back to M620 R (#3206).

An X1C on X1Plus (base 01.08.02.00) answers ams_get_rfid with
``result: FAIL, reason: ERROR STATE``, yet the refresh returned success and
scheduled the K-profile re-apply. The refresh now waits for the answer, sends
the legacy ``M620 R<global tray>`` gcode Bambu Studio uses for printers without
the new protocol when ams_get_rfid is refused, and reports a refusal of both.
"""

import json
from unittest.mock import MagicMock, patch

import pytest

from backend.app.services.bambu_mqtt import BambuMQTTClient
from backend.app.utils.printer_models import uses_legacy_rfid_refresh


def _client(answers: dict[str, dict | None], model: str = "X1C") -> tuple[BambuMQTTClient, list[dict]]:
    """A connected, idle client whose printer answers each command per ``answers``.

    ``answers`` maps a command name to the result fields the printer sends
    back, or None for no answer at all. Returns the client and the list of
    commands it published.
    """
    client = BambuMQTTClient(ip_address="10.0.0.1", serial_number="X1C", access_code="c", model=model)
    client._client = MagicMock()
    client.state.connected = True
    client.state.tray_now = 255
    sent: list[dict] = []

    def publish(_topic, body, **_kw):
        command = json.loads(body)["print"]
        sent.append(command)
        answer = answers.get(command["command"])
        if answer is not None:
            client._process_message({"print": {**command, **answer}})

    client._client.publish.side_effect = publish
    return client, sent


ACCEPT = {"result": "SUCCESS", "reason": "SUCCESS"}
REFUSE = {"result": "FAIL", "reason": "ERROR STATE"}


class TestRfidRefreshAck:
    @pytest.mark.asyncio
    async def test_accepted_sends_only_ams_get_rfid(self):
        client, sent = _client({"ams_get_rfid": ACCEPT})

        ok, message = await client.ams_refresh_tray(0, 3)

        assert ok
        assert message == "Refreshing AMS 0 tray 3"
        assert [c["command"] for c in sent] == ["ams_get_rfid"]
        assert sent[0]["ams_id"] == 0 and sent[0]["slot_id"] == 3

    @pytest.mark.asyncio
    async def test_refused_falls_back_to_m620_with_the_global_tray(self):
        client, sent = _client({"ams_get_rfid": REFUSE, "gcode_line": ACCEPT})

        ok, message = await client.ams_refresh_tray(2, 1)

        assert ok
        assert "legacy" in message
        assert [c["command"] for c in sent] == ["ams_get_rfid", "gcode_line"]
        assert sent[1]["param"] == "M620 R9\n"

    @pytest.mark.asyncio
    async def test_both_refused_reports_the_printer_reason(self):
        client, _ = _client({"ams_get_rfid": REFUSE, "gcode_line": REFUSE})

        ok, message = await client.ams_refresh_tray(0, 3)

        assert not ok
        assert "ERROR STATE" in message

    @pytest.mark.asyncio
    async def test_no_fallback_for_units_without_a_legacy_index(self):
        # AMS-HT (128+) has no M620 R index; only firmware with ams_get_rfid has one.
        client, sent = _client({"ams_get_rfid": REFUSE})

        ok, message = await client.ams_refresh_tray(128, 0)

        assert not ok
        assert "ERROR STATE" in message
        assert [c["command"] for c in sent] == ["ams_get_rfid"]

    @pytest.mark.asyncio
    async def test_no_fallback_on_a_newer_protocol_model(self):
        # Bambu Studio never sends M620 R to these; a refusal there is reported.
        client, sent = _client({"ams_get_rfid": REFUSE}, model="H2D")

        ok, message = await client.ams_refresh_tray(0, 1)

        assert not ok
        assert "ERROR STATE" in message
        assert [c["command"] for c in sent] == ["ams_get_rfid"]

    @pytest.mark.asyncio
    @pytest.mark.parametrize("state", ["RUNNING", "PAUSE", "PREPARE", "SLICING"])
    async def test_no_fallback_during_a_job(self, state):
        # A gcode_line would be executed inside the running print.
        client, sent = _client({"ams_get_rfid": REFUSE})
        client.state.state = state

        ok, _ = await client.ams_refresh_tray(0, 1)

        assert not ok
        assert [c["command"] for c in sent] == ["ams_get_rfid"]

    @pytest.mark.asyncio
    async def test_no_answer_counts_as_accepted(self):
        client, sent = _client({})

        client._rfid_ack_timeout = 0.1
        ok, _ = await client.ams_refresh_tray(0, 0)

        assert ok
        assert [c["command"] for c in sent] == ["ams_get_rfid"]
        assert client._pending_rfid_acks == {}

    @pytest.mark.asyncio
    async def test_unrelated_gcode_line_ack_is_not_taken_for_ours(self):
        client, _ = _client({"ams_get_rfid": REFUSE})
        client._process_message({"print": {"command": "gcode_line", "sequence_id": "0", **REFUSE}})

        client._rfid_ack_timeout = 0.1
        ok, _ = await client.ams_refresh_tray(0, 0)

        # ams_get_rfid refused, M620 R sent but unanswered -> accepted.
        assert ok


class TestRefreshRoute:
    @pytest.mark.asyncio
    async def test_refused_refresh_is_400_and_schedules_no_pa_reapply(self):
        from fastapi import HTTPException

        from backend.app.api.routes import printers as printers_routes

        client, _ = _client({"ams_get_rfid": REFUSE, "gcode_line": REFUSE})
        db = MagicMock()
        result = MagicMock()
        result.scalar_one_or_none.return_value = MagicMock(id=1)

        async def execute(*_a, **_kw):
            return result

        db.execute = execute

        with (
            patch.object(printers_routes, "printer_manager") as pm,
            patch.object(printers_routes, "spawn_background_task") as spawn,
        ):
            pm.get_client.return_value = client
            with pytest.raises(HTTPException) as exc:
                await printers_routes.refresh_ams_slot(1, 0, 3, None, db)

        assert exc.value.status_code == 400
        assert "ERROR STATE" in exc.value.detail
        spawn.assert_not_called()


class TestLegacyRfidModels:
    @pytest.mark.parametrize("model", ["X1C", "X1", "X1E", "P1P", "P1S", "A1", "A1 Mini", "BL-P001", "C12", "N1"])
    def test_legacy(self, model):
        assert uses_legacy_rfid_refresh(model)

    @pytest.mark.parametrize("model", ["H2D", "H2C", "H2S", "P2S", "A2L", "X2D", "O1D", "N7", "", None])
    def test_not_legacy(self, model):
        assert not uses_legacy_rfid_refresh(model)