shanxia2.0/backend/tests/test_pl_contract_plan_qty.py

"""合同产品运输安排数量汇总单元测试。"""
from types import SimpleNamespace

from app.services.pl_contract_plan_qty import (
    _bucket_for_plan,
    accumulate_stop_products,
    merge_product_qty_maps,
    resolve_linked_plan_scopes,
    sum_qtys_for_scope,
)
from collections import defaultdict


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_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}


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

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