mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
529 lines
22 KiB
Python
529 lines
22 KiB
Python
"""API coverage for the spool material number (#2870).
|
|
|
|
The material number is the internal purchasing identifier shared by all
|
|
spools of a product. Pinned here: CRUD round-trip, server-side normalisation,
|
|
inheritance on the create paths, the per-number statistics aggregate and its
|
|
dashboard timeframe, and the CSV round-trip.
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from backend.app.models.spool import Spool
|
|
from backend.app.models.spool_usage_history import SpoolUsageHistory
|
|
|
|
|
|
@pytest.fixture
|
|
async def spool_factory(db_session: AsyncSession):
|
|
async def _create(**kwargs):
|
|
defaults = {
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Jade White",
|
|
"rgba": "FFFFFFFF",
|
|
"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
|
|
|
|
|
|
class TestMaterialNumberCrud:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_persists_and_lists_material_number(self, async_client: AsyncClient):
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={"material": "PLA", "material_number": "15"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "15"
|
|
|
|
listing = await async_client.get("/api/v1/inventory/spools")
|
|
assert listing.status_code == 200
|
|
assert [s["material_number"] for s in listing.json()] == ["15"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_patch_updates_material_number(self, async_client: AsyncClient, spool_factory):
|
|
spool = await spool_factory(material_number="15")
|
|
|
|
resp = await async_client.patch(
|
|
f"/api/v1/inventory/spools/{spool.id}",
|
|
json={"material_number": "16"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "16"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_material_number_longer_than_64_chars_is_rejected(self, async_client: AsyncClient):
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={"material": "PLA", "material_number": "x" * 65},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
class TestMaterialNumberNormalisation:
|
|
"""One validator on the schema, so every write path normalises (#2870).
|
|
|
|
Without it "15" and "15 " are two groups in the statistics aggregate and
|
|
two entries in the inventory filter chip, and the chip's exact match
|
|
never finds the padded one.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_strips_surrounding_whitespace(self, async_client: AsyncClient):
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={"material": "PLA", "material_number": " 15 "},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "15"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_create_maps_blank_to_none(self, async_client: AsyncClient):
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={"material": "PLA", "material_number": " "},
|
|
)
|
|
assert resp.status_code == 200
|
|
# NULL, not "" — "has no number" stays a single state to query for.
|
|
assert resp.json()["material_number"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_patch_strips_surrounding_whitespace(self, async_client: AsyncClient, spool_factory):
|
|
spool = await spool_factory(material_number="15")
|
|
|
|
resp = await async_client.patch(
|
|
f"/api/v1/inventory/spools/{spool.id}",
|
|
json={"material_number": " 16 "},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "16"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_bulk_edit_strips_surrounding_whitespace(self, async_client: AsyncClient, spool_factory):
|
|
spool = await spool_factory()
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [spool.id], "update": {"material_number": " 15 "}},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
listing = await async_client.get("/api/v1/inventory/spools")
|
|
assert [s["material_number"] for s in listing.json()] == ["15"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_csv_import_strips_surrounding_whitespace(self, async_client: AsyncClient):
|
|
csv = "material,brand,material_number\nPLA,Bambu Lab, 15 \n"
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/import",
|
|
files={"file": ("spools.csv", csv.encode("utf-8"), "text/csv")},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
listing = await async_client.get("/api/v1/inventory/spools")
|
|
assert [s["material_number"] for s in listing.json()] == ["15"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_padding_does_not_count_against_the_length_cap(self, async_client: AsyncClient, spool_factory):
|
|
"""The cap applies to what is stored: 63 characters with padding
|
|
around them fit, on every path that goes through the schema."""
|
|
padded = " " + "x" * 63
|
|
created = await async_client.post(
|
|
"/api/v1/inventory/spools", json={"material": "PLA", "material_number": padded}
|
|
)
|
|
assert created.status_code == 200, created.text
|
|
assert created.json()["material_number"] == "x" * 63
|
|
|
|
spool = await spool_factory()
|
|
patched = await async_client.patch(f"/api/v1/inventory/spools/{spool.id}", json={"material_number": padded})
|
|
assert patched.status_code == 200, patched.text
|
|
assert patched.json()["material_number"] == "x" * 63
|
|
|
|
bulk = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [spool.id], "update": {"material_number": " " + "y" * 64 + " "}},
|
|
)
|
|
assert bulk.status_code == 200, bulk.text
|
|
|
|
too_long = await async_client.patch(f"/api/v1/inventory/spools/{spool.id}", json={"material_number": "x" * 65})
|
|
assert too_long.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_padded_duplicate_does_not_become_a_second_group(self, async_client: AsyncClient, spool_factory):
|
|
await async_client.post("/api/v1/inventory/spools", json={"material": "PLA", "material_number": "15"})
|
|
await async_client.post("/api/v1/inventory/spools", json={"material": "PLA", "material_number": "15 "})
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
assert [r["material_number"] for r in resp.json()] == ["15"]
|
|
assert resp.json()[0]["spool_count"] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_blank_number_is_not_offered_as_a_group(self, async_client: AsyncClient):
|
|
await async_client.post("/api/v1/inventory/spools", json={"material": "PLA", "material_number": " "})
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
assert resp.json() == []
|
|
|
|
|
|
class TestMaterialNumberInheritance:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_new_spool_of_same_product_inherits_number(self, async_client: AsyncClient, spool_factory):
|
|
await spool_factory(material_number="15")
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Jade White",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "15"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_different_product_does_not_inherit(self, async_client: AsyncClient, spool_factory):
|
|
await spool_factory(material_number="15")
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Black",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_explicit_number_wins_over_inheritance(self, async_client: AsyncClient, spool_factory):
|
|
await spool_factory(material_number="15")
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Jade White",
|
|
"material_number": "99",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "99"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_bulk_create_inherits_number(self, async_client: AsyncClient, spool_factory):
|
|
await spool_factory(material_number="15")
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk",
|
|
json={
|
|
"spool": {
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Jade White",
|
|
},
|
|
"quantity": 3,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert [s["material_number"] for s in resp.json()] == ["15", "15", "15"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_archived_spool_still_provides_the_number(self, async_client: AsyncClient, spool_factory):
|
|
from datetime import datetime, timezone
|
|
|
|
await spool_factory(material_number="15", archived_at=datetime.now(timezone.utc))
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Jade White",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "15"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_timestamp_tie_goes_to_the_newest_spool(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
"""Two donors touched in the same second must not pick one at random."""
|
|
from datetime import datetime
|
|
|
|
same_moment = datetime(2026, 1, 1, 12, 0, 0)
|
|
older = await spool_factory(material_number="15")
|
|
newer = await spool_factory(material_number="16")
|
|
for spool in (older, newer):
|
|
spool.updated_at = same_moment
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools",
|
|
json={
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu Lab",
|
|
"color_name": "Jade White",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["material_number"] == "16"
|
|
|
|
|
|
class TestMaterialNumberStats:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_stats_group_by_number(self, async_client: AsyncClient, spool_factory, db_session: AsyncSession):
|
|
a = await spool_factory(material_number="15", label_weight=1000, weight_used=200)
|
|
b = await spool_factory(material_number="15", label_weight=1000, weight_used=0)
|
|
c = await spool_factory(material_number="16", color_name="Black", label_weight=1000, weight_used=500)
|
|
await spool_factory(material_number=None, color_name="Gray")
|
|
|
|
db_session.add_all(
|
|
[
|
|
SpoolUsageHistory(spool_id=a.id, weight_used=120, percent_used=12, status="completed", cost=2.4),
|
|
SpoolUsageHistory(spool_id=b.id, weight_used=80, percent_used=8, status="completed", cost=1.6),
|
|
SpoolUsageHistory(spool_id=c.id, weight_used=500, percent_used=50, status="failed", cost=15.0),
|
|
]
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
assert resp.status_code == 200
|
|
rows = {r["material_number"]: r for r in resp.json()}
|
|
|
|
assert set(rows) == {"15", "16"}
|
|
assert rows["15"]["spool_count"] == 2
|
|
assert rows["15"]["remaining_g"] == pytest.approx(1800)
|
|
assert rows["15"]["consumed_g"] == pytest.approx(200)
|
|
assert rows["15"]["cost"] == pytest.approx(4.0)
|
|
assert rows["16"]["consumed_g"] == pytest.approx(500)
|
|
assert rows["16"]["cost"] == pytest.approx(15.0)
|
|
# Heaviest consumption first.
|
|
assert [r["material_number"] for r in resp.json()] == ["16", "15"]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_an_over_consumed_spool_does_not_eat_its_siblings_stock(
|
|
self, async_client: AsyncClient, spool_factory
|
|
):
|
|
"""Remaining stock is clamped per spool, not once over the group.
|
|
|
|
weight_used above label_weight is reachable (a scale reading, an AMS
|
|
sync, or a plain PATCH), and every other remaining-weight computation
|
|
in the codebase clamps each spool at 0. Summing the raw difference
|
|
first would subtract the overshoot from the other spools of the same
|
|
number and report less stock than the inventory list does.
|
|
"""
|
|
await spool_factory(material_number="15", label_weight=1000, weight_used=0)
|
|
await spool_factory(material_number="15", color_name="Black", label_weight=1000, weight_used=1200)
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
assert resp.status_code == 200
|
|
row = resp.json()[0]
|
|
assert row["spool_count"] == 2
|
|
assert row["remaining_g"] == pytest.approx(1000)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_archived_spools_keep_their_recorded_consumption(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
from datetime import datetime, timezone
|
|
|
|
archived = await spool_factory(material_number="15", archived_at=datetime.now(timezone.utc))
|
|
db_session.add(
|
|
SpoolUsageHistory(spool_id=archived.id, weight_used=300, percent_used=30, status="completed", cost=6.0)
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
assert resp.status_code == 200
|
|
rows = {r["material_number"]: r for r in resp.json()}
|
|
# No active spools carry the number, but the consumption is still there.
|
|
assert rows["15"]["spool_count"] == 0
|
|
assert rows["15"]["remaining_g"] == 0
|
|
assert rows["15"]["consumed_g"] == pytest.approx(300)
|
|
|
|
|
|
class TestMaterialNumberStatsTimeframe:
|
|
"""The widget sits in the stats dashboard, so it follows its timeframe.
|
|
|
|
Usage history is the per-period half; stock is point-in-time and stays
|
|
whole — "how much do I hold" has no date range.
|
|
"""
|
|
|
|
@staticmethod
|
|
async def _usage(db_session, spool_id, *, days_ago, grams, cost):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
row = SpoolUsageHistory(
|
|
spool_id=spool_id, weight_used=grams, percent_used=grams / 10, status="completed", cost=cost
|
|
)
|
|
# created_at is a server default, so set it explicitly to age the row.
|
|
row.created_at = (datetime.now(timezone.utc) - timedelta(days=days_ago)).replace(tzinfo=None)
|
|
db_session.add(row)
|
|
await db_session.commit()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_date_from_excludes_older_consumption(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
spool = await spool_factory(material_number="15", label_weight=1000, weight_used=400)
|
|
await self._usage(db_session, spool.id, days_ago=200, grams=1000, cost=20.0)
|
|
await self._usage(db_session, spool.id, days_ago=2, grams=10, cost=0.2)
|
|
|
|
since = (datetime.now(timezone.utc) - timedelta(days=30)).date().isoformat()
|
|
resp = await async_client.get(f"/api/v1/inventory/stats/material-numbers?date_from={since}")
|
|
assert resp.status_code == 200
|
|
row = resp.json()[0]
|
|
assert row["consumed_g"] == pytest.approx(10)
|
|
assert row["cost"] == pytest.approx(0.2)
|
|
# Stock is point-in-time: unaffected by the range.
|
|
assert row["spool_count"] == 1
|
|
assert row["remaining_g"] == pytest.approx(600)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_date_to_excludes_newer_consumption(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
spool = await spool_factory(material_number="15")
|
|
await self._usage(db_session, spool.id, days_ago=200, grams=1000, cost=20.0)
|
|
await self._usage(db_session, spool.id, days_ago=2, grams=10, cost=0.2)
|
|
|
|
until = (datetime.now(timezone.utc) - timedelta(days=30)).date().isoformat()
|
|
resp = await async_client.get(f"/api/v1/inventory/stats/material-numbers?date_to={until}")
|
|
assert resp.json()[0]["consumed_g"] == pytest.approx(1000)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_no_range_still_reports_lifetime_totals(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
spool = await spool_factory(material_number="15")
|
|
await self._usage(db_session, spool.id, days_ago=200, grams=1000, cost=20.0)
|
|
await self._usage(db_session, spool.id, days_ago=2, grams=10, cost=0.2)
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
assert resp.json()[0]["consumed_g"] == pytest.approx(1010)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_a_number_with_no_usage_in_range_still_lists_its_stock(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
spool = await spool_factory(material_number="15", label_weight=1000, weight_used=250)
|
|
await self._usage(db_session, spool.id, days_ago=200, grams=250, cost=5.0)
|
|
|
|
since = (datetime.now(timezone.utc) - timedelta(days=30)).date().isoformat()
|
|
resp = await async_client.get(f"/api/v1/inventory/stats/material-numbers?date_from={since}")
|
|
row = resp.json()[0]
|
|
assert row["consumed_g"] == 0
|
|
assert row["remaining_g"] == pytest.approx(750)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_ties_sort_by_number_so_the_order_is_stable(
|
|
self, async_client: AsyncClient, spool_factory, db_session: AsyncSession
|
|
):
|
|
"""Equal consumption has to fall back to the number, not to row order.
|
|
|
|
The response is assembled in two passes — active spools first, then
|
|
the numbers that only appear in usage history — so "16" (which has a
|
|
live spool) is seeded before "15" (archived, usage only). Without the
|
|
number tie-break the endpoint hands that seeding order straight back.
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
live = await spool_factory(material_number="16", color_name="Black")
|
|
archived = await spool_factory(material_number="15", archived_at=datetime.now(timezone.utc))
|
|
db_session.add_all(
|
|
[
|
|
SpoolUsageHistory(spool_id=live.id, weight_used=100, percent_used=10, status="completed", cost=2.0),
|
|
SpoolUsageHistory(spool_id=archived.id, weight_used=100, percent_used=10, status="completed", cost=2.0),
|
|
]
|
|
)
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.get("/api/v1/inventory/stats/material-numbers")
|
|
rows = resp.json()
|
|
assert [r["consumed_g"] for r in rows] == [pytest.approx(100), pytest.approx(100)]
|
|
assert [r["material_number"] for r in rows] == ["15", "16"]
|
|
|
|
|
|
class TestMaterialNumberCsv:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_export_import_round_trip(self, async_client: AsyncClient, spool_factory, db_session: AsyncSession):
|
|
await spool_factory(material_number="15")
|
|
|
|
export = await async_client.get("/api/v1/inventory/spools/export")
|
|
assert export.status_code == 200
|
|
text = export.text
|
|
header = text.splitlines()[0]
|
|
assert "material_number" in header.split(",")
|
|
assert ",15" in text.splitlines()[1] or text.splitlines()[1].endswith("15")
|
|
|
|
# Wipe and re-import: the number must survive the round trip.
|
|
from sqlalchemy import delete
|
|
|
|
await db_session.execute(delete(Spool))
|
|
await db_session.commit()
|
|
|
|
imported = await async_client.post(
|
|
"/api/v1/inventory/spools/import",
|
|
files={"file": ("spools.csv", text.encode("utf-8"), "text/csv")},
|
|
)
|
|
assert imported.status_code == 200, imported.text
|
|
assert imported.json()["created"] == 1
|
|
|
|
listing = await async_client.get("/api/v1/inventory/spools")
|
|
assert [s["material_number"] for s in listing.json()] == ["15"]
|