|
|
@@ -0,0 +1,926 @@
|
|
|
+"""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]
|