geMoldInsight/tests/conftest.py

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()