134 lines
3.6 KiB
Python
134 lines
3.6 KiB
Python
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(
|
|
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(
|
|
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")),
|
|
("pp1", Decimal("12"), "inbound", Decimal("120")),
|
|
]
|
|
|
|
|
|
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")
|