140 lines
4.2 KiB
Python
140 lines
4.2 KiB
Python
import os
|
|
import sys
|
|
import tempfile
|
|
|
|
vendor_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "_vendor"))
|
|
if os.path.isdir(vendor_dir) and vendor_dir not in sys.path:
|
|
sys.path.insert(0, vendor_dir)
|
|
|
|
src_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))
|
|
if os.path.isdir(src_dir) and src_dir not in sys.path:
|
|
sys.path.insert(0, src_dir)
|
|
|
|
import pytest
|
|
from httpx import AsyncClient, ASGITransport
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
|
from sqlalchemy import select
|
|
|
|
from fastapi import FastAPI, APIRouter
|
|
from api.inventory import inventory_router
|
|
from models.database import Base, User, Customer, Warehouse, Supplier, Product, ProductMaterial, Inventory
|
|
from database.database import get_db_session
|
|
from services.auth_service import get_current_active_user
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def sqlite_db_path():
|
|
fd, path = tempfile.mkstemp(prefix="gemold_test_", suffix=".db")
|
|
os.close(fd)
|
|
yield path
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def async_engine(sqlite_db_path):
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{sqlite_db_path}", future=True)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def db_session(async_engine):
|
|
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
|
|
async with session_factory() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def seeded_db(db_session: AsyncSession):
|
|
user = User(
|
|
id=1,
|
|
username="tester",
|
|
email="tester@example.com",
|
|
hashed_password="x",
|
|
full_name="Tester",
|
|
is_active=True,
|
|
)
|
|
customer = Customer(id=1, code="C001", name="客户A", is_active=True)
|
|
supplier = Supplier(id=1, code="S001", name="供应商A", is_active=True)
|
|
warehouse = Warehouse(id=1, code="W001", name="默认仓库", is_active=True, is_default=True)
|
|
|
|
material = Product(
|
|
id=1,
|
|
sku="MAT-001",
|
|
name="钢材",
|
|
unit="kg",
|
|
item_type="material",
|
|
cost_price=10.0,
|
|
sale_price=0,
|
|
min_stock=0,
|
|
max_stock=100000,
|
|
is_active=True,
|
|
)
|
|
finished = Product(
|
|
id=2,
|
|
sku="MOLD-STD",
|
|
name="标准模具",
|
|
unit="套",
|
|
item_type="finished",
|
|
cost_price=0,
|
|
sale_price=1000.0,
|
|
min_stock=0,
|
|
max_stock=0,
|
|
is_active=True,
|
|
)
|
|
bom = ProductMaterial(
|
|
id=1,
|
|
finished_product_id=finished.id,
|
|
material_product_id=material.id,
|
|
quantity=2.0,
|
|
loss_rate=0.05,
|
|
)
|
|
inv = Inventory(
|
|
id=1,
|
|
product_id=material.id,
|
|
warehouse_id=warehouse.id,
|
|
quantity=1000,
|
|
locked_quantity=0,
|
|
)
|
|
|
|
db_session.add_all([user, customer, supplier, warehouse, material, finished, bom, inv])
|
|
await db_session.commit()
|
|
return {"user": user, "customer": customer, "supplier": supplier, "warehouse": warehouse, "material": material, "finished": finished}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
async def client(async_engine, seeded_db):
|
|
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
test_app = FastAPI()
|
|
test_app.include_router(inventory_router)
|
|
|
|
async def override_get_db_session():
|
|
async with session_factory() as session:
|
|
yield session
|
|
|
|
async def override_get_current_active_user():
|
|
async with session_factory() as session:
|
|
result = await session.execute(select(User).where(User.username == "tester"))
|
|
return result.scalar_one()
|
|
|
|
test_app.dependency_overrides[get_db_session] = override_get_db_session
|
|
test_app.dependency_overrides[get_current_active_user] = override_get_current_active_user
|
|
|
|
transport = ASGITransport(app=test_app, lifespan="off")
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
test_app.dependency_overrides.clear()
|