shanxia2.0/backend/tests/test_pl_dibang_plan_quantity.py

"""地磅月计划量运输联动累加/回减单元测试"""
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