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


def test_match_shared_pool_line_json_package_spec():
    from app.utils.crm_package_spec import encode_package_specs

    products = [
        SimpleNamespace(
            id="L1",
            product_id="SXWL007344",
            package_spec=encode_package_specs(["25kg/桶", "槽罐车"]),
            quantity=715,
        )
    ]
    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 tank is not None
    assert float(bucket.quantity) == 715.0


def test_attach_plan_qty_shared_pool_breakdown():
    from app.utils.crm_package_spec import encode_package_specs

    products = [
        {
            "id": "L1",
            "productId": "pid1",
            "quantity": 715,
            "packageSpec": encode_package_specs(["25kg/桶", "槽罐车"]),
        }
    ]
    qty_map = {
        ("pid1", "25kg/桶"): {
            "archivedQty": 0.0,
            "arrangedUnarchivedQty": 10.0,
            "arrangedTotalQty": 10.0,
        },
        ("pid1", "槽罐车"): {
            "archivedQty": 0.0,
            "arrangedUnarchivedQty": 10.0,
            "arrangedTotalQty": 10.0,
        },
    }
    attach_plan_qty_fields_to_products(products, qty_map)
    assert products[0]["arrangedTotalQty"] == 20.0
    assert products[0]["remainingQty"] == 695.0
    assert products[0]["packageSpecs"] == ["25kg/桶", "槽罐车"]
    assert len(products[0]["packageQtyBreakdown"]) == 2
    assert products[0]["packageQtyBreakdown"][0]["arrangedTotalQty"] == 10.0
    assert products[0]["packageQtyBreakdown"][1]["arrangedTotalQty"] == 10.0


def test_sum_qty_stats_for_keys_shared_pool():
    from app.services.pl_contract_plan_qty import sum_qty_stats_for_keys

    qty_map = {
        ("pid1", "25kg/桶"): {
            "archivedQty": 0.0,
            "arrangedUnarchivedQty": 10.0,
            "arrangedTotalQty": 10.0,
        },
        ("pid1", "槽罐车"): {
            "archivedQty": 0.0,
            "arrangedUnarchivedQty": 10.0,
            "arrangedTotalQty": 10.0,
        },
    }
    stats = sum_qty_stats_for_keys(
        qty_map, [("pid1", "25kg/桶"), ("pid1", "槽罐车")]
    )
    assert stats["arrangedTotalQty"] == 20.0


class DecimalLike:
    def __init__(self, v):
        self.v = v

    def __float__(self):
        return float(self.v)


def test_accumulate_prefers_contract_product_id():
    acc = defaultdict(float)
    accumulate_stop_products(
        acc,
        [
            SimpleNamespace(
                product_id="p1",
                package_spec="槽罐车",
                plan_qty=10,
                contract_product_id="line-a",
            ),
            SimpleNamespace(
                product_id="p1",
                package_spec="槽罐车",
                plan_qty=20,
                contract_product_id="line-b",
            ),
        ],
    )
    from app.services.pl_contract_plan_qty import LINE_OCCUPANCY_PREFIX

    assert acc[(LINE_OCCUPANCY_PREFIX, "line-a")] == 10.0
    assert acc[(LINE_OCCUPANCY_PREFIX, "line-b")] == 20.0
    assert ("p1", "槽罐车") not in acc


def test_attach_plan_qty_independent_duplicate_lines():
    from app.services.pl_contract_plan_qty import LINE_OCCUPANCY_PREFIX

    products = [
        {"id": "line-a", "productId": "pid1", "packageSpec": "槽罐车", "quantity": 50},
        {"id": "line-b", "productId": "pid1", "packageSpec": "槽罐车", "quantity": 40},
    ]
    qty_map = {
        (LINE_OCCUPANCY_PREFIX, "line-a"): {
            "archivedQty": 0.0,
            "arrangedUnarchivedQty": 10.0,
            "arrangedTotalQty": 10.0,
        }
    }
    attach_plan_qty_fields_to_products(products, qty_map)
    assert products[0]["arrangedTotalQty"] == 10.0
    assert products[0]["remainingQty"] == 40.0
    assert products[1]["arrangedTotalQty"] == 0.0
    assert products[1]["remainingQty"] == 40.0


