shanxia2.0/backend/tests/test_pl_plan_contract_gap.py

from sqlalchemy.dialects import mysql

from app.schemas.pl.plan import PlanDeliveryRead, PlanLoadRead
from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES
from app.services.pl_plan_contract_gap import (
    compute_contract_gap_sides,
    compute_plan_gap_sides_from_stops,
    plan_has_contract_gap_exists,
    stop_contract_gap,
)

INTERNAL = next(iter(INTERNAL_CUSTOMER_NAMES))


def test_both_external_no_contracts_gaps_both():
    sides = compute_contract_gap_sides(
        supplier_name="外部供方A",
        customer_name="外部需方B",
        supplier_contract_count=0,
        demander_contract_count=0,
    )
    assert sides == ["supplier", "demander"]


def test_internal_supplier_external_demander_missing_only_demander():
    sides = compute_contract_gap_sides(
        supplier_name=INTERNAL,
        customer_name="外部需方B",
        supplier_contract_count=0,
        demander_contract_count=0,
    )
    assert sides == ["demander"]


def test_both_internal_no_gap():
    sides = compute_contract_gap_sides(
        supplier_name=INTERNAL,
        customer_name=INTERNAL,
        supplier_contract_count=0,
        demander_contract_count=0,
    )
    assert sides == []


def test_external_with_contracts_no_gap():
    sides = compute_contract_gap_sides(
        supplier_name="外部供方A",
        customer_name="外部需方B",
        supplier_contract_count=1,
        demander_contract_count=2,
    )
    assert sides == []


def test_empty_names_treated_as_non_internal_gap_both():
    sides = compute_contract_gap_sides(
        supplier_name=None,
        customer_name="",
        supplier_contract_count=0,
        demander_contract_count=0,
    )
    assert sides == ["supplier", "demander"]


def test_stop_contract_gap_internal_exempt():
    assert stop_contract_gap(INTERNAL, 0) is False
    assert stop_contract_gap("外部客户", 0) is True
    assert stop_contract_gap("外部客户", 1) is False


def test_compute_plan_gap_sides_from_stops_multi_delivery():
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=INTERNAL,
        products=[],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery_ok = PlanDeliveryRead(
        id="d1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName="外部客户A",
        products=[],
        contracts=[{"contractId": "c1", "contractNo": "HT001"}],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery_gap = PlanDeliveryRead(
        id="d2",
        planId="p1",
        tenantNo="t1",
        seq=2,
        customerName="外部客户B",
        products=[],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    sides = compute_plan_gap_sides_from_stops(load, [delivery_ok, delivery_gap])
    assert sides == ["demander"]


def test_compute_plan_gap_sides_from_stops_internal_load_no_gap():
    load = PlanLoadRead(
        id="load1",
        planId="p1",
        tenantNo="t1",
        customerName=INTERNAL,
        products=[],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    delivery = PlanDeliveryRead(
        id="d1",
        planId="p1",
        tenantNo="t1",
        seq=1,
        customerName=INTERNAL,
        products=[],
        contracts=[],
        createTime="2026-08-27T00:00:00",
        updateTime="2026-08-27T00:00:00",
    )
    assert compute_plan_gap_sides_from_stops(load, [delivery]) == []


def _compile(clause) -> str:
    return str(
        clause.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
    ).lower()


def test_plan_has_contract_gap_exists_mentions_contract_and_internal_names():
    clause = plan_has_contract_gap_exists("shanxia2.0")
    sql = _compile(clause)

    assert "pl_plan_contract" in sql
    assert "pl_plan_load" in sql
    assert "pl_plan_delivery" in sql