from __future__ import annotations import unittest from datetime import UTC, datetime from decimal import Decimal from types import SimpleNamespace 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.document_archive # noqa: E402,F401 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.models.base import Base # noqa: E402 from app.models.document_archive import DocumentArchive # noqa: E402 from app.models.master_data import Item, Product, StockBalance, Unit, Warehouse # noqa: E402 from app.models.operations import InventoryTxn, StockLot, WarehouseLocation # noqa: E402 from app.models.org import Department, Employee, User # noqa: E402 from app.schemas.operations import ( # noqa: E402 SpecialWarehouseAdjustmentCreate, SpecialWarehouseAdjustmentLineCreate, ) from app.api.routes.inventory import create_special_warehouse_adjustment # noqa: E402 from app.services.document_archives import ARCHIVE_STATUS_READY, DOCUMENT_TYPE_WAREHOUSE_OPERATION # noqa: E402 from app.services.operations import get_inventory_txn_ledger_query # noqa: E402 from app.services.special_inventory_adjustment import create_special_inventory_adjustment # noqa: E402 class SpecialWarehouseAdjustmentTest(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: now = datetime(2026, 6, 6, 9, 0, tzinfo=UTC) self.department = Department( id=1, dept_code="D-SPECIAL", dept_name="仓库部", dept_type="WAREHOUSE", status="ACTIVE", created_at=now, updated_at=now, ) self.employee = Employee( id=1, employee_code="EMP-SPECIAL-001", employee_name="特殊调整员", dept_id=1, status="ACTIVE", created_at=now, updated_at=now, ) self.user = User( id=1, username="special_tester", password_hash="x", employee_id=1, dept_id=1, nickname="特殊调整员", is_super_admin=1, status="ACTIVE", created_at=now, updated_at=now, ) self.unit = Unit(id=1, unit_code="PCS", unit_name="个", precision_digits=0, created_at=now, updated_at=now) self.raw_item = Item( id=1, item_code="原材料00001", item_name="304不锈钢卷", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("0"), status="ACTIVE", created_at=now, updated_at=now, ) self.aux_item = Item( id=2, item_code="辅料00001", item_name="包装薄膜", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("0"), stock_unit_id=1, status="ACTIVE", created_at=now, updated_at=now, ) self.finished_item = Item( id=3, item_code="产品需规00001", item_name="冲压支架", item_type="FINISHED_GOOD", unit_weight_kg=Decimal("2.5"), stock_unit_id=1, status="ACTIVE", created_at=now, updated_at=now, ) self.finished_product = Product( id=1, item_id=3, drawing_no="FG-DWG-001", product_family="支架", version_no="V1", created_at=now, updated_at=now, ) self.raw_warehouse = Warehouse( id=1, warehouse_code="WH-RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE", created_at=now, updated_at=now, ) self.aux_warehouse = Warehouse( id=2, warehouse_code="WH-AUX", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.finished_warehouse = Warehouse( id=3, warehouse_code="WH-FG", warehouse_name="成品库", warehouse_type="FINISHED", status="ACTIVE", created_at=now, updated_at=now, ) self.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, ) self.aux_location = WarehouseLocation( id=2, warehouse_id=2, location_code="AUX-A", location_name="辅料默认库位", is_default=1, is_locked=0, status="ACTIVE", created_at=now, updated_at=now, ) self.finished_location = WarehouseLocation( id=3, warehouse_id=3, location_code="FG-A", location_name="成品默认库位", is_default=1, is_locked=0, status="ACTIVE", created_at=now, updated_at=now, ) self.raw_lot = StockLot( id=1, lot_no="RAW-LOT-001", lot_role="INBOUND_RAW", item_id=1, warehouse_id=1, location_id=1, source_doc_type="OPENING_STOCK", source_doc_id=1, inbound_qty=Decimal("0"), inbound_weight_kg=Decimal("100"), remaining_qty=Decimal("0"), remaining_weight_kg=Decimal("80"), locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=Decimal("10"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) self.aux_lot = StockLot( id=2, lot_no="AUX-LOT-001", lot_role="INVENTORY", item_id=2, warehouse_id=2, location_id=2, source_doc_type="OPENING_STOCK", source_doc_id=2, inbound_qty=Decimal("10"), inbound_weight_kg=Decimal("0"), remaining_qty=Decimal("6"), remaining_weight_kg=Decimal("0"), locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=Decimal("2.5"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) self.finished_lot = StockLot( id=3, lot_no="FG-LOT-001", lot_role="INBOUND_FINISHED", item_id=3, warehouse_id=3, location_id=3, source_doc_type="FG_RECEIPT", source_doc_id=3, inbound_qty=Decimal("20"), inbound_weight_kg=Decimal("50"), remaining_qty=Decimal("12"), remaining_weight_kg=Decimal("30"), locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=Decimal("18"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) self.raw_balance = StockBalance( id=1, item_id=1, warehouse_id=1, location_id=1, qty_on_hand=Decimal("0"), weight_on_hand_kg=Decimal("80"), qty_available=Decimal("0"), weight_available_kg=Decimal("80"), avg_unit_cost=Decimal("10"), created_at=now, updated_at=now, ) self.aux_balance = StockBalance( id=2, item_id=2, warehouse_id=2, location_id=2, qty_on_hand=Decimal("6"), weight_on_hand_kg=Decimal("0"), qty_available=Decimal("6"), weight_available_kg=Decimal("0"), avg_unit_cost=Decimal("2.5"), created_at=now, updated_at=now, ) self.finished_balance = StockBalance( id=3, item_id=3, warehouse_id=3, location_id=3, qty_on_hand=Decimal("12"), weight_on_hand_kg=Decimal("30"), qty_available=Decimal("12"), weight_available_kg=Decimal("30"), avg_unit_cost=Decimal("18"), created_at=now, updated_at=now, ) self.db.add_all( [ self.department, self.employee, self.user, self.unit, self.raw_item, self.aux_item, self.finished_item, self.finished_product, self.raw_warehouse, self.aux_warehouse, self.finished_warehouse, self.raw_location, self.aux_location, self.finished_location, self.raw_lot, self.aux_lot, self.finished_lot, self.raw_balance, self.aux_balance, self.finished_balance, ] ) self.db.commit() def _adjustment_payload( self, *, direction: str, warehouse_id: int, item_id: int, lot_id: int | None = None, location_id: int | None = None, adjust_qty: Decimal = Decimal("0"), adjust_weight_kg: Decimal = Decimal("0"), reason: str | None = None, unit_cost: Decimal | None = None, line_reason: str | None = None, ) -> SpecialWarehouseAdjustmentCreate: return SpecialWarehouseAdjustmentCreate( warehouse_id=warehouse_id, direction=direction, reason=reason, lines=[ SpecialWarehouseAdjustmentLineCreate( item_id=item_id, lot_id=lot_id, location_id=location_id, adjust_qty=adjust_qty, adjust_weight_kg=adjust_weight_kg, unit_cost=unit_cost, line_reason=line_reason, ) ], ) def _only_txn(self, txn_type: str) -> InventoryTxn: return self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == txn_type)) def _auth_context(self): return SimpleNamespace(user=SimpleNamespace(id=self.user.id)) def _seed_second_raw_lot(self) -> StockLot: now = datetime(2026, 6, 6, 9, 0, tzinfo=UTC) second_lot = StockLot( id=99, lot_no="RAW-LOT-002", lot_role="INBOUND_RAW", item_id=self.raw_item.id, warehouse_id=self.raw_warehouse.id, location_id=self.raw_location.id, source_doc_type="OPENING_STOCK", source_doc_id=99, inbound_qty=Decimal("0"), inbound_weight_kg=Decimal("20"), remaining_qty=Decimal("0"), remaining_weight_kg=Decimal("20"), locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=Decimal("10"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) self.raw_balance.weight_on_hand_kg += Decimal("20") self.raw_balance.weight_available_kg += Decimal("20") self.db.add_all([second_lot, self.raw_balance]) self.db.commit() return second_lot def test_special_out_requires_remark_and_keeps_inventory_unchanged(self) -> None: before_lot_weight = self.raw_lot.remaining_weight_kg before_balance_weight = self.raw_balance.weight_on_hand_kg with self.assertRaises(ValueError) as exc: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="OUT", warehouse_id=self.raw_warehouse.id, item_id=self.raw_item.id, lot_id=self.raw_lot.id, adjust_weight_kg=Decimal("5"), reason="", ), user_id=9, ) self.assertIn("特殊出库必须填写说明", str(exc.exception)) self.db.refresh(self.raw_lot) self.db.refresh(self.raw_balance) self.assertEqual(self.raw_lot.remaining_weight_kg, before_lot_weight) self.assertEqual(self.raw_balance.weight_on_hand_kg, before_balance_weight) self.assertIsNone(self._only_txn("SPECIAL_OUT")) def test_raw_special_out_deducts_lot_balance_by_weight_and_writes_ledger(self) -> None: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="OUT", warehouse_id=self.raw_warehouse.id, item_id=self.raw_item.id, lot_id=self.raw_lot.id, adjust_weight_kg=Decimal("12.5"), reason="样品寄出", ), user_id=9, ) self.db.refresh(self.raw_lot) self.db.refresh(self.raw_balance) txn = self._only_txn("SPECIAL_OUT") self.assertEqual(self.raw_lot.remaining_qty, Decimal("0.000000")) self.assertEqual(self.raw_lot.remaining_weight_kg, Decimal("67.500000")) self.assertEqual(self.raw_balance.qty_on_hand, Decimal("0.000000")) self.assertEqual(self.raw_balance.weight_on_hand_kg, Decimal("67.500000")) self.assertEqual(self.raw_balance.weight_available_kg, Decimal("67.500000")) self.assertIsNotNone(txn) self.assertEqual(txn.lot_id, self.raw_lot.id) self.assertEqual(txn.qty_change, Decimal("0.000000")) self.assertEqual(txn.weight_change_kg, Decimal("-12.500000")) self.assertIn("特殊出库说明：样品寄出", txn.remark) self.assertIn("调整前重量：80", txn.remark) self.assertIn("调整后重量：67.5", txn.remark) def test_raw_special_out_is_visible_in_inventory_transaction_ledger(self) -> None: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="OUT", warehouse_id=self.raw_warehouse.id, item_id=self.raw_item.id, lot_id=self.raw_lot.id, adjust_weight_kg=Decimal("6.5"), reason="研发打样领出", ), user_id=9, ) rows = self.db.execute( get_inventory_txn_ledger_query(warehouse_type="RAW", txn_type="SPECIAL_OUT") ).mappings().all() self.assertEqual(len(rows), 1) row = rows[0] self.assertEqual(row["direction"], "OUT") self.assertEqual(row["txn_type"], "SPECIAL_OUT") self.assertIn("特殊出库说明：研发打样领出", row["remark"]) def test_special_adjustment_generates_archive_for_each_adjustment_transaction(self) -> None: second_lot = self._seed_second_raw_lot() result = create_special_warehouse_adjustment( SpecialWarehouseAdjustmentCreate( warehouse_id=self.raw_warehouse.id, direction="OUT", reason="研发打样多批次领出", lines=[ SpecialWarehouseAdjustmentLineCreate( item_id=self.raw_item.id, lot_id=self.raw_lot.id, adjust_weight_kg=Decimal("5"), ), SpecialWarehouseAdjustmentLineCreate( item_id=self.raw_item.id, lot_id=second_lot.id, adjust_weight_kg=Decimal("7"), ), ], ), context=self._auth_context(), db=self.db, ) txns = self.db.scalars( select(InventoryTxn).where(InventoryTxn.txn_type == "SPECIAL_OUT").order_by(InventoryTxn.id) ).all() self.assertEqual(len(txns), 2) self.assertEqual(result.archive_status, ARCHIVE_STATUS_READY) self.assertEqual(result.archive_document_type, DOCUMENT_TYPE_WAREHOUSE_OPERATION) self.assertEqual(result.archive_business_id, txns[0].id) archive_business_ids = set( self.db.scalars( select(DocumentArchive.business_id).where( DocumentArchive.document_type == DOCUMENT_TYPE_WAREHOUSE_OPERATION, DocumentArchive.business_id.in_([txn.id for txn in txns]), ) ).all() ) self.assertEqual(archive_business_ids, {txn.id for txn in txns}) def test_special_out_rejects_quantity_greater_than_auxiliary_available_quantity(self) -> None: with self.assertRaises(ValueError) as exc: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="OUT", warehouse_id=self.aux_warehouse.id, item_id=self.aux_item.id, lot_id=self.aux_lot.id, adjust_qty=Decimal("7"), reason="盘亏调整", ), user_id=9, ) self.assertIn("特殊出库数量不能超过当前可用数量", str(exc.exception)) self.db.refresh(self.aux_lot) self.db.refresh(self.aux_balance) self.assertEqual(self.aux_lot.remaining_qty, Decimal("6.000000")) self.assertEqual(self.aux_balance.qty_available, Decimal("6.000000")) self.assertIsNone(self._only_txn("SPECIAL_OUT")) def test_auxiliary_special_in_increases_quantity_only_and_writes_ledger(self) -> None: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="IN", warehouse_id=self.aux_warehouse.id, item_id=self.aux_item.id, location_id=self.aux_location.id, adjust_qty=Decimal("4"), adjust_weight_kg=Decimal("99"), reason="供应商补送", unit_cost=Decimal("2.5"), ), user_id=9, ) lot = self.db.scalar( select(StockLot).where( StockLot.item_id == self.aux_item.id, StockLot.warehouse_id == self.aux_warehouse.id, StockLot.source_doc_type == "SPECIAL_ADJUSTMENT", ) ) self.db.refresh(self.aux_balance) txn = self._only_txn("SPECIAL_IN") self.assertIsNotNone(lot) self.assertEqual(lot.inbound_qty, Decimal("4.000000")) self.assertEqual(lot.remaining_qty, Decimal("4.000000")) self.assertEqual(lot.inbound_weight_kg, Decimal("0.000000")) self.assertEqual(lot.remaining_weight_kg, Decimal("0.000000")) self.assertEqual(self.aux_balance.qty_on_hand, Decimal("10.000000")) self.assertEqual(self.aux_balance.qty_available, Decimal("10.000000")) self.assertEqual(self.aux_balance.weight_on_hand_kg, Decimal("0.000000")) self.assertEqual(self.aux_balance.weight_available_kg, Decimal("0.000000")) self.assertIsNotNone(txn) self.assertEqual(txn.qty_change, Decimal("4.000000")) self.assertEqual(txn.weight_change_kg, Decimal("0.000000")) self.assertIn("特殊入库说明：供应商补送", txn.remark) def test_raw_special_in_generates_global_raw_lot_number(self) -> None: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="IN", warehouse_id=self.raw_warehouse.id, item_id=self.raw_item.id, location_id=self.raw_location.id, adjust_weight_kg=Decimal("15"), reason="供应商补送原料", unit_cost=Decimal("10"), ), user_id=9, ) lot = self.db.scalar( select(StockLot).where( StockLot.item_id == self.raw_item.id, StockLot.warehouse_id == self.raw_warehouse.id, StockLot.source_doc_type == "SPECIAL_ADJUSTMENT", ) ) self.assertIsNotNone(lot) self.assertEqual(lot.lot_no, "YL0001") def test_finished_special_out_deducts_quantity_and_weight_together(self) -> None: create_special_inventory_adjustment( self.db, self._adjustment_payload( direction="OUT", warehouse_id=self.finished_warehouse.id, item_id=self.finished_item.id, lot_id=self.finished_lot.id, adjust_qty=Decimal("3"), adjust_weight_kg=Decimal("7.5"), reason="展会样品", ), user_id=9, ) self.db.refresh(self.finished_lot) self.db.refresh(self.finished_balance) txn = self._only_txn("SPECIAL_OUT") self.assertEqual(self.finished_lot.remaining_qty, Decimal("9.000000")) self.assertEqual(self.finished_lot.remaining_weight_kg, Decimal("22.500000")) self.assertEqual(self.finished_balance.qty_on_hand, Decimal("9.000000")) self.assertEqual(self.finished_balance.qty_available, Decimal("9.000000")) self.assertEqual(self.finished_balance.weight_on_hand_kg, Decimal("22.500000")) self.assertEqual(self.finished_balance.weight_available_kg, Decimal("22.500000")) self.assertIsNotNone(txn) self.assertEqual(txn.qty_change, Decimal("-3.000000")) self.assertEqual(txn.weight_change_kg, Decimal("-7.500000")) self.assertIn("特殊出库说明：展会样品", txn.remark) if __name__ == "__main__": unittest.main()