shanxia2.0/backend/tests/test_pl_plan_archive_stop_service.py

from decimal import Decimal

from app.schemas.pl.plan import (
    PlanArchiveDeliveryBlock,
    PlanArchiveLoadBlock,
    PlanArchiveStopProductItem,
)
from app.services.pl_plan_archive_stop_service import (
    iter_stop_archive_inventory_lines,
    sum_delivery_outbound_qty,
)


class _FakeRow:
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)


class _FakeQuery:
    def __init__(self, rows):
        self._rows = rows

    def filter(self, *args, **kwargs):
        return self

    def first(self):
        return self._rows[0] if self._rows else None

    def all(self):
        return self._rows

    def order_by(self, *args, **kwargs):
        return self


class _FakeDb:
    def __init__(self, mapping):
        self._mapping = mapping

    def query(self, model):
        return _FakeQuery(self._mapping.get(model, []))


def test_sum_delivery_outbound_qty():
    from app.models.pl_models import PlPlanDeliveryProduct

    db = _FakeDb(
        {
            PlPlanDeliveryProduct: [
                _FakeRow(outbound_qty=Decimal("10")),
                _FakeRow(outbound_qty=Decimal("5.5")),
                _FakeRow(outbound_qty=None),
            ]
        }
    )
    total = sum_delivery_outbound_qty(db, tenant_no="t1", plan_id="p1")
    assert total == Decimal("15.5")


def test_iter_stop_archive_inventory_lines_internal_load_outbound_and_delivery_inbound():
    from app.models.pl_models import (
        PlPlanDelivery,
        PlPlanDeliveryProduct,
        PlPlanLoad,
        PlPlanLoadProduct,
    )
    from app.services.pl_internal_customer import INTERNAL_CUSTOMER_NAMES

    internal_name = next(iter(INTERNAL_CUSTOMER_NAMES))

    db = _FakeDb(
        {
            PlPlanLoad: [_FakeRow(id="load1", customer_name=internal_name)],
            PlPlanLoadProduct: [
                _FakeRow(
                    id="load-item",
                    plan_product_id="pp1",
                    inbound_qty=Decimal("30"),
                    inbound_price=Decimal("100"),
                )
            ],
            PlPlanDelivery: [
                _FakeRow(id="d1", customer_name="外部客户有限公司"),
                _FakeRow(id="d2", customer_name=internal_name),
            ],
            PlPlanDeliveryProduct: [
                _FakeRow(
                    id="d2-item",
                    delivery_id="d2",
                    plan_product_id="pp1",
                    outbound_qty=Decimal("12"),
                    outbound_price=Decimal("120"),
                )
            ],
        }
    )

    lines = list(
        iter_stop_archive_inventory_lines(db, tenant_no="t1", plan_id="p1")
    )
    assert lines == [
        ("pp1", Decimal("30"), "outbound", Decimal("100"), "load", "load-item"),
        ("pp1", Decimal("12"), "inbound", Decimal("120"), "delivery", "d2-item"),
    ]


def test_archive_stop_product_item_schema_aliases():
    item = PlanArchiveStopProductItem.model_validate(
        {
            "id": "row1",
            "inboundQty": "8",
            "inboundPrice": "99.5",
        }
    )
    assert item.inbound_qty == Decimal("8")
    assert item.inbound_price == Decimal("99.5")

    load_block = PlanArchiveLoadBlock.model_validate(
        {
            "stopId": "load1",
            "products": [{"id": "row1", "inboundQty": 1}],
        }
    )
    assert load_block.stop_id == "load1"
    assert load_block.products[0].id == "row1"

    delivery_block = PlanArchiveDeliveryBlock.model_validate(
        {
            "stopId": "d1",
            "products": [{"id": "row2", "outboundQty": 2}],
        }
    )
    assert delivery_block.stop_id == "d1"
    assert delivery_block.products[0].outbound_qty == Decimal("2")