|
|
@@ -0,0 +1,242 @@
|
|
|
+"""Unit tests for the AMS weight sync calculation logic.
|
|
|
+
|
|
|
+Tests the weight_used calculation and remain% validation extracted from
|
|
|
+the POST /inventory/sync-ams-weights endpoint, without requiring a database.
|
|
|
+"""
|
|
|
+
|
|
|
+import pytest
|
|
|
+
|
|
|
+from backend.app.api.routes.inventory import _find_tray_in_ams_data
|
|
|
+
|
|
|
+
|
|
|
+def _calc_weight_used(label_weight: int | None, remain: int) -> float:
|
|
|
+ """Reproduce the weight calculation from sync_weights_from_ams."""
|
|
|
+ lw = label_weight or 1000
|
|
|
+ return round(lw * (100 - remain) / 100.0, 1)
|
|
|
+
|
|
|
+
|
|
|
+def _is_valid_remain(remain_raw) -> tuple[bool, int]:
|
|
|
+ """Reproduce the remain% validation from sync_weights_from_ams.
|
|
|
+
|
|
|
+ Returns (is_valid, parsed_value). parsed_value is only meaningful
|
|
|
+ when is_valid is True.
|
|
|
+ """
|
|
|
+ if remain_raw is None:
|
|
|
+ return False, 0
|
|
|
+ try:
|
|
|
+ val = int(remain_raw)
|
|
|
+ except (TypeError, ValueError):
|
|
|
+ return False, 0
|
|
|
+ if val < 0 or val > 100:
|
|
|
+ return False, val
|
|
|
+ return True, val
|
|
|
+
|
|
|
+
|
|
|
+class TestWeightCalculation:
|
|
|
+ """Test the weight_used = label_weight * (100 - remain) / 100 formula."""
|
|
|
+
|
|
|
+ def test_remain_100_means_no_usage(self):
|
|
|
+ """A full spool (remain=100) should have weight_used=0."""
|
|
|
+ assert _calc_weight_used(1000, 100) == 0.0
|
|
|
+
|
|
|
+ def test_remain_50_with_1000g_spool(self):
|
|
|
+ """Half-used 1000g spool should have weight_used=500."""
|
|
|
+ assert _calc_weight_used(1000, 50) == 500.0
|
|
|
+
|
|
|
+ def test_remain_0_means_fully_used(self):
|
|
|
+ """An empty spool (remain=0) should have weight_used equal to label_weight.
|
|
|
+
|
|
|
+ Unlike the on_ams_change guard, the sync endpoint processes remain=0
|
|
|
+ since it is a manual recovery tool.
|
|
|
+ """
|
|
|
+ assert _calc_weight_used(1000, 0) == 1000.0
|
|
|
+
|
|
|
+ def test_respects_label_weight_500g(self):
|
|
|
+ """500g spool at remain=50 should have weight_used=250."""
|
|
|
+ assert _calc_weight_used(500, 50) == 250.0
|
|
|
+
|
|
|
+ def test_respects_label_weight_250g(self):
|
|
|
+ """250g spool at remain=75 should have weight_used=62.5."""
|
|
|
+ assert _calc_weight_used(250, 75) == 62.5
|
|
|
+
|
|
|
+ def test_none_label_weight_defaults_to_1000(self):
|
|
|
+ """When label_weight is None, it defaults to 1000g."""
|
|
|
+ assert _calc_weight_used(None, 50) == 500.0
|
|
|
+
|
|
|
+ def test_result_is_rounded_to_one_decimal(self):
|
|
|
+ """Weight used should be rounded to 1 decimal place.
|
|
|
+
|
|
|
+ For a 1000g spool at remain=33, weight_used = 1000 * 67 / 100 = 670.0
|
|
|
+ """
|
|
|
+ assert _calc_weight_used(1000, 33) == 670.0
|
|
|
+
|
|
|
+ def test_odd_fraction_rounds_correctly(self):
|
|
|
+ """750g spool at remain=33 → 750 * 67/100 = 502.5."""
|
|
|
+ assert _calc_weight_used(750, 33) == 502.5
|
|
|
+
|
|
|
+ def test_small_spool_small_remain(self):
|
|
|
+ """200g spool at remain=1 → 200 * 99/100 = 198.0."""
|
|
|
+ assert _calc_weight_used(200, 1) == 198.0
|
|
|
+
|
|
|
+
|
|
|
+class TestRemainValidation:
|
|
|
+ """Test the remain% bounds and type validation."""
|
|
|
+
|
|
|
+ def test_remain_minus_1_is_invalid(self):
|
|
|
+ """remain=-1 (firmware 'unknown') should be skipped."""
|
|
|
+ valid, _ = _is_valid_remain(-1)
|
|
|
+ assert valid is False
|
|
|
+
|
|
|
+ def test_remain_101_is_invalid(self):
|
|
|
+ """remain=101 (out of range) should be skipped."""
|
|
|
+ valid, _ = _is_valid_remain(101)
|
|
|
+ assert valid is False
|
|
|
+
|
|
|
+ def test_remain_negative_large_is_invalid(self):
|
|
|
+ """Large negative remain values should be skipped."""
|
|
|
+ valid, _ = _is_valid_remain(-50)
|
|
|
+ assert valid is False
|
|
|
+
|
|
|
+ def test_remain_200_is_invalid(self):
|
|
|
+ """remain=200 should be skipped."""
|
|
|
+ valid, _ = _is_valid_remain(200)
|
|
|
+ assert valid is False
|
|
|
+
|
|
|
+ def test_remain_none_is_invalid(self):
|
|
|
+ """remain=None (missing from tray data) should be skipped."""
|
|
|
+ valid, _ = _is_valid_remain(None)
|
|
|
+ assert valid is False
|
|
|
+
|
|
|
+ def test_remain_non_numeric_string_is_invalid(self):
|
|
|
+ """Non-numeric string remain should be skipped."""
|
|
|
+ valid, _ = _is_valid_remain("abc")
|
|
|
+ assert valid is False
|
|
|
+
|
|
|
+ def test_remain_0_is_valid(self):
|
|
|
+ """remain=0 should be valid (manual recovery handles empty spools)."""
|
|
|
+ valid, val = _is_valid_remain(0)
|
|
|
+ assert valid is True
|
|
|
+ assert val == 0
|
|
|
+
|
|
|
+ def test_remain_100_is_valid(self):
|
|
|
+ """remain=100 should be valid."""
|
|
|
+ valid, val = _is_valid_remain(100)
|
|
|
+ assert valid is True
|
|
|
+ assert val == 100
|
|
|
+
|
|
|
+ def test_remain_50_is_valid(self):
|
|
|
+ """remain=50 should be valid."""
|
|
|
+ valid, val = _is_valid_remain(50)
|
|
|
+ assert valid is True
|
|
|
+ assert val == 50
|
|
|
+
|
|
|
+ def test_remain_string_number_is_valid(self):
|
|
|
+ """Numeric string remain (e.g. '75') should be parsed as int."""
|
|
|
+ valid, val = _is_valid_remain("75")
|
|
|
+ assert valid is True
|
|
|
+ assert val == 75
|
|
|
+
|
|
|
+
|
|
|
+class TestFindTrayInAmsData:
|
|
|
+ """Test the _find_tray_in_ams_data helper used by the sync endpoint."""
|
|
|
+
|
|
|
+ def test_finds_matching_tray(self):
|
|
|
+ """Should return the matching tray dict."""
|
|
|
+ ams_data = [
|
|
|
+ {
|
|
|
+ "id": 0,
|
|
|
+ "tray": [
|
|
|
+ {"id": 0, "remain": 80},
|
|
|
+ {"id": 1, "remain": 50},
|
|
|
+ ],
|
|
|
+ },
|
|
|
+ ]
|
|
|
+ tray = _find_tray_in_ams_data(ams_data, ams_id=0, tray_id=1)
|
|
|
+ assert tray is not None
|
|
|
+ assert tray["remain"] == 50
|
|
|
+
|
|
|
+ def test_returns_none_for_missing_ams_unit(self):
|
|
|
+ """Should return None when the AMS unit ID is not found."""
|
|
|
+ ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]
|
|
|
+ assert _find_tray_in_ams_data(ams_data, ams_id=1, tray_id=0) is None
|
|
|
+
|
|
|
+ def test_returns_none_for_missing_tray(self):
|
|
|
+ """Should return None when the tray ID is not found."""
|
|
|
+ ams_data = [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]
|
|
|
+ assert _find_tray_in_ams_data(ams_data, ams_id=0, tray_id=3) is None
|
|
|
+
|
|
|
+ def test_returns_none_for_empty_data(self):
|
|
|
+ """Should return None for empty AMS data."""
|
|
|
+ assert _find_tray_in_ams_data([], ams_id=0, tray_id=0) is None
|
|
|
+
|
|
|
+ def test_returns_none_for_none_data(self):
|
|
|
+ """Should return None for None AMS data."""
|
|
|
+ assert _find_tray_in_ams_data(None, ams_id=0, tray_id=0) is None
|
|
|
+
|
|
|
+ def test_multi_ams_unit_lookup(self):
|
|
|
+ """Should find trays across multiple AMS units."""
|
|
|
+ ams_data = [
|
|
|
+ {"id": 0, "tray": [{"id": 0, "remain": 80}]},
|
|
|
+ {"id": 1, "tray": [{"id": 2, "remain": 30}]},
|
|
|
+ ]
|
|
|
+ tray = _find_tray_in_ams_data(ams_data, ams_id=1, tray_id=2)
|
|
|
+ assert tray is not None
|
|
|
+ assert tray["remain"] == 30
|
|
|
+
|
|
|
+ def test_ams_ht_high_id(self):
|
|
|
+ """Should find trays in AMS-HT units (id >= 128)."""
|
|
|
+ ams_data = [{"id": 128, "tray": [{"id": 0, "remain": 65}]}]
|
|
|
+ tray = _find_tray_in_ams_data(ams_data, ams_id=128, tray_id=0)
|
|
|
+ assert tray is not None
|
|
|
+ assert tray["remain"] == 65
|
|
|
+
|
|
|
+
|
|
|
+class TestSyncSkipLogic:
|
|
|
+ """Test combinations that exercise the sync/skip decision path."""
|
|
|
+
|
|
|
+ def test_same_value_is_skipped(self):
|
|
|
+ """When old weight_used matches new, the spool is skipped (no DB write)."""
|
|
|
+ # Simulating the endpoint logic: if round(old_used, 1) == new_used → skip
|
|
|
+ label_weight = 1000
|
|
|
+ remain = 50
|
|
|
+ new_used = _calc_weight_used(label_weight, remain)
|
|
|
+ old_used = 500.0 # Already matches
|
|
|
+ assert round(old_used, 1) == new_used # → would be skipped
|
|
|
+
|
|
|
+ def test_different_value_is_synced(self):
|
|
|
+ """When old weight_used differs from new, the spool is synced."""
|
|
|
+ label_weight = 1000
|
|
|
+ remain = 50
|
|
|
+ new_used = _calc_weight_used(label_weight, remain)
|
|
|
+ old_used = 300.0 # Different
|
|
|
+ assert round(old_used, 1) != new_used # → would be synced
|
|
|
+
|
|
|
+ def test_none_old_used_treated_as_zero(self):
|
|
|
+ """When old weight_used is None (new spool), it defaults to 0."""
|
|
|
+ old_used = None
|
|
|
+ effective_old = old_used or 0
|
|
|
+ new_used = _calc_weight_used(1000, 80) # 200.0
|
|
|
+ assert effective_old == 0
|
|
|
+ assert round(effective_old, 1) != new_used # → would be synced
|
|
|
+
|
|
|
+ def test_remain_0_synced_not_skipped(self):
|
|
|
+ """remain=0 is valid and produces weight_used=label_weight.
|
|
|
+
|
|
|
+ This is distinct from on_ams_change behavior where remain=0 is
|
|
|
+ ignored. The sync endpoint processes it as a manual recovery tool.
|
|
|
+ """
|
|
|
+ valid, val = _is_valid_remain(0)
|
|
|
+ assert valid is True
|
|
|
+ new_used = _calc_weight_used(1000, val)
|
|
|
+ assert new_used == 1000.0
|
|
|
+
|
|
|
+ def test_remain_minus_1_never_reaches_calc(self):
|
|
|
+ """remain=-1 fails validation before weight calculation."""
|
|
|
+ valid, _ = _is_valid_remain(-1)
|
|
|
+ assert valid is False
|
|
|
+ # The endpoint would skip += 1 and continue
|
|
|
+
|
|
|
+ def test_remain_101_never_reaches_calc(self):
|
|
|
+ """remain=101 fails validation before weight calculation."""
|
|
|
+ valid, _ = _is_valid_remain(101)
|
|
|
+ assert valid is False
|
maziggy