858 lines
32 KiB
Python
858 lines
32 KiB
Python
from __future__ import annotations
|
||
|
||
import unittest
|
||
from datetime import UTC, datetime
|
||
from decimal import Decimal
|
||
|
||
from fastapi import HTTPException
|
||
from sqlalchemy import BigInteger, create_engine, select
|
||
from sqlalchemy.ext.compiler import compiles
|
||
from sqlalchemy.orm import Session, sessionmaker
|
||
|
||
|
||
@compiles(BigInteger, "sqlite")
|
||
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
|
||
_ = type_, compiler, kw
|
||
return "INTEGER"
|
||
|
||
|
||
import app.models.master_data # noqa: E402,F401
|
||
import app.models.operations # noqa: E402,F401
|
||
import app.models.org # noqa: E402,F401
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound # noqa: E402
|
||
from app.models.base import Base # noqa: E402
|
||
from app.models.master_data import Bom, BomItem, Item, Warehouse # noqa: E402
|
||
from app.models.operations import InventoryTxn, Process, ProcessRoute, ProcessRouteOperation, StockLot, WarehouseLocation, WorkCenter # noqa: E402
|
||
from app.schemas.operations import WarehouseInboundCreate, WarehouseOutboundCreate # noqa: E402
|
||
|
||
|
||
class User:
|
||
id = 1
|
||
|
||
|
||
class Context:
|
||
user = User()
|
||
|
||
|
||
class SemiFinishedOperationInventoryTest(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
|
||
Base.metadata.create_all(engine)
|
||
self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
|
||
self.db: Session = self.SessionLocal()
|
||
|
||
def tearDown(self) -> None:
|
||
self.db.close()
|
||
|
||
def _seed_route(self) -> tuple[Item, Warehouse, ProcessRoute, ProcessRouteOperation, ProcessRouteOperation]:
|
||
now = datetime.now(UTC)
|
||
product = Item(
|
||
id=1,
|
||
item_code="产品需规00001",
|
||
item_name="五金支架",
|
||
item_type="FINISHED_GOOD",
|
||
unit_weight_kg=Decimal("0.6"),
|
||
safety_stock_weight_kg=Decimal("0"),
|
||
scrap_sale_price=Decimal("0"),
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
process = Process(
|
||
id=1,
|
||
process_code="GX01",
|
||
process_name="冲压",
|
||
process_type="INHOUSE",
|
||
is_report_required=1,
|
||
is_quality_gate=0,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
center = WorkCenter(
|
||
id=1,
|
||
center_code="WC01",
|
||
center_name="一号工位",
|
||
dept_id=1,
|
||
center_type="INHOUSE",
|
||
capacity_hours_per_day=Decimal("8"),
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
warehouse = Warehouse(
|
||
id=1,
|
||
warehouse_code="WH-SEMI-T",
|
||
warehouse_name="半成品测试仓",
|
||
warehouse_type="SEMI",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([product, process, center, warehouse])
|
||
self.db.flush()
|
||
location = WarehouseLocation(
|
||
id=1,
|
||
warehouse_id=warehouse.id,
|
||
location_code="SEMI-A",
|
||
location_name="半成品默认库位",
|
||
is_default=1,
|
||
is_locked=0,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
route = ProcessRoute(
|
||
id=1,
|
||
route_code="R-001",
|
||
route_name="五金支架工艺",
|
||
product_item_id=product.id,
|
||
version_no="V1.0",
|
||
is_default=1,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([location, route])
|
||
self.db.flush()
|
||
op1 = ProcessRouteOperation(
|
||
id=1,
|
||
route_id=route.id,
|
||
seq_no=1,
|
||
process_id=process.id,
|
||
work_center_id=center.id,
|
||
operation_name="冲压",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
op2 = ProcessRouteOperation(
|
||
id=2,
|
||
route_id=route.id,
|
||
seq_no=2,
|
||
process_id=process.id,
|
||
work_center_id=center.id,
|
||
operation_name="折弯",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([op1, op2])
|
||
self.db.commit()
|
||
return product, warehouse, route, op1, op2
|
||
|
||
def _seed_raw_source_lots(self, product: Item) -> tuple[StockLot, StockLot]:
|
||
now = datetime.now(UTC)
|
||
material = Item(
|
||
id=2,
|
||
item_code="原材料00001",
|
||
item_name="冷轧钢板",
|
||
item_type="RAW_MATERIAL",
|
||
unit_weight_kg=Decimal("0"),
|
||
safety_stock_weight_kg=Decimal("0"),
|
||
scrap_sale_price=Decimal("0"),
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
raw_warehouse = Warehouse(
|
||
id=2,
|
||
warehouse_code="WH-RAW-T",
|
||
warehouse_name="原材料测试仓",
|
||
warehouse_type="RAW",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([material, raw_warehouse])
|
||
self.db.flush()
|
||
raw_location = WarehouseLocation(
|
||
id=2,
|
||
warehouse_id=raw_warehouse.id,
|
||
location_code="RAW-A",
|
||
location_name="原材料默认库位",
|
||
is_default=1,
|
||
is_locked=0,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
bom = Bom(
|
||
id=1,
|
||
bom_code="BOM-SEMI-001",
|
||
product_item_id=product.id,
|
||
version_no="V1.0",
|
||
is_default=1,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
bom_item = BomItem(
|
||
id=1,
|
||
bom_id=bom.id,
|
||
seq_no=1,
|
||
material_item_id=material.id,
|
||
usage_type="WEIGHT",
|
||
gross_weight_per_piece_kg=Decimal("0.6"),
|
||
loss_rate=Decimal("0"),
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
lot_a = StockLot(
|
||
id=10,
|
||
lot_no="JH_冷轧钢板00001_0001",
|
||
item_id=material.id,
|
||
warehouse_id=raw_warehouse.id,
|
||
location_id=raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=1,
|
||
source_line_id=1,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("100"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("0"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("4.5"),
|
||
quality_status="PASS",
|
||
status="DEPLETED",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
lot_b = StockLot(
|
||
id=11,
|
||
lot_no="JH_冷轧钢板00001_0002",
|
||
item_id=material.id,
|
||
warehouse_id=raw_warehouse.id,
|
||
location_id=raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=2,
|
||
source_line_id=2,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("200"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("200"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("4.8"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([raw_location, bom, bom_item, lot_a, lot_b])
|
||
self.db.commit()
|
||
return lot_a, lot_b
|
||
|
||
def _seed_finished_warehouse(self) -> Warehouse:
|
||
now = datetime.now(UTC)
|
||
warehouse = Warehouse(
|
||
id=3,
|
||
warehouse_code="WH-FG-T",
|
||
warehouse_name="成品测试仓",
|
||
warehouse_type="FINISHED",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
location = WarehouseLocation(
|
||
id=3,
|
||
warehouse_id=warehouse.id,
|
||
location_code="FG-A",
|
||
location_name="成品默认库位",
|
||
is_default=1,
|
||
is_locked=0,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([warehouse, location])
|
||
self.db.commit()
|
||
return warehouse
|
||
|
||
def test_semi_inbound_records_product_operation_source_line(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=120,
|
||
unit_cost=8,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
source_material_summary="产品：五金支架；已完成工序：第1道 冲压",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.source_line_id, op1.id)
|
||
self.assertEqual(lot.source_doc_id, route.id)
|
||
self.assertIn("第1道", lot.source_material_summary)
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
||
self.assertEqual(txn.source_line_id, op1.id)
|
||
|
||
def test_semi_wip_inbound_accepts_quantity_when_weight_is_empty(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=12,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
source_material_summary="产品：五金支架；已完成工序：第1道 冲压",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.inbound_qty, Decimal("12.000000"))
|
||
self.assertEqual(lot.remaining_qty, Decimal("12.000000"))
|
||
self.assertEqual(lot.inbound_weight_kg, Decimal("0.000000"))
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
||
self.assertEqual(txn.qty_change, Decimal("12.000000"))
|
||
self.assertEqual(txn.weight_change_kg, Decimal("0.000000"))
|
||
|
||
def test_semi_wip_inbound_requires_weight_or_quantity(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=0,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("至少填写一项", caught.exception.detail)
|
||
|
||
def test_semi_wip_inbound_rejects_weight_and_quantity_together(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=60,
|
||
inbound_qty=100,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("只能选择按件或按重一种口径", caught.exception.detail)
|
||
|
||
def test_semi_wip_inbound_follows_existing_piece_basis(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=12,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=30,
|
||
inbound_qty=0,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("已有库存按件管理", caught.exception.detail)
|
||
|
||
def test_semi_wip_inbound_follows_existing_weight_basis(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=30,
|
||
inbound_qty=0,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=12,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("已有库存按重管理", caught.exception.detail)
|
||
|
||
def test_semi_wip_inbound_records_selected_raw_source_lots_without_deducting_them(self) -> None:
|
||
product, warehouse, route, op1, _ = self._seed_route()
|
||
lot_a, lot_b = self._seed_raw_source_lots(product)
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=60,
|
||
inbound_qty=0,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
source_lot_ids=[lot_a.id, lot_b.id],
|
||
source_material_summary="产品：五金支架；已完成工序：第1道 冲压",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.source_material_lot_id, lot_a.id)
|
||
self.assertEqual(lot.source_material_sub_batch_no, f"{lot_a.lot_no}、{lot_b.lot_no}")
|
||
self.assertIn("来源库存批次号", lot.source_material_summary)
|
||
self.assertIn(lot_a.lot_no, lot.source_material_summary)
|
||
self.assertIn(lot_b.lot_no, lot.source_material_summary)
|
||
self.db.refresh(lot_a)
|
||
self.db.refresh(lot_b)
|
||
self.assertEqual(lot_a.remaining_weight_kg, Decimal("0.000000"))
|
||
self.assertEqual(lot_b.remaining_weight_kg, Decimal("200.000000"))
|
||
|
||
def test_semi_outsourcing_inbound_accepts_quantity_when_weight_is_empty(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=8,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
waybill_no="WW-001",
|
||
freight_amount=0,
|
||
source_material_summary="产品：五金支架；已完成工序：第2道 折弯",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.inbound_qty, Decimal("8.000000"))
|
||
self.assertEqual(lot.remaining_qty, Decimal("8.000000"))
|
||
self.assertEqual(lot.inbound_weight_kg, Decimal("0.000000"))
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
||
self.assertEqual(txn.qty_change, Decimal("8.000000"))
|
||
self.assertEqual(txn.weight_change_kg, Decimal("0.000000"))
|
||
|
||
def test_semi_outsourcing_inbound_records_selected_raw_source_lots_without_deducting_them(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
lot_a, lot_b = self._seed_raw_source_lots(product)
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=30,
|
||
inbound_qty=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
source_lot_ids=[lot_a.id, lot_b.id],
|
||
waybill_no="WW-002",
|
||
freight_amount=12,
|
||
provider_name="委外厂甲",
|
||
source_material_summary="产品：五金支架；已完成工序：第2道 折弯",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.source_material_lot_id, lot_a.id)
|
||
self.assertEqual(lot.source_material_sub_batch_no, f"{lot_a.lot_no}、{lot_b.lot_no}")
|
||
self.assertIn("来源库存批次号", lot.source_material_summary)
|
||
self.assertIn(lot_a.lot_no, lot.source_material_summary)
|
||
self.assertIn(lot_b.lot_no, lot.source_material_summary)
|
||
self.db.refresh(lot_a)
|
||
self.db.refresh(lot_b)
|
||
self.assertEqual(lot_a.remaining_weight_kg, Decimal("0.000000"))
|
||
self.assertEqual(lot_b.remaining_weight_kg, Decimal("200.000000"))
|
||
|
||
def test_finished_outsourcing_inbound_uses_quantity_and_optional_raw_source_lots(self) -> None:
|
||
product, _, _, _, _ = self._seed_route()
|
||
warehouse = self._seed_finished_warehouse()
|
||
_, lot_b = self._seed_raw_source_lots(product)
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=12,
|
||
unit_cost=6.5,
|
||
source_lot_ids=[lot_b.id],
|
||
waybill_no="WW-FG-001",
|
||
freight_amount=15,
|
||
provider_name="委外厂甲",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.inbound_qty, Decimal("12.000000"))
|
||
self.assertEqual(lot.remaining_qty, Decimal("12.000000"))
|
||
self.assertEqual(lot.inbound_weight_kg, Decimal("0.000000"))
|
||
self.assertEqual(lot.remaining_weight_kg, Decimal("0.000000"))
|
||
self.assertEqual(lot.source_material_lot_id, lot_b.id)
|
||
self.assertEqual(lot.source_material_sub_batch_no, lot_b.lot_no)
|
||
self.assertIn("来源库存批次号", lot.source_material_summary)
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
||
self.assertEqual(txn.qty_change, Decimal("12.000000"))
|
||
self.assertEqual(txn.weight_change_kg, Decimal("0.000000"))
|
||
self.assertEqual(txn.amount, Decimal("78.00"))
|
||
|
||
def test_finished_outsourcing_inbound_allows_no_raw_source_lot(self) -> None:
|
||
product, _, _, _, _ = self._seed_route()
|
||
warehouse = self._seed_finished_warehouse()
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=5,
|
||
unit_cost=7,
|
||
source_lot_ids=[],
|
||
waybill_no="WW-FG-002",
|
||
freight_amount=0,
|
||
provider_name="委外厂自供原料",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertEqual(lot.inbound_qty, Decimal("5.000000"))
|
||
self.assertEqual(lot.inbound_weight_kg, Decimal("0.000000"))
|
||
self.assertIsNone(lot.source_material_lot_id)
|
||
self.assertIsNone(lot.source_material_sub_batch_no)
|
||
self.assertIn("委外厂自供原料", lot.source_material_summary)
|
||
|
||
def test_semi_outbound_deducts_only_selected_product_operation(self) -> None:
|
||
product, warehouse, route, op1, op2 = self._seed_route()
|
||
first = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=50,
|
||
unit_cost=8,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op1.id,
|
||
source_material_summary="产品：五金支架；已完成工序：第1道 冲压",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
second = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=80,
|
||
unit_cost=9,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
source_material_summary="产品：五金支架；已完成工序：第2道 折弯",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="WIP_OUT",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
outbound_weight_kg=30,
|
||
target_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
target_doc_id=route.id,
|
||
target_doc_line_id=op2.id,
|
||
target_material_sub_batch_no="第2道 折弯",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
first_lot = self.db.get(StockLot, first.lot_id)
|
||
second_lot = self.db.get(StockLot, second.lot_id)
|
||
self.assertEqual(first_lot.remaining_weight_kg, Decimal("50.000000"))
|
||
self.assertEqual(second_lot.remaining_weight_kg, Decimal("50.000000"))
|
||
|
||
def test_semi_wip_outbound_accepts_quantity_when_weight_is_empty(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
inbound = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=20,
|
||
unit_cost=8,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
source_material_summary="产品：五金支架；已完成工序：第2道 折弯",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="WIP_OUT",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
outbound_weight_kg=0,
|
||
outbound_qty=5,
|
||
target_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
target_doc_id=route.id,
|
||
target_doc_line_id=op2.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, inbound.lot_id)
|
||
self.assertEqual(lot.remaining_qty, Decimal("15.000000"))
|
||
self.assertEqual(lot.remaining_weight_kg, Decimal("0.000000"))
|
||
txn = self.db.scalar(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.lot_id == lot.id, InventoryTxn.txn_type == "WIP_OUT")
|
||
.order_by(InventoryTxn.id.desc())
|
||
)
|
||
self.assertEqual(txn.qty_change, Decimal("-5.000000"))
|
||
self.assertEqual(txn.weight_change_kg, Decimal("0.000000"))
|
||
self.assertIn(lot.lot_no, txn.remark)
|
||
|
||
def test_semi_wip_outbound_requires_weight_or_quantity(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="WIP_OUT",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
outbound_weight_kg=0,
|
||
outbound_qty=0,
|
||
target_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
target_doc_id=route.id,
|
||
target_doc_line_id=op2.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("至少填写一项", caught.exception.detail)
|
||
|
||
def test_semi_wip_outbound_follows_existing_piece_basis(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=20,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="WIP_OUT",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
outbound_weight_kg=10,
|
||
outbound_qty=0,
|
||
target_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
target_doc_id=route.id,
|
||
target_doc_line_id=op2.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("按件管理", caught.exception.detail)
|
||
|
||
def test_semi_wip_outbound_follows_existing_weight_basis(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="WIP_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=80,
|
||
inbound_qty=0,
|
||
unit_cost=0,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
with self.assertRaises(HTTPException) as caught:
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="WIP_OUT",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
outbound_weight_kg=0,
|
||
outbound_qty=5,
|
||
target_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
target_doc_id=route.id,
|
||
target_doc_line_id=op2.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(caught.exception.status_code, 400)
|
||
self.assertIn("按重管理", caught.exception.detail)
|
||
|
||
def test_semi_outsourcing_outbound_accepts_quantity_and_records_outsourcing_party(self) -> None:
|
||
product, warehouse, route, _, op2 = self._seed_route()
|
||
inbound = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_IN",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
inbound_weight_kg=0,
|
||
inbound_qty=10,
|
||
source_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
source_doc_id=route.id,
|
||
source_line_id=op2.id,
|
||
waybill_no="WW-IN-001",
|
||
freight_amount=0,
|
||
source_material_summary="产品：五金支架；已完成工序：第2道 折弯",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="OUTSOURCING_OUT",
|
||
item_id=product.id,
|
||
warehouse_id=warehouse.id,
|
||
outbound_weight_kg=0,
|
||
outbound_qty=4,
|
||
target_doc_type="PRODUCT_ROUTE_OPERATION",
|
||
target_doc_id=route.id,
|
||
target_doc_line_id=op2.id,
|
||
waybill_no="WW-OUT-001",
|
||
freight_amount=15,
|
||
outsourcing_party_name="委外厂乙",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, inbound.lot_id)
|
||
self.assertEqual(lot.remaining_qty, Decimal("6.000000"))
|
||
self.assertEqual(lot.remaining_weight_kg, Decimal("0.000000"))
|
||
txn = self.db.scalar(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.lot_id == lot.id, InventoryTxn.txn_type == "OUTSOURCING_OUT")
|
||
.order_by(InventoryTxn.id.desc())
|
||
)
|
||
self.assertEqual(txn.qty_change, Decimal("-4.000000"))
|
||
self.assertEqual(txn.weight_change_kg, Decimal("0.000000"))
|
||
self.assertIn(lot.lot_no, txn.remark)
|
||
self.assertIn("委外方：委外厂乙", txn.remark)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|