Entendendo Testes de Integração: O que são e por que importam
Testes de integração ocupam uma posição estratégica na pirâmide de testes. Enquanto testes unitários validam componentes isolados e testes end-to-end cobrem fluxos completos do usuário, testes de integração fazem o trabalho intermediário crucial: verificar se diferentes módulos, serviços e camadas da sua aplicação funcionam corretamente quando integrados. Em Python, isso significa testar a interação real entre sua lógica de negócio e recursos externos como bancos de dados, APIs e sistemas de fila.
A diferença prática é significativa. Um teste unitário pode passar porque mockamos o banco de dados, mas em produção a aplicação pode quebrar por incompatibilidade de schema ou tipo de dado. Testes de integração pegam justamente esses cenários. No contexto deste artigo, trabalharemos com um banco de dados real (PostgreSQL) em um container Docker, testando com pytest — a ferramenta padrão da comunidade Python que oferece fixture elegantes, parametrização poderosa e relatórios detalhados.
Configurando o Ambiente: Docker, PostgreSQL e pytest
Estruturando o projeto e o Docker Compose
Começamos criando uma estrutura de projeto limpa. O Docker é essencial aqui porque garante que todos os desenvolvedores, testers e pipelines de CI/CD usem exatamente a mesma versão do PostgreSQL, sem divergências do "funciona na minha máquina".
Crie o arquivo docker-compose.yml:
version: '3.8'
services:
postgres:
image: postgres:15-alpine
container_name: test_db
environment:
POSTGRES_USER: testuser
POSTGRES_PASSWORD: testpass
POSTGRES_DB: testdb
ports:
- "5433:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U testuser -d testdb"]
interval: 5s
timeout: 5s
retries: 5
volumes:
postgres_data:
A healthcheck garante que o banco está pronto antes dos testes iniciarem. Use a porta 5433 localmente para evitar conflitos com PostgreSQL que você possa ter rodando. O volume persiste dados durante testes, útil para debug.
Dependências Python
Crie um requirements.txt:
pytest==7.4.3
psycopg2-binary==2.9.9
sqlalchemy==2.0.23
python-dotenv==1.0.0
Use pip install -r requirements.txt. Psycopg2 é o adaptador PostgreSQL, SQLAlchemy fornece a abstração ORM robusta, e python-dotenv facilita gerenciamento de variáveis de ambiente.
Criando a Camada de Dados e Modelos
Estrutura de banco de dados com SQLAlchemy
Vamos criar uma aplicação simples de gerenciamento de usuários e posts. Primeiro, o arquivo app/models.py:
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(80), unique=True, nullable=False, index=True)
email = Column(String(120), unique=True, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
posts = relationship('Post', back_populates='author', cascade='all, delete-orphan')
def __repr__(self):
return f'<User {self.username}>'
class Post(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(200), nullable=False)
content = Column(Text, nullable=False)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
author = relationship('User', back_populates='posts')
def __repr__(self):
return f'<Post {self.title}>'
Este é o mapeamento objeto-relacional. Os modelos definem o contrato entre Python e o banco. As relationships estabelecem os relacionamentos um-para-muitos com suporte a cascade delete.
Repositório para operações de banco
Crie app/repository.py:
from sqlalchemy.orm import Session
from app.models import User, Post
from sqlalchemy.exc import IntegrityError
from datetime import datetime
class UserRepository:
def __init__(self, session: Session):
self.session = session
def create(self, username: str, email: str) -> User:
try:
user = User(username=username, email=email)
self.session.add(user)
self.session.commit()
self.session.refresh(user)
return user
except IntegrityError:
self.session.rollback()
raise ValueError(f"Username ou email já existem")
def get_by_username(self, username: str) -> User:
return self.session.query(User).filter_by(username=username).first()
def get_by_id(self, user_id: int) -> User:
return self.session.query(User).filter_by(id=user_id).first()
def list_all(self) -> list:
return self.session.query(User).all()
def delete(self, user_id: int) -> bool:
user = self.get_by_id(user_id)
if user:
self.session.delete(user)
self.session.commit()
return True
return False
class PostRepository:
def __init__(self, session: Session):
self.session = session
def create(self, title: str, content: str, user_id: int) -> Post:
post = Post(title=title, content=content, user_id=user_id)
self.session.add(post)
self.session.commit()
self.session.refresh(post)
return post
def get_by_id(self, post_id: int) -> Post:
return self.session.query(Post).filter_by(id=post_id).first()
def get_by_user(self, user_id: int) -> list:
return self.session.query(Post).filter_by(user_id=user_id).all()
Os repositórios encapsulam a lógica de acesso ao banco. Não exponha a sessão SQLAlchemy diretamente; sempre use uma camada. Isso facilita testes e futuras migrações de tecnologia.
Escrevendo Testes de Integração com pytest
Fixtures compartilhadas e isolamento de dados
O arquivo tests/conftest.py é onde pytest busca fixtures reutilizáveis. É a peça-chave para testes de integração robustos:
import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker, Session
from app.models import Base
import os
from dotenv import load_dotenv
load_dotenv()
DATABASE_URL = os.getenv(
'TEST_DATABASE_URL',
'postgresql://testuser:testpass@localhost:5433/testdb'
)
@pytest.fixture(scope='session')
def db_engine():
"""Cria a engine do banco uma vez por sessão de testes"""
engine = create_engine(DATABASE_URL, echo=False)
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
@pytest.fixture(scope='function')
def db_session(db_engine):
"""Cria uma nova sessão para cada teste com rollback automático"""
connection = db_engine.connect()
transaction = connection.begin()
session_factory = sessionmaker(bind=connection)
session = session_factory()
yield session
session.close()
transaction.rollback()
connection.close()
@pytest.fixture
def user_repo(db_session):
"""Injeta o repositório de usuários com sessão isolada"""
from app.repository import UserRepository
return UserRepository(db_session)
@pytest.fixture
def post_repo(db_session):
"""Injeta o repositório de posts com sessão isolada"""
from app.repository import PostRepository
return PostRepository(db_session)
A fixture db_session usa transações que são revertidas ao final de cada teste (rollback). Isso garante isolamento — um teste nunca afeta outro. O escopo function (padrão) reinicia o estado antes de cada teste. A engine é reutilizada (scope='session') para performance, mas o schema é recriado com create_all.
Testes unitários da camada de repositório
Crie tests/test_user_integration.py:
import pytest
from app.models import User
from sqlalchemy.exc import IntegrityError
class TestUserRepository:
"""Testes de integração para operações de usuário"""
def test_create_user_successfully(self, user_repo, db_session):
"""Deve criar um usuário e persistir no banco real"""
user = user_repo.create(username='johndoe', email='john@example.com')
assert user.id is not None
assert user.username == 'johndoe'
assert user.email == 'john@example.com'
# Verifica se realmente foi persistido
fetched = db_session.query(User).filter_by(id=user.id).first()
assert fetched is not None
assert fetched.username == 'johndoe'
def test_create_user_duplicate_username(self, user_repo):
"""Deve rejeitar username duplicado"""
user_repo.create(username='alice', email='alice@example.com')
with pytest.raises(ValueError, match='Username ou email já existem'):
user_repo.create(username='alice', email='another@example.com')
def test_get_user_by_username(self, user_repo):
"""Deve recuperar usuário por username"""
created = user_repo.create(username='bob', email='bob@example.com')
fetched = user_repo.get_by_username('bob')
assert fetched is not None
assert fetched.id == created.id
assert fetched.email == 'bob@example.com'
def test_get_user_by_id(self, user_repo):
"""Deve recuperar usuário por ID"""
created = user_repo.create(username='carol', email='carol@example.com')
fetched = user_repo.get_by_id(created.id)
assert fetched is not None
assert fetched.username == 'carol'
def test_list_all_users(self, user_repo):
"""Deve listar todos os usuários"""
user_repo.create(username='user1', email='user1@example.com')
user_repo.create(username='user2', email='user2@example.com')
user_repo.create(username='user3', email='user3@example.com')
users = user_repo.list_all()
assert len(users) == 3
usernames = {u.username for u in users}
assert usernames == {'user1', 'user2', 'user3'}
def test_delete_user(self, user_repo):
"""Deve deletar usuário e suas relações"""
created = user_repo.create(username='todelete', email='delete@example.com')
deleted = user_repo.delete(created.id)
assert deleted is True
fetched = user_repo.get_by_id(created.id)
assert fetched is None
def test_delete_nonexistent_user(self, user_repo):
"""Deve retornar False ao deletar usuário inexistente"""
result = user_repo.delete(9999)
assert result is False
Cada teste é pequeno, focado e testa um comportamento específico. O padrão Arrange-Act-Assert (AAA) é implícito: criamos dados, executamos operações, verificamos resultados. Usamos pytest.raises para testar exceções — mais legível que try-except.
Testes de relacionamentos e integração cross-table
Crie tests/test_post_integration.py:
import pytest
from app.models import User, Post
class TestPostRepository:
"""Testes de integração para posts e relacionamento com usuários"""
def test_create_post_for_existing_user(self, user_repo, post_repo, db_session):
"""Deve criar post vinculado a usuário existente"""
user = user_repo.create(username='postauthor', email='author@example.com')
post = post_repo.create(
title='Meu Primeiro Post',
content='Conteúdo do post',
user_id=user.id
)
assert post.id is not None
assert post.title == 'Meu Primeiro Post'
assert post.user_id == user.id
# Verifica relacionamento
fetched_post = db_session.query(Post).filter_by(id=post.id).first()
assert fetched_post.author.username == 'postauthor'
def test_get_posts_by_user(self, user_repo, post_repo):
"""Deve retornar todos os posts de um usuário"""
user = user_repo.create(username='prolific', email='prolific@example.com')
post1 = post_repo.create('Post 1', 'Content 1', user.id)
post2 = post_repo.create('Post 2', 'Content 2', user.id)
post3 = post_repo.create('Post 3', 'Content 3', user.id)
posts = post_repo.get_by_user(user.id)
assert len(posts) == 3
titles = {p.title for p in posts}
assert titles == {'Post 1', 'Post 2', 'Post 3'}
def test_cascade_delete_posts_when_user_deleted(self, user_repo, post_repo, db_session):
"""Deve deletar posts quando usuário é deletado (cascade)"""
user = user_repo.create(username='todelete', email='del@example.com')
post1 = post_repo.create('Post 1', 'Content', user.id)
post2 = post_repo.create('Post 2', 'Content', user.id)
# Verifica que posts existem
assert len(post_repo.get_by_user(user.id)) == 2
# Deleta usuário
user_repo.delete(user.id)
# Verifica que posts foram deletados em cascata
assert len(post_repo.get_by_user(user.id)) == 0
assert db_session.query(Post).filter_by(user_id=user.id).count() == 0
def test_user_posts_relationship_loaded(self, user_repo, post_repo):
"""Deve carregar posts através do relacionamento do usuário"""
user = user_repo.create(username='reltest', email='rel@example.com')
post_repo.create('Post A', 'Content A', user.id)
post_repo.create('Post B', 'Content B', user.id)
fetched_user = user_repo.get_by_username('reltest')
# Acessa posts através do relacionamento
assert len(fetched_user.posts) == 2
post_titles = {p.title for p in fetched_user.posts}
assert post_titles == {'Post A', 'Post B'}
Esses testes validam comportamentos mais complexos — relacionamentos, cascades e integridade referencial. Tudo isso testa o banco real, descobrindo problemas que mocks nunca revelariam.
Testes parametrizados para validação de dados
Pytest oferece @pytest.mark.parametrize para executar o mesmo teste com múltiplos conjuntos de dados:
class TestDataValidation:
"""Testes parametrizados para validação de entrada"""
@pytest.mark.parametrize('username,email', [
('user', 'user@example.com'),
('a', 'a@a.com'),
('very_long_username_123', 'long@example.com'),
('user-with-dash', 'dash@example.com'),
])
def test_create_user_valid_formats(self, user_repo, username, email):
"""Deve aceitar vários formatos válidos de username e email"""
user = user_repo.create(username=username, email=email)
assert user.id is not None
assert user.username == username
assert user.email == email
@pytest.mark.parametrize('title,content', [
('Título Curto', 'Conteúdo mínimo'),
('A' * 200, 'Conteúdo'),
('Normal', 'A' * 5000), # Teste com conteúdo muito longo
])
def test_create_post_various_lengths(self, user_repo, post_repo, title, content):
"""Deve criar posts com títulos e conteúdos de vários tamanhos"""
user = user_repo.create('testuser', 'test@example.com')
post = post_repo.create(title=title, content=content, user_id=user.id)
assert post.title == title
assert post.content == content
Parametrização reduz repetição de código e aumenta cobertura rapidamente.
Executando Testes e Integração com Docker
Iniciando o container e rodando testes
Use um arquivo .env para configurar variáveis:
TEST_DATABASE_URL=postgresql://testuser:testpass@localhost:5433/testdb
Inicie o banco:
docker-compose up -d postgres
Aguarde o healthcheck passar (use docker-compose logs -f postgres para monitorar). Depois, rode os testes:
pytest tests/ -v
A flag -v (verbose) mostra cada teste. Para mais detalhes:
pytest tests/ -v -s
A flag -s mostra prints durante testes (útil para debug). Para cobertura:
pytest tests/ --cov=app --cov-report=html
Isso gera um relatório HTML mostrando linhas cobertas.
Script de automação para CI/CD
Crie run_tests.sh:
#!/bin/bash
set -e
echo "Iniciando containers..."
docker-compose up -d postgres
echo "Aguardando banco estar pronto..."
for i in {1..30}; do
if docker-compose exec -T postgres pg_isready -U testuser -d testdb; then
echo "Banco pronto!"
break
fi
echo "Tentativa $i/30..."
sleep 2
done
echo "Rodando testes..."
pytest tests/ -v --cov=app
echo "Parando containers..."
docker-compose down -v
echo "Testes completos!"
Torne executável:
chmod +x run_tests.sh
./run_tests.sh
Este script é idempotente — pode rodar múltiplas vezes sem problemas. A flag -v em docker-compose down remove volumes, limpando dados de teste.
Conclusão
Três aprendizados fundamentais sobre testes de integração em Python: Primeiro, fixtures pytest com rollback automático são essenciais para isolamento — cada teste começa com dados limpos sem precisar limpá-los manualmente. Segundo, Docker garante consistência de ambiente; seu test passa localmente, passa no CI/CD e produção porque o banco é idêntico em todos os lugares. Terceiro, testes de integração reais (contra banco real, não mocks) encontram bugs de impedância, constraint violations e race conditions que testes unitários nunca pegam — esse é o retorno do investimento em tempo de escrita.