From cd64463c0ad2e42f49449ee36812ff8f595ab57a Mon Sep 17 00:00:00 2001 From: zlwu Date: Sun, 6 Sep 2026 19:37:34 +0800 Subject: [PATCH] test(pl): expect dibang plate sync even after weighing Co-authored-by: Cursor --- .../tests/test_pl_dibang_upsert_on_arrange.py | 182 +++++++++++++++++- 1 file changed, 181 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_pl_dibang_upsert_on_arrange.py b/backend/tests/test_pl_dibang_upsert_on_arrange.py index c26a7db53..7bbc1d4dd 100644 --- a/backend/tests/test_pl_dibang_upsert_on_arrange.py +++ b/backend/tests/test_pl_dibang_upsert_on_arrange.py @@ -5,7 +5,7 @@ from unittest.mock import MagicMock, patch import pytest -from app.models.db_models import DbInboundOutbound, DbPurchasePlan +from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbWeighingSlip from app.models.pl_models import ( PlPlan, PlPlanDelivery, @@ -278,3 +278,183 @@ def test_upsert_reuses_existing_inventory_id( 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)