shanxia2.0/backend/tests/test_pl_contract_plan_qty.py

"""合同产品运输安排数量汇总单元测试。"""
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)