shanxia2.0/backend/tests/test_pl_plan_archive_export.py

from decimal import Decimal
from unittest.mock import MagicMock, patch

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_paired_product_rows,
    _append_stop_product_rows,
    _plan_common_export_fields,
    build_plan_archive_export_rows,
    resolve_archive_list_io_column,
)


def _stop_product(**kwargs) -> PlanStopProductRead:
    defaults = {
        "id": "sp1",
        "planId": "p1",
        "tenantNo": "t1",
        "productId": "prod1",
        "productNo": "P001",
        "productName": "甲醇",
        "model": "99%",
        "unit": "吨",
        "packageSpec": "散装",
        "planQty": Decimal("10"),
        "createTime": "2026-08-27T00:00:00",
        "updateTime": "2026-08-27T00:00:00",
    }
    defaults.update(kwargs)
    return PlanStopProductRead(**defaults)


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 = _stop_product(
        inboundQty=Decimal("9.5"),
        inboundPrice=Decimal("100"),
        inboundAmount=None,
    )
    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["loadCustomer"] == internal
    assert row["actualQty"] == Decimal("9.5")
    assert row["outboundQty"] == Decimal("9.5")
    assert row["loadOutboundQty"] == Decimal("9.5")
    assert row["inboundQty"] is None
    assert row["unitPrice"] == Decimal("100")
    assert row["loadUnitPrice"] == Decimal("100")
    assert row["amount"] == Decimal("950")
    assert row["loadAmount"] == Decimal("950")


def test_append_stop_product_rows_delivery_external_outbound():
    product = _stop_product(
        id="sp2",
        outboundQty=Decimal("8"),
        outboundPrice=Decimal("120"),
        outboundAmount=Decimal("960"),
    )
    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_no == 1
    assert row["stopKind"] == "送货 #1"
    assert row["deliveryCustomer"] == "外部客户"
    assert row["outboundQty"] == Decimal("8")
    assert row["deliveryOutboundQty"] == Decimal("8")
    assert row["inboundQty"] is None
    assert row["amount"] == Decimal("960")
    assert row["deliveryAmount"] == Decimal("960")


def test_append_paired_product_rows_inner_join_one_row_per_product():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load_a = _stop_product(
        id="lp1",
        productId="prod-a",
        productName="甲醇",
        packageSpec="散装",
        inboundQty=Decimal("10"),
        inboundPrice=Decimal("100"),
    )
    load_b = _stop_product(
        id="lp2",
        productId="prod-b",
        productName="乙醇",
        packageSpec="桶装",
        inboundQty=Decimal("5"),
        inboundPrice=Decimal("80"),
    )
    # 装货独有，送货无匹配 → 不出行
    load_only = _stop_product(
        id="lp3",
        productId="prod-c",
        productName="仅装货",
        packageSpec="散装",
        inboundQty=Decimal("1"),
    )
    delivery_a = _stop_product(
        id="dp1",
        productId="prod-a",
        productName="甲醇",
        packageSpec="散装",
        outboundQty=Decimal("9.5"),
        outboundPrice=Decimal("120"),
        outboundAmount=Decimal("1140"),
    )
    # 同产品不同包装 → 不匹配
    delivery_a_other_pkg = _stop_product(
        id="dp1b",
        productId="prod-a",
        productName="甲醇",
        packageSpec="桶装",
        outboundQty=Decimal("1"),
        outboundPrice=Decimal("1"),
    )
    delivery_b = _stop_product(
        id="dp2",
        productId="prod-b",
        productName="乙醇",
        packageSpec="桶装",
        outboundQty=Decimal("4"),
        outboundPrice=Decimal("90"),
    )
    # 送货独有 → 不出行
    delivery_only = _stop_product(
        id="dp3",
        productId="prod-d",
        productName="仅送货",
        packageSpec="散装",
        outboundQty=Decimal("2"),
    )
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        address="装货地址",
        loadTime="2026-08-27T08:00:00",
        products=[load_a, load_b, load_only],
        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=[delivery_a, delivery_a_other_pkg, delivery_b, delivery_only],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    rows: list[dict] = []
    row_no = _append_paired_product_rows(rows, 0, {"planNo": "PL001"}, load, delivery, 1)
    assert row_no == 2
    assert len(rows) == 2
    by_name = {r["productName"]: r for r in rows}
    assert set(by_name) == {"甲醇", "乙醇"}

    methanol = by_name["甲醇"]
    assert methanol["loadCustomer"] == internal
    assert methanol["deliveryCustomer"] == "外部客户"
    assert methanol["deliveryStopKind"] == "送货 #1"
    assert methanol["packageSpec"] == "散装"
    assert methanol["loadOutboundQty"] == Decimal("10")  # 内部客户装货 → 出库
    assert methanol["loadInboundQty"] is None
    assert methanol["loadUnitPrice"] == Decimal("100")
    assert methanol["loadAmount"] == Decimal("1000")
    assert methanol["deliveryOutboundQty"] == Decimal("9.5")
    assert methanol["deliveryInboundQty"] is None
    assert methanol["deliveryUnitPrice"] == Decimal("120")
    assert methanol["deliveryAmount"] == Decimal("1140")

    ethanol = by_name["乙醇"]
    assert ethanol["loadOutboundQty"] == Decimal("5")
    assert ethanol["deliveryOutboundQty"] == Decimal("4")
    assert ethanol["deliveryAmount"] == Decimal("360")


