ocean_project_manager/backend/tests/conftest.py

import sys
from pathlib import Path
backend_dir = Path(__file__).parent.parent
sys.path.insert(0, str(backend_dir))

import pytest
import asyncio
from typing import AsyncGenerator, Generator
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.pool import StaticPool

try:
    from src.main import app
except ImportError:
    from main import app

try:
    from src.config.database import get_db
except ImportError:
    from config.database import get_db

from src.models.user import User
from src.utils.password import hash_password


# 测试数据库配置
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

# 创建测试引擎
engine = create_async_engine(
    TEST_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

# 创建测试会话工厂
TestingSessionLocal = async_sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)


@pytest.fixture(scope="function")
async def db_session() -> AsyncGenerator[AsyncSession, None]:
    """创建测试数据库会话"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with TestingSessionLocal() as session:
        yield session
        await session.rollback()

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture(scope="function")
async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]:
    """创建测试客户端"""

    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()


@pytest.fixture
async def admin_user(db_session: AsyncSession) -> User:
    """创建管理员用户"""
    user = User(
        username="admin",
        password_hash=hash_password("admin123"),
        real_name="系统管理员",
        department="管理部",
        role="admin",
        email="admin@test.com",
        phone="13800000001",
        is_active=True,
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user


@pytest.fixture
async def market_user(db_session: AsyncSession) -> User:
    """创建市场部用户"""
    user = User(
        username="market",
        password_hash=hash_password("market123"),
        real_name="市场部用户",
        department="市场部",
        role="market",
        email="market@test.com",
        phone="13800000002",
        is_active=True,
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user


@pytest.fixture
async def other_user(db_session: AsyncSession) -> User:
    """创建其他部门用户"""
    user = User(
        username="other",
        password_hash=hash_password("other123"),
        real_name="其他部门用户",
        department="技术部",
        role="other",
        email="other@test.com",
        phone="13800000003",
        is_active=True,
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user


@pytest.fixture
async def inactive_user(db_session: AsyncSession) -> User:
    """创建未激活用户"""
    user = User(
        username="inactive",
        password_hash=hash_password("inactive123"),
        real_name="未激活用户",
        department="市场部",
        role="market",
        email="inactive@test.com",
        phone="13800000004",
        is_active=False,
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user


@pytest.fixture
async def admin_token(client: AsyncClient, admin_user: User) -> str:
    """获取管理员token"""
    response = await client.post(
        "/api/v1/auth/login",
        json={"username": "admin", "password": "admin123"},
    )
    data = response.json()
    return data["data"]["token"]


@pytest.fixture
async def market_token(client: AsyncClient, market_user: User) -> str:
    """获取市场部用户token"""
    response = await client.post(
        "/api/v1/auth/login",
        json={"username": "market", "password": "market123"},
    )
    data = response.json()
    return data["data"]["token"]


@pytest.fixture
async def other_token(client: AsyncClient, other_user: User) -> str:
    """获取其他部门用户token"""
    response = await client.post(
        "/api/v1/auth/login",
        json={"username": "other", "password": "other123"},
    )
    data = response.json()
    return data["data"]["token"]