Integration Testing
Testes de integração para Lambda, SQS, SNS e fluxos completos.
SAM Local Testing
Setup
# Instalar SAM CLI
brew install aws-sam-cli # macOS
# ou
pip install aws-sam-cli
# Verificar instalação
sam --version
Invocar Lambda Localmente
# Gerar evento de teste
sam local generate-event apigateway aws-proxy > event.json
# Editar event.json com dados de teste
# Invocar Lambda
sam local invoke UserFunction -e event.json
# Com debugging
sam local invoke UserFunction -e event.json -d 5858
Testar API Gateway Localmente
# Iniciar API local
sam local start-api
# Em outro terminal, testar
curl http://localhost:3000/api/v1/users
curl -X POST http://localhost:3000/api/v1/users \
-H "Content-Type: application/json" \
-d '{"email":"test@example.com","name":"Test"}'
LocalStack (AWS Services Locais)
Setup
# Docker Compose
docker-compose up localstack
# Ou via Docker
docker run -d -p 4566:4566 localstack/localstack
Testing com LocalStack
import boto3
import pytest
@pytest.fixture
def aws_credentials(monkeypatch):
"""Mock AWS credentials for LocalStack."""
monkeypatch.setenv('AWS_ACCESS_KEY_ID', 'test')
monkeypatch.setenv('AWS_SECRET_ACCESS_KEY', 'test')
monkeypatch.setenv('AWS_ENDPOINT_URL', 'http://localhost:4566')
@pytest.fixture
def sqs_client(aws_credentials):
"""Create SQS client for LocalStack."""
return boto3.client(
'sqs',
region_name='us-east-1',
endpoint_url='http://localhost:4566'
)
@pytest.mark.integration
async def test_message_processing_flow(sqs_client):
"""Test complete SQS message processing flow."""
# Arrange: Create queue
queue_url = sqs_client.create_queue(QueueName='test-queue')['QueueUrl']
# Act: Send message
sqs_client.send_message(
QueueUrl=queue_url,
MessageBody=json.dumps({'user_id': 123, 'action': 'welcome_email'})
)
# Act: Process message (simulate Lambda)
from app.workers.email_worker import lambda_handler
event = {
'Records': [{
'body': '{"user_id": 123, "action": "welcome_email"}'
}]
}
result = await lambda_handler(event, {})
# Assert
assert result['statusCode'] == 200
assert 'processed' in result['body']
Database Integration Tests
TestContainers (Recomendado)
# tests/integration/conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
@pytest.fixture(scope="session")
def postgres_container():
"""Start PostgreSQL container for tests."""
with PostgresContainer("postgres:14") as postgres:
yield postgres
@pytest.fixture(scope="session")
def database_url(postgres_container):
"""Get database URL from container."""
return postgres_container.get_connection_url()
@pytest.fixture
async def db_session(database_url):
"""Create database session for tests."""
engine = create_async_engine(database_url)
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)
Testing com Database Real
@pytest.mark.integration
@pytest.mark.asyncio
async def test_user_crud_operations(db_session):
"""Test complete CRUD flow with real database."""
# Create
user = User(email="test@example.com", name="Test User")
db_session.add(user)
await db_session.commit()
await db_session.refresh(user)
assert user.id is not None
# Read
result = await db_session.execute(
select(User).where(User.email == "test@example.com")
)
found_user = result.scalar_one()
assert found_user.name == "Test User"
# Update
found_user.name = "Updated Name"
await db_session.commit()
await db_session.refresh(found_user)
assert found_user.name == "Updated Name"
# Delete
await db_session.delete(found_user)
await db_session.commit()
result = await db_session.execute(
select(User).where(User.id == user.id)
)
assert result.scalar_one_or_none() is None
E2E Tests (Playwright)
Setup
# Instalar Playwright
npm install --save-dev @playwright/test
npx playwright install
# Configurar
npx playwright init
E2E Test Example
// tests/e2e/user-journey.spec.ts
import { test, expect } from '@playwright/test';
test.describe('User Journey', () => {
test('complete signup to checkout flow', async ({ page }) => {
// 1. Signup
await page.goto('https://staging.seuapp.com/signup');
await page.fill('[name="email"]', 'test@example.com');
await page.fill('[name="password"]', 'SecurePass123!');
await page.click('button:has-text("Sign Up")');
await expect(page).toHaveURL('/dashboard');
// 2. Browse products
await page.click('[data-testid="products-link"]');
await page.click('text=Product Name');
// 3. Add to cart
await page.click('button:has-text("Add to Cart")');
await expect(page.locator('[data-testid="cart-count"]')).toHaveText('1');
// 4. Checkout
await page.click('[data-testid="cart-icon"]');
await page.click('button:has-text("Checkout")');
await page.fill('[name="address"]', '123 Test St');
await page.click('button:has-text("Complete Order")');
// 5. Verify success
await expect(page.locator('text=Order Confirmed')).toBeVisible();
});
});
Playwright Configuration
// playwright.config.ts
import { defineConfig, devices } from '@playwright/test';
export default defineConfig({
testDir: './tests/e2e',
fullyParallel: true,
forbidOnly: !!process.env.CI,
retries: process.env.CI ? 2 : 0,
workers: process.env.CI ? 1 : undefined,
reporter: 'html',
use: {
baseURL: process.env.BASE_URL || 'https://staging.seuapp.com',
trace: 'on-first-retry',
screenshot: 'only-on-failure',
},
projects: [
{
name: 'chromium',
use: { ...devices['Desktop Chrome'] },
},
{
name: 'mobile',
use: { ...devices['iPhone 13'] },
},
],
});
Performance Tests
Load Testing (Locust)
# tests/performance/locustfile.py
from locust import HttpUser, task, between
class APIUser(HttpUser):
wait_time = between(1, 3)
host = "https://staging.seuapp.com"
def on_start(self):
"""Login before tests."""
response = self.client.post("/auth/login", json={
"email": "test@example.com",
"password": "password123"
})
self.token = response.json()["access_token"]
self.headers = {"Authorization": f"Bearer {self.token}"}
@task(3)
def get_users(self):
"""Get users list (more frequent)."""
self.client.get("/api/v1/users", headers=self.headers)
@task(1)
def create_user(self):
"""Create user (less frequent)."""
self.client.post("/api/v1/users", json={
"email": f"user-{uuid.uuid4()}@example.com",
"name": "Test User"
}, headers=self.headers)
# Rodar
# locust -f tests/performance/locustfile.py --host=https://staging.seuapp.com
Smoke Tests
Testes rápidos para verificar que sistema está funcionando.
# tests/smoke/test_health.py
import pytest
import httpx
@pytest.mark.asyncio
async def test_api_health_endpoint():
"""Test that API health endpoint responds."""
async with httpx.AsyncClient() as client:
response = await client.get("https://api.seuapp.com/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
@pytest.mark.asyncio
async def test_database_connection():
"""Test that database is accessible."""
async with httpx.AsyncClient() as client:
response = await client.get("https://api.seuapp.com/health/db")
assert response.status_code == 200
@pytest.mark.asyncio
async def test_critical_endpoint_responds():
"""Test critical endpoints are responding."""
endpoints = [
"/api/v1/users",
"/api/v1/products",
"/api/v1/orders"
]
async with httpx.AsyncClient(base_url="https://api.seuapp.com") as client:
for endpoint in endpoints:
response = await client.get(endpoint)
assert response.status_code in [200, 401] # 401 OK (precisa auth)
CI Integration
GitHub Actions
name: Integration Tests
on:
pull_request:
push:
branches: [dev, main]
jobs:
integration:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:14
env:
POSTGRES_PASSWORD: postgres
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
localstack:
image: localstack/localstack
ports:
- 4566:4566
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements-dev.txt
- name: Run integration tests
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
AWS_ENDPOINT_URL: http://localhost:4566
run: |
pytest tests/integration -v