shanxia2.0/backend/tests/test_pl_dibang_date_sync_on_plan_save.py

"""计划保存时同步地磅出入库日期 — 单元测试"""
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)