bambuddy/backend/tests/integration/test_spoolman_slot_assignments.py

"""Integration tests for Spoolman slot-assignment endpoints.

Tests for:
  POST   /api/v1/spoolman/inventory/slot-assignments
  DELETE /api/v1/spoolman/inventory/slot-assignments/{spoolman_spool_id}
  GET    /api/v1/spoolman/inventory/slot-assignments?printer_id=&ams_id=&tray_id=
  GET    /api/v1/spoolman/inventory/slot-assignments/all[?printer_id=]

Slot assignments are now stored in the local ``spoolman_slot_assignments`` table.
Spoolman's ``spool.location`` field is NOT touched by any of these endpoints.
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy import select

SAMPLE_SPOOL = {
    "id": 10,
    "filament": {
        "id": 1,
        "name": "PLA Basic",
        "material": "PLA",
        "color_hex": "FF0000",
        "weight": 1000,
        "vendor": {"id": 1, "name": "Test Brand"},
    },
    "remaining_weight": 800.0,
    "used_weight": 200.0,
    "location": None,
    "comment": None,
    "first_used": None,
    "last_used": None,
    "registered": "2024-01-01T00:00:00+00:00",
    "archived": False,
    "price": None,
    "extra": {},
}


@pytest.fixture
async def slot_settings(db_session):
    from backend.app.models.settings import Settings

    db_session.add(Settings(key="spoolman_enabled", value="true"))
    db_session.add(Settings(key="spoolman_url", value="http://localhost:7912"))
    await db_session.commit()


@pytest.fixture
async def test_printer(db_session):
    from backend.app.models.printer import Printer

    printer = Printer(
        name="Test Printer",
        serial_number="SLOTTEST001",
        ip_address="192.168.1.100",
        access_code="12345678",
    )
    db_session.add(printer)
    await db_session.commit()
    await db_session.refresh(printer)
    return printer


@pytest.fixture
def mock_client():
    client = MagicMock()
    client.base_url = "http://localhost:7912"
    client.health_check = AsyncMock(return_value=True)
    client.get_spool = AsyncMock(return_value=SAMPLE_SPOOL)
    # #1457: assign route enumerates spools to clear stale fallback-tag links.
    # Default to empty so the cleanup is a no-op for tests that don't exercise it.
    client.get_spools = AsyncMock(return_value=[])
    client.merge_spool_extra = AsyncMock(return_value={"id": 0, "extra": {}})

    with patch(
        "backend.app.api.routes.spoolman_inventory._get_client",
        AsyncMock(return_value=client),
    ):
        yield client


class TestAssignSpoolmanSlot:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_inserts_local_row(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
        """POST /slot-assignments creates a row visible via the /all endpoint."""
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 10,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )

        assert response.status_code == 200
        all_resp = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments/all",
            params={"printer_id": test_printer.id},
        )
        assert all_resp.status_code == 200
        rows = all_resp.json()
        assert len(rows) == 1
        assert rows[0]["spoolman_spool_id"] == 10
        assert rows[0]["ams_id"] == 0
        assert rows[0]["tray_id"] == 0

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_drops_a_held_unlink_on_the_slot(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """#3186: a slot that read empty has its unlink held for a grace period.
        Linking a spool to it is fresher evidence, so the old clock must not
        carry over and delete the user's own link."""
        from backend.app.services import slot_unlink_grace

        hold_key = ("spoolman", 0, 0, 10)
        slot_unlink_grace.removal_confirmed(test_printer.id, hold_key)
        assert slot_unlink_grace.is_held(test_printer.id, hold_key)

        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        assert response.status_code == 200
        assert not slot_unlink_grace.is_held(test_printer.id, hold_key)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_accepts_ams_ht_id(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
        """#1274: AMS-HT units report ams_id 128+. The pre-fix ck_ams_id_range
        only allowed 0-7 / 255, so the upsert blew up with `CHECK constraint
        failed: ck_ams_id_range` and the user couldn't link any spool to the
        H2C/H2D AMS-HT slot. This guards the widened range from regressing.
        """
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 51,
                "printer_id": test_printer.id,
                "ams_id": 128,  # AMS-HT on the left nozzle (matches issue's failing INSERT)
                "tray_id": 0,
            },
        )

        assert response.status_code == 200, response.text
        all_resp = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments/all",
            params={"printer_id": test_printer.id},
        )
        rows = all_resp.json()
        assert any(r["ams_id"] == 128 and r["spoolman_spool_id"] == 51 for r in rows)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_does_not_call_update_spool(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """POST /slot-assignments must NOT write to Spoolman's location field."""
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 10,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )

        assert response.status_code == 200
        mock_client.update_spool.assert_not_called()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_returns_inventory_spool(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """POST /slot-assignments response is mapped to InventorySpool format."""
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 10,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )

        assert response.status_code == 200
        body = response.json()
        assert body["id"] == 10
        assert body["material"] == "PLA"
        assert body["data_origin"] == "spoolman"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_upserts_on_conflict(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """POST /slot-assignments twice for the same slot replaces the old spool ID."""
        # First assign spool 99
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 99,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )
        # Re-assign spool 10 to the same slot
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 10,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )
        assert response.status_code == 200

        # The /all endpoint must report exactly one row for this slot with spool_id=10
        all_resp = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments/all",
            params={"printer_id": test_printer.id},
        )
        assert all_resp.status_code == 200
        rows = all_resp.json()
        matched = [r for r in rows if r["ams_id"] == 0 and r["tray_id"] == 0]
        assert len(matched) == 1
        assert matched[0]["spoolman_spool_id"] == 10

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_printer_not_found(self, async_client: AsyncClient, slot_settings, mock_client):
        """POST /slot-assignments with unknown printer_id returns 404."""
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 10,
                "printer_id": 99999,
                "ams_id": 0,
                "tray_id": 0,
            },
        )

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_assign_invalid_spool_id(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
        """POST /slot-assignments with spool_id=0 returns 422 (gt=0 validation)."""
        response = await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 0,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )

        assert response.status_code == 422


