Eu desenvolvo sistemas que utilizam os serviços da AWS faz algum tempo, e ao longo desse tempo houve mudanças na forma como escrevo testes de código que fazem chamadas a seus serviços. Esse texto tem como objetivo apresentar algumas abordagens para escrita de testes que utilizei, e discutir o que motivou suas evoluções, destacando características de cada abordagem. Ao final, pretendo apresentar um padrão que acredito ser uma forma bastante prática de escrever testes de código que interage com serviços da AWS, usando uma biblioteca em Python que desenvolvi implementando esse padrão, mas que também poderia ser adaptado para outros contextos (serviços) e linguagens.
Código a ser testado
Antes de iniciar a discussão sobre os testes, quero apresentar um exemplo de código para ser testado. Ele é uma função que deve consumir mensagens enviadas para uma fila SQS, realizar um processamento com a informação contida nas mensagens, e enviar o resultado para outra fila SQS.
import json
def funcao1(cliente_sqs, fila1_url, fila2_url):
# Recebe até 10 mensagens do SQS
msgs_recebidas = cliente_sqs.receive_message(
QueueUrl=fila1_url,
MaxNumberOfMessages=10,
)
# Percorre as mensagens recebidas
for msg in msgs_recebidas.get('Messages', []):
# Recupera valor da mensagem
corpo = json.loads(msg['Body'])
n = corpo['n']
# Processa valor
resultado = n * 2
resposta = {'resultado': resultado}
# Envia mensagem com resultado
cliente_sqs.send_message(
QueueUrl=fila2_url,
MessageBody=json.dumps(resposta),
)
# Apaga mensagem atual
cliente_sqs.delete_message(
QueueUrl=fila1_url,
ReceiptHandle=msg['ReceiptHandle'],
)
Esse código é uma função que recebe um objeto para interagir com o serviço SQS (cliente_sqs) e a URL de duas filas (fila1_url e fila2_url). Ele busca até 10 mensagens da fila1_url (o máximo permitido por chamada), cada mensagem é um JSON que possui um número no campo n. Após recuperar esse valor, um processamento é feito (nesse caso multiplicar o valor por 2), um novo JSON é gerado e enviado para a fila2_url. Se tudo isso ocorrer conforme esperado, a mensagem processada é removida da fila1_url, evitando que ela volte para a fila e eventualmente ser feito uma nova tentativa de processá-la. Esse processo é repetido para cada mensagem recebida.
Teste unitário com mock
Uma forma bastante simples de testar é fazer testes unitários (também chamados de testes de unidade) e usar mocks. Para o código de exemplo é possível criar um objeto que simula (mock) o cliente_sqs passado para a função, e registre as funções chamadas dele. Depois de executar a função a ser testada, basta verificar as chamadas feitas no objeto simulado, e assim validar se a função está se comportando como o esperado.
Na biblioteca padrão do Python existe o MagicMock, que facilita a criação de mocks. Bastando definir o que deve ser retornando em cada função (se ela tiver retorno), e depois verificar como suas funções foram chamadas (assert). Caso espera-se que uma função tenha sido chamada de determinada forma e isso não ocorreu, o teste falhará.
Segue a baixo um exemplo de teste unitário com mock para o código apresentado anteriormente:
import json
from exemplo.codigo1 import funcao1
from unittest.mock import MagicMock
from uuid import uuid4
def test_funcao1():
# Mensagens com valores de entrada e resultados esperados
msgs = [
{'n': 0},
{'n': 1},
{'n': 2},
]
respostas = [
{'resultado': 0},
{'resultado': 2},
{'resultado': 4},
]
# Define outros valores auxiliares para o teste
fila1_url = 'http://sqs.aws/fila1'
fila2_url = 'http://sqs.aws/fila2'
msgs_mockadas = [
{
'ReceiptHandle': uuid4().hex,
'Body': json.dumps(msg),
} for msg in msgs
]
# Cria e configura mock do SQS
mock_cliente_sqs = MagicMock()
mock_cliente_sqs.receive_message.return_value = {
'Messages': msgs_mockadas,
}
# Chama função a ser testada
funcao1(mock_cliente_sqs, fila1_url, fila2_url)
# Verifica se a função buscou as mensagens na fila 1
mock_cliente_sqs.receive_message.assert_called_once_with(
QueueUrl=fila1_url,
MaxNumberOfMessages=10,
)
# Verifica se as mensagens da fila 1 foram apagadas
assert mock_cliente_sqs.delete_message.call_count == len(msgs)
for msg in msgs_mockadas:
mock_cliente_sqs.delete_message.assert_any_call(
QueueUrl=fila1_url,
ReceiptHandle=msg['ReceiptHandle'],
)
# Verifica se as respostas estão corretas na fila 2
assert mock_cliente_sqs.send_message.call_count == len(respostas)
for esperado in respostas:
mock_cliente_sqs.send_message.assert_any_call(
QueueUrl=fila2_url,
MessageBody=json.dumps(esperado),
)
Essa abordagem tem algumas vantagens, como: Não precisar configurar e acessar o serviço SQS para rodar os testes, executando tudo localmente; E testa um trecho de código de forma isolada (uma função nesse caso), independente de outras partes do código, como o cliente_sqs. Porém esse método também apresenta desvantagens, como: Ao isolar uma parte do código, perde-se a certeza se ele funcionará junto com as demais partes do sistema, uma vez que podem existir diferenças entre uma resposta real e a resposta do mock (nesse teste o msgs_mockadas possui apenas duas chaves, quando numa resposta verdadeira existiriam várias outras também); Além de ser necessário definir toda vez qual o retorno das funções e validar se elas foram chamadas conforme o esperado; Também é possível fazer um mock que aceite uma chamada inválida para o serviço real; E como a mensagem enviada é uma string com JSON, a ordem dos campos pode mudar, assim como sua formatação, o que pode causar un falso negativo no teste. Então qualquer alteração de parâmetros da função ou resposta, pode gerar uma incompatibilidade do mock com o comportamento real do sistema.
Teste de integração com uma reimplementação
Partindo mais para uma abordagem de testes de integração, existe uma reimplementação dos serviços da AWS em Python para ser usada em testes chamada Moto (que também pode ser utilizada em outras linguagens no seu modo servidor, semelhante ao LocalStack). Assim também é possível rodar os testes localmente, uma vez que ele simula o comportamento dos serviços da AWS. Um exemplo de teste utilizando essa biblioteca pode ser visto a baixo:
import boto3
import json
from exemplo.codigo1 import funcao1
from moto import mock_aws
@mock_aws
def test_funcao1():
# Mensagens com valores de entrada e resultados esperados
msgs = [
{'n': 0},
{'n': 1},
{'n': 2},
]
respostas = [
{'resultado': 0},
{'resultado': 2},
{'resultado': 4},
]
# Conecta no SQS
cliente_sqs = boto3.client('sqs', region_name='us-east-1')
# Cria filas de teste
fila1_url = cliente_sqs.create_queue(QueueName='fila1')['QueueUrl']
fila2_url = cliente_sqs.create_queue(QueueName='fila2')['QueueUrl']
# Envia mensagens de teste
for msg in msgs:
cliente_sqs.send_message(
QueueUrl=fila1_url,
MessageBody=json.dumps(msg),
)
# Chama função a ser testada
funcao1(cliente_sqs, fila1_url, fila2_url)
# Verifica se a função buscou e apagou as mensagens na fila 1
assert sum(int(attr) for attr in cliente_sqs.get_queue_attributes(
QueueUrl=fila1_url,
AttributeNames=[
'ApproximateNumberOfMessages',
'ApproximateNumberOfMessagesNotVisible',
],
)['Attributes'].values()) == 0
# Verifica se as respostas estão corretas na fila 2
msgs_de_resposta = cliente_sqs.receive_message(
QueueUrl=fila2_url,
MaxNumberOfMessages=10,
)['Messages']
for msg in msgs_de_resposta:
cliente_sqs.delete_message(
QueueUrl=fila2_url,
ReceiptHandle=msg['ReceiptHandle'],
)
assert [
json.loads(msg['Body']) for msg in msgs_de_resposta
] == respostas
# Remove filas de teste
cliente_sqs.delete_queue(QueueUrl=fila1_url)
cliente_sqs.delete_queue(QueueUrl=fila2_url)
Diferente do teste anterior, neste não é necessário se preocupar em montar os retornos dos serviços acessados, porém é necessário conhecer melhor o serviço para criar os recursos utilizados pela função a ser testada. Também não existe uma forma de fazer um assert para verificar se uma mensagem foi enviada, é necessário buscá-las da fila, e nesse caso, também chamar a função para removê-las, de forma que elas não voltem para a fila, evitando comportamentos não esperados (embora não seja obrigatório ao usar o Moto como um decorador, mas é bom para evitar problemas). E mesmo que possa existir alguma diferença entre o serviço real e sua reimplementação, espera-se que ele seja confiável, e ao atualizar o Moto, todos os testes já serão validados com os novos comportamentos, terceirizando essa responsabilidade do teste.
Interagindo com outro serviço (SNS)
Pensando em testar outro serviço da AWS, pode-se trocar a fila SQS para qual a resposta do código é enviada por um tópico SNS. O código a ser testado então fica:
import json
def funcao2(cliente_sqs, cliente_sns, fila_url, topico_arn):
# Recebe até 10 mensagens do SQS
msgs_recebidas = cliente_sqs.receive_message(
QueueUrl=fila_url,
MaxNumberOfMessages=10,
)
# Percorre as mensagens recebidas
for msg in msgs_recebidas.get('Messages', []):
# Recupera valor da mensagem
corpo = json.loads(msg['Body'])
n = corpo['n']
# Processa valor
resultado = n * 2
resposta = {'resultado': resultado}
# Publica mensagem com resultado
cliente_sns.publish(
TopicArn=topico_arn,
Message=json.dumps(resposta),
)
# Apaga mensagem atual
cliente_sqs.delete_message(
QueueUrl=fila_url,
ReceiptHandle=msg['ReceiptHandle'],
)
Seu teste pode seguir uma estrutura bastante similar:
import boto3
import json
from exemplo.codigo2 import funcao2
from moto import mock_aws
@mock_aws
def test_funcao2():
# Mensagens com valores de entrada e resultados esperados
msgs = [
{'n': 0},
{'n': 1},
{'n': 2},
]
respostas = [
{'result': 0},
{'result': 2},
{'result': 4},
]
# Conecta no SQS e SNS
cliente_sqs = boto3.client('sqs', region_name='us-east-1')
cliente_sns = boto3.client('sns', region_name='us-east-1')
# Cria fila e tópico de teste
fila_url = cliente_sqs.create_queue(QueueName='fila')['QueueUrl']
topico_arn = cliente_sns.create_topic(Name='topico')['TopicArn']
# Envia mensagens de teste
for msg in msgs:
cliente_sqs.send_message(
QueueUrl=fila_url,
MessageBody=json.dumps(msg),
)
# Chama função a ser testada
funcao2(cliente_sqs, cliente_sns, fila_url, topico_arn)
# Como fazer um assert das mensagens publicadas no SNS?
assert ...
# Remove fila e tópico de teste
cliente_sqs.delete_queue(QueueUrl=fila_url)
cliente_sns.delete_topic(TopicArn=topico_arn)
Porém como verificar o que foi enviado para o tópico SNS? Como mocks não estão sendo utilizados, não tem como fazer um assert para verificar se a função foi chamada e com quais parâmetros. O SNS também não tem uma forma direta de recuperar as mensagens publicadas nele. Uma solução possível é criar uma fila SQS, assinar o tópico SNS, de forma que tudo que for enviado para o tópico seja encaminhado para a fila SQS, e validar o tópico SNS a partir da fila SQS (e após a execução do teste, remover a assinatura, fila e tópico). Embora possível, essa solução é bastante trabalhosa e repetitiva, ainda mais se considerar que isso precisará ser refeito em cada teste do projeto que chamar o SNS.
Simplificando e padronizando (repensando os testes)
Embora a solução apresentada tenha o problema da repetição de código, ela funciona, e para os testes não importa saber criar uma fila ou tópico e removê-los depois (que é uma parte considerável do que é repetido), só importa usá-los. É totalmente plausível (pelo menos na ideia), de na hora de rodar o teste, só perguntar por alguma fila disponível que possa ser utilizada, e depois devolvê-la. Então seria bom se tivesse uma forma de pegar filas e tópicos emprestados para os testes, ou até mesmo criá-los e removê-los, o teste em si não precisa saber o que acontecerá com esses recursos depois de utilizá-los. Na verdade, o teste nem precisaria saber do detalhe de que é necessário primeiro buscar uma mensagem da fila do SQS, e depois deletar essa mensagem, nem de que a mensagem publicada em um tópico SNS precisou passar por uma fila SQS para ser validada.
Pensando dessa forma, só é necessário executar algum código antes do teste para criar os recursos (filas, tópicos...) que serão utilizados, e executar algum código depois dos testes para remover esses recursos criados anteriormente. Além disso, alguma abstração poderia ser criada para simplificar a interação com esses recursos na parte dos testes para facilitar a verificação de sua utilização, que é justamente a parte que importa dos testes.
Para executar algum código antes e depois de algo, no Python pode-se ser usado um contexto gerenciado (aquele criado com a estrutura with ...), que poderia criar os recursos, passá-lo para o teste, e assim que ele acabar, fazer sua remoção. Porém, o que ele deveria retornar?
Para testar um código que envia mensagens para uma fila SQS, só é necessário saber qual a fila e alguma forma de receber as mensagens enviadas. A identificação de uma fila no SQS se dá através de uma URL, e um gerador poderia ser criado para abstrair toda a lógica de consumir a fila. Já para testar um código que recebe mensagens de uma fila SQS é necessário saber a URL da fila, e uma função para enviar as mensagens que serão consumidas pelo código a ser testado.
Agora considerando o caso do tópico SNS, toda vez que um tópico for criado, também poderia se criar uma fila SQS que assina esse tópico. A função geradora também poderia buscar nessa fila as mensagens. Isso deixaria o SQS totalmente transparente para os testes (o teste nem saberia que tem uma fila SQS sendo utilizada). Porém para fazer a assinatura, a URL da fila não serve, é necessário seu ARN, em outros contextos pode ser necessário o nome e não URL ou ARN.
Para o contexto retornar mais de uma coisa, pode-se utilizar as tuplas do Python. Porém a questão continua, o que estará nessa tupla? URL da fila? ARN da fila? Nome? Função para enviar mensagens? Gerador para consumir mensagens? Para não ser necessário criar diferentes tuplas para cada caso, pode-se criar uma tupla retornando tudo. Mas pode ser complicado lembrar em que posição está cada coisa, e um código que recebe uma tupla com vários valores e não utiliza a maioria pode gerar confusão. Então em vez de usar tuplas, pode-se criar um objeto seguindo o modelo de dados do Python, assim informações (como nome, ARN e URL) podem ir como atributos, a função para enviar mensagens poderia ser um método, e esse objeto também poderia se comportar como um iterador, que é um comportamento do gerador. E por seguir o modelo de dados do Python, isso tudo ainda pareceria algo nativo do Python, onde um objeto abstrairia a fila SQS ou tópico SNS (algo semelhante ao page object apresentado no curso de Selenium do Dunossauro).
E para facilitar ainda mais, esses contextos que criam os recursos podem se tornar fixtures do pytest, assim bastaria informar o nome da fixture como parâmetro da função de teste que o recurso já seria criado, e assim que o teste terminar, removido.
Implementando esses padrões nos testes
Os padrões apresentados foram implementados numa biblioteca chamada Read full article
