Backend Testing (Python)
Guia de testes para backend Python com FastAPI, pytest e moto.
Setup
Dependências
requirements-dev.txt:
pytest>=7.4.0
pytest-asyncio>=0.21.0
pytest-cov>=4.1.0
pytest-mock>=3.11.0
moto[all]>=4.2.0 # Mock AWS services
faker>=20.0.0
factory-boy>=3.3.0
httpx>=0.25.0 # For async testing
Configuração pytest
pytest.ini:
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
asyncio_mode = auto
addopts =
-v
--strict-markers
--cov=app
--cov-report=term-missing
--cov-report=html
--cov-fail-under=70
markers =
unit: Unit tests
integration: Integration tests
slow: Slow tests
Unit Tests
Testing Services
# tests/unit/test_services/test_user_service.py
import pytest
from app.services.user_service import UserService
from app.models import User
from app.exceptions import EmailAlreadyExistsError
@pytest.fixture
def user_service(mock_repository):
return UserService(repository=mock_repository)
@pytest.mark.asyncio
async def test_create_user_with_valid_data(user_service, sample_user_data):
"""Test that user is created successfully with valid data."""
# Arrange
user_service.repository.get_by_email.return_value = None
# Act
user = await user_service.create_user(sample_user_data)
# Assert
assert user.email == sample_user_data['email'].lower()
user_service.repository.create.assert_called_once()
@pytest.mark.asyncio
async def test_create_user_fails_with_duplicate_email(user_service, sample_user_data):
"""Test that creating user with duplicate email raises error."""
# Arrange
existing_user = User(id=1, **sample_user_data)
user_service.repository.get_by_email.return_value = existing_user
# Act & Assert
with pytest.raises(EmailAlreadyExistsError):
await user_service.create_user(sample_user_data)
Testing Models
# tests/unit/test_models/test_user.py
from app.models import User
def test_user_model_creation():
"""Test User model instantiation."""
user = User(
email="test@example.com",
name="Test User",
password_hash="hashed"
)
assert user.email == "test@example.com"
assert user.is_active is True # Default value
def test_user_full_name_property():
"""Test User.full_name computed property."""
user = User(name="John Doe")
assert user.full_name == "John Doe"
Testing Validators
# tests/unit/test_validators.py
import pytest
from app.validators import validate_email, validate_password
@pytest.mark.parametrize("email,expected", [
("valid@example.com", True),
("another.valid+tag@domain.co.uk", True),
("invalid", False),
("@invalid.com", False),
("invalid@", False),
])
def test_email_validation(email, expected):
"""Test email validation with various inputs."""
result = validate_email(email)
assert result == expected
def test_password_must_have_minimum_length():
"""Test password validation enforces minimum length."""
with pytest.raises(ValueError, match="at least 8 characters"):
validate_password("short")
Integration Tests
Testing API Endpoints
# tests/integration/test_api/test_users_api.py
import pytest
from httpx import AsyncClient
from app.main import app
@pytest.fixture
async def client():
async with AsyncClient(app=app, base_url="http://test") as client:
yield client
@pytest.fixture
async def auth_headers(client, test_user):
"""Get authentication headers for test user."""
response = await client.post("/auth/login", json={
"email": test_user.email,
"password": "password123"
})
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.mark.asyncio
async def test_create_user_returns_201(client):
"""Test POST /users returns 201 with valid data."""
# Arrange
payload = {
"email": "newuser@example.com",
"name": "New User",
"password": "SecurePass123!"
}
# Act
response = await client.post("/api/v1/users", json=payload)
# Assert
assert response.status_code == 201
data = response.json()
assert data["email"] == payload["email"]
assert "id" in data
assert "password" not in data # Never return password
@pytest.mark.asyncio
async def test_get_user_requires_authentication(client):
"""Test GET /users/me returns 401 without auth."""
response = await client.get("/api/v1/users/me")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_get_current_user_returns_user_data(client, auth_headers, test_user):
"""Test GET /users/me returns current user."""
response = await client.get("/api/v1/users/me", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["email"] == test_user.email
assert data["name"] == test_user.name
Mocking AWS Services (moto)
S3
import boto3
from moto import mock_s3
@mock_s3
def test_upload_file_to_s3():
"""Test file upload to S3."""
# Setup mock S3
s3 = boto3.client('s3', region_name='us-east-1')
s3.create_bucket(Bucket='test-bucket')
# Test upload
s3.put_object(
Bucket='test-bucket',
Key='test.txt',
Body=b'test content'
)
# Assert
response = s3.get_object(Bucket='test-bucket', Key='test.txt')
assert response['Body'].read() == b'test content'
SQS
from moto import mock_sqs
@mock_sqs
@pytest.mark.asyncio
async def test_send_message_to_queue():
"""Test sending message to SQS."""
sqs = boto3.client('sqs', region_name='us-east-1')
# Create queue
queue_url = sqs.create_queue(QueueName='test-queue')['QueueUrl']
# Send message
await send_user_event({'user_id': 123})
# Assert
messages = sqs.receive_message(QueueUrl=queue_url)
assert len(messages['Messages']) == 1
assert 'user_id' in messages['Messages'][0]['Body']
Lambda (SAM Local)
import json
import subprocess
def test_lambda_handler():
"""Test Lambda function locally with SAM."""
event = {
"body": json.dumps({"user_id": 123}),
"headers": {"Content-Type": "application/json"}
}
# Invoke with SAM Local
result = subprocess.run(
["sam", "local", "invoke", "UserFunction", "-e", "event.json"],
capture_output=True
)
response = json.loads(result.stdout)
assert response['statusCode'] == 200
Fixtures e Factories
Fixtures Compartilhadas
# tests/conftest.py
import pytest
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from app.database import Base
@pytest.fixture(scope="session")
def event_loop():
"""Create event loop for async tests."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
async def db_session():
"""Create test database session."""
engine = create_async_engine("postgresql+asyncpg://test:test@localhost/test_db")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with AsyncSession(engine) as session:
yield session
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
def sample_user_data():
return {
"email": "test@example.com",
"name": "Test User",
"password": "SecurePass123!"
}
Factories
# tests/factories.py
import factory
from factory import Faker
from app.models import User
class UserFactory(factory.Factory):
class Meta:
model = User
id = factory.Sequence(lambda n: n)
email = Faker('email')
name = Faker('name')
password_hash = 'hashed_password'
is_active = True
# Uso
user = UserFactory.build()
user_with_email = UserFactory.build(email='specific@example.com')
users = UserFactory.build_batch(10)
Comandos Úteis
# Rodar todos os testes
pytest
# Apenas unit tests
pytest tests/unit
# Com coverage
pytest --cov --cov-report=html
# Específico
pytest tests/unit/test_user_service.py::test_create_user
# Last failed
pytest --lf
# Verbose
pytest -v -s
# Parallel (mais rápido)
pytest -n auto
# Watch mode
ptw # pytest-watch
Debugging
# Adicionar breakpoint
import pdb; pdb.set_trace()
# Ou com pytest
pytest --pdb # Break on failure
# Print output
pytest -s # Show print statements