292 lines
9.1 KiB
Python
292 lines
9.1 KiB
Python
"""合同产品运输安排数量汇总单元测试。"""
|
|
from types import SimpleNamespace
|
|
from collections import defaultdict
|
|
|
|
from app.services.pl_contract_plan_qty import (
|
|
_bucket_for_plan,
|
|
accumulate_stop_products,
|
|
attach_plan_qty_fields_to_products,
|
|
match_contract_product_for_key,
|
|
merge_product_qty_maps,
|
|
normalize_package_spec,
|
|
product_package_key,
|
|
resolve_linked_plan_scopes,
|
|
sum_qtys_for_scope,
|
|
)
|
|
|
|
|
|
def test_bucket_archived():
|
|
assert (
|
|
_bucket_for_plan(
|
|
archive_status="archived",
|
|
follow_status="arranged",
|
|
transport_status="arranged",
|
|
)
|
|
== "archived"
|
|
)
|
|
|
|
|
|
def test_bucket_arranged_unarchived_includes_draft_archive():
|
|
assert (
|
|
_bucket_for_plan(
|
|
archive_status="draft",
|
|
follow_status="arranged",
|
|
transport_status="arranged",
|
|
)
|
|
== "arranged_unarchived"
|
|
)
|
|
|
|
|
|
def test_bucket_skips_pending_and_cancelled():
|
|
assert (
|
|
_bucket_for_plan(
|
|
archive_status="unarchived",
|
|
follow_status="pending_arrange",
|
|
transport_status="pending",
|
|
)
|
|
is None
|
|
)
|
|
assert (
|
|
_bucket_for_plan(
|
|
archive_status="unarchived",
|
|
follow_status="cancelled",
|
|
transport_status="arranged",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_normalize_package_spec():
|
|
from app.services.pl_contract_plan_qty import normalize_package_spec
|
|
|
|
assert normalize_package_spec(None) == ""
|
|
assert normalize_package_spec(" ") == ""
|
|
assert normalize_package_spec(" 槽罐车 ") == "槽罐车"
|
|
assert normalize_package_spec("25kg/桶") != normalize_package_spec("25Kg/袋")
|
|
|
|
|
|
def test_accumulate_splits_same_product_different_package():
|
|
acc = defaultdict(float)
|
|
accumulate_stop_products(
|
|
acc,
|
|
[
|
|
SimpleNamespace(product_id="SXWL007344", package_spec="25kg/桶", plan_qty=15),
|
|
SimpleNamespace(product_id="SXWL007344", package_spec="槽罐车", plan_qty=700),
|
|
SimpleNamespace(product_id="SXWL007344", package_spec=" 槽罐车 ", plan_qty=10),
|
|
],
|
|
)
|
|
k_bucket = ("SXWL007344", "25kg/桶")
|
|
k_tank = ("SXWL007344", "槽罐车")
|
|
assert acc[k_bucket] == 15.0
|
|
assert acc[k_tank] == 710.0
|
|
|
|
|
|
def test_merge_product_qty_maps_composite_keys():
|
|
out = merge_product_qty_maps(
|
|
archived={("p1", "A"): 10.0},
|
|
arranged_unarchived={("p1", "B"): 20.0, ("p1", "A"): 5.0},
|
|
)
|
|
assert out[("p1", "A")]["arrangedTotalQty"] == 15.0
|
|
assert out[("p1", "B")]["arrangedTotalQty"] == 20.0
|
|
|
|
|
|
def test_attach_plan_qty_fields_matches_package():
|
|
qty_map = {
|
|
("pid1", "25kg/桶"): {
|
|
"archivedQty": 1.0,
|
|
"arrangedUnarchivedQty": 2.0,
|
|
"arrangedTotalQty": 3.0,
|
|
},
|
|
("pid1", "槽罐车"): {
|
|
"archivedQty": 10.0,
|
|
"arrangedUnarchivedQty": 20.0,
|
|
"arrangedTotalQty": 30.0,
|
|
},
|
|
}
|
|
products = [
|
|
{"productId": "pid1", "packageSpec": "25kg/桶"},
|
|
{"productId": "pid1", "packageSpec": "槽罐车"},
|
|
]
|
|
attach_plan_qty_fields_to_products(products, qty_map)
|
|
assert products[0]["arrangedTotalQty"] == 3.0
|
|
assert products[1]["arrangedTotalQty"] == 30.0
|
|
|
|
|
|
def test_merge_product_qty_maps_totals():
|
|
out = merge_product_qty_maps(
|
|
archived={("p1", ""): 120.0, ("p2", ""): 50.0},
|
|
arranged_unarchived={("p1", ""): 80.0, ("p3", ""): 100.0},
|
|
)
|
|
assert out[("p1", "")]["archivedQty"] == 120.0
|
|
assert out[("p1", "")]["arrangedUnarchivedQty"] == 80.0
|
|
assert out[("p1", "")]["arrangedTotalQty"] == 200.0
|
|
assert out[("p2", "")]["arrangedTotalQty"] == 50.0
|
|
assert out[("p3", "")]["arrangedTotalQty"] == 100.0
|
|
|
|
|
|
def test_sum_qtys_explicit_delivery_stop_only():
|
|
load_products = {
|
|
"L1": [SimpleNamespace(product_id="p1", plan_qty=10)],
|
|
}
|
|
delivery_products = {
|
|
"D1": [SimpleNamespace(product_id="p1", plan_qty=40)],
|
|
"D2": [SimpleNamespace(product_id="p1", plan_qty=50)],
|
|
}
|
|
qtys = sum_qtys_for_scope(
|
|
load_products_by_load=load_products,
|
|
delivery_products_by_delivery=delivery_products,
|
|
all_load_ids={"L1"},
|
|
all_delivery_ids={"D1", "D2"},
|
|
scope={"load_ids": set(), "delivery_ids": {"D1"}, "use_all_stops": False},
|
|
)
|
|
assert qtys == {("p1", ""): 40.0}
|
|
|
|
|
|
def test_legacy_use_all_stops_prefers_delivery():
|
|
load_products = {
|
|
"L1": [SimpleNamespace(product_id="p1", plan_qty=99)],
|
|
}
|
|
delivery_products = {
|
|
"D1": [SimpleNamespace(product_id="p1", plan_qty=40)],
|
|
}
|
|
qtys = sum_qtys_for_scope(
|
|
load_products_by_load=load_products,
|
|
delivery_products_by_delivery=delivery_products,
|
|
all_load_ids={"L1"},
|
|
all_delivery_ids={"D1"},
|
|
scope={"load_ids": set(), "delivery_ids": set(), "use_all_stops": True},
|
|
)
|
|
assert qtys == {("p1", ""): 40.0}
|
|
|
|
|
|
def test_resolve_scopes_marks_legacy():
|
|
links = [
|
|
SimpleNamespace(plan_id="A", side="delivery", stop_id="D1"),
|
|
]
|
|
scopes = resolve_linked_plan_scopes(links, ["B"])
|
|
assert scopes["A"]["delivery_ids"] == {"D1"}
|
|
assert scopes["A"]["has_explicit"] is True
|
|
assert scopes["B"]["use_all_stops"] is True
|
|
|
|
|
|
def test_accumulate_stop_products():
|
|
acc = defaultdict(float)
|
|
accumulate_stop_products(
|
|
acc,
|
|
[
|
|
SimpleNamespace(product_id="p1", plan_qty=DecimalLike(12.5)),
|
|
SimpleNamespace(product_id="p1", plan_qty=7.5),
|
|
SimpleNamespace(product_id="", plan_qty=9),
|
|
],
|
|
)
|
|
assert acc[("p1", "")] == 20.0
|
|
|
|
|
|
def test_accumulate_archive_actual_load_uses_inbound_qty():
|
|
acc = defaultdict(float)
|
|
accumulate_stop_products(
|
|
acc,
|
|
[
|
|
SimpleNamespace(product_id="p1", plan_qty=100, inbound_qty=88),
|
|
SimpleNamespace(product_id="p1", plan_qty=50, inbound_qty=None),
|
|
],
|
|
stop_kind="load",
|
|
qty_mode="archive_actual",
|
|
)
|
|
# 88 + 回退 plan_qty 50
|
|
assert acc[("p1", "")] == 138.0
|
|
|
|
|
|
def test_accumulate_archive_actual_delivery_uses_outbound_qty():
|
|
acc = defaultdict(float)
|
|
accumulate_stop_products(
|
|
acc,
|
|
[SimpleNamespace(product_id="p1", plan_qty=100, outbound_qty=66)],
|
|
stop_kind="delivery",
|
|
qty_mode="archive_actual",
|
|
)
|
|
assert acc[("p1", "")] == 66.0
|
|
|
|
|
|
def test_accumulate_archive_actual_zero_does_not_fallback():
|
|
acc = defaultdict(float)
|
|
accumulate_stop_products(
|
|
acc,
|
|
[SimpleNamespace(product_id="p1", plan_qty=100, inbound_qty=0)],
|
|
stop_kind="load",
|
|
qty_mode="archive_actual",
|
|
)
|
|
assert acc[("p1", "")] == 0.0
|
|
|
|
|
|
def test_sum_qtys_archive_actual_mode_on_scoped_stops():
|
|
load_products = {
|
|
"L1": [SimpleNamespace(product_id="p1", plan_qty=10, inbound_qty=7)],
|
|
}
|
|
delivery_products = {
|
|
"D1": [SimpleNamespace(product_id="p1", plan_qty=40, outbound_qty=33)],
|
|
}
|
|
qtys = sum_qtys_for_scope(
|
|
load_products_by_load=load_products,
|
|
delivery_products_by_delivery=delivery_products,
|
|
all_load_ids={"L1"},
|
|
all_delivery_ids={"D1"},
|
|
scope={"load_ids": {"L1"}, "delivery_ids": {"D1"}, "use_all_stops": False},
|
|
qty_mode="archive_actual",
|
|
)
|
|
assert qtys == {("p1", ""): 40.0} # 7 + 33
|
|
|
|
|
|
def test_sum_qtys_default_mode_ignores_actual_fields():
|
|
load_products = {
|
|
"L1": [SimpleNamespace(product_id="p1", plan_qty=10, inbound_qty=7)],
|
|
}
|
|
delivery_products = {
|
|
"D1": [SimpleNamespace(product_id="p1", plan_qty=40, outbound_qty=33)],
|
|
}
|
|
qtys = sum_qtys_for_scope(
|
|
load_products_by_load=load_products,
|
|
delivery_products_by_delivery=delivery_products,
|
|
all_load_ids={"L1"},
|
|
all_delivery_ids={"D1"},
|
|
scope={"load_ids": set(), "delivery_ids": {"D1"}, "use_all_stops": False},
|
|
)
|
|
assert qtys == {("p1", ""): 40.0}
|
|
|
|
|
|
def test_match_contract_product_for_key_picks_package():
|
|
products = [
|
|
SimpleNamespace(product_id="SXWL007344", package_spec="25kg/桶", quantity=15),
|
|
SimpleNamespace(product_id="SXWL007344", package_spec="槽罐车", quantity=700),
|
|
]
|
|
bucket = match_contract_product_for_key(
|
|
products, product_package_key("SXWL007344", "25kg/桶")
|
|
)
|
|
tank = match_contract_product_for_key(
|
|
products, product_package_key("SXWL007344", " 槽罐车 ")
|
|
)
|
|
assert bucket is not None and float(bucket.quantity) == 15.0
|
|
assert tank is not None and float(tank.quantity) == 700.0
|
|
|
|
|
|
def test_match_contract_product_for_key_sums_duplicate_keys():
|
|
products = [
|
|
SimpleNamespace(product_id="p1", package_spec="槽罐车", quantity=100),
|
|
SimpleNamespace(product_id="p1", package_spec="槽罐车", quantity=50),
|
|
]
|
|
row = match_contract_product_for_key(products, ("p1", "槽罐车"))
|
|
assert row is not None and float(row.quantity) == 150.0
|
|
|
|
|
|
def test_match_contract_product_for_key_missing():
|
|
products = [SimpleNamespace(product_id="p1", package_spec="槽罐车", quantity=10)]
|
|
assert match_contract_product_for_key(products, ("p1", "25kg/桶")) is None
|
|
|
|
|
|
class DecimalLike:
|
|
def __init__(self, v):
|
|
self.v = v
|
|
|
|
def __float__(self):
|
|
return float(self.v)
|