124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
from decimal import Decimal
|
|
|
|
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_stop_product_rows,
|
|
_plan_common_export_fields,
|
|
resolve_archive_list_io_column,
|
|
)
|
|
|
|
|
|
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 = PlanStopProductRead(
|
|
id="sp1",
|
|
planId="p1",
|
|
tenantNo="t1",
|
|
productId="prod1",
|
|
productNo="P001",
|
|
productName="甲醇",
|
|
model="99%",
|
|
unit="吨",
|
|
packageSpec="散装",
|
|
planQty=Decimal("10"),
|
|
inboundQty=Decimal("9.5"),
|
|
inboundPrice=Decimal("100"),
|
|
inboundAmount=None,
|
|
createTime="2026-08-27T00:00:00",
|
|
updateTime="2026-08-27T00:00:00",
|
|
)
|
|
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["actualQty"] == Decimal("9.5")
|
|
assert row["outboundQty"] == Decimal("9.5")
|
|
assert row["inboundQty"] is None
|
|
assert row["unitPrice"] == Decimal("100")
|
|
assert row["amount"] == Decimal("950")
|
|
|
|
|
|
def test_append_stop_product_rows_delivery_external_outbound():
|
|
product = PlanStopProductRead(
|
|
id="sp2",
|
|
planId="p1",
|
|
tenantNo="t1",
|
|
productId="prod1",
|
|
productNo="P001",
|
|
productName="甲醇",
|
|
outboundQty=Decimal("8"),
|
|
outboundPrice=Decimal("120"),
|
|
outboundAmount=Decimal("960"),
|
|
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=[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["stopKind"] == "送货 #1"
|
|
assert row["outboundQty"] == Decimal("8")
|
|
assert row["inboundQty"] is None
|
|
assert row["amount"] == Decimal("960")
|
|
|
|
|
|
class _FakePlan:
|
|
plan_no = "PL001"
|
|
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["planNo"] == "PL001"
|
|
assert fields["archiveStatus"] == "已归档"
|
|
assert fields["followStatus"] == "已安排"
|
|
assert fields["planType"] == "直销单"
|
|
assert fields["licensePlate"] == "皖A12345挂001"
|
|
assert fields["erpEntered"] == "是"
|