<p><strong>Meta Description:</strong> Aprenda a implementar testes de integração eficazes: APIs, serviços, bancos de dados e microsserviços. Guia completo com estratégias e exemplos práticos.</p>

<hr>
<h2>O Que São Testes de Integração?</h2>

<p>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).</p>

<h3>Quando Usar Testes de Integração</h3>

<figure class="wp-block-table"><table>
<tr><th>Situação</th><th>Teste Recomendado</th></tr>
<tr><td>Testar uma função isolada</td><td>Unit</td></tr>
<tr><td>Testar integração API ↔ Database</td><td>Integração</td></tr>
<tr><td>Testar múltiplos serviços juntos</td><td>Integração</td></tr>
<tr><td>Testar fluxo completo do usuário</td><td>E2E</td></tr>
<tr><td>Testar sistema inteiro em produção-like</td><td>Sistema</td></tr>
</table></figure>

<hr>
<h2>Tipos de Testes de Integração</h2>

<h3>1. Testes de Integração de APIs</h3>

<pre><code># test_integration_api.py
import requests
import pytest

<p>class TestOrderAPI:</p>

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

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

<p>def test_create_order_flow(self):<br>        """Testa fluxo completo: criar usuário -&gt; adicionar produto -&gt; fazer pedido"""</p>

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

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

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

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

<p># 5. Verificar pedido criado<br>        assert order["status"] == "pending"<br>        assert order["total"] == products[0]["price"] * 2<br></code></pre></p>

<h3>2. Testes de Integração com Banco de Dados</h3>

<pre><code># test_integration_database.py
import psycopg2
from decimal import Decimal

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

<p>@pytest.fixture<br>def clean_db(db_connection):<br>    """Limpa tabelas antes de cada teste"""<br>    cursor = db_connection.cursor()</p>

<p>cursor.execute("""<br>        TRUNCATE TABLE orders, cart_items, products, users<br>        RESTART IDENTITY CASCADE<br>    """)<br>    db_connection.commit()<br>    cursor.close()</p>

<p>yield db_connection</p>

<p>def test_user_order_integration(clean_db):<br>    """Testa integração usuário -&gt; pedido"""<br>    cursor = clean_db.cursor()</p>

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

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

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

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

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

<p>clean_db.commit()</p>

<p>def test_product_inventory_update(clean_db):<br>    """Testa que estoque é decrementado"""<br>    cursor = clean_db.cursor()</p>

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

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

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

<h3>3. Testes de Microsserviços</h3>

<pre><code># test_integration_microservices.py
import requests
from unittest.mock import patch, Mock

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

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

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

<p>assert response.status_code == 201<br>        order = response.json()</p>

<p># Verifica que pagamento foi chamado<br>        mock_payment_gateway.assert_called_once()</p>

<p># Verifica resultado<br>        assert order["status"] == "confirmed"<br>        assert "transaction_id" in order["payment"]</p>

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

<p>wait_for_services(containers)</p>

<p>yield containers</p>

<p>stop_containers(containers)<br></code></pre></p>

<h3>4. Testes de Contrato (Consumer-Driven)</h3>

<pre><code># test_contract_consumer.py
from pact import Consumer, Provider

<p>def test_consumer_driven_contract():<br>    """Testa contrato definido pelo consumer"""</p>

<p>pact = Consumer('OrderService').has_pact_with(<br>        Provider('ProductService')<br>    )<br>    pact.start_service()</p>

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

<hr>
<h2>TestContainers para Integração</h2>

<pre><code># conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

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

<p>@pytest.fixture(scope="module")<br>def redis():<br>    with RedisContainer("redis:7") as rd:<br>        yield rd</p>

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

<p>with app.app_context():<br>        yield app</p>

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

<p>yield postgres<br></code></pre></p>

<hr>
<h2>Testes de Integração de Mensageria</h2>

<pre><code># test_integration_messaging.py
import pika
import json

<p>@pytest.fixture<br>def rabbitmq():<br>    with RabbitMQContainer("rabbitmq:3") as rmq:<br>        yield rmq</p>

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

<p># Declare queue<br>    channel.queue_declare(queue='order_events', durable=True)</p>

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

<p># Consume message<br>    method, properties, body = channel.basic_get(queue='order_events')<br>    received = json.loads(body)</p>

<p>assert received['event'] == 'order_created'<br>    assert received['order_id'] == 123</p>

<p>connection.close()<br></code></pre></p>

<hr>
<h2>Estratégias de Integração</h2>

<h3>Bottom-Up</h3>

<pre><code>[DAL] → [BLL] → [API]
  ↓
 testes de unidade
</code></pre>

<h3>Top-Down</h3>

<pre><code>[UI] → [API] → [Services]
  ↑
 stubs
</code></pre>

<h3>Sandwich</h3>

<pre><code>[UI] → [API] → [Services] → [DAL]
   ↑          ↓
  stubs       drivers
</code></pre>

<hr>
<h2>Conclusão</h2>

<p>Testes de integração são essenciais para garantir que componentes funcionam juntos. As chaves são:<br><ol><li><strong>Selecionar escopo adequado</strong> - Nem demais, nem de menos</li><br><li><strong>Usar TestContainers</strong> - Isolamento e repeatability</li><br><li><strong>Testar contratos</strong> - Consumer-driven para microservices</li><br><li><strong>Manter dados limpos</strong> - Setup e teardown</li><br><li><strong>Executar frequentemente</strong> - CI/CD integrado</li><br></ol><br><hr><br><h3>FAQ</h3></p>

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

<p><strong>P: Testes de integração são lentos?</strong>  <br>R: Sim, mas TestContainers e paralelização ajudam. Priorize velocidade.</p>

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