149 lines
4.8 KiB
Python
149 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import UTC, datetime
|
|
|
|
from fastapi import HTTPException
|
|
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
|
|
from app.models.base import Base # noqa: E402
|
|
from app.models.master_data import Warehouse # noqa: E402
|
|
from app.models.operations import Stocktake, StocktakeWarehouse # noqa: E402
|
|
from app.services.stocktake import ensure_warehouses_unlocked, get_locked_warehouse_ids # noqa: E402
|
|
|
|
|
|
class StocktakeLockingTest(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()
|
|
|
|
def tearDown(self) -> None:
|
|
self.db.close()
|
|
|
|
def _seed(self) -> None:
|
|
now = datetime.now(UTC)
|
|
self.raw = Warehouse(
|
|
id=1,
|
|
warehouse_code="WH-RAW-01",
|
|
warehouse_name="原材料库",
|
|
warehouse_type="RAW",
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.finished = Warehouse(
|
|
id=2,
|
|
warehouse_code="WH-FG-01",
|
|
warehouse_name="成品库",
|
|
warehouse_type="FINISHED",
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add_all([self.raw, self.finished])
|
|
self.db.flush()
|
|
stocktake = Stocktake(
|
|
stocktake_no="PK-20260530-001",
|
|
status="LOCKED",
|
|
started_by_user_id=1,
|
|
started_at=now,
|
|
remark="测试锁库",
|
|
)
|
|
self.db.add(stocktake)
|
|
self.db.flush()
|
|
self.db.add(
|
|
StocktakeWarehouse(
|
|
stocktake_id=stocktake.id,
|
|
warehouse_id=self.raw.id,
|
|
warehouse_type="RAW",
|
|
warehouse_name="原材料库",
|
|
status="LOCKED",
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
def test_get_locked_warehouse_ids_returns_only_active_stocktakes(self) -> None:
|
|
locked = get_locked_warehouse_ids(self.db, [self.raw.id, self.finished.id])
|
|
|
|
self.assertEqual(locked, {self.raw.id})
|
|
|
|
def test_ensure_warehouses_unlocked_raises_for_locked_warehouse(self) -> None:
|
|
with self.assertRaises(HTTPException) as exc:
|
|
ensure_warehouses_unlocked(self.db, [self.raw.id], "入库")
|
|
|
|
self.assertEqual(exc.exception.status_code, 423)
|
|
self.assertIn("原材料库正在盘库", exc.exception.detail)
|
|
|
|
def test_ensure_warehouses_unlocked_allows_unlocked_warehouse(self) -> None:
|
|
ensure_warehouses_unlocked(self.db, [self.finished.id], "销售出库")
|
|
|
|
def test_inventory_inbound_is_blocked_when_warehouse_is_locked(self) -> None:
|
|
from decimal import Decimal
|
|
|
|
from app.api.routes.inventory import create_warehouse_inbound
|
|
from app.models.master_data import Item
|
|
from app.schemas.operations import WarehouseInboundCreate
|
|
|
|
now = datetime.now(UTC)
|
|
item = Item(
|
|
id=1,
|
|
item_code="原材料00001",
|
|
item_name="冷轧钢板",
|
|
item_type="RAW_MATERIAL",
|
|
unit_weight_kg=Decimal("1"),
|
|
safety_stock_weight_kg=Decimal("0"),
|
|
scrap_sale_price=Decimal("0"),
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add(item)
|
|
self.db.commit()
|
|
|
|
class User:
|
|
id = 1
|
|
|
|
class Context:
|
|
user = User()
|
|
|
|
with self.assertRaises(HTTPException) as exc:
|
|
create_warehouse_inbound(
|
|
WarehouseInboundCreate(
|
|
biz_type="OPENING",
|
|
item_id=item.id,
|
|
warehouse_id=self.raw.id,
|
|
inbound_weight_kg=10,
|
|
unit_cost=1,
|
|
),
|
|
context=Context(),
|
|
db=self.db,
|
|
)
|
|
|
|
self.assertEqual(exc.exception.status_code, 423)
|
|
self.assertIn("正在盘库", exc.exception.detail)
|
|
|
|
def test_lock_guard_message_names_action(self) -> None:
|
|
with self.assertRaises(HTTPException) as exc:
|
|
ensure_warehouses_unlocked(self.db, [self.raw.id], "到货入库")
|
|
|
|
self.assertEqual(exc.exception.status_code, 423)
|
|
self.assertIn("到货入库", exc.exception.detail)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|