bambuddy/backend/tests/unit/test_fts_kprofile_routing.py

"""K-profiles follow an AMS when it moves between Filament Track Switch inlets.

Measured on the maintainer's H2C, 2026-08-16. Spool 75, "HF Bambu PLA Matte
Black", is calibrated on both hotends and Bambuddy stores both:

    extruder 1 (left)   K 0.018   cali_idx 16
    extruder 0 (right)  K 0.020   cali_idx 15

A tray holds exactly one ``cali_idx``, and the printer's calibration table is
numbered per nozzle — so index 16 exists on both hotends and means a different
profile on each. Moving that AMS from In-A to In-B therefore leaves the slot
bound to the *left* profile while it now feeds the right nozzle. The printer
does not re-resolve it, and an RFID re-read only re-asserts the same wrong
index. That is the state the live printer was in when this was written:

    AMS 1  info=10002E03  inlet=IN-B   tray 0: PLA 000000  cali_idx=16
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from backend.app.utils.fts_routing import FTS_INLET_EXTRUDER, extruder_for_inlet, slot_extruder


class TestInletToNozzle:
    def test_in_a_is_the_left_hotend(self):
        assert extruder_for_inlet("A") == 1

    def test_in_b_is_the_right_hotend(self):
        assert extruder_for_inlet("B") == 0

    @pytest.mark.parametrize("value", [None, "", "C", "left"])
    def test_anything_else_is_unknown(self, value):
        assert extruder_for_inlet(value) is None

    def test_the_two_inlets_do_not_share_a_nozzle(self):
        """A table that mapped both inlets to one hotend would silently bind
        every slot to the same K-profile."""
        assert sorted(FTS_INLET_EXTRUDER.values()) == [0, 1]


class TestSlotExtruder:
    def test_a_real_extruder_id_wins(self):
        """A non-0xE binding is authoritative even on a machine with a switch,
        which is how BambuStudio treats it too."""
        assert slot_extruder(2, 0, {"2": 1}, {"2": "B"}) == 1

    def test_falls_back_to_the_switch_inlet(self):
        """The FTS case: every AMS reports 0xE, so the map is empty."""
        assert slot_extruder(1, 0, {}, {"1": "B"}) == 0
        assert slot_extruder(2, 3, {}, {"2": "A"}) == 1

    def test_unknown_is_none_not_zero(self):
        """The bug this replaced: three call sites ended in `else 0`, which on a
        switch machine filed every profile under the right-hand nozzle."""
        assert slot_extruder(1, 0, {}, {}) is None
        assert slot_extruder(1, 0, None, None) is None

    def test_external_slots_name_their_own_side(self):
        assert slot_extruder(255, 0, {}, {}) == 1  # Ext-L
        assert slot_extruder(255, 1, {}, {}) == 0  # Ext-R

    def test_an_unmapped_ams_does_not_borrow_another_units_inlet(self):
        assert slot_extruder(3, 0, {}, {"1": "A", "2": "B"}) is None

    def test_the_maintainers_h2c(self):
        """The live layout: AMS 0/1/128 on In-B, AMS 2 on In-A, empty map."""
        inlets = {"0": "B", "1": "B", "2": "A", "128": "B"}
        assert slot_extruder(0, 0, {}, inlets) == 0
        assert slot_extruder(1, 0, {}, inlets) == 0
        assert slot_extruder(2, 1, {}, inlets) == 1
        assert slot_extruder(128, 0, {}, inlets) == 0


def _profile(cali_idx: int, extruder: int, k: float):
    return MagicMock(
        cali_idx=cali_idx, extruder=extruder, k_value=k, name="HF Bambu PLA Matte Black", filament_id="GFA01"
    )


class TestReSelectOnInletMove:
    """The callback that re-points a moved AMS's slots."""

    @staticmethod
    def _run(inlet, tray, profile, *, connected=True):
        """Drive on_fts_inlet_change for one AMS holding one tray."""
        from backend.app import main as main_module

        client = MagicMock()
        client.extrusion_cali_sel = MagicMock(return_value=True)
        state = MagicMock()
        state.raw_data = {"ams": [{"id": "1", "tray": [tray]}]}
        state.nozzles = [MagicMock(nozzle_diameter="0.4")]

        pm = MagicMock()
        pm.get_client.return_value = client if connected else None
        pm.get_status.return_value = state

        session = AsyncMock()
        session.__aenter__ = AsyncMock(return_value=session)
        session.__aexit__ = AsyncMock()

        with (
            patch("backend.app.main.printer_manager", pm),
            patch("backend.app.main.async_session", return_value=session),
            patch(
                "backend.app.main.find_slot_kprofile_for_extruder",
                new=AsyncMock(return_value=profile),
            ) as lookup,
        ):
            import asyncio

            asyncio.run(main_module.on_fts_inlet_change(7, 1, inlet))
        return client, lookup

    def test_moving_to_in_b_selects_the_right_hotends_profile(self):
        """The reported case, with the maintainer's real numbers."""
        tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
        client, lookup = self._run("B", tray, _profile(15, 0, 0.020))

        assert lookup.await_args.args[4] == 0, "must look up the RIGHT hotend's profile"
        client.extrusion_cali_sel.assert_called_once()
        assert client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 15

    def test_moving_to_in_a_selects_the_left_hotends_profile(self):
        tray = {"id": "0", "tray_type": "PLA", "cali_idx": 15, "tray_info_idx": "GFA01"}
        client, lookup = self._run("A", tray, _profile(16, 1, 0.018))

        assert lookup.await_args.args[4] == 1
        assert client.extrusion_cali_sel.call_args.kwargs["cali_idx"] == 16

    def test_an_already_correct_slot_is_left_alone(self):
        """No point re-sending a binding the printer already holds, and every
        avoided write is one less chance of the firmware mislinking it."""
        tray = {"id": "0", "tray_type": "PLA", "cali_idx": 15, "tray_info_idx": "GFA01"}
        client, _ = self._run("B", tray, _profile(15, 0, 0.020))

        client.extrusion_cali_sel.assert_not_called()

    def test_a_spool_with_no_profile_for_that_nozzle_is_untouched(self):
        """Calibrated on one hotend only: keep whatever the operator set by hand
        rather than swapping it for a guess."""
        tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
        client, _ = self._run("B", tray, None)

        client.extrusion_cali_sel.assert_not_called()

    def test_an_empty_slot_is_skipped(self):
        tray = {"id": "0", "tray_type": "", "cali_idx": -1}
        client, lookup = self._run("B", tray, _profile(15, 0, 0.020))

        lookup.assert_not_awaited()
        client.extrusion_cali_sel.assert_not_called()

    def test_a_disconnected_printer_is_a_no_op(self):
        tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
        client, _ = self._run("B", tray, _profile(15, 0, 0.020), connected=False)

        client.extrusion_cali_sel.assert_not_called()

    def test_an_unknown_inlet_is_a_no_op(self):
        tray = {"id": "0", "tray_type": "PLA", "cali_idx": 16, "tray_info_idx": "GFA01"}
        client, lookup = self._run("C", tray, _profile(15, 0, 0.020))

        lookup.assert_not_awaited()
        client.extrusion_cali_sel.assert_not_called()


