bambuddy/backend/tests/integration/test_bulk_reads_api.py

"""Bulk reads (/api/v1/bulk/...) answer exactly what their single-id routes do.

The Printers page asked for each printer's status, slot presets, AMS labels,
plugs, sensor readings, firmware and queue once per printer, and the Archives
page for each card's folders once per archive; the frontend now batches those
calls into these routes. Each test compares a bulk answer with the single
route's answer for the same ids, and pins what a bulk answer leaves out: ids
the caller may not see, which the client then asks the single route for.
"""

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from httpx import AsyncClient

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

BULK = "/api/v1/bulk"


def _auth(jwt: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {jwt}"}


async def _admin_token(async_client: AsyncClient) -> str:
    await async_client.post(
        "/api/v1/auth/setup",
        json={"auth_enabled": True, "admin_username": "bulkadmin", "admin_password": "AdminPass1!"},
    )
    login = await async_client.post("/api/v1/auth/login", json={"username": "bulkadmin", "password": "AdminPass1!"})
    assert login.status_code == 200, login.text
    return login.json()["access_token"]


async def _user(async_client, admin_jwt, username, permissions, printer_ids=None) -> tuple[str, int]:
    groups = []
    perms = await async_client.post(
        "/api/v1/groups/", headers=_auth(admin_jwt), json={"name": f"perms_{username}", "permissions": permissions}
    )
    assert perms.status_code == 201, perms.text
    groups.append(perms.json()["id"])
    if printer_ids is not None:
        team = await async_client.post(
            "/api/v1/groups/",
            headers=_auth(admin_jwt),
            json={"name": f"team_{username}", "permissions": [], "restrict_printers": True, "printer_ids": printer_ids},
        )
        assert team.status_code == 201, team.text
        groups.append(team.json()["id"])
    created = await async_client.post(
        "/api/v1/users/",
        headers=_auth(admin_jwt),
        json={"username": username, "password": "UserPass1!", "group_ids": groups},
    )
    assert created.status_code in (200, 201), created.text
    login = await async_client.post("/api/v1/auth/login", json={"username": username, "password": "UserPass1!"})
    assert login.status_code == 200, login.text
    return login.json()["access_token"], created.json()["id"]


async def _get(async_client: AsyncClient, url: str, **kwargs):
    response = await async_client.get(url, **kwargs)
    assert response.status_code == 200, f"{url}: {response.text}"
    return response.json()


def _live_state(**overrides) -> MagicMock:
    """A connected printer's state, enough for the status route to build its answer."""
    state = MagicMock(
        connected=True,
        state="IDLE",
        current_print=None,
        subtask_name=None,
        gcode_file=None,
        progress=0,
        remaining_time=None,
        layer_num=0,
        total_layers=0,
        temperatures={"nozzle": 25.0, "bed": 25.0},
        raw_data={"ams": [{"id": 0, "sn": "AMS-SN-1", "tray": []}]},
        hms_errors=[],
        firmware_version="01.08.00.00",
        developer_mode=True,
        ams_filament_backup=None,
        fila_switch=None,
        subtask_id=None,
        kprofiles=[],
        nozzles=[],
        ams_switch_inlet={},
    )
    for key, value in overrides.items():
        setattr(state, key, value)
    return state


class TestPrinterStatuses:
    async def test_matches_the_single_route(self, async_client, printer_factory):
        a, b = await printer_factory(name="A"), await printer_factory(name="B")

        bulk = await _get(async_client, f"{BULK}/printer-statuses", params={"ids": f"{b.id},{a.id}"})

        # In the order asked for, each the single route's answer
        assert [row["id"] for row in bulk] == [b.id, a.id]
        for row in bulk:
            assert row == await _get(async_client, f"/api/v1/printers/{row['id']}/status")

    async def test_matches_the_single_route_for_a_printing_printer(
        self, async_client, printer_factory, archive_factory
    ):
        # The full status path: live state, the cover, and the active print's archive
        from backend.app.services.bambu_mqtt import PrinterState

        printing, idle = await printer_factory(name="Busy"), await printer_factory(name="Idle")
        archive = await archive_factory(printing.id, subtask_id="sub-42")
        states = {}
        for printer, run_state in ((printing, "RUNNING"), (idle, "IDLE")):
            state = PrinterState()
            state.connected = True
            state.state = run_state
            states[printer.id] = state
        states[printing.id].gcode_file = "/data/Metadata/plate_1.gcode"
        states[printing.id].subtask_id = "sub-42"
        states[printing.id].progress = 40

        with patch("backend.app.api.routes.printers.printer_manager") as manager:
            manager.get_status = MagicMock(side_effect=lambda pid: states.get(pid))
            manager.is_awaiting_plate_clear = MagicMock(return_value=False)
            bulk = await _get(async_client, f"{BULK}/printer-statuses", params={"ids": f"{printing.id},{idle.id}"})
            singles = [await _get(async_client, f"/api/v1/printers/{pid}/status") for pid in (printing.id, idle.id)]

        assert bulk == singles
        assert bulk[0]["current_archive_id"] == archive.id
        assert bulk[0]["cover_url"] == f"/api/v1/printers/{printing.id}/cover"
        assert bulk[1]["connected"] is True

    async def test_unknown_printer_is_left_out(self, async_client, printer_factory):
        printer = await printer_factory()

        bulk = await _get(async_client, f"{BULK}/printer-statuses", params={"ids": f"{printer.id},99999"})

        assert [row["id"] for row in bulk] == [printer.id]

    async def test_printer_outside_the_scope_is_left_out(self, async_client, printer_factory):
        a, b = await printer_factory(name="A"), await printer_factory(name="B")
        admin = await _admin_token(async_client)
        jwt, _ = await _user(async_client, admin, "member", ["printers:read"], printer_ids=[a.id])

        bulk = await _get(
            async_client, f"{BULK}/printer-statuses", params={"ids": f"{a.id},{b.id}"}, headers=_auth(jwt)
        )

        assert [row["id"] for row in bulk] == [a.id]

    async def test_needs_printers_read(self, async_client, printer_factory):
        printer = await printer_factory()
        admin = await _admin_token(async_client)
        jwt, _ = await _user(async_client, admin, "nobody", ["queue:read_all"])

        response = await async_client.get(
            f"{BULK}/printer-statuses", params={"ids": str(printer.id)}, headers=_auth(jwt)
        )

        assert response.status_code == 403

    @pytest.mark.parametrize("ids", ["abc", "1,x", "-1", "0", ",".join(str(i) for i in range(1, 502))])
    async def test_rejects_bad_id_lists(self, async_client, ids):
        response = await async_client.get(f"{BULK}/printer-statuses", params={"ids": ids})

        assert response.status_code == 422


class TestSlotPresetsAndAmsLabels:
    async def test_slot_presets_match_the_single_route(self, async_client, db_session, printer_factory):
        from backend.app.models.slot_preset import SlotPresetMapping

        a, b = await printer_factory(), await printer_factory()
        db_session.add_all(
            [
                SlotPresetMapping(printer_id=a.id, ams_id=0, tray_id=1, preset_id="GFA00", preset_name="PLA"),
                SlotPresetMapping(printer_id=a.id, ams_id=255, tray_id=0, preset_id="GFB00", preset_name="ABS"),
            ]
        )
        await db_session.commit()

        bulk = await _get(async_client, f"{BULK}/slot-presets", params={"ids": f"{a.id},{b.id}"})

        assert set(bulk) == {str(a.id), str(b.id)}
        for printer_id in (a.id, b.id):
            assert bulk[str(printer_id)] == await _get(async_client, f"/api/v1/printers/{printer_id}/slot-presets")
        assert bulk[str(b.id)] == {}

    async def test_ams_labels_match_the_single_route(self, async_client, db_session, printer_factory):
        from backend.app.models.ams_label import AmsLabel

        a, b = await printer_factory(), await printer_factory()
        db_session.add(AmsLabel(ams_serial_number="AMS-SN-1", label="Left rack"))
        await db_session.commit()

        with patch("backend.app.api.routes.printers.printer_manager") as manager:
            manager.get_status = MagicMock(
                side_effect=lambda pid: _live_state() if pid == a.id else None,
            )
            bulk = await _get(async_client, f"{BULK}/ams-labels", params={"ids": f"{a.id},{b.id}"})
            singles = {pid: await _get(async_client, f"/api/v1/printers/{pid}/ams-labels") for pid in (a.id, b.id)}

        assert bulk == {str(pid): answer for pid, answer in singles.items()}
        assert bulk[str(a.id)] == {"0": "Left rack"}


class TestCardPlugs:
    async def test_matches_the_two_single_routes(self, async_client, printer_factory, smart_plug_factory):
        a, b = await printer_factory(), await printer_factory()
        await smart_plug_factory(name="Outlet", printer_id=a.id, ip_address="10.0.0.5")
        await smart_plug_factory(
            name="Lights",
            plug_type="homeassistant",
            ha_entity_id="script.lights",
            printer_id=a.id,
            show_on_printer_card=True,
        )

        bulk = await _get(async_client, f"{BULK}/card-plugs", params={"ids": f"{a.id},{b.id}"})

        for printer_id in (a.id, b.id):
            assert bulk[str(printer_id)]["plug"] == await _get(
                async_client, f"/api/v1/smart-plugs/by-printer/{printer_id}"
            )
            assert bulk[str(printer_id)]["scripts"] == await _get(
                async_client, f"/api/v1/smart-plugs/by-printer/{printer_id}/scripts"
            )
        assert bulk[str(a.id)]["plug"]["name"] == "Outlet"
        assert [p["name"] for p in bulk[str(a.id)]["scripts"]] == ["Lights"]
        assert bulk[str(b.id)] == {"plug": None, "scripts": []}


class TestHaSensorReadings:
    async def test_matches_the_single_route(self, async_client, db_session, printer_factory):
        from backend.app.models.printer_ha_sensor import PrinterHASensor

        a, b = await printer_factory(), await printer_factory()
        db_session.add_all(
            [
                PrinterHASensor(printer_id=a.id, name="Door", entity_id="binary_sensor.door", sort_order=2),
                PrinterHASensor(printer_id=a.id, name="Smoke", entity_id="binary_sensor.smoke", sort_order=1),
                PrinterHASensor(
                    printer_id=a.id, name="Hidden", entity_id="binary_sensor.hidden", show_on_printer_card=False
                ),
            ]
        )
        await db_session.commit()

        bulk = await _get(async_client, f"{BULK}/ha-sensor-readings", params={"ids": f"{a.id},{b.id}"})

        for printer_id in (a.id, b.id):
            assert bulk[str(printer_id)] == await _get(
                async_client, f"/api/v1/ha-sensors/by-printer/{printer_id}/readings"
            )
        assert [r["name"] for r in bulk[str(a.id)]] == ["Smoke", "Door"]
        assert bulk[str(b.id)] == []


class TestFirmwareUpdates:
    async def test_matches_the_single_route(self, async_client, printer_factory):
        a, b = await printer_factory(model="X1C"), await printer_factory(model="P1S")
        service = MagicMock()
        service.check_for_update = AsyncMock(
            return_value={
                "update_available": True,
                "latest_version": "01.09.00.00",
                "download_url": None,
                "release_notes": None,
                "available_versions": [],
            }
        )

        with (
            patch("backend.app.api.routes.bulk.get_firmware_service", return_value=service),
            patch("backend.app.api.routes.firmware.get_firmware_service", return_value=service),
        ):
            bulk = await _get(async_client, f"{BULK}/firmware-updates", params={"ids": f"{a.id},{b.id},99999"})
            singles = [await _get(async_client, f"/api/v1/firmware/updates/{pid}") for pid in (a.id, b.id)]

        assert bulk == singles

    async def test_switched_off_asks_nobody(self, async_client, db_session, printer_factory):
        from backend.app.models.settings import Settings

        a, b = await printer_factory(), await printer_factory()
        db_session.add(Settings(key="check_printer_firmware", value="false"))
        await db_session.commit()
        service = MagicMock()
        service.check_for_update = AsyncMock()

        with patch("backend.app.api.routes.bulk.get_firmware_service", return_value=service):
            bulk = await _get(async_client, f"{BULK}/firmware-updates", params={"ids": f"{a.id},{b.id}"})

        assert [row["update_available"] for row in bulk] == [False, False]
        service.check_for_update.assert_not_called()


class TestPrinterQueues:
    async def test_matches_the_single_route_including_model_jobs(
        self, async_client, db_session, printer_factory, archive_factory
    ):
        from backend.app.models.print_queue import PrintQueueItem

        x1_a = await printer_factory(model="X1C")
        x1_b = await printer_factory(model="X1C")
        p1 = await printer_factory(model="P1S")
        archive = await archive_factory(x1_a.id)
        db_session.add_all(
            [
                PrintQueueItem(archive_id=archive.id, printer_id=x1_a.id, status="pending", position=3),
                PrintQueueItem(
                    archive_id=archive.id, printer_id=None, target_model="x1c", status="pending", position=1
                ),
                PrintQueueItem(archive_id=archive.id, printer_id=p1.id, status="pending", position=2),
                PrintQueueItem(archive_id=archive.id, printer_id=x1_a.id, status="printing", position=4),
                PrintQueueItem(archive_id=archive.id, printer_id=x1_b.id, status="completed", position=5),
            ]
        )
        await db_session.commit()
        ids = [x1_a.id, x1_b.id, p1.id]

        for status in ("pending", "printing"):
            bulk = await _get(
                async_client,
                f"{BULK}/printer-queues",
                params={"ids": ",".join(map(str, ids)), "status": status},
            )
            for printer_id in ids:
                single = await _get(async_client, "/api/v1/queue/", params={"printer_id": printer_id, "status": status})
                assert bulk[str(printer_id)] == single, (status, printer_id)

        pending = await _get(
            async_client, f"{BULK}/printer-queues", params={"ids": ",".join(map(str, ids)), "status": "pending"}
        )
        # The "Any X1C" job sits under both X1Cs, ahead of the pinned one
        assert [i["printer_id"] for i in pending[str(x1_a.id)]] == [None, x1_a.id]
        assert [i["printer_id"] for i in pending[str(x1_b.id)]] == [None]
        assert [i["printer_id"] for i in pending[str(p1.id)]] == [p1.id]

    async def test_read_own_sees_only_own_jobs(self, async_client, db_session, printer_factory, archive_factory):
        from backend.app.models.print_queue import PrintQueueItem

        printer = await printer_factory()
        archive = await archive_factory(printer.id)
        admin = await _admin_token(async_client)
        jwt, user_id = await _user(async_client, admin, "owner", ["queue:read_own"])
        db_session.add_all(
            [
                PrintQueueItem(archive_id=archive.id, printer_id=printer.id, status="pending", created_by_id=user_id),
                PrintQueueItem(archive_id=archive.id, printer_id=printer.id, status="pending"),
            ]
        )
        await db_session.commit()

        bulk = await _get(
            async_client,
            f"{BULK}/printer-queues",
            params={"ids": str(printer.id), "status": "pending"},
            headers=_auth(jwt),
        )

        assert [i["created_by_id"] for i in bulk[str(printer.id)]] == [user_id]


class TestArchiveFolders:
    async def test_matches_the_single_route(self, async_client, db_session, printer_factory, archive_factory):
        from backend.app.models.library import LibraryFolder

        printer = await printer_factory()
        linked, unlinked = await archive_factory(printer.id), await archive_factory(printer.id)
        db_session.add_all(
            [
                LibraryFolder(name="Zeta", archive_id=linked.id),
                LibraryFolder(name="Alpha", archive_id=linked.id),
                LibraryFolder(name="Elsewhere"),
            ]
        )
        await db_session.commit()

        bulk = await _get(async_client, f"{BULK}/archive-folders", params={"ids": f"{linked.id},{unlinked.id}"})

        for archive_id in (linked.id, unlinked.id):
            assert bulk[str(archive_id)] == await _get(async_client, f"/api/v1/library/folders/by-archive/{archive_id}")
        assert [f["name"] for f in bulk[str(linked.id)]] == ["Alpha", "Zeta"]
        assert bulk[str(unlinked.id)] == []