shanxia2.0/backend/tests/test_pl_plan_contract_gap_export.py

"""合同完善导出：按段次×产品展开，缺口类型跟本行段。"""
from datetime import datetime

from app.api.pl.plan import (
    CONTRACT_GAP_EXPORT_COLUMNS,
    build_contract_gap_export_rows,
)
from app.schemas.pl.plan import (
    PlanContractGapRead,
    PlanContractLink,
    PlanDeliveryRead,
    PlanLoadRead,
    PlanStopProductRead,
)
from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES

NOW = datetime(2026, 10, 11, 0, 0, 0)
INTERNAL = next(iter(INTERNAL_CUSTOMER_NAMES))


def _stop_product(name: str, *, product_id: str) -> PlanStopProductRead:
    return PlanStopProductRead(
        id=f"sp-{product_id}",
        planId="p1",
        tenantNo="t1",
        productId=product_id,
        productNo=f"NO-{product_id}",
        productName=name,
        createTime=NOW,
        updateTime=NOW,
    )


def _load(*, customer_name: str, contracts: list, products: list) -> PlanLoadRead:
    return PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=customer_name,
        products=products,
        contracts=contracts,
        createTime=NOW,
        updateTime=NOW,
    )


def _delivery(
    *,
    stop_id: str,
    seq: int,
    customer_name: str,
    delivery_time: str,
    contracts: list,
    products: list,
) -> PlanDeliveryRead:
    return PlanDeliveryRead(
        id=stop_id,
        planId="p1",
        tenantNo="t1",
        seq=seq,
        customerName=customer_name,
        deliveryTime=delivery_time,
        products=products,
        contracts=contracts,
        createTime=NOW,
        updateTime=NOW,
    )


def _gap_item() -> PlanContractGapRead:
    return PlanContractGapRead(
        id="p1",
        planNo="PL001",
        planDate="2026-10-11",
        planType="direct",
        followStatus="arranged",
        supplierName="供方甲",
        customerName="需方乙",
        supplierContractExempt=False,
        demanderContractExempt=False,
    )


def test_contract_gap_export_columns_match_segment_rows():
    assert [column.field for column in CONTRACT_GAP_EXPORT_COLUMNS] == [
        "planNo",
        "stopSegment",
        "planDate",
        "deliveryTimes",
        "planType",
        "followStatus",
        "loadCustomer",
        "deliveryCustomer",
        "productName",
        "supplierName",
        "customerName",
        "gapType",
    ]
    assert [column.title for column in CONTRACT_GAP_EXPORT_COLUMNS] == [
        "计划编号",
        "段次",
        "计划日期",
        "送货时间",
        "单据类型",
        "跟进状态",
        "装货客户",
        "送货客户",
        "产品名称",
        "供方",
        "需方",
        "缺口类型",
    ]


def test_build_contract_gap_export_rows_expands_multi_segment_products():
    """装货缺合同 + 两送货段（一段有合同、一段两产品缺合同）展开为 4 行。"""
    load = _load(
        customer_name="外部装货客户",
        contracts=[],
        products=[_stop_product("甲醇", product_id="prod-load")],
    )
    delivery_ok = _delivery(
        stop_id="d1",
        seq=1,
        customer_name="外部送货客户A",
        delivery_time="2026-10-12",
        contracts=[PlanContractLink(contractId="c1", contractNo="HT001")],
        products=[_stop_product("甲醇", product_id="prod-d1")],
    )
    delivery_gap = _delivery(
        stop_id="d2",
        seq=2,
        customer_name="外部送货客户B",
        delivery_time="2026-10-13",
        contracts=[],
        products=[
            _stop_product("乙醇", product_id="prod-d2a"),
            _stop_product("丙酮", product_id="prod-d2b"),
        ],
    )

    rows = build_contract_gap_export_rows(
        [_gap_item()],
        {"p1": (load, [delivery_ok, delivery_gap])},
    )

    assert len(rows) == 4
    assert [row["stopSegment"] for row in rows] == ["装货", "送货 #1", "送货 #2", "送货 #2"]
    assert [row["productName"] for row in rows] == ["甲醇", "甲醇", "乙醇", "丙酮"]
    assert [row["gapType"] for row in rows] == ["缺供方合同", "", "缺需方合同", "缺需方合同"]

    load_row = rows[0]
    assert load_row["loadCustomer"] == "外部装货客户"
    assert load_row["deliveryCustomer"] == ""
    assert load_row["deliveryTimes"] == ""

    ok_row = rows[1]
    assert ok_row["loadCustomer"] == ""
    assert ok_row["deliveryCustomer"] == "外部送货客户A"
    assert ok_row["deliveryTimes"] == "2026-10-12"

    gap_row = rows[3]
    assert gap_row["deliveryCustomer"] == "外部送货客户B"
    assert gap_row["deliveryTimes"] == "2026-10-13"
    assert gap_row["loadCustomer"] == ""

    for row in rows:
        assert row["planNo"] == "PL001"
        assert row["planDate"] == "2026-10-11"
        assert row["planType"] == "直销单"
        assert row["followStatus"] == "已安排"
        assert row["supplierName"] == "供方甲"
        assert row["customerName"] == "需方乙"


def test_build_contract_gap_export_rows_internal_side_has_empty_gap_type():
    """内置客户免合同：装货行缺口为空，外送段仍标缺需方合同。"""
    load = _load(
        customer_name=INTERNAL,
        contracts=[],
        products=[_stop_product("甲醇", product_id="prod-load")],
    )
    delivery = _delivery(
        stop_id="d1",
        seq=1,
        customer_name="外部送货客户",
        delivery_time="2026-10-12T08:30:00",
        contracts=[],
        products=[_stop_product("甲醇", product_id="prod-d1")],
    )

    rows = build_contract_gap_export_rows(
        [_gap_item()],
        {"p1": (load, [delivery])},
    )

    assert [row["stopSegment"] for row in rows] == ["装货", "送货 #1"]
    assert [row["gapType"] for row in rows] == ["", "缺需方合同"]
    assert rows[1]["deliveryTimes"] == "2026-10-12"