mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
fb3821630f
Bulk operations on the Inventory page in both built-in and Spoolman modes.
Reporter wanted ten-of-the-same-spool edits without ten round-trips through
the per-spool editor.
Frontend
- New checkbox column on the inventory table (header / row / group). Sticky
toolbar appears when at least one row is selected with Edit / Print labels /
Reset usage / Archive (or Restore in the Archived tab) / Delete / Clear.
Selection clears on any filter / tab / search change so the count can't
drift from what is on screen.
- BulkEditSpoolsModal is a three-state-per-field form. The user opts in per
field by ticking its checkbox or just typing into it; only ticked + non-
empty fields are sent. Clearing fields in bulk is intentionally NOT
supported per the issue discussion.
- A new SearchableSelect renders all categorical fields (material, sub-type,
brand, category, slicer preset name, slicer filament, storage location)
with the same dropdown pattern the per-spool editor uses - text input +
chevron + filtered button list, click-outside / Escape closes. No native
select anywhere in the modal. Options merge the canonical constants from
spool-form/constants.ts with whatever already exists in the user's
inventory. Slicer-preset dropdowns fetch the same sources as the per-spool
form (Bambu Cloud + Orca Cloud + local + built-in) through buildFilament
Options() and three useQuery calls gated on isOpen.
- onSuccess handlers surface three outcomes: all-succeeded (green toast),
partial-success (yellow toast with ok / failed counts), all-failed (red
toast that keeps the selection and modal open so the user can retry).
The first cut silently dropped errors / not_found arrays - audited and
fixed before merge.
- Invalid rgba hex is flagged inline with a red border + helper text and
the Apply button is gated on a hasDroppedTickedField guard, so silently
dropping a ticked field is no longer possible.
- bulkResetConsumedCounterMutation.onSuccess now closes the confirm modal +
clears selection, matching the other three bulk mutations.
Backend
- Four new endpoints per inventory mode (eight total):
POST /api/v1/inventory/spools/bulk-update INVENTORY_UPDATE
POST /api/v1/inventory/spools/bulk-delete INVENTORY_UPDATE
POST /api/v1/inventory/spools/bulk-archive INVENTORY_UPDATE
POST /api/v1/inventory/spools/bulk-restore INVENTORY_UPDATE
POST /api/v1/spoolman/inventory/spools/bulk-* FILAMENTS_UPDATE
- Built-in update runs the same prepare_internal_spool_payload(...) +
weight_used / weight_locked auto-stamp as the per-spool PATCH.
- Spoolman update loops the per-spool update_spool route function so the
filament re-linking / extra-dict / extra-lock / shared-filament rules
stay byte-identical to single-spool edits.
- Per-spool failures inside the batch are collected. Spoolman bulk-delete /
archive / restore now catch non-HTTPException too (matches bulk-update) -
a mid-batch httpx.ConnectError or TimeoutError no longer aborts the route
with a 500 and skips the WS broadcast.
- Both modes broadcast a single inventory_changed WS event at the end of
the batch.
213 lines
7.1 KiB
Python
213 lines
7.1 KiB
Python
"""Bulk inventory endpoint coverage for the batch-edit feature (#1795).
|
|
|
|
Endpoints under test:
|
|
- POST /api/v1/inventory/spools/bulk-update
|
|
- POST /api/v1/inventory/spools/bulk-delete
|
|
- POST /api/v1/inventory/spools/bulk-archive
|
|
- POST /api/v1/inventory/spools/bulk-restore
|
|
|
|
The Spoolman-mode equivalents live in test_spoolman_inventory_api.py.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from backend.app.models.spool import Spool
|
|
|
|
|
|
@pytest.fixture
|
|
async def spool_factory(db_session: AsyncSession):
|
|
async def _create(**kwargs):
|
|
defaults = {
|
|
"material": "PLA",
|
|
"subtype": "Basic",
|
|
"brand": "Bambu",
|
|
"color_name": "Red",
|
|
"rgba": "FF0000FF",
|
|
"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 TestBulkUpdate:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_applies_patch_to_all_listed_spools(self, async_client: AsyncClient, spool_factory, db_session):
|
|
a = await spool_factory(brand="Bambu", note=None)
|
|
b = await spool_factory(brand="Bambu", note=None)
|
|
c = await spool_factory(brand="Bambu", note=None)
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [a.id, b.id, c.id], "update": {"brand": "Sunlu", "note": "From bulk edit"}},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["updated"] == 3
|
|
assert body["not_found"] == []
|
|
|
|
for spool in (a, b, c):
|
|
await db_session.refresh(spool)
|
|
assert spool.brand == "Sunlu"
|
|
assert spool.note == "From bulk edit"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_reports_unknown_ids_in_not_found(self, async_client: AsyncClient, spool_factory, db_session):
|
|
real = await spool_factory(brand="Bambu")
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [real.id, 999_999], "update": {"brand": "Sunlu"}},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["updated"] == 1
|
|
assert body["not_found"] == [999_999]
|
|
|
|
await db_session.refresh(real)
|
|
assert real.brand == "Sunlu"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_empty_update_rejected(self, async_client: AsyncClient, spool_factory):
|
|
a = await spool_factory()
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [a.id], "update": {}},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_setting_weight_used_auto_locks(self, async_client: AsyncClient, spool_factory, db_session):
|
|
a = await spool_factory(weight_locked=False, weight_used=0.0)
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [a.id], "update": {"weight_used": 250.5}},
|
|
)
|
|
assert resp.status_code == 200
|
|
await db_session.refresh(a)
|
|
assert a.weight_used == 250.5
|
|
assert a.weight_locked is True
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_empty_ids_list_rejected(self, async_client: AsyncClient):
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-update",
|
|
json={"ids": [], "update": {"brand": "X"}},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
class TestBulkDelete:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_deletes_listed_spools(self, async_client: AsyncClient, spool_factory, db_session):
|
|
a = await spool_factory()
|
|
b = await spool_factory()
|
|
kept = await spool_factory()
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-delete",
|
|
json={"ids": [a.id, b.id]},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["deleted"] == 2
|
|
assert body["not_found"] == []
|
|
|
|
remaining = (await db_session.execute(select(Spool.id))).scalars().all()
|
|
assert kept.id in remaining
|
|
assert a.id not in remaining
|
|
assert b.id not in remaining
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_reports_unknown_ids(self, async_client: AsyncClient, spool_factory):
|
|
a = await spool_factory()
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-delete",
|
|
json={"ids": [a.id, 999_999]},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["deleted"] == 1
|
|
assert body["not_found"] == [999_999]
|
|
|
|
|
|
class TestBulkArchiveRestore:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_bulk_archive_sets_archived_at(self, async_client: AsyncClient, spool_factory, db_session):
|
|
a = await spool_factory()
|
|
b = await spool_factory()
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-archive",
|
|
json={"ids": [a.id, b.id]},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["archived"] == 2
|
|
assert body["already_archived"] == []
|
|
assert body["not_found"] == []
|
|
|
|
for s in (a, b):
|
|
await db_session.refresh(s)
|
|
assert s.archived_at is not None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_bulk_archive_skips_already_archived(self, async_client: AsyncClient, spool_factory, db_session):
|
|
active = await spool_factory()
|
|
already = await spool_factory(archived_at=datetime.now(timezone.utc))
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-archive",
|
|
json={"ids": [active.id, already.id]},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["archived"] == 1
|
|
assert body["already_archived"] == [already.id]
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_bulk_restore_clears_archived_at(self, async_client: AsyncClient, spool_factory, db_session):
|
|
archived = await spool_factory(archived_at=datetime.now(timezone.utc))
|
|
active = await spool_factory(archived_at=None)
|
|
|
|
resp = await async_client.post(
|
|
"/api/v1/inventory/spools/bulk-restore",
|
|
json={"ids": [archived.id, active.id]},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["restored"] == 1
|
|
assert body["already_active"] == [active.id]
|
|
|
|
await db_session.refresh(archived)
|
|
assert archived.archived_at is None
|