shanxia2.0/backend/tests/test_pl_plan_archive_export.py

from decimal import Decimal

from app.schemas.pl.plan import PlanDeliveryRead, PlanLoadRead, PlanStopProductRead
from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
from app.services.pl_plan_archive_export import (
    _append_stop_product_rows,
    _plan_common_export_fields,
    resolve_archive_list_io_column,
)


def test_resolve_archive_list_io_column():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    assert resolve_archive_list_io_column("load", internal) == "outbound"
    assert resolve_archive_list_io_column("load", "外部客户") == "inbound"
    assert resolve_archive_list_io_column("delivery", internal) == "inbound"
    assert resolve_archive_list_io_column("delivery", "外部客户") == "outbound"


def test_append_stop_product_rows_io_and_amount():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    product = PlanStopProductRead(
        id="sp1",
        planId="p1",
        tenantNo="t1",
        productId="prod1",
        productNo="P001",
        productName="甲醇",
        model="99%",
        unit="吨",
        packageSpec="散装",
        planQty=Decimal("10"),
        inboundQty=Decimal("9.5"),
        inboundPrice=Decimal("100"),
        inboundAmount=None,
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        address="装货地址",
        loadTime="2026-08-27T08:00:00",
        products=[product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    rows: list[dict] = []
    plan_common = {
        "planNo": "PL20260827001",
        "planDate": "2026-08-27",
    }
    row_no = _append_stop_product_rows(rows, 0, plan_common, "load", load, None, load.products)
    assert row_no == 1
    assert len(rows) == 1
    row = rows[0]
    assert row["stopKind"] == "装货"
    assert row["stopCustomer"] == internal
    assert row["actualQty"] == Decimal("9.5")
    assert row["outboundQty"] == Decimal("9.5")
    assert row["inboundQty"] is None
    assert row["unitPrice"] == Decimal("100")
    assert row["amount"] == Decimal("950")


def test_append_stop_product_rows_delivery_external_outbound():
    product = PlanStopProductRead(
        id="sp2",
        planId="p1",
        tenantNo="t1",
        productId="prod1",
        productNo="P001",
        productName="甲醇",
        outboundQty=Decimal("8"),
        outboundPrice=Decimal("120"),
        outboundAmount=Decimal("960"),
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName="外部客户",
        address="送货地址",
        deliveryTime="2026-08-28T10:00:00",
        products=[product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    rows: list[dict] = []
    row_no = _append_stop_product_rows(rows, 0, {"planNo": "P1"}, "delivery", delivery, 1, delivery.products)
    row = rows[0]
    assert row["stopKind"] == "送货 #1"
    assert row["outboundQty"] == Decimal("8")
    assert row["inboundQty"] is None
    assert row["amount"] == Decimal("960")


class _FakePlan:
    plan_no = "PL001"
    plan_date = "2026-08-27"
    archive_status = "archived"
    follow_status = "arranged"
    plan_type = "direct"
    license_plate = "皖A12345"
    trailer_no = "挂001"
    erp_entered = 1
    freight_amount = Decimal("500")
    remark = "测试"


def test_plan_common_export_fields():
    fields = _plan_common_export_fields(_FakePlan(), None)
    assert fields["planNo"] == "PL001"
    assert fields["archiveStatus"] == "已归档"
    assert fields["followStatus"] == "已安排"
    assert fields["planType"] == "直销单"
    assert fields["licensePlate"] == "皖A12345挂001"
    assert fields["erpEntered"] == "是"