class TestTheMoveIsDetected:
    """The MQTT side: only a genuine move fires the callback."""

    @pytest.fixture
    def mqtt_client(self):
        from backend.app.services.bambu_mqtt import BambuMQTTClient

        return BambuMQTTClient(ip_address="192.168.1.100", serial_number="TEST123", access_code="12345678")

    @staticmethod
    def _info(inlet_bits: int) -> str:
        return f"{(inlet_bits << 24) | (0xE << 8) | 1:08X}"

    def _push(self, client, inlet_bits):
        client._process_message(
            {
                "print": {
                    "gcode_state": "IDLE",
                    "device": {"fila_switch": {"in": [-1, -1], "out": [1, 1], "stat": 1, "info": 0}},
                    "ams": {"ams": [{"id": "1", "info": self._info(inlet_bits), "tray": []}]},
                }
            }
        )

    def test_a_move_fires_once(self, mqtt_client):
        seen = []
        mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))

        self._push(mqtt_client, 1)  # In-A
        self._push(mqtt_client, 0)  # In-B

        assert seen == [(1, "B")]

    def test_the_first_sighting_does_not_fire(self, mqtt_client):
        """Every reconnect learns the bindings afresh. Re-applying K-profiles
        there would fight a binding the operator set deliberately."""
        seen = []
        mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))

        self._push(mqtt_client, 1)

        assert seen == []

    def test_repeated_frames_do_not_fire(self, mqtt_client):
        seen = []
        mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))

        for _ in range(4):
            self._push(mqtt_client, 1)

        assert seen == []

    def test_moving_back_fires_again(self, mqtt_client):
        seen = []
        mqtt_client.on_fts_inlet_change = lambda ams_id, inlet: seen.append((ams_id, inlet))

        self._push(mqtt_client, 1)
        self._push(mqtt_client, 0)
        self._push(mqtt_client, 1)

        assert seen == [(1, "B"), (1, "A")]