class TestUnassignSpoolmanSlot:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unassign_deletes_local_row(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """DELETE /slot-assignments/{id} removes the row so /all no longer lists it."""
        # First assign spool 10
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={
                "spoolman_spool_id": 10,
                "printer_id": test_printer.id,
                "ams_id": 0,
                "tray_id": 0,
            },
        )
        # Then unassign
        response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")
        assert response.status_code == 200

        # The /all endpoint must now return an empty list for this printer
        all_resp = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments/all",
            params={"printer_id": test_printer.id},
        )
        assert all_resp.status_code == 200
        assert all_resp.json() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unassign_does_not_call_update_spool(self, async_client: AsyncClient, slot_settings, mock_client):
        """DELETE /slot-assignments/{id} must NOT touch Spoolman's location field."""
        response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")

        assert response.status_code == 200
        mock_client.update_spool.assert_not_called()

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unassign_returns_inventory_spool(self, async_client: AsyncClient, slot_settings, mock_client):
        """DELETE /slot-assignments/{id} returns the spool in InventorySpool format."""
        response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")

        assert response.status_code == 200
        body = response.json()
        assert body["id"] == 10
        assert body["data_origin"] == "spoolman"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unassign_invalid_id(self, async_client: AsyncClient, slot_settings, mock_client):
        """DELETE /slot-assignments/0 returns 422 (gt=0 path validation)."""
        response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/0")

        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unassign_succeeds_when_spool_deleted_in_spoolman(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
    ):
        """DELETE /slot-assignments/{id} returns 200 even when the spool no longer exists in Spoolman.

        The local row must be removed regardless — the caller should not see an error just
        because Spoolman has already discarded the spool.
        """
        from sqlalchemy import select

        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
        from backend.app.services.spoolman import SpoolmanNotFoundError

        # Create the assignment first
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        # Spool 10 has since been deleted from Spoolman
        mock_client.get_spool = AsyncMock(side_effect=SpoolmanNotFoundError("spool 10 not found"))

        response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")

        assert response.status_code == 200
        assert response.json().get("id") == 10

        # Local row must be gone
        result = await db_session.execute(
            select(SpoolmanSlotAssignment).where(
                SpoolmanSlotAssignment.printer_id == test_printer.id,
                SpoolmanSlotAssignment.ams_id == 0,
                SpoolmanSlotAssignment.tray_id == 0,
            )
        )
        assert result.scalar_one_or_none() is None


