Como fazer stress test em integrações com API de CPF antes de ir para produção

Para fazer stress test em uma integração com API de CPF, você precisa simular carga realista com mocks, medir latência em percentis (P50, P95, P99) e validar se o seu código lida corretamente com timeouts, retries e circuit breakers — sem consumir consultas reais do plano. A API da CPFHub.io tem latência média de ~900ms, então os thresholds de latência aceitável precisam ser calibrados para essa realidade. Colocar a integração em produção sem esse teste é aceitar o risco de cascatas de falha que comprometem funcionalidades que nem dependem da consulta de CPF diretamente. A documentação do k6 é uma boa referência para estruturar testes de carga contra dependências externas.

Colocar uma integração em produção sem testar seu comportamento sob carga é um risco que pode resultar em timeouts, erros em cascata e experiência degradada para o usuário final. Quando sua aplicação depende de uma API externa para validar CPFs em tempo real, é fundamental entender como ela se comporta sob diferentes níveis de demanda -- e, mais importante, como o seu código lida com cenários de falha.

O que testar em um stress test

O stress test de uma integração com API externa não testa apenas a API em si. O objetivo principal é validar como o seu código se comporta sob pressão.

Pontos críticos

• Tratamento de timeout -- O que acontece quando a API demora mais que o esperado para responder?

• Tratamento de erros temporários -- Como o código reage quando recebe um erro 500 ou 503?

• Pool de conexões -- O número de conexões simultâneas está configurado corretamente?

• Retries e backoff -- A lógica de retry funciona sem sobrecarregar a API?

• Circuit breaker -- O sistema tem um mecanismo para parar de fazer chamadas quando a API está fora do ar?

• Consumo de memória -- O código não está acumulando respostas em memória sem liberar?

Respeitando os rate limits

Antes de executar qualquer stress test, é fundamental conhecer os rate limits do plano contratado.

O stress test deve simular a carga que sua aplicação realmente enfrentará em produção, respeitando os limites do plano. Para testes de carga que excedam esses limites, use mocks no lugar da API real.

Stress test com mocks em Python

O teste a seguir simula múltiplas consultas simultâneas usando mocks, sem consumir consultas reais da API.

import time
import random
import threading
from unittest.mock import patch, MagicMock
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def consultar_cpf(cpf: str) -> dict:
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': 'SUA_CHAVE_DE_API',
    'Accept': 'application/json'
    }
    response = requests.get(url, headers=headers, timeout=10)
    return response.json()

def mock_api_response(*args, **kwargs):
    """Simula latencia e respostas da API"""
    mock_response = MagicMock()
    latencia = random.uniform(0.5, 1.5)
    time.sleep(latencia)

    # Simular 5% de erros
    if random.random() < 0.05:
    mock_response.status_code = 503
    mock_response.json.return_value = {'success': False, 'error': 'Service Unavailable'}
    else:
    mock_response.status_code = 200
    mock_response.json.return_value = {
    'success': True,
    'data': {
    'cpf': '12345678900',
    'name': 'Teste Stress',
    'nameUpper': 'TESTE STRESS',
    'gender': 'M',
    'birthDate': '01/01/1990',
    'day': 1,
    'month': 1,
    'year': 1990
    }
    }

    return mock_response

@patch('requests.get', side_effect=mock_api_response)
def executar_stress_test(mock_get, num_requisicoes=100, max_workers=10):
    resultados = {
    'sucesso': 0,
    'erro': 0,
    'tempos': [],
    'erros': []
    }

    def fazer_consulta(i):
    inicio = time.time()
    try:
    resultado = consultar_cpf(f'{i:011d}')
    tempo = (time.time() - inicio) * 1000

    if resultado.get('success'):
    return ('sucesso', tempo)
    else:
    return ('erro', tempo, resultado.get('error', 'Desconhecido'))
    except Exception as e:
    tempo = (time.time() - inicio) * 1000
    return ('excecao', tempo, str(e))

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = [executor.submit(fazer_consulta, i) for i in range(num_requisicoes)]

    for future in as_completed(futures):
    resultado = future.result()
    if resultado[0] == 'sucesso':
    resultados['sucesso'] += 1
    resultados['tempos'].append(resultado[1])
    else:
    resultados['erro'] += 1
    resultados['erros'].append(resultado[2] if len(resultado) > 2 else 'N/A')

    # Metricas
    tempos = resultados['tempos']
    if tempos:
    tempos.sort()
    print(f'Total: {num_requisicoes} requisicoes')
    print(f'Sucesso: {resultados["sucesso"]}')
    print(f'Erros: {resultados["erro"]}')
    print(f'Media: {sum(tempos)/len(tempos):.0f}ms')
    print(f'P50: {tempos[len(tempos)//2]:.0f}ms')
    print(f'P95: {tempos[int(len(tempos)*0.95)]:.0f}ms')
    print(f'P99: {tempos[int(len(tempos)*0.99)]:.0f}ms')
    print(f'Max: {max(tempos):.0f}ms')

