"""计划保存时同步地磅出入库日期 — 单元测试""" 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 from app.models.pl_models import PlPlan, PlPlanDelivery, PlPlanLoad, PlTransportDbLink from app.services.pl_dibang_bridge_service import sync_linked_dibang_dates_for_plan 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=False) def test_sync_same_month_updates_inventory_date_only(_mock_weighing): plan = PlPlan( id="plan-1", plan_no="PO1", plan_type="outbound", plan_date=date(2026, 8, 31), pickup_time=datetime(2026, 8, 31, 0, 0), tenant_no="t1", ) load = PlPlanLoad( id="load-1", plan_id="plan-1", load_time=datetime(2026, 8, 31, 0, 0), tenant_no="t1", del_flag=0, ) db_plan = DbPurchasePlan( id="dbp-1", tenant_no="t1", plan_no="SX-OLD", plan_date=date(2026, 8, 15), start_date=date(2026, 8, 1), end_date=date(2026, 8, 31), business_type="出库", product_id="prod-1", product="硫酸", planned_quantity=31.0, customer_name="客户A", del_flag=0, ) inventory = DbInboundOutbound( id="inv-1", tenant_no="t1", date=date(2026, 8, 15), partner_company="客户A", business_type="出库", product_id="prod-1", product="硫酸", specification="98%", quantity=31.0, plan_no="SX-OLD", status="待出库", purchase_plan_id="dbp-1", del_flag=0, ) link = PlTransportDbLink( id="link-1", tenant_no="t1", plan_id="plan-1", transport_id="tr-1", stop_type="load", stop_id="load-1", db_purchase_plan_id="dbp-1", db_inventory_id="inv-1", contributed_qty=Decimal("31"), del_flag=0, ) db = MagicMock() def link_q(m): m.filter.return_value.all.return_value = [link] 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 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, DbInboundOutbound: inv_q, DbPurchasePlan: plan_q, } ) warnings = sync_linked_dibang_dates_for_plan( db, tenant_no="t1", plan=plan, user_id="u1", dept_id="d1", ) assert warnings == [] assert inventory.date == date(2026, 8, 31) assert inventory.update_by == "u1" assert inventory.purchase_plan_id == "dbp-1" assert db_plan.planned_quantity == 31.0 @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_plan") def test_sync_cross_month_repoints_purchase_plan(mock_create, mock_find, _mock_weighing): plan = PlPlan( id="plan-1", plan_no="PO20260830002", plan_type="outbound", plan_date=date(2026, 9, 1), pickup_time=datetime(2026, 9, 1, 0, 0), tenant_no="t1", ) load = PlPlanLoad( id="load-1", plan_id="plan-1", load_time=datetime(2026, 9, 1, 0, 0), tenant_no="t1", del_flag=0, ) old_db_plan = DbPurchasePlan( id="dbp-aug", tenant_no="t1", plan_no="SX-AUG", plan_date=date(2026, 8, 31), start_date=date(2026, 8, 1), end_date=date(2026, 8, 31), business_type="出库", product_id="prod-1", product="硫酸", planned_quantity=31.0, customer_name="客户A", del_flag=0, ) new_db_plan = DbPurchasePlan( id="dbp-sep", tenant_no="t1", plan_no="SX-SEP", plan_date=date(2026, 9, 1), start_date=date(2026, 9, 1), end_date=date(2026, 9, 30), business_type="出库", product_id="prod-1", product="硫酸", planned_quantity=10.0, customer_name="客户A", del_flag=0, ) inventory = DbInboundOutbound( id="inv-1", tenant_no="t1", date=date(2026, 8, 31), partner_company="客户A", business_type="出库", product_id="prod-1", product="硫酸", specification="98%", quantity=31.0, plan_no="SX-AUG", status="待出库", purchase_plan_id="dbp-aug", del_flag=0, ) link = PlTransportDbLink( id="link-1", tenant_no="t1", plan_id="plan-1", transport_id="tr-1", stop_type="load", stop_id="load-1", db_purchase_plan_id="dbp-aug", db_inventory_id="inv-1", contributed_qty=Decimal("31"), del_flag=0, ) mock_find.return_value = new_db_plan mock_create.side_effect = AssertionError("should reuse existing September plan") db = MagicMock() def link_q(m): m.filter.return_value.all.return_value = [link] 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 def plan_q(m): m.filter.return_value.first.return_value = old_db_plan return m db.query.side_effect = _make_query_router( { PlTransportDbLink: link_q, PlPlanLoad: load_q, PlPlanDelivery: delivery_q, DbInboundOutbound: inv_q, DbPurchasePlan: plan_q, } ) warnings = sync_linked_dibang_dates_for_plan( db, tenant_no="t1", plan=plan, user_id="u1", dept_id="d1", ) assert warnings == [] assert inventory.date == date(2026, 9, 1) assert inventory.purchase_plan_id == "dbp-sep" assert inventory.plan_no == "SX-SEP" assert link.db_purchase_plan_id == "dbp-sep" assert old_db_plan.planned_quantity == 0.0 assert new_db_plan.planned_quantity == 41.0 @patch("app.services.pl_dibang_bridge_service._inventory_has_weighing_slip", return_value=True) def test_sync_skips_when_weighing_exists(_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), tenant_no="t1", ) load = PlPlanLoad( id="load-1", plan_id="plan-1", 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, 8, 31), partner_company="客户A", business_type="出库", product="硫酸", quantity=31.0, batch_no="CRKDH0002400", status="待出库", del_flag=0, ) link = PlTransportDbLink( id="link-1", tenant_no="t1", plan_id="plan-1", transport_id="tr-1", stop_type="load", stop_id="load-1", db_purchase_plan_id="dbp-1", db_inventory_id="inv-1", contributed_qty=Decimal("31"), del_flag=0, ) db = MagicMock() def link_q(m): m.filter.return_value.all.return_value = [link] 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, } ) warnings = sync_linked_dibang_dates_for_plan( db, tenant_no="t1", plan=plan, user_id="u1", dept_id="d1", ) assert len(warnings) == 1 assert "CRKDH0002400" in warnings[0] assert inventory.date == date(2026, 8, 31)