bambuddy/backend/tests/integration/test_suppliers_api.py

"""API coverage for the supplier master list and spool assignments (#2988).

The master list lives under /api/v1/inventory/suppliers (Locations pattern),
gated by the plain inventory permissions. Assignments exist for both
inventories: `spool_suppliers` for built-in spools and the
`spoolman_spool_suppliers` twin keyed by the remote spool id.
"""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.api.routes.inventory import DUPLICATE_SUPPLIER_NAME
from backend.app.models.spool import Spool
from backend.app.models.spool_usage_history import SpoolUsageHistory
from backend.app.models.supplier import SpoolmanSpoolSupplier, SpoolSupplier, Supplier

SAMPLE_SPOOLMAN_SPOOL = {
    "id": 7,
    "filament": {
        "id": 1,
        "name": "PETG CF",
        "material": "PETG",
        "weight": 1000,
        "color_hex": "000000",
        "vendor": {"id": 1, "name": "BrandX"},
    },
    "remaining_weight": 600.0,
    "used_weight": 400.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 spool_factory(db_session: AsyncSession):
    async def _create(**kwargs):
        defaults = {
            "material": "PLA",
            "subtype": "Matte",
            "brand": "Bambu Lab",
            "color_name": "Charcoal",
            "rgba": "333333FF",
            "label_weight": 1000,
            "core_weight": 250,
            "weight_used": 0,
            "weight_used_baseline": 0,
            "weight_locked": False,
        }
        defaults.update(kwargs)
        spool = Spool(**defaults)
        db_session.add(spool)
        await db_session.commit()
        await db_session.refresh(spool)
        return spool

    return _create


@pytest.fixture
async def supplier_factory(db_session: AsyncSession):
    _counter = [0]

    async def _create(**kwargs):
        _counter[0] += 1
        defaults = {"name": f"Supplier {_counter[0]}"}
        defaults.update(kwargs)
        supplier = Supplier(**defaults)
        db_session.add(supplier)
        await db_session.commit()
        await db_session.refresh(supplier)
        return supplier

    return _create


@pytest.fixture
async def spoolman_settings(db_session: AsyncSession):
    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
def mock_spoolman_client():
    client = MagicMock()
    client.base_url = "http://localhost:7912"
    client.health_check = AsyncMock(return_value=True)
    client.get_spool = AsyncMock(return_value=SAMPLE_SPOOLMAN_SPOOL)
    client.get_all_spools = AsyncMock(return_value=[SAMPLE_SPOOLMAN_SPOOL])
    client.get_distinct_locations = AsyncMock(return_value=[])
    client.delete_spool = AsyncMock(return_value=None)

    with (
        patch(
            "backend.app.api.routes.spoolman_inventory._get_client",
            AsyncMock(return_value=client),
        ),
        # The supplier delete reconciles twin rows against Spoolman before it
        # refuses (#2988); inventory.py resolves its own client.
        patch("backend.app.api.routes.inventory.get_spoolman_client", AsyncMock(return_value=client)),
    ):
        yield client


@pytest.mark.unit
def test_supplier_relationships_use_the_default_loader():
    """No relationship-level eager loader (#2988).

    ``Spool.supplier_links`` used to be ``lazy="selectin"``, which made every
    ``select(Spool)`` in the app — usage tracker, AMS sync, labels, backup —
    pay two extra round trips for assignments it never reads. The routes that
    embed them ask for ``selectinload()`` at the query site instead.
    """
    from sqlalchemy import inspect as sa_inspect

    assert sa_inspect(Spool).relationships["supplier_links"].lazy == "select"
    assert sa_inspect(SpoolSupplier).relationships["supplier"].lazy == "select"
    assert sa_inspect(SpoolmanSpoolSupplier).relationships["supplier"].lazy == "select"


class TestSupplierCrud:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_create_and_list(self, async_client: AsyncClient):
        resp = await async_client.post(
            "/api/v1/inventory/suppliers",
            json={"name": "Filament24", "website": "https://filament24.example", "customer_number": "C-1042"},
        )
        assert resp.status_code == 201
        body = resp.json()
        assert body["name"] == "Filament24"
        assert body["spool_count"] == 0

        listing = await async_client.get("/api/v1/inventory/suppliers")
        assert listing.status_code == 200
        assert [s["name"] for s in listing.json()] == ["Filament24"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_update(self, async_client: AsyncClient, supplier_factory):
        supplier = await supplier_factory(name="Old Name")
        resp = await async_client.patch(f"/api/v1/inventory/suppliers/{supplier.id}", json={"name": "New Name"})
        assert resp.status_code == 200
        assert resp.json()["name"] == "New Name"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_unreferenced(self, async_client: AsyncClient, supplier_factory):
        supplier = await supplier_factory()
        resp = await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")
        assert resp.status_code == 200
        assert (await async_client.get("/api/v1/inventory/suppliers")).json() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_referenced_is_refused(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        supplier = await supplier_factory()
        spool = await spool_factory()
        db_session.add(SpoolSupplier(spool_id=spool.id, supplier_id=supplier.id))
        await db_session.commit()

        resp = await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")
        assert resp.status_code == 409
        assert "cannot be deleted" in resp.json()["detail"]

        # The listing surfaces the usage count behind the refusal.
        listing = await async_client.get("/api/v1/inventory/suppliers")
        assert listing.json()[0]["spool_count"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_refused_for_spoolman_only_reference(
        self, async_client: AsyncClient, supplier_factory, db_session: AsyncSession
    ):
        """A supplier used only by Spoolman-mode assignments is still protected."""
        supplier = await supplier_factory()
        db_session.add(SpoolmanSpoolSupplier(spoolman_spool_id=7, supplier_id=supplier.id))
        await db_session.commit()

        resp = await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")
        assert resp.status_code == 409
        assert (await async_client.get("/api/v1/inventory/suppliers")).json()[0]["spool_count"] == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_list_reads_case_insensitively(self, async_client: AsyncClient, supplier_factory):
        """Ordered on the folded name, so "extrudr" sits between "Bambu Store"
        and "Zultrat" rather than after every capitalised name (#2988)."""
        for name in ["Zultrat", "extrudr", "Bambu Store"]:
            await supplier_factory(name=name)
        listing = (await async_client.get("/api/v1/inventory/suppliers")).json()
        assert [s["name"] for s in listing] == ["Bambu Store", "extrudr", "Zultrat"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_a_padded_name_at_the_cap_is_measured_after_trimming(
        self, async_client: AsyncClient, supplier_factory
    ):
        """The length limit applies to the stored name, not to surrounding
        whitespace that is stripped anyway (#2988)."""
        name = "S" * 200
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": f"  {name}  "})
        assert resp.status_code == 201
        assert resp.json()["name"] == name

        other = await supplier_factory(name="Other")
        resp = await async_client.patch(f"/api/v1/inventory/suppliers/{other.id}", json={"name": f" {'T' * 200} "})
        assert resp.status_code == 200
        assert resp.json()["name"] == "T" * 200

        # A name that really is too long is still refused.
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": "U" * 201})
        assert resp.status_code == 422


class TestSupplierNameUniqueness:
    """Supplier names are the feature's key (#2988): CSV import resolves
    against them and a rename re-points every assignment, so two rows with
    the same name silently send an import to the wrong supplier."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_duplicate_name_is_refused(self, async_client: AsyncClient):
        assert (await async_client.post("/api/v1/inventory/suppliers", json={"name": "Extrudr"})).status_code == 201
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": "Extrudr"})
        assert resp.status_code == 409
        assert resp.json()["detail"] == DUPLICATE_SUPPLIER_NAME
        assert len((await async_client.get("/api/v1/inventory/suppliers")).json()) == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_duplicate_is_case_and_whitespace_insensitive(self, async_client: AsyncClient):
        """The CSV map is keyed on the trimmed lower-cased name, so a case
        variant would be just as ambiguous as an exact duplicate."""
        await async_client.post("/api/v1/inventory/suppliers", json={"name": "Extrudr"})
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": "  eXtRuDr "})
        assert resp.status_code == 409

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_duplicate_is_refused_for_non_ascii_names(self, async_client: AsyncClient):
        """The fold has to be the Python one to be worth anything here.

        SQLite's lower() folds ASCII only, so a unique index on lower(name)
        saw these as two different names and let both in — while the import
        map, which folds in Python, collapsed them onto a single entry and
        resolved to whichever row it built last. That is exactly the silent
        wrong-supplier assignment the rule exists to prevent.
        """
        assert (await async_client.post("/api/v1/inventory/suppliers", json={"name": "Ökofilament"})).status_code == 201
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": "ökofilament"})
        assert resp.status_code == 409
        assert resp.json()["detail"] == DUPLICATE_SUPPLIER_NAME
        assert len((await async_client.get("/api/v1/inventory/suppliers")).json()) == 1

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_rename_onto_an_existing_name_is_refused(self, async_client: AsyncClient, supplier_factory):
        a = await supplier_factory(name="Extrudr")
        b = await supplier_factory(name="Filament24")
        resp = await async_client.patch(f"/api/v1/inventory/suppliers/{b.id}", json={"name": "extrudr"})
        assert resp.status_code == 409
        # Renaming a supplier to the name it already has is not a conflict.
        assert (
            await async_client.patch(f"/api/v1/inventory/suppliers/{a.id}", json={"name": "Extrudr"})
        ).status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_name_is_trimmed_on_write(self, async_client: AsyncClient):
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": "  Extrudr  "})
        assert resp.status_code == 201
        assert resp.json()["name"] == "Extrudr"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_explicit_null_name_is_a_validation_error(self, async_client: AsyncClient, supplier_factory):
        """422, not the 500 a NOT NULL violation used to produce."""
        supplier = await supplier_factory()
        resp = await async_client.patch(f"/api/v1/inventory/suppliers/{supplier.id}", json={"name": None})
        assert resp.status_code == 422

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_csv_separator_in_name_is_refused(self, async_client: AsyncClient):
        """A ';' in the name would split into unknown names on CSV import and
        silently drop every assignment that used it."""
        resp = await async_client.post("/api/v1/inventory/suppliers", json={"name": "Extrudr; GmbH"})
        assert resp.status_code == 422


class TestSpoolSupplierAssignments:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_replace_all_and_embed_in_spool_response(
        self, async_client: AsyncClient, supplier_factory, spool_factory
    ):
        a = await supplier_factory(name="Supplier A")
        b = await supplier_factory(name="Supplier B")
        spool = await spool_factory()

        resp = await async_client.put(
            f"/api/v1/inventory/spools/{spool.id}/suppliers",
            json=[
                {
                    "supplier_id": a.id,
                    "supplier_article_number": "A-100",
                    "quoted_price_per_kg": 19.99,
                    "is_purchase_source": True,
                },
                {"supplier_id": b.id, "quoted_price_per_kg": 22.5},
            ],
        )
        assert resp.status_code == 200
        body = resp.json()
        assert {row["supplier_name"] for row in body} == {"Supplier A", "Supplier B"}
        assert [row["is_purchase_source"] for row in sorted(body, key=lambda r: r["supplier_id"])] == [True, False]

        # Embedded in the inventory listing.
        listing = await async_client.get("/api/v1/inventory/spools")
        spool_row = next(s for s in listing.json() if s["id"] == spool.id)
        assert {row["supplier_name"] for row in spool_row["suppliers"]} == {"Supplier A", "Supplier B"}

        # Replace-all: shrinking the list removes the other assignment.
        resp = await async_client.put(
            f"/api/v1/inventory/spools/{spool.id}/suppliers",
            json=[{"supplier_id": b.id, "is_purchase_source": True}],
        )
        assert resp.status_code == 200
        assert [row["supplier_name"] for row in resp.json()] == ["Supplier B"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_two_purchase_sources_are_refused(self, async_client: AsyncClient, supplier_factory, spool_factory):
        a = await supplier_factory()
        b = await supplier_factory()
        spool = await spool_factory()

        resp = await async_client.put(
            f"/api/v1/inventory/spools/{spool.id}/suppliers",
            json=[
                {"supplier_id": a.id, "is_purchase_source": True},
                {"supplier_id": b.id, "is_purchase_source": True},
            ],
        )
        assert resp.status_code == 400

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_unknown_supplier_is_refused(self, async_client: AsyncClient, spool_factory):
        spool = await spool_factory()
        resp = await async_client.put(
            f"/api/v1/inventory/spools/{spool.id}/suppliers",
            json=[{"supplier_id": 999999}],
        )
        assert resp.status_code == 404

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_duplicate_supplier_is_refused(self, async_client: AsyncClient, supplier_factory, spool_factory):
        a = await supplier_factory()
        spool = await spool_factory()
        resp = await async_client.put(
            f"/api/v1/inventory/spools/{spool.id}/suppliers",
            json=[{"supplier_id": a.id}, {"supplier_id": a.id}],
        )
        assert resp.status_code == 400


class TestSpoolmanSupplierAssignments:
    """Spoolman parity (#2988): same endpoints, same shape, twin table."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_replace_all_and_embed(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client
    ):
        a = await supplier_factory(name="Supplier A")

        resp = await async_client.put(
            "/api/v1/spoolman/inventory/spools/7/suppliers",
            json=[
                {
                    "supplier_id": a.id,
                    "supplier_article_number": "A-100",
                    "quoted_price_per_kg": 19.99,
                    "is_purchase_source": True,
                }
            ],
        )
        assert resp.status_code == 200
        body = resp.json()
        assert body[0]["supplier_name"] == "Supplier A"
        assert body[0]["quoted_price_per_kg"] == 19.99

        # Same rows via GET, embedded in the single-spool and list responses.
        assert (await async_client.get("/api/v1/spoolman/inventory/spools/7/suppliers")).json() == body
        single = await async_client.get("/api/v1/spoolman/inventory/spools/7")
        assert [row["supplier_name"] for row in single.json()["suppliers"]] == ["Supplier A"]
        listing = await async_client.get("/api/v1/spoolman/inventory/spools")
        spool_row = next(s for s in listing.json() if s["id"] == 7)
        assert [row["supplier_name"] for row in spool_row["suppliers"]] == ["Supplier A"]

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_validation_matches_builtin_inventory(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client
    ):
        a = await supplier_factory()
        b = await supplier_factory()

        resp = await async_client.put(
            "/api/v1/spoolman/inventory/spools/7/suppliers",
            json=[{"supplier_id": a.id, "is_purchase_source": True}, {"supplier_id": b.id, "is_purchase_source": True}],
        )
        assert resp.status_code == 400

        resp = await async_client.put(
            "/api/v1/spoolman/inventory/spools/7/suppliers",
            json=[{"supplier_id": 999999}],
        )
        assert resp.status_code == 404


class TestSupplierInheritance:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_new_spool_of_same_product_inherits_sources(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        supplier = await supplier_factory(name="Supplier A")
        donor = await spool_factory()
        db_session.add(
            SpoolSupplier(
                spool_id=donor.id,
                supplier_id=supplier.id,
                supplier_article_number="A-100",
                quoted_price_per_kg=19.99,
                is_purchase_source=True,
            )
        )
        await db_session.commit()

        resp = await async_client.post(
            "/api/v1/inventory/spools",
            json={"material": "PLA", "subtype": "Matte", "brand": "Bambu Lab", "color_name": "Charcoal"},
        )
        assert resp.status_code == 200
        suppliers = resp.json()["suppliers"]
        assert [row["supplier_name"] for row in suppliers] == ["Supplier A"]
        assert suppliers[0]["supplier_article_number"] == "A-100"
        assert suppliers[0]["quoted_price_per_kg"] == 19.99
        # Where THIS spool was bought is unknown — never inherited.
        assert suppliers[0]["is_purchase_source"] is False

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_different_product_inherits_nothing(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        supplier = await supplier_factory()
        donor = await spool_factory()
        db_session.add(SpoolSupplier(spool_id=donor.id, supplier_id=supplier.id))
        await db_session.commit()

        resp = await async_client.post(
            "/api/v1/inventory/spools",
            json={"material": "PETG", "subtype": "Matte", "brand": "Bambu Lab", "color_name": "Charcoal"},
        )
        assert resp.status_code == 200
        assert resp.json()["suppliers"] == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_bulk_create_gives_every_copy_its_own_links(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        """One donor lookup for the batch, one set of rows per copy."""
        supplier = await supplier_factory(name="Supplier A")
        donor = await spool_factory()
        db_session.add(SpoolSupplier(spool_id=donor.id, supplier_id=supplier.id, supplier_article_number="A-100"))
        await db_session.commit()

        resp = await async_client.post(
            "/api/v1/inventory/spools/bulk",
            json={
                "spool": {"material": "PLA", "subtype": "Matte", "brand": "Bambu Lab", "color_name": "Charcoal"},
                "quantity": 3,
            },
        )
        assert resp.status_code == 200
        created = resp.json()
        assert len(created) == 3
        for row in created:
            assert [link["supplier_name"] for link in row["suppliers"]] == ["Supplier A"]
            assert row["suppliers"][0]["supplier_article_number"] == "A-100"
            assert row["suppliers"][0]["is_purchase_source"] is False

        # Own rows, not shared ones.
        assert len({row["suppliers"][0]["id"] for row in created}) == 3


class TestSupplierLifecycle:
    """What happens to assignments when the spool they hang on goes away."""

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleting_a_spool_frees_its_supplier(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        supplier = await supplier_factory()
        spool = await spool_factory()
        db_session.add(SpoolSupplier(spool_id=spool.id, supplier_id=supplier.id))
        await db_session.commit()

        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 409
        assert (await async_client.delete(f"/api/v1/inventory/spools/{spool.id}")).status_code == 200

        # delete-orphan on Spool.supplier_links takes the assignment with it,
        # so the supplier stops being referenced and becomes deletable.
        assert (await async_client.get("/api/v1/inventory/suppliers")).json()[0]["spool_count"] == 0
        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_deleting_a_spoolman_spool_drops_the_twin_rows(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client, db_session
    ):
        """Spoolman owns the spool, Bambuddy owns the assignment, and nothing
        in the database can cascade it. A leaked row keeps the supplier's
        reference count non-zero, so the delete guard would answer 409 for a
        spool the user can no longer see (#2988)."""
        supplier = await supplier_factory()
        assert (
            await async_client.put(
                "/api/v1/spoolman/inventory/spools/7/suppliers",
                json=[{"supplier_id": supplier.id, "is_purchase_source": True}],
            )
        ).status_code == 200
        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 409

        assert (await async_client.delete("/api/v1/spoolman/inventory/spools/7")).status_code == 200

        rows = await db_session.execute(select(SpoolmanSpoolSupplier))
        assert rows.scalars().all() == []
        assert (await async_client.get("/api/v1/inventory/suppliers")).json()[0]["spool_count"] == 0
        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_bulk_delete_drops_the_twin_rows_too(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client, db_session
    ):
        supplier = await supplier_factory()
        assert (
            await async_client.put("/api/v1/spoolman/inventory/spools/7/suppliers", json=[{"supplier_id": supplier.id}])
        ).status_code == 200
        # The row has to be proven present before the delete, or the empty
        # assertion below holds whether or not the purge did anything.
        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 409

        resp = await async_client.post("/api/v1/spoolman/inventory/spools/bulk-delete", json={"ids": [7]})
        assert resp.status_code == 200
        assert resp.json()["deleted"] == 1
        rows = await db_session.execute(select(SpoolmanSpoolSupplier))
        assert rows.scalars().all() == []
        assert (await async_client.get("/api/v1/inventory/suppliers")).json()[0]["spool_count"] == 0
        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 200

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_delete_reconciles_a_spool_deleted_in_spoolman_itself(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client, db_session
    ):
        """Spoolman is a separate application with its own UI, and Bambuddy
        only hears about the deletes it performs itself. A spool removed over
        there leaves its assignment behind, and that phantom reference used to
        make the supplier permanently undeletable with nothing on any screen
        that could show or clear it (#2988)."""
        supplier = await supplier_factory()
        assert (
            await async_client.put("/api/v1/spoolman/inventory/spools/7/suppliers", json=[{"supplier_id": supplier.id}])
        ).status_code == 200
        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 409

        # Spool 7 disappears from Spoolman without Bambuddy doing anything.
        mock_spoolman_client.get_all_spools = AsyncMock(return_value=[])

        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 200
        rows = await db_session.execute(select(SpoolmanSpoolSupplier))
        assert rows.scalars().all() == []

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_reconcile_keeps_assignments_of_an_archived_spool(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client
    ):
        """Archiving is a soft delete: the spool is still there and its
        assignment has to survive, so the reconcile asks for the archived ones
        too and the delete still answers 409."""
        supplier = await supplier_factory()
        assert (
            await async_client.put("/api/v1/spoolman/inventory/spools/7/suppliers", json=[{"supplier_id": supplier.id}])
        ).status_code == 200

        async def _all_spools(allow_archived: bool = False):
            return [dict(SAMPLE_SPOOLMAN_SPOOL, archived=True)] if allow_archived else []

        mock_spoolman_client.get_all_spools = AsyncMock(side_effect=_all_spools)

        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 409

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_reconcile_keeps_the_rows_when_spoolman_is_unreachable(
        self, async_client: AsyncClient, supplier_factory, spoolman_settings, mock_spoolman_client, db_session
    ):
        """A failed lookup is not evidence that the spool is gone."""
        supplier = await supplier_factory()
        assert (
            await async_client.put("/api/v1/spoolman/inventory/spools/7/suppliers", json=[{"supplier_id": supplier.id}])
        ).status_code == 200

        mock_spoolman_client.get_all_spools = AsyncMock(side_effect=RuntimeError("Cannot reach Spoolman"))

        assert (await async_client.delete(f"/api/v1/inventory/suppliers/{supplier.id}")).status_code == 409
        rows = await db_session.execute(select(SpoolmanSpoolSupplier))
        assert len(rows.scalars().all()) == 1


class TestSupplierStats:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_groups_by_purchase_source(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        a = await supplier_factory(name="Supplier A")
        b = await supplier_factory(name="Supplier B")
        bought_at_a = await spool_factory(label_weight=1000, weight_used=200)
        alt_only = await spool_factory(color_name="Red")
        db_session.add_all(
            [
                SpoolSupplier(spool_id=bought_at_a.id, supplier_id=a.id, is_purchase_source=True),
                # Alternative source only — must NOT count toward supplier B.
                SpoolSupplier(spool_id=alt_only.id, supplier_id=b.id, is_purchase_source=False),
                SpoolUsageHistory(
                    spool_id=bought_at_a.id, weight_used=150, percent_used=15, status="completed", cost=3.0
                ),
            ]
        )
        await db_session.commit()

        resp = await async_client.get("/api/v1/inventory/stats/suppliers")
        assert resp.status_code == 200
        rows = resp.json()
        assert len(rows) == 1
        assert rows[0]["supplier_name"] == "Supplier A"
        assert rows[0]["spool_count"] == 1
        assert rows[0]["remaining_g"] == pytest.approx(800)
        assert rows[0]["consumed_g"] == pytest.approx(150)
        # Cost comes from the recorded usage history (spool.cost_per_kg based),
        # never from quoted_price_per_kg.
        assert rows[0]["cost"] == pytest.approx(3.0)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_usage_half_honours_the_date_window(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        """The dashboard timeframe scopes consumption and cost; stock is
        point-in-time and stays out of the window (#2988)."""
        from datetime import datetime, timedelta, timezone

        supplier = await supplier_factory(name="Supplier A")
        spool = await spool_factory(label_weight=1000, weight_used=300)
        now = datetime.now(timezone.utc)
        db_session.add_all(
            [
                SpoolSupplier(spool_id=spool.id, supplier_id=supplier.id, is_purchase_source=True),
                SpoolUsageHistory(
                    spool_id=spool.id,
                    weight_used=100,
                    percent_used=10,
                    status="completed",
                    cost=2.0,
                    created_at=now - timedelta(days=90),
                ),
                SpoolUsageHistory(
                    spool_id=spool.id,
                    weight_used=200,
                    percent_used=20,
                    status="completed",
                    cost=4.0,
                    created_at=now - timedelta(days=2),
                ),
            ]
        )
        await db_session.commit()

        lifetime = (await async_client.get("/api/v1/inventory/stats/suppliers")).json()
        assert lifetime[0]["consumed_g"] == pytest.approx(300)
        assert lifetime[0]["cost"] == pytest.approx(6.0)

        date_from = (now - timedelta(days=30)).date().isoformat()
        windowed = (await async_client.get(f"/api/v1/inventory/stats/suppliers?date_from={date_from}")).json()
        assert windowed[0]["consumed_g"] == pytest.approx(200)
        assert windowed[0]["cost"] == pytest.approx(4.0)
        # Stock is not windowed.
        assert windowed[0]["spool_count"] == 1
        assert windowed[0]["remaining_g"] == pytest.approx(700)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_an_over_consumed_spool_counts_as_empty_not_negative(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        """Each spool is clamped before the sum, as the inventory page shows it:
        a full 1000 g spool next to one over-consumed by 200 g is 1000 g in
        stock, not 800 g (#2988)."""
        supplier = await supplier_factory(name="Supplier A")
        full = await spool_factory(label_weight=1000, weight_used=0)
        over = await spool_factory(label_weight=1000, weight_used=1200, color_name="Red")
        db_session.add_all(
            [
                SpoolSupplier(spool_id=full.id, supplier_id=supplier.id, is_purchase_source=True),
                SpoolSupplier(spool_id=over.id, supplier_id=supplier.id, is_purchase_source=True),
            ]
        )
        await db_session.commit()

        rows = (await async_client.get("/api/v1/inventory/stats/suppliers")).json()
        assert rows[0]["spool_count"] == 2
        assert rows[0]["remaining_g"] == pytest.approx(1000)

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_equal_consumption_falls_back_to_the_name(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        """A narrow date range leaves most rows at 0 g; they read by name
        instead of in whatever order the GROUP BY produced (#2988)."""
        names = ["zultrat", "Extrudr", "Bambu Store"]
        for name in names:
            supplier = await supplier_factory(name=name)
            spool = await spool_factory(color_name=name)
            db_session.add(SpoolSupplier(spool_id=spool.id, supplier_id=supplier.id, is_purchase_source=True))
        heavy = await supplier_factory(name="Yet Another")
        heavy_spool = await spool_factory(color_name="Heavy")
        db_session.add_all(
            [
                SpoolSupplier(spool_id=heavy_spool.id, supplier_id=heavy.id, is_purchase_source=True),
                SpoolUsageHistory(spool_id=heavy_spool.id, weight_used=50, percent_used=5, status="completed"),
            ]
        )
        await db_session.commit()

        rows = (await async_client.get("/api/v1/inventory/stats/suppliers")).json()
        assert [r["supplier_name"] for r in rows] == ["Yet Another", "Bambu Store", "Extrudr", "zultrat"]


class TestSupplierCsv:
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_export_carries_both_columns(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        a = await supplier_factory(name="Supplier A")
        b = await supplier_factory(name="Supplier B")
        spool = await spool_factory()
        db_session.add_all(
            [
                SpoolSupplier(spool_id=spool.id, supplier_id=a.id, is_purchase_source=True),
                SpoolSupplier(spool_id=spool.id, supplier_id=b.id),
            ]
        )
        await db_session.commit()

        export = await async_client.get("/api/v1/inventory/spools/export")
        assert export.status_code == 200
        header, row = export.text.splitlines()[:2]
        columns = header.split(",")
        assert "suppliers" in columns
        assert "purchase_supplier" in columns
        assert "Supplier A; Supplier B" in row
        assert row.split(",")[columns.index("purchase_supplier")].strip('"') == "Supplier A"

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_import_matches_by_name_and_creates_nothing(
        self, async_client: AsyncClient, supplier_factory, db_session: AsyncSession
    ):
        await supplier_factory(name="Supplier A")
        await supplier_factory(name="Supplier B")
        csv_text = (
            "material,brand,color_name,suppliers,purchase_supplier\n"
            'PLA,Bambu Lab,Charcoal,"  supplier a ; Supplier B",Supplier B\n'
            "PETG,Bambu Lab,Red,Unknown Corp,\n"
        )

        # Dry run: the unknown name is a warning, not a row error.
        preview = await async_client.post(
            "/api/v1/inventory/spools/import?dry_run=true",
            files={"file": ("spools.csv", csv_text.encode(), "text/csv")},
        )
        assert preview.status_code == 200
        body = preview.json()
        assert body["valid_count"] == 2
        assert body["error_count"] == 0
        assert any("Unknown Corp" in w for w in body["warnings"])

        # Real import: both rows land; assignments match by name, trimmed and
        # case-insensitive; the unknown name is dropped and NOT created.
        result = await async_client.post(
            "/api/v1/inventory/spools/import",
            files={"file": ("spools.csv", csv_text.encode(), "text/csv")},
        )
        assert result.status_code == 200
        assert result.json()["created"] == 2

        listing = await async_client.get("/api/v1/inventory/spools")
        by_color = {s["color_name"]: s for s in listing.json()}
        charcoal = by_color["Charcoal"]["suppliers"]
        assert {row["supplier_name"] for row in charcoal} == {"Supplier A", "Supplier B"}
        assert [row["supplier_name"] for row in charcoal if row["is_purchase_source"]] == ["Supplier B"]
        assert by_color["Red"]["suppliers"] == []
        suppliers = (await async_client.get("/api/v1/inventory/suppliers")).json()
        assert {s["name"] for s in suppliers} == {"Supplier A", "Supplier B"}

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_round_trip_preserves_assignments(
        self, async_client: AsyncClient, supplier_factory, spool_factory, db_session: AsyncSession
    ):
        a = await supplier_factory(name="Supplier A")
        spool = await spool_factory()
        db_session.add(SpoolSupplier(spool_id=spool.id, supplier_id=a.id, is_purchase_source=True))
        await db_session.commit()

        export = await async_client.get("/api/v1/inventory/spools/export")
        result = await async_client.post(
            "/api/v1/inventory/spools/import",
            files={"file": ("spools.csv", export.content, "text/csv")},
        )
        assert result.status_code == 200
        assert result.json()["created"] == 1

        listing = (await async_client.get("/api/v1/inventory/spools")).json()
        assert len(listing) == 2
        for row in listing:
            assert [link["supplier_name"] for link in row["suppliers"]] == ["Supplier A"]
            assert row["suppliers"][0]["is_purchase_source"] is True


class TestFromSlotInheritance:
    """The RFID "+ Add to inventory" path (#2988).

    POST /spools/from-slot builds the spool through create_spool_from_tray,
    which pre-initialises spool.supplier_links to []. The inheritance rows are
    added afterwards, so the closing query has to repopulate the collection —
    otherwise the identity-mapped instance answers with the stale empty list
    and the caller sees no suppliers until the next fetch.
    """

    @staticmethod
    def _status_for_tray(ams_id: int, tray_id: int, tray: dict):
        status = MagicMock()
        status.raw_data = {"ams": {"ams": [{"id": ams_id, "tray": [{"id": tray_id, **tray}]}]}}
        return status

    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_from_slot_response_carries_inherited_suppliers(
        self,
        async_client: AsyncClient,
        printer_factory,
        supplier_factory,
        spool_factory,
        db_session: AsyncSession,
    ):
        printer = await printer_factory(name="X1C-supplier-inherit")
        supplier = await supplier_factory(name="Supplier A")
        donor = await spool_factory(material="PLA", subtype=None, brand="Bambu Lab", color_name="Clear")
        db_session.add(SpoolSupplier(spool_id=donor.id, supplier_id=supplier.id, supplier_article_number="A-100"))
        await db_session.commit()

        # alpha=00 → create_spool_from_tray names the colour "Clear", matching
        # the donor product without needing a colour-catalogue row.
        tray = {
            "tray_type": "PLA",
            "tray_color": "11223300",
            "tag_uid": "1122334455667788",
            "tray_uuid": "0123456789ABCDEF0123456789ABCDEF",
        }
        with patch(
            "backend.app.services.printer_manager.printer_manager.get_status",
            return_value=self._status_for_tray(0, 1, tray),
        ):
            resp = await async_client.post(
                "/api/v1/inventory/spools/from-slot",
                json={"printer_id": printer.id, "ams_id": 0, "tray_id": 1},
            )

        assert resp.status_code == 200, resp.text
        body = resp.json()
        assert [row["supplier_name"] for row in body["suppliers"]] == ["Supplier A"]
        assert body["suppliers"][0]["supplier_article_number"] == "A-100"
        assert body["suppliers"][0]["is_purchase_source"] is False

        # The row was always written — the defect was the response reading a
        # stale collection off the identity-mapped instance.
        rows = await db_session.execute(select(SpoolSupplier).where(SpoolSupplier.spool_id == body["id"]))
        assert [row.supplier_id for row in rows.scalars().all()] == [supplier.id]