166 lines
4.4 KiB
Python
166 lines
4.4 KiB
Python
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
|
|
|
|
from src.main import app
|
|
from src.config.database import Base, 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"]
|