bambuddy/backend/tests/integration/test_slicer_loaded_spools_3172.py

"""GET /slicer/loaded-spools — what the Slice dialog may filter on (#3172).

The dialog offers to narrow its lists to the printers that are online and the
spools loaded in them. The endpoint behind it has to:

* list only printers with a live connection, since an offline printer can't
  say what it has loaded;
* report every AMS slot as the printer does, empty ones included, with the
  profile Bambuddy saved for it, and an external holder only while it holds
  a spool;
* keep to the caller's printers (#1727) and to callers who may read printer
  status.
"""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import patch

import pytest
from httpx import AsyncClient

from backend.tests.integration.test_printer_scope_1727 import _admin_token, _auth, _group, _team_member, _user

pytestmark = [pytest.mark.asyncio, pytest.mark.integration]

URL = "/api/v1/slicer/loaded-spools"


def _status(raw_data: dict, connected: bool = True) -> SimpleNamespace:
    return SimpleNamespace(connected=connected, raw_data=raw_data)


def _ams(ams_id: int, trays: list[dict]) -> dict:
    return {"id": ams_id, "tray": trays}


def _tray(tray_id: int, tray_type: str = "", **kw) -> dict:
    return {"id": tray_id, "tray_type": tray_type, **kw}


def _patched_status(statuses: dict[int, SimpleNamespace | None]):
    return patch(
        "backend.app.api.routes.slicer_presets.printer_manager.get_status",
        side_effect=lambda pid: statuses.get(pid),
    )


class TestLoadedSpools:
    async def test_lists_only_connected_printers(self, async_client: AsyncClient, printer_factory):
        online = await printer_factory(name="A online", model="BL-P001")
        disconnected = await printer_factory(name="B disconnected")
        never_seen = await printer_factory(name="C never seen")
        inactive = await printer_factory(name="D inactive", is_active=False)

        statuses = {
            online.id: _status({"ams": []}),
            disconnected.id: _status({"ams": [_ams(0, [_tray(0, "PLA")])]}, connected=False),
            never_seen.id: None,
            inactive.id: _status({"ams": []}),
        }
        with _patched_status(statuses):
            response = await async_client.get(URL)

        assert response.status_code == 200, response.text
        printers = response.json()["printers"]
        assert [p["id"] for p in printers] == [online.id]
        # The SSDP model code is turned into the short name @BBL tags use.
        assert printers[0]["model"] == "X1C"

    async def test_no_connected_printer_is_an_empty_list(self, async_client: AsyncClient, printer_factory):
        printer = await printer_factory()
        with _patched_status({printer.id: None}):
            response = await async_client.get(URL)
        assert response.status_code == 200
        assert response.json() == {"printers": []}

    async def test_reports_slots_with_their_saved_profile(self, async_client: AsyncClient, printer_factory, db_session):
        from backend.app.models.slot_preset import SlotPresetMapping

        printer = await printer_factory(model="H2D")
        db_session.add_all(
            [
                SlotPresetMapping(
                    printer_id=printer.id,
                    ams_id=0,
                    tray_id=1,
                    preset_id="local_7",
                    preset_name="Overture PLA Matte",
                    preset_source="local",
                    tray_info_idx="GFL99",
                ),
                SlotPresetMapping(
                    printer_id=printer.id,
                    ams_id=255,
                    tray_id=1,
                    preset_id="GFSG99",
                    preset_name="Generic PETG",
                    preset_source="cloud",
                ),
            ]
        )
        await db_session.commit()

        raw = {
            "ams": [
                _ams(
                    0,
                    [
                        _tray(0, "PLA", tray_color="FF0000FF", tray_info_idx="GFA00", tray_sub_brands="PLA Basic"),
                        _tray(1, "PLA", tray_color="00FF00FF", tray_info_idx="GFL99"),
                        _tray(2, exists=True),
                        _tray(3, state=9),
                    ],
                ),
                _ams(128, [_tray(0, "PETG", tray_color="0000FFFF")]),
            ],
            "vt_tray": [_tray(254, ""), _tray(255, "PETG", tray_color="FFFFFFFF")],
        }
        with _patched_status({printer.id: _status(raw)}):
            response = await async_client.get(URL)

        assert response.status_code == 200, response.text
        [entry] = response.json()["printers"]
        regular, ht = entry["ams"]

        assert regular["is_ams_ht"] is False
        assert [t["tray_id"] for t in regular["trays"]] == [0, 1, 2, 3]
        first, second, unknown, empty = regular["trays"]
        assert first["tray_sub_brands"] == "PLA Basic"
        assert first["saved_preset"] is None
        assert second["saved_preset"] == {
            "preset_id": "local_7",
            "preset_name": "Overture PLA Matte",
            "preset_source": "local",
            "tray_info_idx": "GFL99",
        }
        # Empty slots stay in the list, so the grid matches the unit.
        assert unknown["tray_type"] is None and unknown["exists"] is True
        assert empty["tray_type"] is None and empty["state"] == 9

        assert ht["is_ams_ht"] is True and ht["id"] == 128

        # The empty left holder is left out; the right one is AMS 255, tray 1.
        [ext] = entry["external"]
        assert (ext["ams_id"], ext["tray_id"]) == (255, 1)
        assert ext["saved_preset"]["preset_name"] == "Generic PETG"
        assert entry["external_holders"] == 2

    async def test_tolerates_malformed_status(self, async_client: AsyncClient, printer_factory):
        printer = await printer_factory()
        raw = {"ams": ["garbage", {"id": 1, "tray": [None, _tray(0, "ABS")]}], "vt_tray": None}
        with _patched_status({printer.id: _status(raw)}):
            response = await async_client.get(URL)
        assert response.status_code == 200, response.text
        [entry] = response.json()["printers"]
        assert [u["id"] for u in entry["ams"]] == [1]
        assert entry["external"] == []


class TestLoadedSpoolsAccess:
    async def test_limited_user_sees_only_their_printers(self, async_client: AsyncClient, printer_factory):
        mine = await printer_factory(name="Mine")
        other = await printer_factory(name="Other")
        admin = await _admin_token(async_client)
        jwt, _ = await _team_member(async_client, admin, "slicer3172", [mine.id])
        # The team permissions don't include slicing; add it in a second group.
        slicing = await _group(async_client, admin, "slicing3172", permissions=["library:upload"])
        me = await async_client.get("/api/v1/auth/me", headers=_auth(jwt))
        groups = [g["id"] for g in me.json()["groups"]] + [slicing]
        updated = await async_client.patch(
            f"/api/v1/users/{me.json()['id']}", headers=_auth(admin), json={"group_ids": groups}
        )
        assert updated.status_code == 200, updated.text

        statuses = {mine.id: _status({"ams": []}), other.id: _status({"ams": []})}
        with _patched_status(statuses):
            response = await async_client.get(URL, headers=_auth(jwt))

        assert response.status_code == 200, response.text
        assert [p["id"] for p in response.json()["printers"]] == [mine.id]

    async def test_needs_printer_read(self, async_client: AsyncClient, printer_factory):
        """The slots are printer status; slicing alone doesn't grant it."""
        await printer_factory()
        admin = await _admin_token(async_client)
        group = await _group(async_client, admin, "uploadonly3172", permissions=["library:upload"])
        jwt, _ = await _user(async_client, admin, "uploadonly3172", [group])

        response = await async_client.get(URL, headers=_auth(jwt))
        assert response.status_code == 403