executar_stress_test(num_requisicoes=200, max_workers=20)

Testando o circuit breaker

Um circuit breaker é um padrão que interrompe as chamadas à API quando ela começa a falhar repetidamente, evitando sobrecarga no sistema.

import requests
import time

class CircuitBreaker:
    def __init__(self, falhas_max=5, tempo_reset=30):
    self.falhas_max = falhas_max
    self.tempo_reset = tempo_reset
    self.falhas = 0
    self.estado = 'fechado' # fechado, aberto, meio_aberto
    self.ultimo_erro = None

    def pode_executar(self) -> bool:
    if self.estado == 'fechado':
    return True
    if self.estado == 'aberto':
    if time.time() - self.ultimo_erro > self.tempo_reset:
    self.estado = 'meio_aberto'
    return True
    return False
    return True # meio_aberto

    def registrar_sucesso(self):
    self.falhas = 0
    self.estado = 'fechado'

    def registrar_falha(self):
    self.falhas += 1
    self.ultimo_erro = time.time()
    if self.falhas >= self.falhas_max:
    self.estado = 'aberto'

circuit_breaker = CircuitBreaker(falhas_max=3, tempo_reset=60)

def consultar_cpf_com_circuit_breaker(cpf: str) -> dict:
    if not circuit_breaker.pode_executar():
    return {
    'success': False,
    'error': 'Circuit breaker aberto -- API temporariamente indisponivel'
    }

    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': 'SUA_CHAVE_DE_API',
    'Accept': 'application/json'
    }

    try:
    response = requests.get(url, headers=headers, timeout=10)

    if response.status_code == 200:
    circuit_breaker.registrar_sucesso()
    return response.json()
    else:
    circuit_breaker.registrar_falha()
    return {'success': False, 'error': f'Status {response.status_code}'}

    except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
    circuit_breaker.registrar_falha()
    return {'success': False, 'error': str(e)}

Métricas que importam

Após executar os testes, analise as seguintes métricas para determinar se a integração está pronta para produção.

Checklist pré-produção

Antes de colocar a integração em produção, confirme que todos estes itens foram verificados:

• Timeout configurado -- Todas as chamadas à API têm timeout definido (recomendado: 10 segundos).

• Retry com backoff -- Erros temporários (500, 503) são retentados com intervalo crescente.

• Circuit breaker implementado -- O sistema para de fazer chamadas quando a API está fora do ar.

• Rate limits respeitados -- O código não excede os limites do plano contratado.

• Logs e métricas -- Todas as chamadas são registradas com tempo de resposta e status.

• Fallback definido -- O sistema tem um comportamento alternativo quando a API não está disponível.

• Stress test executado -- O código foi testado sob carga simulada com resultados documentados.

Perguntas frequentes

Qual a latência esperada da API de CPF em produção?

A latência típica da API da CPFHub.io é de aproximadamente 900ms no P50. Para dimensionar corretamente os timeouts, configure um valor de 10 segundos — o que cobre tranquilamente P99 e deixa margem para retries. Valores menores que 2 segundos costumam causar timeouts espúrios em horários de pico.

O que testar com mocks versus com a API real durante o stress test?

Use mocks para simular carga, erros (timeout, 503) e comportamento de circuit breaker sem consumir consultas reais. Use a API real apenas para validar o fluxo de ponta a ponta com um volume controlado — entre 5 e 10 chamadas — antes de ir para produção.

Quantas conexões simultâneas posso fazer no plano gratuito?

O plano gratuito permite 1 requisição a cada 2 segundos. Isso significa que paralelismo real não se aplica nesse plano — a lógica correta é uma fila sequencial com delay de 2 segundos entre chamadas. Para cargas com concorrência real, o plano Pro (1 req/s) ou Corporativo (personalizado) é o caminho.

Como documentar os resultados do stress test para o time?

Registre P50, P95, P99, taxa de sucesso e taxa de timeout em uma tabela simples. Compare com os valores aceitáveis da tabela de métricas acima e inclua o código do teste no repositório. A OWASP recomenda documentar testes de performance como artefatos do processo de segurança em integrações com terceiros.

Conclusão

O stress test é uma etapa indispensável antes de colocar qualquer integração com API em produção. Ao testar como o seu código se comporta sob carga, você identifica gargalos, valida mecanismos de resiliência e garante que a experiência do usuário final não será comprometida em dias de pico.

A API da CPFHub.io foi projetada para uso em produção, com SLA de 99% no plano Pro e infraestrutura estável para validação de CPF em tempo real. Mas o comportamento da integração — timeouts, retries, circuit breaker — é responsabilidade do seu código, e só um bom stress test revela isso antes que o usuário final descubra.

Cadastre-se em cpfhub.io e comece com 50 consultas gratuitas para validar sua integração antes de ir para produção.