shanxia2.0/backend/tests/test_pl_plan_print_service.py

"""送(提)货单按客户聚合打印"""
from datetime import datetime
from decimal import Decimal

from app.schemas.pl.plan import PlanDeliveryRead, PlanStopProductRead
from app.services.pl_plan_print_service import (
    build_delivery_slip_print,
    list_delivery_print_customers,
)


def _stop_product(**kwargs) -> PlanStopProductRead:
    now = datetime(2026, 8, 26, 10, 0, 0)
    return PlanStopProductRead(
        id=kwargs.get("id", "sp1"),
        plan_id="p1",
        product_id=kwargs["product_id"],
        product_no=kwargs.get("product_no", "PN"),
        product_name=kwargs.get("product_name", "产品"),
        unit=kwargs.get("unit", "吨"),
        plan_qty=kwargs.get("plan_qty"),
        outbound_qty=kwargs.get("outbound_qty"),
        tenant_no="t1",
        create_time=now,
        update_time=now,
    )


def _delivery(**kwargs) -> PlanDeliveryRead:
    now = datetime(2026, 8, 26, 10, 0, 0)
    return PlanDeliveryRead(
        id=kwargs["id"],
        plan_id="p1",
        seq=kwargs.get("seq", 1),
        customer_id=kwargs.get("customer_id"),
        customer_name=kwargs.get("customer_name"),
        address=kwargs.get("address"),
        contact=kwargs.get("contact"),
        products=kwargs.get("products") or [],
        tenant_no="t1",
        create_time=now,
        update_time=now,
    )


def test_list_delivery_print_customers_three_distinct():
    deliveries = [
        _delivery(id="d1", seq=1, customer_id="c1", customer_name="客户A"),
        _delivery(id="d2", seq=2, customer_id="c2", customer_name="客户B"),
        _delivery(id="d3", seq=3, customer_id="c3", customer_name="客户C"),
    ]
    rows = list_delivery_print_customers(deliveries)
    assert len(rows) == 3
    assert [r.customer_id for r in rows] == ["c1", "c2", "c3"]
    assert all(r.delivery_count == 1 for r in rows)


def test_list_delivery_print_customers_dedupes_by_customer_id():
    deliveries = [
        _delivery(id="d1", seq=2, customer_id="c1", customer_name="客户A"),
        _delivery(id="d2", seq=1, customer_id="c1", customer_name="客户A"),
        _delivery(id="d3", seq=3, customer_id="c2", customer_name="客户B"),
    ]
    rows = list_delivery_print_customers(deliveries)
    assert len(rows) == 2
    assert rows[0].customer_id == "c1"
    assert rows[0].delivery_count == 2
    assert rows[1].customer_id == "c2"


def test_build_delivery_slip_print_merges_same_customer_and_uses_min_seq_contact(monkeypatch):
    deliveries = [
        _delivery(
            id="d2",
            seq=2,
            customer_id="c1",
            customer_name="客户A",
            address="地址B",
            contact="联系B",
            products=[
                _stop_product(id="sp2", product_id="prod1", plan_qty=Decimal("5"), outbound_qty=Decimal("5"))
            ],
        ),
        _delivery(
            id="d1",
            seq=1,
            customer_id="c1",
            customer_name="客户A",
            address="地址A",
            contact="联系A",
            products=[
                _stop_product(id="sp1", product_id="prod1", plan_qty=Decimal("10"), outbound_qty=Decimal("10"))
            ],
        ),
    ]

    monkeypatch.setattr(
        "app.services.pl_plan_print_service.load_plan_stops_grouped",
        lambda db, tenant_no, plan_ids: {"p1": (None, deliveries)},
    )

    result = build_delivery_slip_print(
        db=None,  # type: ignore[arg-type]
        tenant_no="t1",
        plan_id="p1",
        plan_no="PL001",
        plan_date="2026-08-26",
        customer_id="c1",
    )
    assert result.receiving_unit == "客户A"
    assert result.receive_contact == "联系A"
    assert result.delivery_address == "地址A"
    assert len(result.products) == 1
    assert result.products[0].plan_qty == Decimal("15")
    assert result.products[0].outbound_qty == Decimal("15")