from __future__ import annotations import unittest from datetime import UTC, datetime, timedelta from decimal import Decimal from sqlalchemy import BigInteger, create_engine 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.models.base import Base # noqa: E402 class WarehouseTransactionLedgerTest(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_ledger_data(self) -> None: from app.models.master_data import Item, Warehouse from app.models.operations import InventoryTxn, StockLot, WarehouseLocation from app.models.org import Department, Employee, User now = datetime(2026, 6, 1, 9, 0, tzinfo=UTC) department = Department( id=1, dept_code="D001", dept_name="仓储部", dept_type="WAREHOUSE", status="ACTIVE", created_at=now, updated_at=now, ) employee = Employee( id=1, employee_code="EMP001", employee_name="张仓管", dept_id=1, status="ACTIVE", created_at=now, updated_at=now, ) user = User( id=1, username="keeper", password_hash="x", employee_id=1, dept_id=1, nickname="张仓管", status="ACTIVE", created_at=now, updated_at=now, ) raw_wh = Warehouse( id=1, warehouse_code="WH-RAW-01", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE", created_at=now, updated_at=now, ) finished_wh = Warehouse( id=2, warehouse_code="WH-FG-01", warehouse_name="成品库", warehouse_type="FINISHED", status="ACTIVE", created_at=now, updated_at=now, ) raw_location = WarehouseLocation( id=1, warehouse_id=1, location_code="RAW-A", location_name="原料主库位", is_default=1, is_locked=0, status="ACTIVE", created_at=now, updated_at=now, ) finished_location = WarehouseLocation( id=2, warehouse_id=2, location_code="FG-A", location_name="成品主库位", is_default=1, is_locked=0, status="ACTIVE", created_at=now, updated_at=now, ) raw_item = Item( id=1, item_code="原材料00001", item_name="304不锈钢", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE", created_at=now, updated_at=now, ) finished_item = Item( id=2, item_code="产品需规00001", item_name="冲压支架", item_type="FINISHED_GOOD", unit_weight_kg=Decimal("2.5"), status="ACTIVE", created_at=now, updated_at=now, ) raw_lot = StockLot( id=1, lot_no="JH_原材料00001_0001", lot_role="INBOUND_RAW", item_id=1, warehouse_id=1, location_id=1, source_doc_type="OPENING_STOCK", source_doc_id=1, inbound_qty=0, inbound_weight_kg=1000, remaining_qty=0, remaining_weight_kg=850, locked_qty=0, locked_weight_kg=0, unit_cost=Decimal("10"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) finished_lot = StockLot( id=2, lot_no="FG-20260601-001", lot_role="INBOUND_FINISHED", item_id=2, warehouse_id=2, location_id=2, source_doc_type="FG_RECEIPT", source_doc_id=9, inbound_qty=100, inbound_weight_kg=250, remaining_qty=80, remaining_weight_kg=200, locked_qty=0, locked_weight_kg=0, unit_cost=Decimal("18"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) transactions = [ InventoryTxn( id=1, txn_no="TXN-RAW-IN", txn_type="OPENING_IN", item_id=1, warehouse_id=1, location_id=1, lot_id=1, qty_change=0, weight_change_kg=1000, unit_cost=10, amount=10000, source_doc_type="OPENING_STOCK", source_doc_id=1, biz_time=now, operator_user_id=1, remark="期初入库 304", created_at=now, updated_at=now, ), InventoryTxn( id=2, txn_no="TXN-RAW-OUT", txn_type="MATERIAL_ISSUE", item_id=1, warehouse_id=1, location_id=1, lot_id=1, qty_change=0, weight_change_kg=-150, unit_cost=10, amount=1500, source_doc_type="WORK_ORDER", source_doc_id=2, biz_time=now + timedelta(hours=1), operator_user_id=1, remark="生产出库", created_at=now, updated_at=now, ), InventoryTxn( id=3, txn_no="TXN-FG-IN", txn_type="FG_IN", item_id=2, warehouse_id=2, location_id=2, lot_id=2, qty_change=100, weight_change_kg=250, unit_cost=18, amount=1800, source_doc_type="FG_RECEIPT", source_doc_id=9, biz_time=now + timedelta(hours=2), operator_user_id=1, remark="成品入库", created_at=now, updated_at=now, ), ] self.db.add_all( [ department, employee, user, raw_wh, finished_wh, raw_location, finished_location, raw_item, finished_item, raw_lot, finished_lot, *transactions, ] ) self.db.commit() def test_transaction_ledger_filters_by_warehouse_type_and_returns_summary(self) -> None: from app.api.routes.inventory import list_inventory_transaction_ledger self.seed_ledger_data() result = list_inventory_transaction_ledger( warehouse_type="RAW", page=1, page_size=10, db=self.db, ) self.assertEqual(result.total, 2) self.assertEqual([row.warehouse_type for row in result.items], ["RAW", "RAW"]) self.assertEqual(result.summary.in_count, 1) self.assertEqual(result.summary.out_count, 1) self.assertEqual(Decimal(str(result.summary.in_weight_kg)), Decimal("1000.0")) self.assertEqual(Decimal(str(result.summary.out_weight_kg)), Decimal("150.0")) self.assertEqual(result.items[0].txn_type, "MATERIAL_ISSUE") self.assertEqual(result.items[0].direction, "OUT") def test_transaction_ledger_supports_keyword_direction_pagination_and_raw_qty_zero(self) -> None: from app.api.routes.inventory import list_inventory_transaction_ledger self.seed_ledger_data() result = list_inventory_transaction_ledger( warehouse_type="RAW", keyword="304", direction="OUT", page=1, page_size=1, db=self.db, ) self.assertEqual(result.total, 1) self.assertEqual(len(result.items), 1) row = result.items[0] self.assertEqual(row.txn_no, "TXN-RAW-OUT") self.assertEqual(row.item_code, "原材料00001") self.assertEqual(row.lot_no, "JH_原材料00001_0001") self.assertEqual(row.qty_change, 0) self.assertEqual(row.operator_name, "张仓管") self.assertEqual(result.summary.out_count, 1) def test_transaction_ledger_rejects_unknown_warehouse_type(self) -> None: from app.api.routes.inventory import list_inventory_transaction_ledger from fastapi import HTTPException self.seed_ledger_data() with self.assertRaises(HTTPException) as exc: list_inventory_transaction_ledger( warehouse_type="UNKNOWN", page=1, page_size=10, db=self.db, ) self.assertEqual(exc.exception.status_code, 400) self.assertIn("未知仓库类型", exc.exception.detail)