mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
927 lines
40 KiB
Python
927 lines
40 KiB
Python
"""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]
|