class TestGetSpoolmanSlotAssignment:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_returns_matched_spool(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
        """GET /slot-assignments returns the spool whose ID is in the local table."""
        # First assign so the row exists
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )
        mock_client.get_spool.reset_mock()

        response = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments",
            params={"printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        assert response.status_code == 200
        body = response.json()
        assert body is not None
        assert body["id"] == 10
        mock_client.get_spool.assert_awaited_once_with(10)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_returns_null_when_no_assignment(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """GET /slot-assignments returns null when no local row exists for the slot."""
        response = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments",
            params={"printer_id": test_printer.id, "ams_id": 1, "tray_id": 0},
        )

        assert response.status_code == 200
        assert response.json() is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_printer_not_found(self, async_client: AsyncClient, slot_settings, mock_client):
        """GET /slot-assignments with unknown printer_id returns 404."""
        response = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments",
            params={"printer_id": 99999, "ams_id": 0, "tray_id": 0},
        )

        assert response.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_missing_params(self, async_client: AsyncClient, slot_settings, mock_client):
        """GET /slot-assignments without required params returns 422."""
        response = await async_client.get("/api/v1/spoolman/inventory/slot-assignments")

        assert response.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_returns_null_and_cleans_stale_when_spool_deleted_in_spoolman(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
    ):
        """GET /slot-assignments returns null and removes the stale row when Spoolman returns 404."""
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
        from backend.app.services.spoolman import SpoolmanNotFoundError

        # Assign spool 10 first
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        # Simulate spool 10 being deleted from Spoolman (404 via SpoolmanNotFoundError)
        mock_client.get_spool = AsyncMock(side_effect=SpoolmanNotFoundError("spool 10 not found"))

        response = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments",
            params={"printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        assert response.status_code == 200
        assert response.json() is None

        # Stale row must have been removed
        await db_session.refresh(test_printer)  # ensure session is fresh
        result = await db_session.execute(
            select(SpoolmanSlotAssignment).where(
                SpoolmanSlotAssignment.printer_id == test_printer.id,
                SpoolmanSlotAssignment.ams_id == 0,
                SpoolmanSlotAssignment.tray_id == 0,
            )
        )
        assert result.scalar_one_or_none() is None

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_propagates_503_from_spoolman(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client
    ):
        """GET /slot-assignments propagates a 503 from Spoolman instead of silently returning null."""
        from backend.app.services.spoolman import SpoolmanUnavailableError

        # Assign spool 10 first so a local row exists
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        # Simulate Spoolman being unreachable
        mock_client.get_spool = AsyncMock(side_effect=SpoolmanUnavailableError("timeout"))

        response = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments",
            params={"printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )

        assert response.status_code == 503


class TestGetAllSpoolmanSlotAssignments:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_all_returns_empty_list(self, async_client: AsyncClient, slot_settings, mock_client):
        """GET /slot-assignments/all returns [] when no assignments exist."""
        response = await async_client.get("/api/v1/spoolman/inventory/slot-assignments/all")

        assert response.status_code == 200
        assert response.json() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_all_returns_all_rows(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
        """GET /slot-assignments/all returns all existing assignments."""
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 20, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 1},
        )

        response = await async_client.get("/api/v1/spoolman/inventory/slot-assignments/all")

        assert response.status_code == 200
        body = response.json()
        assert len(body) == 2
        spool_ids = {r["spoolman_spool_id"] for r in body}
        assert spool_ids == {10, 20}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_get_all_filters_by_printer(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
    ):
        """GET /slot-assignments/all?printer_id=X only returns that printer's rows."""
        from backend.app.models.printer import Printer

        # Create a second printer via DB directly (no Spoolman mock needed for printer creation)
        other = Printer(
            name="Other Printer",
            serial_number="SLOTTEST002",
            ip_address="192.168.1.101",
            access_code="87654321",
        )
        db_session.add(other)
        await db_session.commit()
        await db_session.refresh(other)

        # Assign via API for test_printer
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
        )
        # Assign via API for other printer
        await async_client.post(
            "/api/v1/spoolman/inventory/slot-assignments",
            json={"spoolman_spool_id": 99, "printer_id": other.id, "ams_id": 0, "tray_id": 0},
        )

        response = await async_client.get(
            "/api/v1/spoolman/inventory/slot-assignments/all",
            params={"printer_id": test_printer.id},
        )

        assert response.status_code == 200
        body = response.json()
        assert len(body) == 1
        assert body[0]["spoolman_spool_id"] == 10
        assert body[0]["printer_id"] == test_printer.id