def test_attach_plan_qty_line_bound_multi_package_breakdown():
    """绑定合同行后数量记在 __line__；多包装拆开展示须带上对应包装的归档量。"""
    from app.utils.crm_package_spec import encode_package_specs
    from app.services.pl_contract_plan_qty import LINE_OCCUPANCY_PREFIX, package_display_key

    acc = defaultdict(float)
    accumulate_stop_products(
        acc,
        [
            SimpleNamespace(
                product_id="pid1",
                package_spec="20kg/袋",
                plan_qty=0.55,
                inbound_qty=0.55,
                contract_product_id="line-2",
            )
        ],
        stop_kind="load",
        qty_mode="archive_actual",
    )
    assert acc[(LINE_OCCUPANCY_PREFIX, "line-2")] == 0.55
    assert ("pid1", "20kg/袋") not in acc
    dkey = package_display_key("line-2", "20kg/袋")
    assert dkey in acc and acc[dkey] == 0.55

    products = [
        {
            "id": "line-2",
            "productId": "pid1",
            "quantity": 1,
            "packageSpec": encode_package_specs(["20kg/袋", "50Kg/桶", "1.2吨/桶"]),
        }
    ]
    qty_map = {
        (LINE_OCCUPANCY_PREFIX, "line-2"): {
            "archivedQty": 0.55,
            "arrangedUnarchivedQty": 0.0,
            "arrangedTotalQty": 0.55,
        },
        dkey: {
            "archivedQty": 0.55,
            "arrangedUnarchivedQty": 0.0,
            "arrangedTotalQty": 0.55,
        },
    }
    attach_plan_qty_fields_to_products(products, qty_map)
    assert products[0]["arrangedTotalQty"] == 0.55
    assert products[0]["remainingQty"] == 0.45
    by_spec = {x["packageSpec"]: x for x in products[0]["packageQtyBreakdown"]}
    assert by_spec["20kg/袋"]["archivedQty"] == 0.55
    assert by_spec["50Kg/桶"]["archivedQty"] == 0.0
    assert by_spec["1.2吨/桶"]["archivedQty"] == 0.0


def test_package_display_qty_isolated_per_contract_line():
    from app.utils.crm_package_spec import encode_package_specs
    from app.services.pl_contract_plan_qty import LINE_OCCUPANCY_PREFIX

    acc = defaultdict(float)
    accumulate_stop_products(
        acc,
        [
            SimpleNamespace(
                product_id="pid1",
                package_spec="20kg/袋",
                plan_qty=5.1,
                inbound_qty=5.1,
                contract_product_id="line-1",
            ),
            SimpleNamespace(
                product_id="pid1",
                package_spec="20kg/袋",
                plan_qty=0.55,
                inbound_qty=0.55,
                contract_product_id="line-2",
            ),
        ],
        stop_kind="load",
        qty_mode="archive_actual",
    )
    qty_map = merge_product_qty_maps(archived=dict(acc), arranged_unarchived={})
    products = [
        {
            "id": "line-1",
            "productId": "pid1",
            "quantity": 11,
            "packageSpec": encode_package_specs(["20kg/袋"]),
        },
        {
            "id": "line-2",
            "productId": "pid1",
            "quantity": 1,
            "packageSpec": encode_package_specs(["20kg/袋", "50Kg/桶", "1.2吨/桶"]),
        },
    ]
    attach_plan_qty_fields_to_products(products, qty_map)
    assert products[0]["archivedQty"] == 5.1
    assert products[0]["remainingQty"] == 5.9
    assert products[0]["packageQtyBreakdown"][0]["archivedQty"] == 5.1
    assert products[1]["archivedQty"] == 0.55
    assert products[1]["remainingQty"] == 0.45
    by_spec = {x["packageSpec"]: x for x in products[1]["packageQtyBreakdown"]}
    assert by_spec["20kg/袋"]["archivedQty"] == 0.55
    assert (LINE_OCCUPANCY_PREFIX, "line-1") in acc


def test_occupancy_keys_do_not_share_package_across_duplicate_lines():
    from app.services.pl_contract_plan_qty import occupancy_keys_for_contract_line

    lines = [
        SimpleNamespace(id="a", product_id="p1", package_spec="槽罐车"),
        SimpleNamespace(id="b", product_id="p1", package_spec="槽罐车"),
    ]
    keys_a = occupancy_keys_for_contract_line(lines[0], lines)
    keys_b = occupancy_keys_for_contract_line(lines[1], lines)
    assert ("__line__", "a") in keys_a
    assert ("p1", "槽罐车") in keys_a
    assert ("__line__", "b") in keys_b
    assert ("p1", "槽罐车") not in keys_b