|
|
@@ -0,0 +1,135 @@
|
|
|
+"""The built-in inventory's AMS weight sync, the weight lock and a printing printer.
|
|
|
+
|
|
|
+A weight set by hand or measured on the SpoolBuddy scale is locked, so the
|
|
|
+automatic remain% sync leaves it alone. "Sync Weights from AMS" is the explicit
|
|
|
+override: it also updates locked spools and releases the lock. It leaves the
|
|
|
+spools of a printing printer alone, whose usage the print's completion charges.
|
|
|
+"""
|
|
|
+
|
|
|
+from datetime import datetime, timezone
|
|
|
+from types import SimpleNamespace
|
|
|
+from unittest.mock import MagicMock, patch
|
|
|
+
|
|
|
+import pytest
|
|
|
+from httpx import AsyncClient
|
|
|
+from sqlalchemy import select
|
|
|
+
|
|
|
+from backend.app.models.spool import Spool
|
|
|
+from backend.app.models.spool_assignment import SpoolAssignment
|
|
|
+from backend.app.models.spoolbuddy_device import SpoolBuddyDevice
|
|
|
+
|
|
|
+
|
|
|
+async def _assigned_spool(db_session, printer_factory, *, locked: bool) -> int:
|
|
|
+ printer = await printer_factory()
|
|
|
+ spool = Spool(material="PLA", label_weight=1000, core_weight=250, weight_used=100, weight_locked=locked)
|
|
|
+ db_session.add(spool)
|
|
|
+ await db_session.flush()
|
|
|
+ db_session.add(SpoolAssignment(spool_id=spool.id, printer_id=printer.id, ams_id=0, tray_id=0))
|
|
|
+ await db_session.commit()
|
|
|
+ return spool.id
|
|
|
+
|
|
|
+
|
|
|
+def _printer(state: str, remain: int = 50, connected: bool = True):
|
|
|
+ return SimpleNamespace(
|
|
|
+ state=state, connected=connected, raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": remain}]}]}
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def _sync(async_client, state):
|
|
|
+ with patch("backend.app.services.printer_manager.printer_manager") as pm:
|
|
|
+ pm.get_status = MagicMock(return_value=state)
|
|
|
+ return await async_client.post("/api/v1/inventory/sync-ams-weights")
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.integration
|
|
|
+async def test_the_sync_updates_a_locked_spool_and_releases_the_lock(
|
|
|
+ async_client: AsyncClient, db_session, printer_factory
|
|
|
+):
|
|
|
+ spool_id = await _assigned_spool(db_session, printer_factory, locked=True)
|
|
|
+
|
|
|
+ response = await _sync(async_client, _printer("IDLE"))
|
|
|
+
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert response.json()["synced"] == 1
|
|
|
+ spool = await db_session.get(Spool, spool_id)
|
|
|
+ await db_session.refresh(spool)
|
|
|
+ assert (spool.weight_used, spool.weight_locked) == (500, False)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.integration
|
|
|
+@pytest.mark.parametrize("gcode_state", ["PREPARE", "RUNNING", "PAUSE"])
|
|
|
+async def test_a_printing_printers_spool_is_left_alone(
|
|
|
+ async_client: AsyncClient, db_session, printer_factory, gcode_state
|
|
|
+):
|
|
|
+ spool_id = await _assigned_spool(db_session, printer_factory, locked=False)
|
|
|
+
|
|
|
+ response = await _sync(async_client, _printer(gcode_state))
|
|
|
+
|
|
|
+ assert (response.json()["synced"], response.json()["printing"]) == (0, 1)
|
|
|
+ spool = await db_session.get(Spool, spool_id)
|
|
|
+ await db_session.refresh(spool)
|
|
|
+ assert spool.weight_used == 100
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.integration
|
|
|
+async def test_weighing_on_the_scale_locks_the_spool(async_client: AsyncClient, db_session):
|
|
|
+ db_session.add(
|
|
|
+ SpoolBuddyDevice(
|
|
|
+ device_id="sb-1",
|
|
|
+ hostname="spoolbuddy-1",
|
|
|
+ ip_address="10.0.0.1",
|
|
|
+ last_seen=datetime.now(timezone.utc),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ spool = Spool(material="PLA", label_weight=1000, core_weight=250, weight_used=100)
|
|
|
+ db_session.add(spool)
|
|
|
+ await db_session.commit()
|
|
|
+
|
|
|
+ response = await async_client.post(
|
|
|
+ "/api/v1/spoolbuddy/scale/update-spool-weight", json={"spool_id": spool.id, "weight_grams": 600}
|
|
|
+ )
|
|
|
+
|
|
|
+ assert response.status_code == 200
|
|
|
+ await db_session.refresh(spool)
|
|
|
+ assert (spool.weight_used, spool.weight_locked) == (650, True)
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.integration
|
|
|
+async def test_a_print_whose_completion_is_still_being_charged_counts_as_printing(
|
|
|
+ async_client: AsyncClient, db_session, printer_factory
|
|
|
+):
|
|
|
+ """The printer already reports FINISH; the usage session is still open
|
|
|
+ until completion has charged the print."""
|
|
|
+ from backend.app.services.usage_tracker import PrintSession, _active_sessions
|
|
|
+
|
|
|
+ spool_id = await _assigned_spool(db_session, printer_factory, locked=False)
|
|
|
+ printer_id = await db_session.scalar(select(SpoolAssignment.printer_id).where(SpoolAssignment.spool_id == spool_id))
|
|
|
+ _active_sessions[printer_id] = PrintSession(
|
|
|
+ printer_id=printer_id, print_name="x", started_at=datetime.now(timezone.utc)
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ response = await _sync(async_client, _printer("FINISH"))
|
|
|
+ finally:
|
|
|
+ _active_sessions.pop(printer_id, None)
|
|
|
+
|
|
|
+ assert (response.json()["synced"], response.json()["printing"]) == (0, 1)
|
|
|
+ spool = await db_session.get(Spool, spool_id)
|
|
|
+ await db_session.refresh(spool)
|
|
|
+ assert spool.weight_used == 100
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.asyncio
|
|
|
+@pytest.mark.integration
|
|
|
+async def test_a_disconnected_printers_last_report_is_not_used(async_client: AsyncClient, db_session, printer_factory):
|
|
|
+ spool_id = await _assigned_spool(db_session, printer_factory, locked=False)
|
|
|
+
|
|
|
+ response = await _sync(async_client, _printer("unknown", connected=False))
|
|
|
+
|
|
|
+ assert response.json()["synced"] == 0
|
|
|
+ spool = await db_session.get(Spool, spool_id)
|
|
|
+ await db_session.refresh(spool)
|
|
|
+ assert spool.weight_used == 100
|