提交

backend/tests/test_pl_dibang_plan_quantity.py

@@ -0,0 +1,136 @@
+"""地磅月计划量运输联动累加/回减单元测试"""
+from decimal import Decimal
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.models.db_models import DbInboundOutbound, DbPurchasePlan
+from app.models.pl_models import PlTransportDbLink
+from app.services.pl_dibang_bridge_service import (
+    _apply_plan_quantity_delta,
+    _resolve_link_contributed_qty,
+    revoke_regeneratable_dibang_for_transport,
+)
+
+
+def test_apply_plan_quantity_delta_accumulates():
+    db_plan = DbPurchasePlan(planned_quantity=30.0)
+    _apply_plan_quantity_delta(db_plan, 31, user_id="u1")
+    assert db_plan.planned_quantity == 61.0
+    assert db_plan.update_by == "u1"
+
+
+def test_apply_plan_quantity_delta_never_below_zero():
+    db_plan = DbPurchasePlan(planned_quantity=30.0)
+    _apply_plan_quantity_delta(db_plan, -50, user_id="u1")
+    assert db_plan.planned_quantity == 0.0
+
+
+def test_resolve_link_contributed_qty_prefers_link_field():
+    link = PlTransportDbLink(contributed_qty=Decimal("30"))
+    inventory = DbInboundOutbound(quantity=Decimal("99"))
+    assert _resolve_link_contributed_qty(link, inventory) == 30.0
+
+
+def test_resolve_link_contributed_qty_falls_back_to_inventory():
+    link = PlTransportDbLink(contributed_qty=None)
+    inventory = DbInboundOutbound(quantity=Decimal("31"))
+    assert _resolve_link_contributed_qty(link, inventory) == 31.0
+
+
+def test_resolve_link_contributed_qty_zero_when_missing():
+    link = PlTransportDbLink(contributed_qty=None)
+    assert _resolve_link_contributed_qty(link, None) == 0.0
+
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
+def test_revoke_subtracts_plan_quantity_on_cancel(_mock_weighing):
+    db_plan = DbPurchasePlan(id="plan-1", tenant_no="t1", planned_quantity=61.0, del_flag=0)
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        del_flag=0,
+        status="待入库",
+        quantity=Decimal("30"),
+        batch_no="CRKDH0000001",
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        transport_id="tr-1",
+        db_inventory_id="inv-1",
+        db_purchase_plan_id="plan-1",
+        contributed_qty=Decimal("30"),
+        del_flag=0,
+    )
+
+    db = MagicMock()
+    link_query = MagicMock()
+    link_query.filter.return_value.all.return_value = [link]
+
+    inv_query = MagicMock()
+    inv_query.filter.return_value.first.return_value = inventory
+
+    plan_query = MagicMock()
+    plan_query.filter.return_value.first.return_value = db_plan
+
+    def query_router(model):
+        if model is PlTransportDbLink:
+            return link_query
+        if model is DbInboundOutbound:
+            return inv_query
+        if model is DbPurchasePlan:
+            return plan_query
+        return MagicMock()
+
+    db.query.side_effect = query_router
+
+    revoke_regeneratable_dibang_for_transport(
+        db,
+        tenant_no="t1",
+        transport_id="tr-1",
+        user_id="u1",
+    )
+
+    assert db_plan.planned_quantity == 31.0
+    assert inventory.del_flag == 1
+    assert link.del_flag == 1
+
+
+@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True)
+def test_revoke_blocks_when_weighing_exists(_mock_weighing):
+    inventory = DbInboundOutbound(
+        id="inv-1",
+        tenant_no="t1",
+        del_flag=0,
+        status="待入库",
+        batch_no="CRKDH0000001",
+    )
+    link = PlTransportDbLink(
+        id="link-1",
+        tenant_no="t1",
+        transport_id="tr-1",
+        db_inventory_id="inv-1",
+        db_purchase_plan_id="plan-1",
+        contributed_qty=Decimal("30"),
+        del_flag=0,
+    )
+
+    db = MagicMock()
+    link_query = MagicMock()
+    link_query.filter.return_value.all.return_value = [link]
+    inv_query = MagicMock()
+    inv_query.filter.return_value.first.return_value = inventory
+    db.query.side_effect = lambda model: link_query if model is PlTransportDbLink else inv_query
+
+    with pytest.raises(ValueError, match="已关联地磅"):
+        revoke_regeneratable_dibang_for_transport(
+            db,
+            tenant_no="t1",
+            transport_id="tr-1",
+            user_id="u1",
+        )
+
+    assert inventory.del_flag == 0
+    assert link.del_flag == 0