class TestCascadeDeletePrinter:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_printer_removes_slot_assignments(
        self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
    ):
        """DELETE /printers/{id} removes all slot assignments for that printer.

        SQLite does not enforce FK cascades automatically. The delete_printer
        endpoint must explicitly delete SpoolmanSlotAssignment rows so no
        orphaned rows survive after the printer record is gone.
        """
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment

        # Assign two spools to different AMS slots on the test printer
        for tray_id, spool_id in [(0, 10), (1, 20)]:
            await async_client.post(
                "/api/v1/spoolman/inventory/slot-assignments",
                json={
                    "spoolman_spool_id": spool_id,
                    "printer_id": test_printer.id,
                    "ams_id": 0,
                    "tray_id": tray_id,
                },
            )

        # Verify both rows exist
        pre = await db_session.execute(
            select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == test_printer.id)
        )
        assert len(pre.scalars().all()) == 2

        # Delete the printer
        del_resp = await async_client.delete(f"/api/v1/printers/{test_printer.id}")
        assert del_resp.status_code == 200

        # All slot assignment rows for the deleted printer must be gone
        post = await db_session.execute(
            select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == test_printer.id)
        )
        assert post.scalars().all() == []


class TestModeSwitchKeepsAssignments:
    """#2812 — the mode toggle no longer deletes anything.

    It used to empty the other mode's table on every switch, so opening the
    settings page to see what Spoolman mode did destroyed the built-in slot
    assignments, with no confirmation and no way back. Both tables are kept now
    and the readers that could be confused by a row in the inactive one ask
    which mode is active instead (``spoolman_owns_assignments``).
    """

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_switch_to_internal_mode_keeps_spoolman_slot_assignments(
        self, async_client: AsyncClient, db_session, test_printer
    ):
        from backend.app.models.settings import Settings
        from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment

        db_session.add(Settings(key="spoolman_enabled", value="true"))
        db_session.add(SpoolmanSlotAssignment(printer_id=test_printer.id, ams_id=0, tray_id=0, spoolman_spool_id=1))
        await db_session.commit()

        resp = await async_client.put("/api/v1/settings/spoolman", json={"spoolman_enabled": "false"})
        assert resp.status_code == 200

        rows = await db_session.execute(
            select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == test_printer.id)
        )
        assert len(rows.scalars().all()) == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_switch_to_spoolman_mode_keeps_builtin_assignments(
        self, async_client: AsyncClient, db_session, test_printer
    ):
        """The reported case: four toggles in 85 seconds took the reporter's
        built-in assignments and never gave them back."""
        from backend.app.models.settings import Settings
        from backend.app.models.spool import Spool
        from backend.app.models.spool_assignment import SpoolAssignment

        db_session.add(Settings(key="spoolman_enabled", value="false"))
        spool = Spool(material="PLA", rgba="000000FF")
        db_session.add(spool)
        await db_session.flush()
        db_session.add(SpoolAssignment(printer_id=test_printer.id, ams_id=0, tray_id=0, spool_id=spool.id))
        await db_session.commit()

        on = await async_client.put("/api/v1/settings/spoolman", json={"spoolman_enabled": "true"})
        assert on.status_code == 200
        off = await async_client.put("/api/v1/settings/spoolman", json={"spoolman_enabled": "false"})
        assert off.status_code == 200

        rows = await db_session.execute(select(SpoolAssignment).where(SpoolAssignment.printer_id == test_printer.id))
        assert len(rows.scalars().all()) == 1