<h2>Entendendo Testes de Integração: O que são e por que importam</h2>
<p>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.</p>
<p>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.</p>
<h2>Configurando o Ambiente: Docker, PostgreSQL e pytest</h2>
<h3>Estruturando o projeto e o Docker Compose</h3>
<p>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".</p>
<p>Crie o arquivo <code>docker-compose.yml</code>:</p>
<pre><code class="language-yaml">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:</code></pre>
<p>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.</p>
<h3>Dependências Python</h3>
<p>Crie um <code>requirements.txt</code>:</p>
<pre><code>pytest==7.4.3
psycopg2-binary==2.9.9
sqlalchemy==2.0.23
python-dotenv==1.0.0</code></pre>
<p>Use <code>pip install -r requirements.txt</code>. Psycopg2 é o adaptador PostgreSQL, SQLAlchemy fornece a abstração ORM robusta, e python-dotenv facilita gerenciamento de variáveis de ambiente.</p>
<h2>Criando a Camada de Dados e Modelos</h2>
<h3>Estrutura de banco de dados com SQLAlchemy</h3>
<p>Vamos criar uma aplicação simples de gerenciamento de usuários e posts. Primeiro, o arquivo <code>app/models.py</code>:</p>
<pre><code class="language-python">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}>'</code></pre>
<p>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.</p>
<h3>Repositório para operações de banco</h3>
<p>Crie <code>app/repository.py</code>:</p>
<pre><code class="language-python">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()</code></pre>
<p>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.</p>
<h2>Escrevendo Testes de Integração com pytest</h2>
<h3>Fixtures compartilhadas e isolamento de dados</h3>
<p>O arquivo <code>tests/conftest.py</code> é onde pytest busca fixtures reutilizáveis. É a peça-chave para testes de integração robustos:</p>
<pre><code class="language-python">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)</code></pre>
<p>A fixture <code>db_session</code> usa transações que são revertidas ao final de cada teste (<code>rollback</code>). Isso garante isolamento — um teste nunca afeta outro. O escopo <code>function</code> (padrão) reinicia o estado antes de cada teste. A engine é reutilizada (<code>scope='session'</code>) para performance, mas o schema é recriado com <code>create_all</code>.</p>
<h3>Testes unitários da camada de repositório</h3>
<p>Crie <code>tests/test_user_integration.py</code>:</p>
<pre><code class="language-python">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</code></pre>
<p>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 <code>pytest.raises</code> para testar exceções — mais legível que try-except.</p>
<h3>Testes de relacionamentos e integração cross-table</h3>
<p>Crie <code>tests/test_post_integration.py</code>:</p>
<pre><code class="language-python">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'}</code></pre>
<p>Esses testes validam comportamentos mais complexos — relacionamentos, cascades e integridade referencial. Tudo isso testa o banco real, descobrindo problemas que mocks nunca revelariam.</p>
<h3>Testes parametrizados para validação de dados</h3>
<p>Pytest oferece <code>@pytest.mark.parametrize</code> para executar o mesmo teste com múltiplos conjuntos de dados:</p>
<pre><code class="language-python">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</code></pre>
<p>Parametrização reduz repetição de código e aumenta cobertura rapidamente.</p>
<h2>Executando Testes e Integração com Docker</h2>
<h3>Iniciando o container e rodando testes</h3>
<p>Use um arquivo <code>.env</code> para configurar variáveis:</p>
<pre><code>TEST_DATABASE_URL=postgresql://testuser:testpass@localhost:5433/testdb</code></pre>
<p>Inicie o banco:</p>
<pre><code class="language-bash">docker-compose up -d postgres</code></pre>
<p>Aguarde o healthcheck passar (use <code>docker-compose logs -f postgres</code> para monitorar). Depois, rode os testes:</p>
<pre><code class="language-bash">pytest tests/ -v</code></pre>
<p>A flag <code>-v</code> (verbose) mostra cada teste. Para mais detalhes:</p>
<pre><code class="language-bash">pytest tests/ -v -s</code></pre>
<p>A flag <code>-s</code> mostra prints durante testes (útil para debug). Para cobertura:</p>
<pre><code class="language-bash">pytest tests/ --cov=app --cov-report=html</code></pre>
<p>Isso gera um relatório HTML mostrando linhas cobertas.</p>
<h3>Script de automação para CI/CD</h3>
<p>Crie <code>run_tests.sh</code>:</p>
<pre><code class="language-bash">#!/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!"</code></pre>
<p>Torne executável:</p>
<pre><code class="language-bash">chmod +x run_tests.sh
./run_tests.sh</code></pre>
<p>Este script é idempotente — pode rodar múltiplas vezes sem problemas. A flag <code>-v</code> em <code>docker-compose down</code> remove volumes, limpando dados de teste.</p>
<h2>Conclusão</h2>
<p>Três aprendizados fundamentais sobre testes de integração em Python: <strong>Primeiro</strong>, fixtures pytest com rollback automático são essenciais para isolamento — cada teste começa com dados limpos sem precisar limpá-los manualmente. <strong>Segundo</strong>, 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. <strong>Terceiro</strong>, 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.</p>
<h2>Referências</h2>
<ul>
<li><a href="https://docs.pytest.org/" target="_blank" rel="noopener noreferrer">pytest Documentation</a></li>
<li><a href="https://docs.sqlalchemy.org/en/20/orm/" target="_blank" rel="noopener noreferrer">SQLAlchemy ORM Documentation</a></li>
<li><a href="https://docs.docker.com/compose/" target="_blank" rel="noopener noreferrer">Docker Compose Documentation</a></li>
<li><a href="https://hub.docker.com/_/postgres" target="_blank" rel="noopener noreferrer">PostgreSQL Docker Image</a></li>
<li><a href="https://realpython.com/python-mock-library/" target="_blank" rel="noopener noreferrer">Python unittest.mock vs Real Integration Testing</a></li>
</ul>
<p><!-- FIM --></p>