Então como sabemos sempre devevemos fazer testes unitários nos nossos projetos.

Então nesse post vou mostrar como fazer testes unitarios em sistemas que utilizam o sistema de filas da Amazon o SQS.

Uma das formas de se fazer testes é fazer com que o Python mock o serviço de SQS através do MagicMock, para isso basta deixarmos a assinatura do nosso metodo de teste como o exemplo abaixo.

@patch('boto.connect_sqs')
def test_mock_success(self, sqs_connect_mock):
   ....

No exemplo acima estou fazendo com que o python mock o método connect_sqs.

Mas nem sempre precisamos que o mock retorne sucesso, as vezes precisamos testar uma resposta de exception, para isso vamos ver o exemplo abaixo.

class ExternalAPIMock:
    def __init__(self, status_code, response=None):
        self.status_code = status_code
        self.response = response

    def json(self):
        return self.response

    def raise_for_status(self):
        if self.status_code >= 400:
            raise Exception('error http code')


@patch('boto.connect_sqs')
def test_mock_exception(self, sqs_connect_mock):
    sqs_connect_mock.return_value = ExternalAPIMock(400)

No código acima estou fazendo que sempre que o método connect_sqs for chamado, uma excessão será lançada.

Temos mais uma forma de fazermos testes utilizando SQS, que é criando um serviço local de envio de mensagens para quando for necessário testar um consumo real de fila.

No exemplo abaixo está uma classe que tem um método de teste que cria o serviço escreve na fila e consome.

import unittest

import boto3
from moto import mock_sqs

QUEUE_NAME = 'queue_name'


class SqsTestCase(unittest.TestCase):

    def write_message(self, queue_url, daily_message, channel):
        sqs = boto3.client('sqs')
        message = str({'msg': f'{daily_message}', 'channel': channel})
        sqs.send_message(QueueUrl=queue_url, MessageBody=message)

    @mock_sqs
    def test_write_message_valid(self):
        sqs = boto3.resource('sqs')
        queue = sqs.create_queue(QueueName=QUEUE_NAME)
        queue_url = queue.url
        message = 'test message to send on SQS'
        channel = 'test'
        expected_message = str({'msg': f'{message}', 'channel': f'{channel}'})
        self.write_message(queue_url, message, channel)
        sqs_messages = queue.receive_messages()

        self.assertEqual(sqs_messages[0].body, expected_message)