Teste uma integração de Consulta da operadora offline com fixtures de resposta

Injete o transporte e teste fixtures de resposta sintéticas explícitas antes de permitir que um consumidor da CarrierLookup utilize a rede.

Os testes de mock da API de consulta de operadora devem substituir o transporte HTTP, negar acesso acidental à rede e validar como sua aplicação lida com cada resultado documentado. Fixtures sintéticas podem testar a lógica do consumidor sem enviar números de telefone ou gastar saldo.

Coloque a camada de teste antes da requisição HTTP

Um importador de CRM pode passar em seu teste de "caminho feliz" e ainda rotular um número Indeterminado como inválido. Ele também pode passar em um teste de navegador com mock enquanto seu backend faz uma requisição paga real. Coloque sua camada de teste na função de backend que realiza a troca HTTP de saída.

O consumidor deve aceitar uma dependência de transporte em vez de descobrir um cliente real internamente. O transporte de teste injetado recebe o caminho e o corpo da requisição pretendidos, valida-os e retorna um envelope sintético. Não há chave de API no teste. Mantenha credenciais reais fora do job de CI também.

Use uma restrição de rede separada como proteção. Um mock que retorna dados convenientes não é prova de que toda chamada de saída foi interceptada. Negue o tráfego de saída na fronteira do container ou executor de CI e falhe o job caso haja tentativa de acesso à rede. Uma proteção em nível de linguagem ajuda a detectar erros comuns, mas não pode cobrir subprocessos ou todas as bibliotecas de rede.

Defina os resultados antes de escrever as fixtures

Nosso serviço de operadora síncrono usa service_type=carrier. Respostas únicas possuem campos data planos, incluindo identifier, carrier, underlying_carrier, number_type, country_code, region e city. Ler data.extra.carrier testa um formato inventado.

Os valores da operadora descrevem a alocação original. Eles não devem preencher um campo de CRM chamado rede atual. Da mesma forma, nem um resultado de alocação vazio nem exists=false estabelecem a validade do telefone. Use rótulos de resultado de aplicação separados para que uma consulta de relatório posterior não possa mesclar esses estados acidentalmente.

Execute uma pequena suíte de fixtures sintéticas

Salve o seguinte como carrier_consumer_test.py e execute python3 carrier_consumer_test.py. Ele usa a biblioteca padrão do Python. O exemplo é uma camada de consumidor mínima, com identificadores sintéticos e valores escolhidos deliberadamente; ele nunca contata a CarrierLookup. Substitua o client e o classificador de demonstração pelo seu próprio adaptador, mantendo as validações das fixtures.

import socket
import unittest
from unittest.mock import patch

FIELDS = ("carrier", "underlying_carrier", "number_type",
          "country_code", "region", "city")
A, B = "+12025550101", "+12025550102"  # Apenas dados de teste sintéticos.

def classify(envelope):
    code = envelope["code"]
    if code != 0:
        return {42200: "undetermined", 42901: "busy",
                50400: "timeout"}.get(code, "error")
    data = envelope["data"]
    if data["service_type"] != "carrier":
        raise ValueError("serviço inesperado")
    return "completed" if data["carrier"] else "completed_empty"

def client(transport, identifiers):
    multi = len(identifiers) > 1
    path = "/api/v1/batch-check" if multi else "/api/v1/check"
    body = {"service_type": "carrier"}
    body["identifiers" if multi else "identifier"] = (
        identifiers if multi else identifiers[0])
    return transport(path, body)

def completed(identifier, carrier=""):
    return {"code": 0, "msg": "fixture sintética", "data": {
        "service_type": "carrier", "identifier": identifier,
        **dict.fromkeys(FIELDS, ""), "carrier": carrier}}

