はじめに

boto3で実装したSQSメッセージ送信処理のテストコードを書くために、AWSサービスをモックするmotoを使うことにしました。
テストフレームワークは、pytestを使います。
その時にはまったポイントをご紹介します。

バージョン

それぞれのバージョンは、以下の通りです。


Python	3.9.11
boto3	1.21.31
pytest	7.1.2
moto	3.1.12

プロダクトコード

テスト対象となるプロダクトコードは、以下のようにクラスで実装しました。

sqs.py

# -*- coding: utf-8 -*-

import os
import uuid
import logging
import boto3

logger = logging.getLogger()


class Sqs():
    def __init__(self):
        self.__queue_url    = os.environ.get('QUEUE_URL')
        self.__sqs_client   = boto3.client('sqs')
        self.__msg_group_id = "group_id_1"


    def send_msg(self, message: str):
        logger.info('Start sending sqs message')
        logger.debug(f'sqs queue url: {self.__queue_url}')
        logger.debug(f'sqs message body: {message}')

        response = self.__sqs_client.send_message(
            QueueUrl               = self.__queue_url,
            MessageBody            = message,
            MessageGroupId         = self.__msg_group_id,
            MessageDeduplicationId = str(uuid.uuid4())
        )

        logger.info('Success sending sqs message')
        logger.debug(f'sqs message response: {response}')

キューURLは、環境変数QUEUE_URLで指定しています。
メッセージグループIDは、group_id_1の固定。
メッセージ重複排除ID(MessageDeduplicationId)は、UUIDで生成します。

SQSキュー

SQSキューは、以下のように作成しました。
タイプはFIFOにしています。
デッドレターキューの設定はしていません。

テストコード

テストコードは、以下のように作成しました。

test_sqs.py

import pytest
import boto3

from sqs import Sqs
from moto import mock_sqs


@pytest.fixture
def message() -> str:
    return 'あああいいいうううう'


@mock_sqs
class TestSqs:
    def setup_method(self, method):
        self.sqs_client = boto3.client('sqs')
        response        = self.sqs_client.create_queue(QueueName='some-queue.fifo')
        self.queue_url  = response['QueueUrl']


    # メッセージ送信成功
    def test_send_msg_success(self, mocker, message):
        # 存在するキューURLを設定する
        mocker.patch.dict('os.environ', {'QUEUE_URL': self.queue_url})

        sqs = Sqs()

        # 送信実行
        sqs.send_msg(message)

        # 送信したメッセージを受信して取得
        recv_msg = self.sqs_client.receive_message(QueueUrl=self.queue_url)

        # 受信したメッセージボディと送信したメッセージボディが一致している
        assert recv_msg['Messages'][0]['Body'] == message


    # メッセージ送信失敗
    def test_send_msg_fail(self, mocker, message):
        # 存在しないキューURLを設定する
        mocker.patch.dict('os.environ', {'QUEUE_URL': 'https://hogefuga.com'})

        sqs = Sqs()

        # 送信実行
        with pytest.raises(Exception) as e:
            sqs.send_msg(message)

        assert str(e.value) == "An error occurred (AWS.SimpleQueueService.NonExistentQueue) when calling the SendMessage operation: The specified queue does not exist for this wsdl version."

setup_methodでキューを作成して、キューのURLを取得しています。
テストケースとしては、存在するキューURLで成功、存在しないキューURLで失敗の2パターンです。

エラー内容

このテストコードを実行すると、以下のエラーが発生してハマりました。

botocore.exceptions.ClientError: An error occurred (InvalidParameterValue) when calling the SendMessage operation: Value group_id_1 for parameter MessageGroupId is invalid. Reason: The request include parameter that is not valid for this queue type.

「メッセージグループIDが無効」と言われてしまいます。
プロダクトコードの方は成功しているので、テストコードの書き方の問題っぽい。
FIFOだとメッセージグループIDは必要なはずなのになぜ？
メッセージグループIDって形式決まってたっけ？
と思いながらエラーメッセージでググったりしましたが、目ぼしいのは見当たらない。
motoの公式ドキュメントを見ても書いてない。。。
- https://docs.getmoto.org/en/latest/docs/services/sqs.html

解消方法

motoのソースコードを追っていくと、こんな風に書いてました。
- https://github.com/spulec/moto/blob/3.1.12/tests/test_sqs/test_sqs.py#L52
```
sqs.create_queue(QueueName=str(uuid4())[0:6], Attributes={"FifoQueue": "true"})
```
どうやら、FIFOということを明示的に指定しないといけない模様。
「FIFOじゃないならメッセージグループIDいらないよ」ってなって落ちてたんですね。

というわけで

response        = self.sqs_client.create_queue(QueueName='some-queue.fifo', Attributes={"FifoQueue": "true"})

と書いたら、無事にオールグリーンになりました♪

終わりに

こういうのは公式ドキュメントに書いておいて欲しいなと思いつつも、やっぱり使ってるライブラリのソースコードをちゃんと追っていくのって大事ですね〜

motoによるSQSキュー(FIFO)のモック作成

はじめに

バージョン

プロダクトコード

SQSキュー

テストコード

エラー内容

解消方法

終わりに