Testes de Integração: Conectando Componentes

---

O Que São Testes de Integração?

Testes de integração verificam que diferentes módulos ou serviços funcionam corretamente quando combinados. São o meio termo entre unit tests (testam unidades isoladas) e E2E tests (testam sistema completo).

Quando Usar Testes de Integração

Situação	Teste Recomendado
Testar uma função isolada	Unit
Testar integração API ↔ Database	Integração
Testar múltiplos serviços juntos	Integração
Testar fluxo completo do usuário	E2E
Testar sistema inteiro em produção-like	Sistema

---

Tipos de Testes de Integração

1. Testes de Integração de APIs

# test_integration_api.py
import requests
import pytest

class TestOrderAPI:

@pytest.fixture(autouse=True)
    def setup(self):
        self.base_url = "http://localhost:8000/api/v1"
        self.session = requests.Session()

# Setup: criar usuário de teste
        response = self.session.post(f"{self.base_url}/users", json={
            "name": "Test User",
            "email": f"test_{uuid.uuid4()}@example.com",
            "password": "testpass123"
        })
        self.user_id = response.json()["id"]
        self.token = response.json()["token"]
        self.session.headers.update({"Authorization": f"Bearer {self.token}"})

def test_create_order_flow(self):
        """Testa fluxo completo: criar usuário -> adicionar produto -> fazer pedido"""

# 1. Listar produtos
        products_response = self.session.get(f"{self.base_url}/products")
        assert products_response.status_code == 200
        products = products_response.json()["items"]
        product_id = products[0]["id"]

# 2. Adicionar ao carrinho
        cart_response = self.session.post(f"{self.base_url}/cart/items", json={
            "product_id": product_id,
            "quantity": 2
        })
        assert cart_response.status_code == 201

# 3. Verificar carrinho
        cart = self.session.get(f"{self.base_url}/cart").json()
        assert len(cart["items"]) == 1
        assert cart["items"][0]["quantity"] == 2

# 4. Criar pedido
        order_response = self.session.post(f"{self.base_url}/orders", json={
            "shipping_address": "Rua Teste, 123",
            "payment_method": "credit_card"
        })
        assert order_response.status_code == 201
        order = order_response.json()

# 5. Verificar pedido criado
        assert order["status"] == "pending"
        assert order["total"] == products[0]["price"] * 2

2. Testes de Integração com Banco de Dados

# test_integration_database.py
import psycopg2
from decimal import Decimal

@pytest.fixture
def db_connection():
    connection = psycopg2.connect(
        host="localhost",
        database="testdb",
        user="test",
        password="test"
    )
    yield connection
    connection.close()

@pytest.fixture
def clean_db(db_connection):
    """Limpa tabelas antes de cada teste"""
    cursor = db_connection.cursor()

cursor.execute("""
        TRUNCATE TABLE orders, cart_items, products, users
        RESTART IDENTITY CASCADE
    """)
    db_connection.commit()
    cursor.close()

yield db_connection

def test_user_order_integration(clean_db):
    """Testa integração usuário -> pedido"""
    cursor = clean_db.cursor()

# 1. Criar usuário
    cursor.execute("""
        INSERT INTO users (name, email, password_hash)
        VALUES ({6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s)
        RETURNING id
    """, ("João", "joao@example.com", "hash123"))
    user_id = cursor.fetchone()[0]

# 2. Criar produto
    cursor.execute("""
        INSERT INTO products (name, price, stock)
        VALUES ({6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s)
        RETURNING id
    """, ("Camiseta", Decimal("49.90"), 100))
    product_id = cursor.fetchone()[0]

# 3. Criar pedido
    cursor.execute("""
        INSERT INTO orders (user_id, total, status)
        VALUES ({6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s)
        RETURNING id
    """, (user_id, Decimal("49.90"), "pending"))
    order_id = cursor.fetchone()[0]

# 4. Verificar integridade
    cursor.execute("""
        SELECT u.name, o.id, o.total
        FROM users u
        JOIN orders o ON u.id = o.user_id
        WHERE u.id = {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s
    """, (user_id,))

result = cursor.fetchone()
    assert result[0] == "João"
    assert result[1] == order_id
    assert result[2] == Decimal("49.90")

clean_db.commit()

def test_product_inventory_update(clean_db):
    """Testa que estoque é decrementado"""
    cursor = clean_db.cursor()

# Setup: criar produto com estoque 10
    cursor.execute("""
        INSERT INTO products (name, price, stock)
        VALUES ({6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s, {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s)
        RETURNING id
    """, ("Tênis", Decimal("199.90"), 10))
    product_id = cursor.fetchone()[0]

# Action: decrementar estoque (simula venda)
    cursor.execute("""
        UPDATE products 
        SET stock = stock - 2
        WHERE id = {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s
    """, (product_id,))

# Assert: estoque correto
    cursor.execute("SELECT stock FROM products WHERE id = {6727d158e4474b0847515bd23a8ee4bebd5f1aac7a18bc968e51d8985dccb442}s", (product_id,))
    assert cursor.fetchone()[0] == 8

3. Testes de Microsserviços

# test_integration_microservices.py
import requests
from unittest.mock import patch, Mock

