Transactional тесты, фикстуры, изоляция тестов
Тестирование кода с SQLAlchemy требует особого подхода: изоляция тестов, transactional тесты, правильные фикстуры и быстрое выполнение.
# tests/conftest.py
import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker
from myapp.models import Base
@pytest.fixture(scope='session')
def db_engine():
"""Создать engine для тестов."""
engine = create_engine(
'postgresql+psycopg://postgres:postgres@localhost/test_db',
echo=False, # Отключить логирование для скорости
)
# Создать все таблицы
Base.metadata.create_all(engine)
yield engine
# Удалить все таблицы после тестов
Base.metadata.drop_all(engine)
@pytest.fixture
def db_session(db_engine):
"""
Transactional фикстура: каждый тест в транзакции.
Транзакция откатывается после теста — полная изоляция.
"""
connection = db_engine.connect()
transaction = connection.begin()
session = Session(bind=connection)
yield session
# Откатить все изменения
session.rollback()
session.close()
transaction.rollback()
connection.close()# tests/test_users.py
def test_create_user(db_session):
"""Тест создания пользователя."""
from myapp.models import User
user = User(email='test@example.com', name='Test')
db_session.add(user)
db_session.commit()
assert user.id is not None
assert user.email == 'test@example.com'
def test_user_relationships(db_session):
"""Тест relationships."""
from myapp.models import User, Post
user = User(email='test@example.com', name='Test')
post = Post(title='Test Post', author=user)
db_session.add_all([user, post])
db_session.commit()
assert len(user.posts) == 1
assert user.posts[0].title == 'Test Post'
def test_user_deletion_cascade(db_session):
"""Тест каскадного удаления."""
from myapp.models import User, Post
user = User(email='test@example.com', name='Test')
post = Post(title='Test Post', author=user)
db_session.add_all([user, post])
db_session.commit()
# Удалить пользователя
db_session.delete(user)
db_session.commit()
# Пост должен удалиться каскадом
posts_count = db_session.query(Post).count()
assert posts_count == 0# tests/conftest.py
import pytest
from myapp.models import User, Post
@pytest.fixture
def test_user(db_session):
"""Создать тестового пользователя."""
user = User(email='test@example.com', name='Test User')
db_session.add(user)
db_session.commit()
db_session.refresh(user)
return user
@pytest.fixture
def test_post(db_session, test_user):
"""Создать тестовый пост."""
post = Post(title='Test Post', content='Content', author=test_user)
db_session.add(post)
db_session.commit()
db_session.refresh(post)
return post
# Использование
def test_get_user(db_session, test_user):
user = db_session.get(User, test_user.id)
assert user.email == test_user.email# tests/factories.py
import factory
from factory.alchemy import SQLAlchemyModelFactory
from myapp.models import User, Post
from myapp.database import SessionLocal
class BaseFactory(SQLAlchemyModelFactory):
class Meta:
abstract = True
sqlalchemy_session = SessionLocal()
sqlalchemy_session_persistence = 'commit'
class UserFactory(BaseFactory):
class Meta:
model = User
email = factory.Sequence(lambda n: f'user{n}@example.com')
name = factory.Faker('name')
active = True
class PostFactory(BaseFactory):
class Meta:
model = Post
title = factory.Faker('sentence')
content = factory.Faker('text')
user = factory.SubFactory(UserFactory)
# Использование в тестах
def test_many_users(db_session):
from tests.factories import UserFactory
users = UserFactory.create_batch(10)
assert len(users) == 10# tests/conftest.py
import pytest
from datetime import datetime, timedelta
from myapp.models import User, Post, Comment
@pytest.fixture
def users_batch(db_session):
"""Создать пачку пользователей."""
users = []
for i in range(10):
user = User(
email=f'user{i}@example.com',
name=f'User {i}',
created_at=datetime.utcnow() - timedelta(days=i)
)
db_session.add(user)
users.append(user)
db_session.commit()
return users
@pytest.fixture
def posts_with_comments(db_session, test_user):
"""Создать посты с комментариями."""
posts = []
for i in range(5):
post = Post(
title=f'Post {i}',
content=f'Content {i}',
author=test_user,
created_at=datetime.utcnow() - timedelta(days=i)
)
db_session.add(post)
# Добавить комментарии
for j in range(3):
comment = Comment(
content=f'Comment {j}',
post=post,
author=test_user
)
db_session.add(comment)
posts.append(post)
db_session.commit()
return posts# ПЛОХО: тесты влияют друг на друга
class TestUserBad:
def test_create(self):
user = User(email='test@example.com')
db_session.add(user)
db_session.commit() # Данные остаются!
def test_count(self):
count = db_session.query(User).count()
assert count == 1 # Может упасть если test_create запустился# ХОРОШО: каждый тест в своей транзакции
@pytest.fixture
def db_session(db_engine):
connection = db_engine.connect()
transaction = connection.begin()
session = Session(bind=connection)
yield session
transaction.rollback() # Откат всех изменений
session.close()
connection.close()
class TestUserGood:
def test_create(self, db_session):
user = User(email='test@example.com')
db_session.add(user)
db_session.commit()
count = db_session.query(User).count()
assert count == 1
def test_count(self, db_session):
count = db_session.query(User).count()
assert count == 0 # Всегда 0, предыдущий тест откатился# pytest.ini
[pytest]
addopts = -n auto # Автоматическое количество процессов
testpaths = tests
python_files = test_*.py# Запуск параллельно
poetry run pytest -n 4 # 4 процесса
poetry run pytest -n auto # Авто (CPU cores)# tests/conftest.py
import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
@pytest.fixture(scope='session')
def db_engine():
"""Создать engine с уникальной БД для параллельных тестов."""
import os
worker_id = os.environ.get('PYTEST_XDIST_WORKER', 'gw0')
# Уникальная БД для каждого worker
db_name = f'test_db_{worker_id}'
# Создать БД если не существует
master_engine = create_engine('postgresql+psycopg://postgres:postgres@localhost/postgres')
with master_engine.connect() as conn:
conn.execution_options(isolation_level='AUTOCOMMIT').execute(
text(f"DROP DATABASE IF EXISTS {db_name}")
)
conn.execution_options(isolation_level='AUTOCOMMIT').execute(
text(f"CREATE DATABASE {db_name}")
)
engine = create_engine(f'postgresql+psycopg://postgres:postgres@localhost/{db_name}')
Base.metadata.create_all(engine)
yield engine
# Очистить БД
Base.metadata.drop_all(engine)
with master_engine.connect() as conn:
conn.execution_options(isolation_level='AUTOCOMMIT').execute(
text(f"DROP DATABASE {db_name}")
)# tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
@pytest.fixture(scope='session')
def async_engine():
"""Создать async engine для тестов."""
engine = create_async_engine(
'postgresql+asyncpg://postgres:postgres@localhost/test_db',
echo=False,
)
return engine
@pytest.fixture
async def async_db_session(async_engine):
"""Async transactional фикстура."""
async with async_engine.connect() as connection:
# Создать таблицы
async with connection.begin():
await connection.run_sync(Base.metadata.create_all)
async_session_maker = async_sessionmaker(
connection,
class_=AsyncSession,
expire_on_commit=False,
)
async with async_session_maker() as session:
yield session
# Откатить всё
async with connection.begin():
await connection.run_sync(Base.metadata.drop_all)# tests/test_async_users.py
import pytest
from sqlalchemy import select
from myapp.models import User
@pytest.mark.asyncio
async def test_create_user_async(async_db_session):
"""Тест создания пользователя в async."""
user = User(email='test@example.com', name='Test')
async_db_session.add(user)
await async_db_session.commit()
assert user.id is not None
@pytest.mark.asyncio
async def test_get_user_async(async_db_session, test_user):
"""Тест получения пользователя в async."""
result = await async_db_session.execute(
select(User).where(User.id == test_user.id)
)
user = result.scalar()
assert user.email == test_user.email
@pytest.mark.asyncio
async def test_user_relationships_async(async_db_session, test_user):
"""Тест relationships в async."""
from myapp.models import Post
from sqlalchemy.orm import selectinload
result = await async_db_session.execute(
select(User)
.options(selectinload(User.posts))
.where(User.id == test_user.id)
)
user = result.scalar()
assert len(user.posts) == len(test_user.posts)# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import inspect, text
@pytest.fixture
def alembic_config(db_engine):
"""Alembic конфиг для тестовой БД."""
config = Config("alembic.ini")
config.set_main_option("sqlalchemy.url", str(db_engine.url))
return config
@pytest.fixture
def alembic_upgraded(alembic_config):
"""Применить все миграции перед тестом."""
command.upgrade(alembic_config, "head")
yield alembic_config
command.downgrade(alembic_config, "base")
def test_all_migrations_apply(alembic_upgraded):
"""Тест что все миграции применяются."""
inspector = inspect(alembic_upgraded.get_section('sqlalchemy.url'))
tables = inspector.get_table_names()
assert 'users' in tables
assert 'posts' in tables
def test_migration_downgrade(alembic_config, db_engine):
"""Тест что downgrade работает."""
command.upgrade(alembic_config, "head")
command.downgrade(alembic_config, "-1")
inspector = inspect(db_engine)
tables = inspector.get_table_names()
# Последняя таблица должна исчезнуть
assert 'last_table' not in tables
def test_data_migration(alembic_upgraded, db_engine):
"""Тест data migration."""
with db_engine.connect() as conn:
# Вставить тестовые данные
conn.execute(text("""
INSERT INTO users (email, name)
VALUES ('test@example.com', 'Test User')
"""))
conn.commit()
# Применить миграцию
command.upgrade(alembic_config, "next_revision")
# Проверить результат
result = conn.execute(text("""
SELECT full_name FROM users
WHERE email = 'test@example.com'
"""))
full_name = result.scalar()
assert full_name == 'Test User'# tests/test_services.py
import pytest
from unittest.mock import Mock, MagicMock
from myapp.services import UserService
def test_get_user_by_id():
"""Тест сервиса с мок сессией."""
mock_session = Mock()
# Настроить mock
mock_user = Mock()
mock_user.id = 1
mock_user.email = 'test@example.com'
mock_session.get.return_value = mock_user
service = UserService(mock_session)
user = service.get_user_by_id(1)
assert user.email == 'test@example.com'
mock_session.get.assert_called_once()
def test_create_user():
"""Тест создания с мок сессией."""
mock_session = Mock()
service = UserService(mock_session)
user = service.create_user('test@example.com', 'Test')
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()# Используйте transactional фикстуры
@pytest.fixture
def db_session(db_engine):
connection = db_engine.connect()
transaction = connection.begin()
session = Session(bind=connection)
yield session
transaction.rollback()
# Изолируйте тесты
# Каждый тест должен работать с чистой БД
# Используйте фикстуры для данных
@pytest.fixture
def test_user(db_session):
user = User(email='test@example.com')
db_session.add(user)
db_session.commit()
return user
# Тестируйте async с async фикстурами
@pytest.mark.asyncio
async def test_async(async_db_session):
...
# Тестируйте миграции
def test_migrations(alembic_upgraded):
...# Не используйте общую сессию между тестами
db_session = Session(engine) # ПЛОХО: глобальная сессия
# Не забывайте откатывать транзакции
def test_something(db_session):
db_session.add(obj)
db_session.commit()
# ПЛОХО: нет rollback, данные остаются
# Не тестируйте только happy path
def test_only_success():
...
# Добавьте тесты ошибок
def test_failure():
...
# Не используйте time.sleep() в тестах
# Используйте mock для времени
from unittest.mock import patch
with patch('myapp.datetime') as mock_dt:
mock_dt.utcnow.return_value = datetime(2024, 1, 1)Далее: Data Validation