71 lines
1.8 KiB
Python
71 lines
1.8 KiB
Python
import logging
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base
|
|
from app.models.user import User
|
|
from app.middleware.logging_filter import APIKeyFilter
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def add_api_key_filter():
|
|
"""自动为每个测试添加APIKeyFilter到root logger"""
|
|
root_logger = logging.getLogger()
|
|
api_key_filter = APIKeyFilter()
|
|
root_logger.addFilter(api_key_filter)
|
|
yield
|
|
root_logger.removeFilter(api_key_filter)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_engine():
|
|
"""Create async engine for testing with SQLite."""
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def async_session(async_engine):
|
|
"""Create async session for testing."""
|
|
async_session_maker = async_sessionmaker(
|
|
async_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
autoflush=False,
|
|
autocommit=False,
|
|
)
|
|
async with async_session_maker() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def test_user(async_session):
|
|
"""Create a test user."""
|
|
user = User(
|
|
id=uuid.uuid4(),
|
|
email="test@example.com",
|
|
password_hash="hashed_password",
|
|
name="Test User",
|
|
plan="free",
|
|
max_queries=5,
|
|
is_active=True,
|
|
email_verified=True,
|
|
)
|
|
async_session.add(user)
|
|
await async_session.commit()
|
|
await async_session.refresh(user)
|
|
return user
|