class TestOrderServiceIntegration:
    """Testa integração entre Order Service e Payment/Gateway Service"""

@pytest.fixture
    def mock_payment_gateway(self):
        """Mock do serviço de pagamento"""
        with patch('services.payment_gateway.requests.post') as mock:
            mock.return_value = Mock(
                status_code=200,
                json=lambda: {
                    "transaction_id": "tx_123",
                    "status": "approved",
                    "amount": 99.90
                }
            )
            yield mock

def test_order_with_payment(self, mock_payment_gateway):
        """Testa que pedido processa corretamente com pagamento"""
        # Call order service
        response = requests.post(
            "http://localhost:8001/api/orders",
            json={
                "user_id": 1,
                "items": [{"product_id": 1, "quantity": 1}],
                "payment": {
                    "method": "credit_card",
                    "card_token": "tok_test"
                }
            }
        )

assert response.status_code == 201
        order = response.json()

# Verifica que pagamento foi chamado
        mock_payment_gateway.assert_called_once()

# Verifica resultado
        assert order["status"] == "confirmed"
        assert "transaction_id" in order["payment"]

@pytest.fixture(scope="module")
def service_containers():
    """Start containers para serviços reais"""
    containers = {
        'user_service': start_container('user-service:latest'),
        'product_service': start_container('product-service:latest'),
        'order_service': start_container('order-service:latest'),
    }

wait_for_services(containers)

yield containers

stop_containers(containers)

4. Testes de Contrato (Consumer-Driven)

# test_contract_consumer.py
from pact import Consumer, Provider

def test_consumer_driven_contract():
    """Testa contrato definido pelo consumer"""

pact = Consumer('OrderService').has_pact_with(
        Provider('ProductService')
    )
    pact.start_service()

try:
        (pact
         .given('product with id 1 exists')
         .upon_receiving('a request for product details')
         .with_request('GET', '/products/1')
         .will_respond_with(200, body={
             'id': 1,
             'name': 'Test Product',
             'price': 99.99,
             'in_stock': True
         })
         .with_request('GET', '/products/999')
         .will_respond_with(404, body={
             'error': 'Product not found'
         })
         .verify())
    finally:
        pact.stop_service()

---

TestContainers para Integração

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="module")
def postgres():
    with PostgresContainer("postgres:15") as pg:
        # Run migrations
        run_migrations(pg.get_connection_url())
        yield pg

@pytest.fixture(scope="module")
def redis():
    with RedisContainer("redis:7") as rd:
        yield rd

@pytest.fixture
def integration_app(postgres, redis):
    """Aplica com todas dependências"""
    app = create_app(
        database_url=postgres.get_connection_url(),
        redis_url=redis.get_connection_url()
    )
    app.config['TESTING'] = True

with app.app_context():
        yield app

@pytest.fixture
def clean_data(postgres):
    """Limpa dados entre testes"""
    cursor = postgres.get_connection().cursor()
    cursor.execute("""
        TRUNCATE TABLE orders, users, products, sessions
        RESTART IDENTITY CASCADE
    """)
    postgres.get_connection().commit()
    cursor.close()

yield postgres

---

Testes de Integração de Mensageria

# test_integration_messaging.py
import pika
import json

@pytest.fixture
def rabbitmq():
    with RabbitMQContainer("rabbitmq:3") as rmq:
        yield rmq

def test_message_publishing(rabbitmq):
    """Testa publicação e consumo de mensagens"""
    connection = pika.BlockingConnection(
        pika.URLParameters(rabbitmq.get_connection_url())
    )
    channel = connection.channel()

# Declare queue
    channel.queue_declare(queue='order_events', durable=True)

# Publish message
    message = {
        'event': 'order_created',
        'order_id': 123,
        'user_id': 456,
        'total': 99.90
    }
    channel.basic_publish(
        exchange='',
        routing_key='order_events',
        body=json.dumps(message)
    )

# Consume message
    method, properties, body = channel.basic_get(queue='order_events')
    received = json.loads(body)

assert received['event'] == 'order_created'
    assert received['order_id'] == 123

connection.close()

---

Estratégias de Integração

Bottom-Up

[DAL] → [BLL] → [API]
  ↓
 testes de unidade

Top-Down

[UI] → [API] → [Services]
  ↑
 stubs

Sandwich

[UI] → [API] → [Services] → [DAL]
   ↑          ↓
  stubs       drivers

---

Conclusão

Testes de integração são essenciais para garantir que componentes funcionam juntos. As chaves são:

1. Selecionar escopo adequado – Nem demais, nem de menos
2. Usar TestContainers – Isolamento e repeatability
3. Testar contratos – Consumer-driven para microservices
4. Manter dados limpos – Setup e teardown
5. Executar frequentemente – CI/CD integrado

---

FAQ

P: Quantos testes de integração devo ter?
R: Cobertura adequada das interfaces críticas. Priorize integrações com maior risco.

P: Testes de integração são lentos?
R: Sim, mas TestContainers e paralelização ajudam. Priorize velocidade.

P: Usar mocks ou serviços reais?
R: Mocks para dependências externas lentas/indisponíveis. Reais para DB e cache locais.