Add integration tests for new measure endpoint

spoolman/api/v1/spool.py

@@ -17,7 +17,7 @@ from spoolman.api.v1.models import Message, Spool, SpoolEvent
 from spoolman.database import spool
 from spoolman.database.database import get_db_session
 from spoolman.database.utils import SortOrder
-from spoolman.exceptions import ItemCreateError
+from spoolman.exceptions import ItemCreateError, SpoolMeasureError
 from spoolman.extra_fields import EntityType, get_extra_fields, validate_extra_field_dict
 from spoolman.ws import websocket_manager
 
@@ -514,5 +514,12 @@ async def measure(  # noqa: ANN201
     spool_id: int,
     body: SpoolMeasureParameters,
 ):
+    try:
         db_item = await spool.measure(db, spool_id, body.weight)
         return Spool.from_db(db_item)
+    except SpoolMeasureError as e:
+        logger.exception("Failed to update spool measurement.")
+        return JSONResponse(
+            status_code=400,
+            content={"message": e.args[0]},
+        )

spoolman/database/spool.py

@@ -20,7 +20,7 @@ from spoolman.database.utils import (
     add_where_clause_str_opt,
     parse_nested_field,
 )
-from spoolman.exceptions import ItemCreateError, ItemNotFoundError
+from spoolman.exceptions import ItemCreateError, ItemNotFoundError, SpoolMeasureError
 from spoolman.math import weight_from_length
 from spoolman.ws import websocket_manager
 
@@ -339,17 +339,19 @@ async def measure(db: AsyncSession, spool_id: int, weight: float) -> models.Spoo
 
     """
     spool_result = await db.execute(
-        sqlalchemy.select(models.Spool.initial_weight, models.Spool.used_weight).where(models.Spool.id == spool_id),
+        sqlalchemy.select(models.Spool.initial_weight, models.Spool.used_weight, models.Spool.empty_weight).where(
+            models.Spool.id == spool_id,
+        ),
     )
 
     try:
         spool_info = spool_result.one()
     except NoResultFound as exc:
-        raise ItemNotFoundError("Spool not found.") from exc
+        raise SpoolMeasureError("Spool not found.") from exc
 
     initial_weight = spool_info[0]
-
+    empty_weight = spool_info[2]
-    if initial_weight is None:
+    if initial_weight is None or empty_weight is None:
         # Get filament weight and spool_weight
         result = await db.execute(
             sqlalchemy.select(models.Filament.weight, models.Filament.spool_weight)
@@ -360,13 +362,25 @@ async def measure(db: AsyncSession, spool_id: int, weight: float) -> models.Spoo
             filament_info = result.one()
         except NoResultFound as exc:
             raise ItemNotFoundError("Filament not found for spool.") from exc
-        initial_weight = filament_info[0] + filament_info[1]
+        if empty_weight is None:
+            empty_weight = filament_info[1]
+
+        if initial_weight is None:
+            initial_weight = filament_info[0] + empty_weight
+
+    # if the measurement is greater than the initial weight, raise an error
+    if weight > initial_weight:
+        raise SpoolMeasureError("Measured weight is greater than the initial weight of the spool.")
 
     # Calculate the current gross weight (initial_weight - used_weight)
     current_use = initial_weight - spool_info[1]
     # Calculate the weight used since last measure
     weight_to_use = current_use - weight
 
+    # If the measured weight is less than the empty weight, use the rest of the spool
+    if (initial_weight - weight_to_use) < empty_weight:
+        weight_to_use = current_use - empty_weight
+
     return await use_weight(db, spool_id, weight_to_use)
 
 

spoolman/exceptions.py

@@ -11,3 +11,7 @@ class ItemDeleteError(Exception):
 
 class ItemCreateError(Exception):
     pass
+
+
+class SpoolMeasureError(Exception):
+    pass

tests_integration/tests/spool/test_measure.py

@@ -0,0 +1,91 @@
+"""Integration tests for the Spool API endpoint."""
+
+from datetime import datetime, timezone
+from typing import Any
+
+import httpx
+import pytest
+
+from ..conftest import URL
+
+
+@pytest.mark.parametrize("measurement", [0, 0.05, -0.05, 1000])
+def test_measure_spool(random_filament: dict[str, Any], measurement: float):
+    """Test using a spool in the database."""
+    # Setup
+    random_filament["weight"]
+    initial_weight = 1255
+    empty_weight = 246
+    start_weight = 1000
+    result = httpx.post(
+        f"{URL}/api/v1/spool",
+        json={
+            "filament_id": random_filament["id"],
+            "remaining_weight": start_weight,
+            "initial_weight": initial_weight,
+            "empty_weight": empty_weight,
+        },
+    )
+    result.raise_for_status()
+    spool = result.json()
+
+    # Execute
+    result = httpx.put(
+        f"{URL}/api/v1/spool/{spool['id']}/measure",
+        json={
+            "weight": measurement,
+        },
+    )
+    result.raise_for_status()
+
+    # Verify
+    spool = result.json()
+    # remaining_weight should be clamped so it's never negative, but used_weight should not be clamped to the net weight
+    expected_use = min(initial_weight - measurement, initial_weight - empty_weight)
+    assert spool["used_weight"] == pytest.approx(expected_use)
+    expected_remaining = max(measurement - empty_weight, 0)
+    assert spool["remaining_weight"] == pytest.approx(expected_remaining)
+    # Verify that first_used has been updated
+    diff = abs((datetime.now(tz=timezone.utc) - datetime.fromisoformat(spool["first_used"])).total_seconds())
+    assert diff < 60
+
+    # Verify that last_used has been updated
+    diff = abs((datetime.now(tz=timezone.utc) - datetime.fromisoformat(spool["last_used"])).total_seconds())
+    assert diff < 60
+
+    # Clean up
+    httpx.delete(f"{URL}/api/v1/spool/{spool['id']}").raise_for_status()
+
+
+@pytest.mark.parametrize("measurement", [1500])  # 1500 is more than the initial weight
+def test_measure_spool_invalid(random_filament: dict[str, Any], measurement: float):
+    """Test using a spool in the database."""
+    # Setup
+    random_filament["weight"]
+    initial_weight = 1255
+    empty_weight = 246
+    start_weight = 1000
+    result = httpx.post(
+        f"{URL}/api/v1/spool",
+        json={
+            "filament_id": random_filament["id"],
+            "remaining_weight": start_weight,
+            "initial_weight": initial_weight,
+            "empty_weight": empty_weight,
+        },
+    )
+    result.raise_for_status()
+    spool = result.json()
+
+    # Execute
+    result = httpx.put(
+        f"{URL}/api/v1/spool/{spool['id']}/measure",
+        json={
+            "weight": measurement,
+        },
+    )
+    # Update is invalid if the weight is more than the initial weight
+    assert result.status_code == 400
+
+    # Clean up
+    httpx.delete(f"{URL}/api/v1/spool/{spool['id']}").raise_for_status()