mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-10-17 19:36:28 +02:00
maziggy
b06f8f6951
notify_missing_spool_assignments_on_print_start queried only the legacy
SpoolAssignment table. In Spoolman mode that table is empty -- bindings
live in spoolman_slot_assignments -- so every used tray was flagged
missing, firing a false-positive notification on every print.
- spool_assignment_notifications.py: the assigned-tray set is now the
union of SpoolAssignment + SpoolmanSlotAssignment rows. Union-only,
so legacy-mode behavior cannot regress.
- settings.py: the Spoolman toggle cleared SpoolAssignment on switch-on
but never cleared SpoolmanSlotAssignment on switch-off. Added the
symmetric clear so stale Spoolman rows can't leak into a later
internal-mode session and mask a real missing-assignment warning.
Adds 3 notification tests + 1 mode-switch integration test. An audit
of the remaining SpoolAssignment consumers confirmed usage_tracker,
spool_tag_matcher and routes/inventory are correctly internal-mode-only.
610 lines
24 KiB
Python
610 lines
24 KiB
Python
"""Integration tests for Spoolman slot-assignment endpoints.
|
|
|
|
Tests for:
|
|
POST /api/v1/spoolman/inventory/slot-assignments
|
|
DELETE /api/v1/spoolman/inventory/slot-assignments/{spoolman_spool_id}
|
|
GET /api/v1/spoolman/inventory/slot-assignments?printer_id=&ams_id=&tray_id=
|
|
GET /api/v1/spoolman/inventory/slot-assignments/all[?printer_id=]
|
|
|
|
Slot assignments are now stored in the local ``spoolman_slot_assignments`` table.
|
|
Spoolman's ``spool.location`` field is NOT touched by any of these endpoints.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
SAMPLE_SPOOL = {
|
|
"id": 10,
|
|
"filament": {
|
|
"id": 1,
|
|
"name": "PLA Basic",
|
|
"material": "PLA",
|
|
"color_hex": "FF0000",
|
|
"weight": 1000,
|
|
"vendor": {"id": 1, "name": "Test Brand"},
|
|
},
|
|
"remaining_weight": 800.0,
|
|
"used_weight": 200.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 slot_settings(db_session):
|
|
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
|
|
async def test_printer(db_session):
|
|
from backend.app.models.printer import Printer
|
|
|
|
printer = Printer(
|
|
name="Test Printer",
|
|
serial_number="SLOTTEST001",
|
|
ip_address="192.168.1.100",
|
|
access_code="12345678",
|
|
)
|
|
db_session.add(printer)
|
|
await db_session.commit()
|
|
await db_session.refresh(printer)
|
|
return printer
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_client():
|
|
client = MagicMock()
|
|
client.base_url = "http://localhost:7912"
|
|
client.health_check = AsyncMock(return_value=True)
|
|
client.get_spool = AsyncMock(return_value=SAMPLE_SPOOL)
|
|
# #1457: assign route enumerates spools to clear stale fallback-tag links.
|
|
# Default to empty so the cleanup is a no-op for tests that don't exercise it.
|
|
client.get_spools = AsyncMock(return_value=[])
|
|
client.merge_spool_extra = AsyncMock(return_value={"id": 0, "extra": {}})
|
|
|
|
with patch(
|
|
"backend.app.api.routes.spoolman_inventory._get_client",
|
|
AsyncMock(return_value=client),
|
|
):
|
|
yield client
|
|
|
|
|
|
class TestAssignSpoolmanSlot:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_inserts_local_row(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
|
|
"""POST /slot-assignments creates a row visible via the /all endpoint."""
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 10,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
all_resp = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments/all",
|
|
params={"printer_id": test_printer.id},
|
|
)
|
|
assert all_resp.status_code == 200
|
|
rows = all_resp.json()
|
|
assert len(rows) == 1
|
|
assert rows[0]["spoolman_spool_id"] == 10
|
|
assert rows[0]["ams_id"] == 0
|
|
assert rows[0]["tray_id"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_accepts_ams_ht_id(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
|
|
"""#1274: AMS-HT units report ams_id 128+. The pre-fix ck_ams_id_range
|
|
only allowed 0-7 / 255, so the upsert blew up with `CHECK constraint
|
|
failed: ck_ams_id_range` and the user couldn't link any spool to the
|
|
H2C/H2D AMS-HT slot. This guards the widened range from regressing.
|
|
"""
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 51,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 128, # AMS-HT on the left nozzle (matches issue's failing INSERT)
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
all_resp = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments/all",
|
|
params={"printer_id": test_printer.id},
|
|
)
|
|
rows = all_resp.json()
|
|
assert any(r["ams_id"] == 128 and r["spoolman_spool_id"] == 51 for r in rows)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_does_not_call_update_spool(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client
|
|
):
|
|
"""POST /slot-assignments must NOT write to Spoolman's location field."""
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 10,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_client.update_spool.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_returns_inventory_spool(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client
|
|
):
|
|
"""POST /slot-assignments response is mapped to InventorySpool format."""
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 10,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == 10
|
|
assert body["material"] == "PLA"
|
|
assert body["data_origin"] == "spoolman"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_upserts_on_conflict(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client
|
|
):
|
|
"""POST /slot-assignments twice for the same slot replaces the old spool ID."""
|
|
# First assign spool 99
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 99,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
# Re-assign spool 10 to the same slot
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 10,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# The /all endpoint must report exactly one row for this slot with spool_id=10
|
|
all_resp = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments/all",
|
|
params={"printer_id": test_printer.id},
|
|
)
|
|
assert all_resp.status_code == 200
|
|
rows = all_resp.json()
|
|
matched = [r for r in rows if r["ams_id"] == 0 and r["tray_id"] == 0]
|
|
assert len(matched) == 1
|
|
assert matched[0]["spoolman_spool_id"] == 10
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_printer_not_found(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""POST /slot-assignments with unknown printer_id returns 404."""
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 10,
|
|
"printer_id": 99999,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_assign_invalid_spool_id(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
|
|
"""POST /slot-assignments with spool_id=0 returns 422 (gt=0 validation)."""
|
|
response = await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 0,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
class TestUnassignSpoolmanSlot:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unassign_deletes_local_row(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client
|
|
):
|
|
"""DELETE /slot-assignments/{id} removes the row so /all no longer lists it."""
|
|
# First assign spool 10
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": 10,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": 0,
|
|
},
|
|
)
|
|
# Then unassign
|
|
response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")
|
|
assert response.status_code == 200
|
|
|
|
# The /all endpoint must now return an empty list for this printer
|
|
all_resp = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments/all",
|
|
params={"printer_id": test_printer.id},
|
|
)
|
|
assert all_resp.status_code == 200
|
|
assert all_resp.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unassign_does_not_call_update_spool(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""DELETE /slot-assignments/{id} must NOT touch Spoolman's location field."""
|
|
response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")
|
|
|
|
assert response.status_code == 200
|
|
mock_client.update_spool.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unassign_returns_inventory_spool(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""DELETE /slot-assignments/{id} returns the spool in InventorySpool format."""
|
|
response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["id"] == 10
|
|
assert body["data_origin"] == "spoolman"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unassign_invalid_id(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""DELETE /slot-assignments/0 returns 422 (gt=0 path validation)."""
|
|
response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/0")
|
|
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_unassign_succeeds_when_spool_deleted_in_spoolman(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
|
|
):
|
|
"""DELETE /slot-assignments/{id} returns 200 even when the spool no longer exists in Spoolman.
|
|
|
|
The local row must be removed regardless — the caller should not see an error just
|
|
because Spoolman has already discarded the spool.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
|
|
from backend.app.services.spoolman import SpoolmanNotFoundError
|
|
|
|
# Create the assignment first
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
# Spool 10 has since been deleted from Spoolman
|
|
mock_client.get_spool = AsyncMock(side_effect=SpoolmanNotFoundError("spool 10 not found"))
|
|
|
|
response = await async_client.delete("/api/v1/spoolman/inventory/slot-assignments/10")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json().get("id") == 10
|
|
|
|
# Local row must be gone
|
|
result = await db_session.execute(
|
|
select(SpoolmanSlotAssignment).where(
|
|
SpoolmanSlotAssignment.printer_id == test_printer.id,
|
|
SpoolmanSlotAssignment.ams_id == 0,
|
|
SpoolmanSlotAssignment.tray_id == 0,
|
|
)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
|
|
class TestGetSpoolmanSlotAssignment:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_returns_matched_spool(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
|
|
"""GET /slot-assignments returns the spool whose ID is in the local table."""
|
|
# First assign so the row exists
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
mock_client.get_spool.reset_mock()
|
|
|
|
response = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
params={"printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body is not None
|
|
assert body["id"] == 10
|
|
mock_client.get_spool.assert_awaited_once_with(10)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_returns_null_when_no_assignment(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client
|
|
):
|
|
"""GET /slot-assignments returns null when no local row exists for the slot."""
|
|
response = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
params={"printer_id": test_printer.id, "ams_id": 1, "tray_id": 0},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_printer_not_found(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""GET /slot-assignments with unknown printer_id returns 404."""
|
|
response = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
params={"printer_id": 99999, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_missing_params(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""GET /slot-assignments without required params returns 422."""
|
|
response = await async_client.get("/api/v1/spoolman/inventory/slot-assignments")
|
|
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_returns_null_and_cleans_stale_when_spool_deleted_in_spoolman(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
|
|
):
|
|
"""GET /slot-assignments returns null and removes the stale row when Spoolman returns 404."""
|
|
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
|
|
from backend.app.services.spoolman import SpoolmanNotFoundError
|
|
|
|
# Assign spool 10 first
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
# Simulate spool 10 being deleted from Spoolman (404 via SpoolmanNotFoundError)
|
|
mock_client.get_spool = AsyncMock(side_effect=SpoolmanNotFoundError("spool 10 not found"))
|
|
|
|
response = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
params={"printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() is None
|
|
|
|
# Stale row must have been removed
|
|
await db_session.refresh(test_printer) # ensure session is fresh
|
|
result = await db_session.execute(
|
|
select(SpoolmanSlotAssignment).where(
|
|
SpoolmanSlotAssignment.printer_id == test_printer.id,
|
|
SpoolmanSlotAssignment.ams_id == 0,
|
|
SpoolmanSlotAssignment.tray_id == 0,
|
|
)
|
|
)
|
|
assert result.scalar_one_or_none() is None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_propagates_503_from_spoolman(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client
|
|
):
|
|
"""GET /slot-assignments propagates a 503 from Spoolman instead of silently returning null."""
|
|
from backend.app.services.spoolman import SpoolmanUnavailableError
|
|
|
|
# Assign spool 10 first so a local row exists
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
# Simulate Spoolman being unreachable
|
|
mock_client.get_spool = AsyncMock(side_effect=SpoolmanUnavailableError("timeout"))
|
|
|
|
response = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
params={"printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
|
|
|
|
class TestGetAllSpoolmanSlotAssignments:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_all_returns_empty_list(self, async_client: AsyncClient, slot_settings, mock_client):
|
|
"""GET /slot-assignments/all returns [] when no assignments exist."""
|
|
response = await async_client.get("/api/v1/spoolman/inventory/slot-assignments/all")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == []
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_all_returns_all_rows(self, async_client: AsyncClient, slot_settings, test_printer, mock_client):
|
|
"""GET /slot-assignments/all returns all existing assignments."""
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 20, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 1},
|
|
)
|
|
|
|
response = await async_client.get("/api/v1/spoolman/inventory/slot-assignments/all")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert len(body) == 2
|
|
spool_ids = {r["spoolman_spool_id"] for r in body}
|
|
assert spool_ids == {10, 20}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_get_all_filters_by_printer(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
|
|
):
|
|
"""GET /slot-assignments/all?printer_id=X only returns that printer's rows."""
|
|
from backend.app.models.printer import Printer
|
|
|
|
# Create a second printer via DB directly (no Spoolman mock needed for printer creation)
|
|
other = Printer(
|
|
name="Other Printer",
|
|
serial_number="SLOTTEST002",
|
|
ip_address="192.168.1.101",
|
|
access_code="87654321",
|
|
)
|
|
db_session.add(other)
|
|
await db_session.commit()
|
|
await db_session.refresh(other)
|
|
|
|
# Assign via API for test_printer
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 10, "printer_id": test_printer.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
# Assign via API for other printer
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={"spoolman_spool_id": 99, "printer_id": other.id, "ams_id": 0, "tray_id": 0},
|
|
)
|
|
|
|
response = await async_client.get(
|
|
"/api/v1/spoolman/inventory/slot-assignments/all",
|
|
params={"printer_id": test_printer.id},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert len(body) == 1
|
|
assert body[0]["spoolman_spool_id"] == 10
|
|
assert body[0]["printer_id"] == test_printer.id
|
|
|
|
|
|
class TestCascadeDeletePrinter:
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_delete_printer_removes_slot_assignments(
|
|
self, async_client: AsyncClient, slot_settings, test_printer, mock_client, db_session
|
|
):
|
|
"""DELETE /printers/{id} removes all slot assignments for that printer.
|
|
|
|
SQLite does not enforce FK cascades automatically. The delete_printer
|
|
endpoint must explicitly delete SpoolmanSlotAssignment rows so no
|
|
orphaned rows survive after the printer record is gone.
|
|
"""
|
|
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
|
|
|
|
# Assign two spools to different AMS slots on the test printer
|
|
for tray_id, spool_id in [(0, 10), (1, 20)]:
|
|
await async_client.post(
|
|
"/api/v1/spoolman/inventory/slot-assignments",
|
|
json={
|
|
"spoolman_spool_id": spool_id,
|
|
"printer_id": test_printer.id,
|
|
"ams_id": 0,
|
|
"tray_id": tray_id,
|
|
},
|
|
)
|
|
|
|
# Verify both rows exist
|
|
pre = await db_session.execute(
|
|
select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == test_printer.id)
|
|
)
|
|
assert len(pre.scalars().all()) == 2
|
|
|
|
# Delete the printer
|
|
del_resp = await async_client.delete(f"/api/v1/printers/{test_printer.id}")
|
|
assert del_resp.status_code == 200
|
|
|
|
# All slot assignment rows for the deleted printer must be gone
|
|
post = await db_session.execute(
|
|
select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == test_printer.id)
|
|
)
|
|
assert post.scalars().all() == []
|
|
|
|
|
|
class TestModeSwitchClearsAssignments:
|
|
"""#1473 follow-up — the Spoolman mode toggle clears the other mode's
|
|
slot-assignment table so stale rows can't bleed across a mode switch."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_switch_to_internal_mode_clears_spoolman_slot_assignments(
|
|
self, async_client: AsyncClient, db_session, test_printer
|
|
):
|
|
"""Switching Spoolman OFF deletes spoolman_slot_assignments rows — the
|
|
symmetric counterpart of clearing legacy spool_assignment rows when
|
|
switching ON. Stale rows would otherwise wrongly count as 'assigned'
|
|
in mode-agnostic checks (e.g. the missing-spool-assignment notification,
|
|
which unions both tables)."""
|
|
from backend.app.models.settings import Settings
|
|
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
|
|
|
|
db_session.add(Settings(key="spoolman_enabled", value="true"))
|
|
db_session.add(SpoolmanSlotAssignment(printer_id=test_printer.id, ams_id=0, tray_id=0, spoolman_spool_id=1))
|
|
await db_session.commit()
|
|
|
|
resp = await async_client.put("/api/v1/settings/spoolman", json={"spoolman_enabled": "false"})
|
|
assert resp.status_code == 200
|
|
|
|
rows = await db_session.execute(
|
|
select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.printer_id == test_printer.id)
|
|
)
|
|
assert rows.scalars().all() == []
|