160 lines
5.0 KiB
Python
160 lines
5.0 KiB
Python
"""地磅月计划量运输联动累加/回减单元测试"""
|
|
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,
|
|
resolve_stop_dibang_partner,
|
|
revoke_regeneratable_dibang_for_transport,
|
|
)
|
|
|
|
|
|
class _StopStub:
|
|
def __init__(self, customer_name: str):
|
|
self.customer_name = customer_name
|
|
|
|
|
|
def test_resolve_stop_dibang_partner_load_uses_first_delivery():
|
|
load = _StopStub("安庆市山峡化工有限公司")
|
|
deliveries = [_StopStub("外部客户A"), _StopStub("外部客户B")]
|
|
assert resolve_stop_dibang_partner("load", load, deliveries) == "外部客户A"
|
|
|
|
|
|
def test_resolve_stop_dibang_partner_delivery_uses_load():
|
|
load = _StopStub("外部客户A")
|
|
deliveries = [_StopStub("安庆市山峡化工有限公司")]
|
|
assert resolve_stop_dibang_partner("delivery", load, deliveries) == "外部客户A"
|
|
|
|
|
|
def test_resolve_stop_dibang_partner_load_fallback_without_delivery():
|
|
load = _StopStub("安庆市山峡化工有限公司")
|
|
assert resolve_stop_dibang_partner("load", load, []) == "安庆市山峡化工有限公司"
|
|
|
|
|
|
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
|