274 lines
11 KiB
Python
274 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
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.miniapp # noqa: E402,F401
|
|
import app.models.operations # noqa: E402,F401
|
|
import app.models.org # noqa: E402,F401
|
|
import app.models.planning # noqa: E402,F401
|
|
import app.models.sales # noqa: E402,F401
|
|
from app.api.routes.inventory import create_warehouse_outbound # noqa: E402
|
|
from app.api.routes.purchase import create_purchase_order, create_purchase_receipt, lock_purchase_order # noqa: E402
|
|
from app.api.routes.quality import inspect_receipt_item # noqa: E402
|
|
from app.models.base import Base # noqa: E402
|
|
from app.models.master_data import Item, Material, StockBalance, Unit, Warehouse # noqa: E402
|
|
from app.models.operations import InventoryTxn, PurchaseOrderItem, PurchaseReceiptItem, StockLot, Supplier # noqa: E402
|
|
from app.models.org import Department, Employee, Permission, User # noqa: E402
|
|
from app.schemas.operations import ( # noqa: E402
|
|
PurchaseOrderCreate,
|
|
PurchaseOrderItemCreate,
|
|
PurchaseReceiptCreate,
|
|
PurchaseReceiptItemCreate,
|
|
QualityInspectionAction,
|
|
WarehouseOutboundCreate,
|
|
)
|
|
from app.services.auth import AuthContext # noqa: E402
|
|
from app.services.operations import ( # noqa: E402
|
|
get_inventory_txn_ledger_query,
|
|
get_inventory_txn_query,
|
|
get_stock_balances_query,
|
|
get_stock_lots_query,
|
|
)
|
|
|
|
|
|
class AuxiliaryQuantityFlowTest(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()
|
|
self._seed_master_data()
|
|
|
|
def tearDown(self) -> None:
|
|
self.db.close()
|
|
|
|
def _seed_master_data(self) -> None:
|
|
self.department = Department(id=1, dept_code="D-AUX", dept_name="仓库部", dept_type="WAREHOUSE", status="ACTIVE")
|
|
self.employee = Employee(id=1, employee_code="EMP-AUX-001", employee_name="辅料仓管", dept_id=1, status="ACTIVE")
|
|
self.user = User(
|
|
id=1,
|
|
username="aux_tester",
|
|
password_hash="x",
|
|
employee_id=1,
|
|
dept_id=1,
|
|
nickname="辅料测试员",
|
|
is_super_admin=1,
|
|
status="ACTIVE",
|
|
)
|
|
self.purchase_receipt_permission = Permission(
|
|
id=1,
|
|
permission_code="MENU_PURCHASE_RECEIPT",
|
|
permission_name="到货入库",
|
|
module_code="PURCHASE",
|
|
action_code="MENU",
|
|
status="ACTIVE",
|
|
created_at=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
)
|
|
self.unit = Unit(id=1, unit_code="PCS_BOX", unit_name="个", precision_digits=0)
|
|
self.aux_item = Item(
|
|
id=1,
|
|
item_code="辅料00001",
|
|
item_name="包装薄膜",
|
|
item_type="RAW_MATERIAL",
|
|
specification="500*300",
|
|
unit_weight_kg=Decimal("0"),
|
|
sales_unit_id=1,
|
|
stock_unit_id=1,
|
|
purchase_unit_id=1,
|
|
status="ACTIVE",
|
|
)
|
|
self.material = Material(
|
|
item_id=1,
|
|
default_supplier_id=1,
|
|
purchase_calc_mode="BY_QTY",
|
|
min_purchase_qty=Decimal("0"),
|
|
purchase_multiple_qty=Decimal("0"),
|
|
purchase_multiple_weight_kg=Decimal("0"),
|
|
lead_time_days=0,
|
|
)
|
|
self.supplier = Supplier(
|
|
id=1,
|
|
supplier_code="SUP-AUX-001",
|
|
supplier_name="辅料供应商",
|
|
short_name="辅料供应商",
|
|
lead_time_days=0,
|
|
default_tax_rate=Decimal("0"),
|
|
status="ACTIVE",
|
|
)
|
|
self.warehouse = Warehouse(id=1, warehouse_code="WH-AUX", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE")
|
|
self.db.add_all([
|
|
self.department,
|
|
self.employee,
|
|
self.user,
|
|
self.purchase_receipt_permission,
|
|
self.unit,
|
|
self.aux_item,
|
|
self.material,
|
|
self.supplier,
|
|
self.warehouse,
|
|
])
|
|
self.db.commit()
|
|
|
|
def _auth_context(self) -> AuthContext:
|
|
return AuthContext(
|
|
user=self.user,
|
|
employee=self.employee,
|
|
department=self.department,
|
|
role_codes=["ADMIN"],
|
|
role_names=["管理员"],
|
|
permission_codes=[],
|
|
)
|
|
|
|
def test_auxiliary_purchase_receipt_quality_and_production_out_use_quantity(self) -> None:
|
|
order = create_purchase_order(
|
|
PurchaseOrderCreate(
|
|
supplier_id=self.supplier.id,
|
|
expected_date=date.today(),
|
|
target_warehouse_type="AUX",
|
|
items=[
|
|
PurchaseOrderItemCreate(
|
|
material_item_id=self.aux_item.id,
|
|
order_qty=Decimal("10"),
|
|
unit_price=Decimal("2.5"),
|
|
)
|
|
],
|
|
),
|
|
db=self.db,
|
|
)
|
|
|
|
po_item = self.db.scalar(select(PurchaseOrderItem).where(PurchaseOrderItem.purchase_order_id == order.purchase_order_id))
|
|
self.assertEqual(float(po_item.order_qty), 10)
|
|
self.assertEqual(float(po_item.order_weight_kg), 0)
|
|
self.assertEqual(float(po_item.line_amount), 25)
|
|
self.assertEqual(float(order.total_amount), 25)
|
|
self.assertEqual(float(order.total_order_qty), 10)
|
|
self.assertEqual(float(order.total_order_weight_kg), 0)
|
|
|
|
lock_purchase_order(order.purchase_order_id, db=self.db)
|
|
create_purchase_receipt(
|
|
PurchaseReceiptCreate(
|
|
purchase_order_id=order.purchase_order_id,
|
|
warehouse_id=self.warehouse.id,
|
|
receiver_employee_id=self.employee.id,
|
|
items=[
|
|
PurchaseReceiptItemCreate(
|
|
purchase_order_item_id=po_item.id,
|
|
material_item_id=self.aux_item.id,
|
|
received_qty=Decimal("6"),
|
|
unit_cost=Decimal("2.5"),
|
|
)
|
|
],
|
|
),
|
|
context=self._auth_context(),
|
|
db=self.db,
|
|
)
|
|
|
|
receipt_item = self.db.scalar(select(PurchaseReceiptItem))
|
|
lot = self.db.scalar(select(StockLot))
|
|
balance = self.db.scalar(select(StockBalance))
|
|
inbound_txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "PURCHASE_IN"))
|
|
|
|
self.assertEqual(float(receipt_item.received_qty), 6)
|
|
self.assertEqual(float(receipt_item.received_weight_kg), 0)
|
|
self.assertEqual(float(lot.inbound_qty), 6)
|
|
self.assertEqual(float(lot.remaining_qty), 6)
|
|
self.assertEqual(float(lot.locked_qty), 6)
|
|
self.assertEqual(float(lot.inbound_weight_kg), 0)
|
|
self.assertEqual(float(balance.qty_on_hand), 6)
|
|
self.assertEqual(float(balance.qty_available), 0)
|
|
self.assertEqual(float(balance.weight_on_hand_kg), 0)
|
|
self.assertEqual(float(inbound_txn.qty_change), 6)
|
|
self.assertEqual(float(inbound_txn.weight_change_kg), 0)
|
|
self.assertEqual(float(inbound_txn.amount), 15)
|
|
|
|
inspect_receipt_item(
|
|
receipt_item.id,
|
|
QualityInspectionAction(result="PASS", remark="合格"),
|
|
context=self._auth_context(),
|
|
db=self.db,
|
|
)
|
|
|
|
self.db.refresh(po_item)
|
|
self.db.refresh(receipt_item)
|
|
self.db.refresh(lot)
|
|
self.db.refresh(balance)
|
|
self.assertEqual(float(receipt_item.accepted_qty), 6)
|
|
self.assertEqual(float(receipt_item.accepted_weight_kg), 0)
|
|
self.assertEqual(float(po_item.received_qty), 6)
|
|
self.assertEqual(float(po_item.received_weight_kg), 0)
|
|
self.assertEqual(po_item.status, "PARTIAL")
|
|
self.assertEqual(float(lot.locked_qty), 0)
|
|
self.assertEqual(float(balance.qty_available), 6)
|
|
|
|
balance_rows = self.db.execute(get_stock_balances_query()).mappings().all()
|
|
aux_balance_row = next(row for row in balance_rows if row["warehouse_type"] == "AUX")
|
|
self.assertEqual(float(aux_balance_row["qty_on_hand"]), 6)
|
|
self.assertEqual(float(aux_balance_row["qty_available"]), 6)
|
|
self.assertEqual(aux_balance_row["specification"], "500*300")
|
|
self.assertEqual(aux_balance_row["stock_unit_name"], "个")
|
|
|
|
lot_rows = self.db.execute(get_stock_lots_query()).mappings().all()
|
|
aux_lot_row = next(row for row in lot_rows if row["warehouse_type"] == "AUX")
|
|
self.assertEqual(float(aux_lot_row["inbound_qty"]), 6)
|
|
self.assertEqual(float(aux_lot_row["remaining_qty"]), 6)
|
|
self.assertEqual(float(aux_lot_row["locked_qty"]), 0)
|
|
|
|
txn_rows = self.db.execute(get_inventory_txn_query()).mappings().all()
|
|
aux_purchase_txn_row = next(row for row in txn_rows if row["warehouse_type"] == "AUX" and row["txn_type"] == "PURCHASE_IN")
|
|
self.assertEqual(float(aux_purchase_txn_row["qty_change"]), 6)
|
|
|
|
ledger_rows = self.db.execute(get_inventory_txn_ledger_query(warehouse_type="AUX")).mappings().all()
|
|
aux_purchase_ledger_row = next(row for row in ledger_rows if row["txn_type"] == "PURCHASE_IN")
|
|
self.assertEqual(float(aux_purchase_ledger_row["qty_change"]), 6)
|
|
|
|
create_warehouse_outbound(
|
|
WarehouseOutboundCreate(
|
|
biz_type="PRODUCTION_OUT",
|
|
item_id=self.aux_item.id,
|
|
warehouse_id=self.warehouse.id,
|
|
outbound_qty=Decimal("2"),
|
|
remark="包装一线使用",
|
|
),
|
|
context=self._auth_context(),
|
|
db=self.db,
|
|
)
|
|
|
|
self.db.refresh(lot)
|
|
self.db.refresh(balance)
|
|
outbound_txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "PRODUCTION_OUT"))
|
|
self.assertEqual(float(lot.remaining_qty), 4)
|
|
self.assertEqual(float(lot.remaining_weight_kg), 0)
|
|
self.assertEqual(float(balance.qty_on_hand), 4)
|
|
self.assertEqual(float(balance.qty_available), 4)
|
|
self.assertEqual(float(outbound_txn.qty_change), -2)
|
|
self.assertEqual(float(outbound_txn.weight_change_kg), 0)
|
|
self.assertEqual(float(outbound_txn.amount), 5)
|
|
self.assertIn("用途：包装一线使用", outbound_txn.remark)
|
|
|
|
lot_rows_after_outbound = self.db.execute(get_stock_lots_query()).mappings().all()
|
|
aux_lot_after_outbound = next(row for row in lot_rows_after_outbound if row["warehouse_type"] == "AUX")
|
|
self.assertEqual(float(aux_lot_after_outbound["remaining_qty"]), 4)
|
|
|
|
txn_rows_after_outbound = self.db.execute(get_inventory_txn_query()).mappings().all()
|
|
aux_outbound_txn_row = next(row for row in txn_rows_after_outbound if row["warehouse_type"] == "AUX" and row["txn_type"] == "PRODUCTION_OUT")
|
|
self.assertEqual(float(aux_outbound_txn_row["qty_change"]), -2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|