660 lines
18 KiB
Python
660 lines
18 KiB
Python
"""地磅按段 upsert — 取消勾选阻断 / 复用出入库 id"""
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip
|
|
from app.models.pl_models import (
|
|
PlPlan,
|
|
PlPlanDelivery,
|
|
PlPlanDeliveryProduct,
|
|
PlPlanLoad,
|
|
PlPlanLoadProduct,
|
|
PlTransport,
|
|
PlTransportDbLink,
|
|
)
|
|
from app.services.pl_dibang_bridge_service import (
|
|
StopDibangOption,
|
|
upsert_stop_dibang_links,
|
|
)
|
|
|
|
|
|
def _make_query_router(mapping):
|
|
def query_router(model):
|
|
mock = MagicMock()
|
|
handler = mapping.get(model)
|
|
if handler is None:
|
|
return MagicMock()
|
|
return handler(mock)
|
|
|
|
return query_router
|
|
|
|
|
|
@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True)
|
|
def test_cancel_generate_dibang_blocked_when_weighing_exists(_mock_weighing):
|
|
plan = PlPlan(
|
|
id="plan-1",
|
|
plan_no="PO1",
|
|
plan_type="outbound",
|
|
plan_date=date(2026, 9, 1),
|
|
tenant_no="t1",
|
|
)
|
|
transport = PlTransport(id="tr-1", tenant_no="t1", license_plate="皖A12345", del_flag=0)
|
|
load = PlPlanLoad(
|
|
id="load-1",
|
|
plan_id="plan-1",
|
|
customer_name="客户A",
|
|
generate_dibang=1,
|
|
dibang_direction="outbound",
|
|
load_time=datetime(2026, 9, 1, 0, 0),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
inventory = DbInboundOutbound(
|
|
id="inv-1",
|
|
tenant_no="t1",
|
|
date=date(2026, 9, 1),
|
|
partner_company="客户A",
|
|
business_type="出库",
|
|
product_id="prod-1",
|
|
product="硫酸",
|
|
quantity=30.0,
|
|
batch_no="CRKDH0002400",
|
|
status="待出库",
|
|
del_flag=0,
|
|
)
|
|
link = PlTransportDbLink(
|
|
id="link-1",
|
|
tenant_no="t1",
|
|
plan_id="plan-1",
|
|
transport_id="tr-1",
|
|
plan_product_id="pp-1",
|
|
stop_type="load",
|
|
stop_id="load-1",
|
|
db_purchase_plan_id="dbp-1",
|
|
db_inventory_id="inv-1",
|
|
contributed_qty=Decimal("30"),
|
|
del_flag=0,
|
|
)
|
|
|
|
db = MagicMock()
|
|
|
|
def link_q(m):
|
|
filtered = MagicMock()
|
|
filtered.all.return_value = [link]
|
|
filtered.first.return_value = link
|
|
m.filter.return_value = filtered
|
|
return m
|
|
|
|
def load_q(m):
|
|
m.filter.return_value.first.return_value = load
|
|
return m
|
|
|
|
def delivery_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = []
|
|
return m
|
|
|
|
def inv_q(m):
|
|
m.filter.return_value.first.return_value = inventory
|
|
return m
|
|
|
|
db.query.side_effect = _make_query_router(
|
|
{
|
|
PlTransportDbLink: link_q,
|
|
PlPlanLoad: load_q,
|
|
PlPlanDelivery: delivery_q,
|
|
DbInboundOutbound: inv_q,
|
|
}
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="CRKDH0002400"):
|
|
upsert_stop_dibang_links(
|
|
db,
|
|
tenant_no="t1",
|
|
plan=plan,
|
|
transport=transport,
|
|
stop_options=[
|
|
StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=False),
|
|
],
|
|
user_id="u1",
|
|
dept_id="d1",
|
|
)
|
|
|
|
|
|
@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
|
|
@patch("app.services.pl_dibang_bridge_service.find_monthly_db_plan")
|
|
@patch("app.services.pl_dibang_bridge_service._create_db_inventory")
|
|
@patch("app.services.pl_dibang_bridge_service.resolve_crm_product")
|
|
def test_upsert_reuses_existing_inventory_id(
|
|
_mock_crm,
|
|
mock_create_inv,
|
|
mock_find_plan,
|
|
_mock_weighing,
|
|
):
|
|
plan = PlPlan(
|
|
id="plan-1",
|
|
plan_no="PO1",
|
|
plan_type="outbound",
|
|
plan_date=date(2026, 9, 1),
|
|
pickup_time=datetime(2026, 9, 1, 0, 0),
|
|
license_plate="皖A12345",
|
|
tenant_no="t1",
|
|
)
|
|
transport = PlTransport(id="tr-1", tenant_no="t1", license_plate="皖A12345", del_flag=0)
|
|
load = PlPlanLoad(
|
|
id="load-1",
|
|
plan_id="plan-1",
|
|
customer_name="安庆市山峡化工有限公司",
|
|
generate_dibang=1,
|
|
dibang_direction="outbound",
|
|
load_time=datetime(2026, 9, 1, 0, 0),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
delivery = PlPlanDelivery(
|
|
id="del-1",
|
|
plan_id="plan-1",
|
|
customer_name="客户B",
|
|
seq=1,
|
|
generate_dibang=0,
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
stop_product = PlPlanLoadProduct(
|
|
id="lp-1",
|
|
load_id="load-1",
|
|
plan_product_id="pp-1",
|
|
product_id="prod-1",
|
|
product_name="硫酸",
|
|
model="98%",
|
|
unit="吨",
|
|
plan_qty=Decimal("30"),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
order_num=1,
|
|
)
|
|
db_plan = DbPurchasePlan(
|
|
id="dbp-1",
|
|
tenant_no="t1",
|
|
plan_no="SX-1",
|
|
planned_quantity=30.0,
|
|
del_flag=0,
|
|
)
|
|
inventory = DbInboundOutbound(
|
|
id="inv-1",
|
|
tenant_no="t1",
|
|
date=date(2026, 9, 1),
|
|
partner_company="客户B",
|
|
business_type="出库",
|
|
product_id="prod-1",
|
|
product="硫酸",
|
|
specification="98%",
|
|
quantity=30.0,
|
|
batch_no="CRKDH0000001",
|
|
vehicle_no="皖A12345",
|
|
status="待出库",
|
|
purchase_plan_id="dbp-1",
|
|
plan_no="SX-1",
|
|
del_flag=0,
|
|
)
|
|
link = PlTransportDbLink(
|
|
id="link-1",
|
|
tenant_no="t1",
|
|
plan_id="plan-1",
|
|
transport_id="tr-1",
|
|
plan_product_id="pp-1",
|
|
stop_type="load",
|
|
stop_id="load-1",
|
|
db_purchase_plan_id="dbp-1",
|
|
db_inventory_id="inv-1",
|
|
contributed_qty=Decimal("30"),
|
|
del_flag=0,
|
|
)
|
|
|
|
mock_create_inv.side_effect = AssertionError("must reuse existing inventory")
|
|
mock_find_plan.return_value = db_plan
|
|
|
|
db = MagicMock()
|
|
|
|
def link_q(m):
|
|
filtered = MagicMock()
|
|
filtered.all.return_value = [link]
|
|
filtered.first.return_value = link
|
|
m.filter.return_value = filtered
|
|
return m
|
|
|
|
def load_q(m):
|
|
m.filter.return_value.first.return_value = load
|
|
return m
|
|
|
|
def delivery_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = [delivery]
|
|
return m
|
|
|
|
def load_prod_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = [stop_product]
|
|
return m
|
|
|
|
def delivery_prod_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = []
|
|
return m
|
|
|
|
def inv_q(m):
|
|
m.filter.return_value.first.return_value = inventory
|
|
return m
|
|
|
|
def plan_q(m):
|
|
m.filter.return_value.first.return_value = db_plan
|
|
return m
|
|
|
|
db.query.side_effect = _make_query_router(
|
|
{
|
|
PlTransportDbLink: link_q,
|
|
PlPlanLoad: load_q,
|
|
PlPlanDelivery: delivery_q,
|
|
PlPlanLoadProduct: load_prod_q,
|
|
PlPlanDeliveryProduct: delivery_prod_q,
|
|
DbInboundOutbound: inv_q,
|
|
DbPurchasePlan: plan_q,
|
|
}
|
|
)
|
|
|
|
results, warnings = upsert_stop_dibang_links(
|
|
db,
|
|
tenant_no="t1",
|
|
plan=plan,
|
|
transport=transport,
|
|
stop_options=[
|
|
StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=True),
|
|
],
|
|
user_id="u1",
|
|
dept_id="d1",
|
|
)
|
|
|
|
assert warnings == []
|
|
assert len(results) == 1
|
|
assert results[0].db_inventory_id == "inv-1"
|
|
assert results[0].batch_no == "CRKDH0000001"
|
|
mock_create_inv.assert_not_called()
|
|
|
|
|
|
@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True)
|
|
@patch("app.services.pl_dibang_bridge_service.resolve_crm_product")
|
|
def test_upsert_updates_inventory_and_weighing_plate_when_already_weighed(
|
|
mock_crm, _mock_weighing
|
|
):
|
|
"""已过磅：仍强制回写出入库与过磅单车牌；称重不动；出入库 id 不变。"""
|
|
mock_crm.return_value = MagicMock(
|
|
product_id="prod-1",
|
|
product_name="硫酸",
|
|
specification="98%",
|
|
)
|
|
plan = PlPlan(
|
|
id="plan-1",
|
|
plan_no="PO1",
|
|
plan_type="outbound",
|
|
plan_date=date(2026, 9, 1),
|
|
pickup_time=datetime(2026, 9, 1, 0, 0),
|
|
license_plate="皖HA7429",
|
|
tenant_no="t1",
|
|
)
|
|
transport = PlTransport(
|
|
id="tr-1", tenant_no="t1", license_plate="皖HA7429", del_flag=0
|
|
)
|
|
load = PlPlanLoad(
|
|
id="load-1",
|
|
plan_id="plan-1",
|
|
customer_name="安庆市山峡化工有限公司",
|
|
generate_dibang=1,
|
|
dibang_direction="outbound",
|
|
load_time=datetime(2026, 9, 1, 0, 0),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
delivery = PlPlanDelivery(
|
|
id="del-1",
|
|
plan_id="plan-1",
|
|
customer_name="江峰",
|
|
seq=1,
|
|
generate_dibang=0,
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
stop_product = PlPlanLoadProduct(
|
|
id="lp-1",
|
|
load_id="load-1",
|
|
plan_product_id="pp-1",
|
|
product_id="prod-1",
|
|
product_name="硫酸",
|
|
model="98%",
|
|
unit="吨",
|
|
plan_qty=Decimal("22"),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
order_num=1,
|
|
)
|
|
inventory = DbInboundOutbound(
|
|
id="2861",
|
|
tenant_no="t1",
|
|
date=date(2026, 9, 6),
|
|
partner_company="江峰",
|
|
business_type="出库",
|
|
product_id="prod-1",
|
|
product="硫酸",
|
|
specification="98%",
|
|
quantity=22.0,
|
|
batch_no="CRKDH0002488",
|
|
vehicle_no="皖H28926",
|
|
status="待出库",
|
|
purchase_plan_id="dbp-1",
|
|
plan_no="SX-1",
|
|
del_flag=0,
|
|
)
|
|
slip = DbWeighingSlip(
|
|
id="ws-1",
|
|
tenant_no="t1",
|
|
vehicle_no="皖H28926",
|
|
plan_no="SX-1",
|
|
cargo_name="硫酸",
|
|
specification="98%",
|
|
business_type="出库",
|
|
batch_no="CRKDH0002488",
|
|
supplier="",
|
|
consignee="江峰",
|
|
gross_weight=0.0,
|
|
tare_weight=18730.0,
|
|
net_weight=-18730.0,
|
|
inbound_outbound_id="2861",
|
|
purchase_plan_id="dbp-1",
|
|
is_voided=0,
|
|
del_flag=0,
|
|
)
|
|
link = PlTransportDbLink(
|
|
id="link-1",
|
|
tenant_no="t1",
|
|
plan_id="plan-1",
|
|
transport_id="tr-1",
|
|
plan_product_id="pp-1",
|
|
stop_type="load",
|
|
stop_id="load-1",
|
|
db_purchase_plan_id="dbp-1",
|
|
db_inventory_id="2861",
|
|
contributed_qty=Decimal("22"),
|
|
del_flag=0,
|
|
)
|
|
db = MagicMock()
|
|
|
|
def link_q(m):
|
|
filtered = MagicMock()
|
|
filtered.all.return_value = []
|
|
filtered.first.return_value = link
|
|
m.filter.return_value = filtered
|
|
return m
|
|
|
|
def load_q(m):
|
|
m.filter.return_value.first.return_value = load
|
|
return m
|
|
|
|
def delivery_q(m):
|
|
m.filter.return_value.first.return_value = delivery
|
|
m.filter.return_value.order_by.return_value.all.return_value = [delivery]
|
|
return m
|
|
|
|
def load_prod_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = [stop_product]
|
|
return m
|
|
|
|
def delivery_prod_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = []
|
|
return m
|
|
|
|
def inv_q(m):
|
|
m.filter.return_value.first.return_value = inventory
|
|
return m
|
|
|
|
def slip_q(m):
|
|
filtered = MagicMock()
|
|
filtered.all.return_value = [slip]
|
|
m.filter.return_value = filtered
|
|
return m
|
|
|
|
def db_plan_q(m):
|
|
filtered = MagicMock()
|
|
filtered.first.return_value = None
|
|
m.filter.return_value = filtered
|
|
return m
|
|
|
|
db.query.side_effect = _make_query_router(
|
|
{
|
|
PlTransportDbLink: link_q,
|
|
PlPlanLoad: load_q,
|
|
PlPlanDelivery: delivery_q,
|
|
PlPlanLoadProduct: load_prod_q,
|
|
PlPlanDeliveryProduct: delivery_prod_q,
|
|
DbInboundOutbound: inv_q,
|
|
DbWeighingSlip: slip_q,
|
|
DbPurchasePlan: db_plan_q,
|
|
}
|
|
)
|
|
|
|
results, warnings = upsert_stop_dibang_links(
|
|
db,
|
|
tenant_no="t1",
|
|
plan=plan,
|
|
transport=transport,
|
|
stop_options=[
|
|
StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=True),
|
|
StopDibangOption(stop_type="delivery", stop_id="del-1", generate_dibang=False),
|
|
],
|
|
user_id="u1",
|
|
dept_id="d1",
|
|
)
|
|
|
|
assert results and results[0].db_inventory_id == "2861"
|
|
assert inventory.vehicle_no == "皖HA7429"
|
|
assert slip.vehicle_no == "皖HA7429"
|
|
assert slip.tare_weight == 18730.0
|
|
assert slip.net_weight == -18730.0
|
|
assert not any("未同步" in w for w in warnings)
|
|
|
|
|
|
@patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=False)
|
|
@patch("app.services.pl_dibang_bridge_service._create_db_plan")
|
|
@patch("app.services.pl_dibang_bridge_service.find_monthly_db_plan")
|
|
@patch("app.services.pl_dibang_bridge_service.resolve_crm_product")
|
|
def test_upsert_remounts_monthly_plan_when_partner_changes(
|
|
mock_crm,
|
|
mock_find_plan,
|
|
mock_create_plan,
|
|
_mock_weighing,
|
|
):
|
|
"""往来单位变更：同月利旧月计划并改挂，出入库 id 不变。"""
|
|
mock_crm.return_value = MagicMock(
|
|
product_id="prod-1",
|
|
product_name="硫酸",
|
|
specification="98%",
|
|
)
|
|
old_partner = "客户旧"
|
|
new_partner = "客户新"
|
|
old_plan = DbPurchasePlan(
|
|
id="dbp-old",
|
|
tenant_no="t1",
|
|
plan_no="SX-OLD",
|
|
customer_name=old_partner,
|
|
planned_quantity=30.0,
|
|
del_flag=0,
|
|
)
|
|
new_plan = DbPurchasePlan(
|
|
id="dbp-new",
|
|
tenant_no="t1",
|
|
plan_no="SX-NEW",
|
|
customer_name=new_partner,
|
|
planned_quantity=0.0,
|
|
del_flag=0,
|
|
)
|
|
mock_find_plan.return_value = new_plan
|
|
|
|
plan = PlPlan(
|
|
id="plan-1",
|
|
plan_no="PO1",
|
|
plan_type="outbound",
|
|
plan_date=date(2026, 9, 1),
|
|
pickup_time=datetime(2026, 9, 1, 0, 0),
|
|
license_plate="皖A12345",
|
|
tenant_no="t1",
|
|
)
|
|
transport = PlTransport(
|
|
id="tr-1", tenant_no="t1", license_plate="皖A12345", del_flag=0
|
|
)
|
|
load = PlPlanLoad(
|
|
id="load-1",
|
|
plan_id="plan-1",
|
|
customer_name="安庆市山峡化工有限公司",
|
|
generate_dibang=1,
|
|
dibang_direction="outbound",
|
|
load_time=datetime(2026, 9, 1, 0, 0),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
# 装货段往来取送货客户：设为新往来，触发改挂
|
|
delivery = PlPlanDelivery(
|
|
id="del-1",
|
|
plan_id="plan-1",
|
|
customer_name=new_partner,
|
|
seq=1,
|
|
generate_dibang=0,
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
)
|
|
stop_product = PlPlanLoadProduct(
|
|
id="lp-1",
|
|
load_id="load-1",
|
|
plan_product_id="pp-1",
|
|
product_id="prod-1",
|
|
product_name="硫酸",
|
|
model="98%",
|
|
unit="吨",
|
|
plan_qty=Decimal("30"),
|
|
tenant_no="t1",
|
|
del_flag=0,
|
|
order_num=1,
|
|
)
|
|
inventory = DbInboundOutbound(
|
|
id="inv-1",
|
|
tenant_no="t1",
|
|
date=date(2026, 9, 1),
|
|
partner_company=old_partner,
|
|
business_type="出库",
|
|
product_id="prod-1",
|
|
product="硫酸",
|
|
specification="98%",
|
|
quantity=30.0,
|
|
batch_no="CRKDH0000001",
|
|
vehicle_no="皖A12345",
|
|
status="待出库",
|
|
purchase_plan_id="dbp-old",
|
|
plan_no="SX-OLD",
|
|
del_flag=0,
|
|
)
|
|
link = PlTransportDbLink(
|
|
id="link-1",
|
|
tenant_no="t1",
|
|
plan_id="plan-1",
|
|
transport_id="tr-1",
|
|
plan_product_id="pp-1",
|
|
stop_type="load",
|
|
stop_id="load-1",
|
|
db_purchase_plan_id="dbp-old",
|
|
db_inventory_id="inv-1",
|
|
contributed_qty=Decimal("30"),
|
|
del_flag=0,
|
|
)
|
|
|
|
db = MagicMock()
|
|
|
|
def link_q(m):
|
|
filtered = MagicMock()
|
|
filtered.all.return_value = [link]
|
|
filtered.first.return_value = link
|
|
m.filter.return_value = filtered
|
|
return m
|
|
|
|
def load_q(m):
|
|
m.filter.return_value.first.return_value = load
|
|
return m
|
|
|
|
def delivery_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = [delivery]
|
|
return m
|
|
|
|
def load_prod_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = [stop_product]
|
|
return m
|
|
|
|
def delivery_prod_q(m):
|
|
m.filter.return_value.order_by.return_value.all.return_value = []
|
|
return m
|
|
|
|
def inv_q(m):
|
|
m.filter.return_value.first.return_value = inventory
|
|
return m
|
|
|
|
def plan_q(m):
|
|
# _subtract_plan_contribution 查旧计划；后续可能再查
|
|
m.filter.return_value.first.return_value = old_plan
|
|
return m
|
|
|
|
db.query.side_effect = _make_query_router(
|
|
{
|
|
PlTransportDbLink: link_q,
|
|
PlPlanLoad: load_q,
|
|
PlPlanDelivery: delivery_q,
|
|
PlPlanLoadProduct: load_prod_q,
|
|
PlPlanDeliveryProduct: delivery_prod_q,
|
|
DbInboundOutbound: inv_q,
|
|
DbPurchasePlan: plan_q,
|
|
}
|
|
)
|
|
|
|
results, warnings = upsert_stop_dibang_links(
|
|
db,
|
|
tenant_no="t1",
|
|
plan=plan,
|
|
transport=transport,
|
|
stop_options=[
|
|
StopDibangOption(stop_type="load", stop_id="load-1", generate_dibang=True),
|
|
],
|
|
user_id="u1",
|
|
dept_id="d1",
|
|
)
|
|
|
|
assert warnings == []
|
|
assert len(results) == 1
|
|
assert inventory.id == "inv-1"
|
|
assert inventory.partner_company == new_partner
|
|
assert inventory.purchase_plan_id == new_plan.id
|
|
assert link.db_purchase_plan_id == new_plan.id
|
|
assert inventory.plan_no == "SX-NEW"
|
|
mock_find_plan.assert_called()
|
|
mock_create_plan.assert_not_called()
|
|
|
|
|
|
def test_arrange_transport_source_has_no_revoke_call():
|
|
import inspect
|
|
from app.api.pl import transport as transport_api
|
|
|
|
src = inspect.getsource(transport_api.arrange_transport)
|
|
assert "revoke_regeneratable_dibang_for_transport" not in src
|
|
|
|
|
|
def test_arrange_transport_syncs_stop_dibang_even_when_all_unchecked():
|
|
"""全部取消勾选时仍应进入按段 sync，以便取消/已过磅阻断生效。"""
|
|
import inspect
|
|
from app.api.pl import transport as transport_api
|
|
|
|
src = inspect.getsource(transport_api.arrange_transport)
|
|
assert "bool(stop_dibang_options)" in src
|
|
assert "item.generate_dibang for item in stop_dibang_options" not in src
|