473 lines
15 KiB
Python
473 lines
15 KiB
Python
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"] == "入库"
|