def test_build_plan_archive_export_rows_pair_mode_and_single_side():
    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load_product = _stop_product(
        inboundQty=Decimal("10"),
        inboundPrice=Decimal("100"),
    )
    delivery_product = _stop_product(
        id="dp1",
        outboundQty=Decimal("9"),
        outboundPrice=Decimal("110"),
        outboundAmount=Decimal("990"),
    )
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        address="装货地址",
        loadTime="2026-08-27T08:00:00",
        products=[load_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=2,
        customerName="外部客户",
        address="送货地址",
        deliveryTime="2026-08-28T10:00:00",
        products=[delivery_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )

    plan = MagicMock()
    plan.id = "p1"
    plan.plan_no = "PL001"
    plan.contract_no = "C1"
    plan.plan_date = "2026-08-27"
    plan.archive_status = "archived"
    plan.follow_status = "arranged"
    plan.plan_type = "direct"
    plan.license_plate = "皖A12345"
    plan.trailer_no = None
    plan.erp_entered = 0
    plan.freight_amount = None
    plan.remark = ""
    plan.transport_id = None
    plan.customer_name = "外部客户"
    plan.supplier_name = None
    plan.destination_name = None
    plan.delivery_time = None

    with patch(
        "app.services.pl_plan_archive_export.load_plan_stops_grouped",
        return_value={"p1": (load, [delivery])},
    ), patch(
        "app.services.pl_plan_archive_export._load_legacy_products_by_plan",
        return_value={},
    ), patch(
        "app.services.pl_plan_archive_export._load_contract_nos_by_plan",
        return_value={"p1": []},
    ), patch(
        "app.services.pl_plan_archive_export.load_crm_product_alias_maps",
        return_value=({}, {}),
    ):
        paired = build_plan_archive_export_rows(
            MagicMock(), "t1", [plan], {}, stop_kind=None
        )
        load_only = build_plan_archive_export_rows(
            MagicMock(), "t1", [plan], {}, stop_kind="load"
        )
        delivery_only = build_plan_archive_export_rows(
            MagicMock(), "t1", [plan], {}, stop_kind="delivery"
        )

    assert len(paired) == 1
    assert paired[0]["productName"] == "甲醇"
    assert paired[0]["loadCustomer"] == internal
    assert paired[0]["deliveryCustomer"] == "外部客户"
    assert paired[0]["deliveryStopKind"] == "送货 #2"
    assert "stopKind" not in paired[0]

    assert len(load_only) == 1
    assert load_only[0]["stopKind"] == "装货"
    assert load_only[0]["loadCustomer"] == internal
    assert load_only[0]["deliveryCustomer"] == ""

    assert len(delivery_only) == 1
    assert delivery_only[0]["stopKind"] == "送货 #2"
    assert delivery_only[0]["deliveryCustomer"] == "外部客户"
    assert delivery_only[0]["loadCustomer"] == ""


class _FakePlan:
    id = "p1"
    plan_no = "PL001"
    contract_no = "SX/X-JT20260827001"
    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["planId"] == "p1"
    assert fields["planNo"] == "PL001"
    assert fields["contractNo"] == "SX/X-JT20260827001"
    assert fields["archiveStatus"] == "已归档"
    assert fields["followStatus"] == "已安排"
    assert fields["planType"] == "直销单"
    assert fields["licensePlate"] == "皖A12345挂001"
    assert fields["erpEntered"] == "是"


def test_append_paired_product_rows_includes_brand_and_alias():
    plan_product = MagicMock()
    plan_product.id = "pp1"
    plan_product.product_id = "prod1"
    plan_product.product_name = "甲醇别名"
    plan_product.brand_mode = "required"
    plan_product.brand_name = "君正"

    load_product = _stop_product(planProductId="pp1", productName="甲醇别名")
    delivery_product = _stop_product(
        id="dp1",
        planProductId="pp1",
        productName="甲醇别名",
        outboundQty=Decimal("9"),
        outboundPrice=Decimal("110"),
    )
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName="装货客户",
        products=[load_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName="送货客户",
        products=[delivery_product],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    rows: list[dict] = []
    _append_paired_product_rows(
        rows,
        0,
        {"planNo": "PL001"},
        load,
        delivery,
        1,
        plan_products_by_id={"pp1": plan_product},
        plan_products_by_product_id={"prod1": plan_product},
        standard_by_id={"prod1": "甲醇"},
        alias_options_by_id={"prod1": ["甲醇别名"]},
    )
    assert len(rows) == 1
    assert rows[0]["productName"] == "甲醇"
    assert rows[0]["brandName"] == "君正"
    assert rows[0]["aliasName"] == "甲醇别名"


def test_plan_common_export_fields_merges_link_contract_nos():
    fields = _plan_common_export_fields(
        _FakePlan(),
        None,
        contract_nos=["SX/X-JT20260827001", "SX/X-JT20260827002"],
    )
    assert fields["contractNo"] == "SX/X-JT20260827001、SX/X-JT20260827002"


def test_append_paired_rows_internal_customer_contract_placeholder():
    from app.services.pl_plan_archive_export import format_row_contract_export_label

    internal = next(iter(INTERNAL_CUSTOMER_NAMES))
    load_product = _stop_product(id="lp1", inboundQty=Decimal("10"))
    delivery_product = _stop_product(id="dp1", outboundQty=Decimal("10"))
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=internal,
        products=[load_product],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="del1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName=internal,
        products=[delivery_product],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    assert format_row_contract_export_label(load=load, delivery=delivery) == "出库、入库"

    rows: list[dict] = []
    _append_paired_product_rows(
        rows,
        0,
        {"planNo": "PL001", "contractNo": ""},
        load,
        delivery,
        1,
    )
    assert rows[0]["contractNo"] == "出库、入库"

    load_only_rows: list[dict] = []
    _append_stop_product_rows(
        load_only_rows,
        0,
        {"planNo": "PL001", "contractNo": ""},
        "load",
        load,
        None,
        load.products,
    )
    assert load_only_rows[0]["contractNo"] == "出库"

    delivery_only_rows: list[dict] = []
    _append_stop_product_rows(
        delivery_only_rows,
        0,
        {"planNo": "PL001", "contractNo": ""},
        "delivery",
        delivery,
        1,
        delivery.products,
    )
    assert delivery_only_rows[0]["contractNo"] == "入库"