# tests/conftest.py
import asyncio
from collections.abc import AsyncGenerator
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from app.core.database import Base, get_db
from app.main import create_app
# ---- Engine & Session ----
TEST_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
@pytest.fixture(scope="session")
def event_loop():
"""Tạo event loop cho toàn bộ test session."""
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(autouse=True)
async def setup_db():
"""Tạo tables trước mỗi test, drop sau khi xong."""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
async def db_session() -> AsyncGenerator[AsyncSession]:
"""Cung cấp database session cho mỗi test."""
async with TestSessionLocal() as session:
yield session
await session.rollback()
# ---- App & Client ----
@pytest.fixture
async def app(db_session: AsyncSession):
"""Tạo FastAPI app với test database."""
_app = create_app()
async def override_get_db():
yield db_session
_app.dependency_overrides[get_db] = override_get_db
return _app
@pytest.fixture
async def client(app) -> AsyncGenerator[AsyncClient]:
"""HTTP client cho e2e tests."""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
@pytest.fixture
async def auth_client(client: AsyncClient) -> AsyncClient:
"""Authenticated client."""
response = await client.post("/api/v1/auth/register", json={
"name": "Test User",
"email": "[email protected]",
"password": "StrongPass123!",
})
token = response.json()["access_token"]
client.headers["Authorization"] = f"Bearer {token}"
return client
3. テストデータ用のファクトリーボーイ
# tests/factories.py
import factory
from factory import fuzzy
from app.core.security import hash_password
from app.modules.users.models import User
from app.modules.posts.models import Post
class UserFactory(factory.Factory):
class Meta:
model = User
id = factory.Sequence(lambda n: n + 1)
name = factory.Faker("name")
email = factory.LazyAttribute(lambda o: f"user{o.id}@example.com")
hashed_password = factory.LazyFunction(lambda: hash_password("Test1234!"))
is_active = True
class PostFactory(factory.Factory):
class Meta:
model = Post
id = factory.Sequence(lambda n: n + 1)
title = factory.Faker("sentence", nb_words=5)
content = factory.Faker("paragraph")
author_id = factory.SubFactory(UserFactory)
is_published = True
4. 単体テスト
# tests/unit/test_schemas.py
import pytest
from pydantic import ValidationError
from app.modules.users.schemas import UserCreate
class TestUserCreateSchema:
def test_valid_user(self):
user = UserCreate(
name="John Doe",
email="[email protected]",
password="StrongPass123!",
)
assert user.name == "John Doe"
assert user.email == "[email protected]"
def test_short_name(self):
with pytest.raises(ValidationError) as exc_info:
UserCreate(name="J", email="[email protected]", password="StrongPass123!")
errors = exc_info.value.errors()
assert errors[0]["loc"] == ("name",)
assert "min_length" in errors[0]["type"]
def test_short_password(self):
with pytest.raises(ValidationError):
UserCreate(name="John", email="[email protected]", password="123")
# tests/unit/test_user_service.py
from unittest.mock import AsyncMock, MagicMock
import pytest
from fastapi import HTTPException
from app.modules.users.schemas import UserCreate
from app.modules.users.service import UserService
from tests.factories import UserFactory
class TestUserService:
@pytest.fixture
def mock_repo(self):
repo = MagicMock()
repo.get_by_email = AsyncMock(return_value=None)
repo.get_by_id = AsyncMock()
repo.create = AsyncMock()
return repo
@pytest.fixture
def service(self, mock_repo):
return UserService(repo=mock_repo)
async def test_create_user_success(self, service, mock_repo):
data = UserCreate(name="John", email="[email protected]", password="Test1234!")
mock_repo.create.return_value = UserFactory.build(
name="John", email="[email protected]"
)
result = await service.create_user(data)
assert result.name == "John"
assert result.email == "[email protected]"
mock_repo.create.assert_awaited_once()
async def test_create_user_email_taken(self, service, mock_repo):
mock_repo.get_by_email.return_value = UserFactory.build()
data = UserCreate(name="John", email="[email protected]", password="Test1234!")
with pytest.raises(HTTPException) as exc_info:
await service.create_user(data)
assert exc_info.value.status_code == 409
async def test_get_user_not_found(self, service, mock_repo):
mock_repo.get_by_id.return_value = None
with pytest.raises(HTTPException) as exc_info:
await service.get_user(999)
assert exc_info.value.status_code == 404
5. 結合テスト
# tests/integration/test_user_repository.py
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from app.modules.users.models import User
from app.modules.users.repository import UserRepository
@pytest.mark.integration
class TestUserRepository:
@pytest.fixture
def repo(self, db_session: AsyncSession):
return UserRepository(session=db_session)
async def test_create_and_get(self, repo: UserRepository, db_session):
user = User(
name="Alice",
email="[email protected]",
hashed_password="hashed",
)
created = await repo.create(user)
await db_session.commit()
found = await repo.get_by_id(created.id)
assert found is not None
assert found.name == "Alice"
async def test_get_by_email(self, repo: UserRepository, db_session):
user = User(
name="Bob",
email="[email protected]",
hashed_password="hashed",
)
await repo.create(user)
await db_session.commit()
found = await repo.get_by_email("[email protected]")
assert found is not None
assert found.name == "Bob"
async def test_get_nonexistent(self, repo: UserRepository):
found = await repo.get_by_id(99999)
assert found is None