class ConsumerTests(unittest.TestCase):
    def setUp(self):
        self.guard = patch.object(
            socket.socket, "connect",
            side_effect=AssertionError("rede proibida"))
        self.guard.start()
        self.addCleanup(self.guard.stop)

    def test_single_flat_data_and_outcomes(self):
        for value, expected in [("Exemplo de Alocação", "completed"),
                                ("", "completed_empty")]:
            def fake(path, body):
                self.assertEqual(path, "/api/v1/check")
                self.assertEqual(body, {"service_type": "carrier",
                                        "identifier": A})
                return completed(A, value)
            result = client(fake, [A])
            self.assertNotIn("extra", result["data"])
            self.assertNotIn("registered", result["data"])
            self.assertEqual(classify(result), expected)

    def test_error_branches(self):
        for code, expected in [(42200, "undetermined"),
                               (42901, "busy"), (50400, "timeout")]:
            with self.subTest(code=code):
                result = client(lambda path, body: {
                    "code": code, "msg": "erro sintético",
                    "data": None}, [A])
                self.assertEqual(classify(result), expected)

    def test_multi_order_and_no_result(self):
        def fake(path, body):
            self.assertEqual(path, "/api/v1/batch-check")
            self.assertEqual(body, {"service_type": "carrier",
                                    "identifiers": [B, A]})
            return {"code": 0, "msg": "fixture sintética", "data": {
                "service_type": "carrier", "total": 2,
                "succeeded": 1, "failed": 1, "results": [
                    {"identifier": B, "exists": True,
                     **dict.fromkeys(FIELDS, "")},
                    {"identifier": A, "exists": False}]}}
        data = client(fake, [B, A])["data"]
        rows = data["results"]
        self.assertEqual(data["service_type"], "carrier")
        self.assertEqual([r["identifier"] for r in rows], [B, A])
        self.assertEqual(len(rows), data["total"])
        outcomes = ["completed" if r["exists"] else "no_result"
                    for r in rows]
        self.assertEqual(outcomes, ["completed", "no_result"])
        self.assertNotIn("carrier", rows[1])

    def test_guard_rejects_an_escape(self):
        with socket.socket() as sock:
            with self.assertRaisesRegex(AssertionError, "rede proibida"):
                sock.connect(("127.0.0.1", 9))

if __name__ == "__main__":
    unittest.main()

O exemplo retorna envelopes já decodificados do seu transporte. Ele testa a construção da requisição e o tratamento de resultados, em vez da análise HTTP. Mantenha fixtures para status HTTP, cabeçalhos e JSON malformado na sua camada de transporte real também. Em particular, uma resposta 42901 inclui Retry-After; uma fixture de cabeçalho de resposta deve verificar se seu transporte a preserva para o chamador. Não codifique um novo intervalo de repetição no classificador de resultados.

A proteção de socket testa intencionalmente seu próprio caminho de falha. Ela bloqueia o método de conexão de socket deste processo Python; a restrição de saída de CI separada permanece necessária para outros caminhos. Uma chamada paga acidental deve falhar o job, não se tornar uma fixture extra despercebida.

Torne as validações de resultados múltiplos sensíveis a erros de ordenação

A fixture múltipla envia B antes de A. Isso importa: um teste que envia identificadores ordenados alfabeticamente e recebe a mesma ordem pode perder um adaptador que ordena respostas. Combine cada identificador retornado com sua posição de entrada e valide o comprimento total.

O primeiro item múltiplo produz um resultado concluído com valores de alocação vazios. O segundo tem exists=false e nenhum campo de operadora. Preserve essa diferença ao mapear para registros de aplicação. Se o seu adaptador de produção descartar linhas com falha, a fixture deve falhar, pois linhas posteriores poderiam, de outra forma, deslocar-se para os contatos errados.

Adicione uma validação na função de mapeamento de CRM real: uma consulta Indeterminado ou com falha deve deixar a observação de alocação anterior intacta ou marcá-la explicitamente como obsoleta conforme sua política. Ela não deve sobrescrever o registro com "número inválido". O mesmo teste deve rejeitar a derivação da localização atual a partir da região ou cidade de alocação.

Um tempo limite síncrono múltiplo falha toda a requisição sem resultados parciais. Teste a ramificação de tempo limite de toda a requisição separadamente de um envelope bem-sucedido contendo um item malsucedido. Essas são entradas diferentes para o consumidor.

Mantenha a confiança offline separada da verificação ao vivo

Esses testes estabelecem como seu código responde ao contrato representado pelas fixtures. Eles não demonstram disponibilidade atual do provedor, aceitação de país, permissões de conta ou faturamento. Uma fixture fabricada também não pode provar que um provedor ao vivo retornará um campo de alocação específico para um número específico.

Mantenha a procedência da fixture ao lado do teste: a CarrierLookup API reference, a data de revisão e os campos que seu adaptador consome. Atualize as fixtures quando um contrato documentado mudar, então examine se os dados de aplicação antigos precisam de um caminho de leitura compatível.

Execute um teste de fumaça ao vivo autorizado separadamente quando a verificação de integração exigir, usando entrada aprovada e seu próprio orçamento de saldo. Não faça o CI comum depender dessa chamada. Comece substituindo uma importação de cliente ao vivo em sua suíte de testes por um transporte injetado e confirmando que uma tentativa de escape de rede falha imediatamente.

Saiba mais

Escolha as informações do produto que se ajustam ao próximo passo em seu fluxo de trabalho.

• Ver a documentação do produto →
• Falar com a equipe →