146 lines
4.0 KiB
Python
146 lines
4.0 KiB
Python
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
|