はじめに
個人の備忘録用です。
各種バージョン
- Python : 3.7
- pytest : 6.2.4
- boto3 : 1.17.71
- moto : 2.0.6
問題点
motoを使って単体テストを実装した時に、boto3がなぜか実際のAWSに接続していた。
実際のコード
- 実際の処理
import boto3
class S3Service:
def __init__(self):
self.client = s3.client('s3',
aws_access_key_id="aws_key",
aws_secret_access_key="aws_secret",
region_name="ap-northeast-1")
def upload(self, key: str, file_path: str):
with open(file_path, "rb") as file:
self.client.put_object(Body=file.read(),
Key=key,
Bucket="my-bucket")
- テストコード
import pytest
from moto import mock_s3
import boto3
from app.services import S3Service
s3_service = S3Service()
class TestSample:
@classmethod
def setup_class(cls):
cls.mock_s3 = mock_s3()
cls.mock_s3.start()
cls.client_s3 = boto3.client('s3', region_name="ap-northeast-1")
cls.client_s3.create_bucket(Bucket="my-bucket", CreateBucketConfiguration={ 'LocationConstraint': "my-bucket" })
@classmethod
def teardown_class(cls):
cls.mock_s3.stop()
def test_success(self):
s3_service.upload("key1", "path_to_test_file")
- エラー内容
botocore.exceptions.ClientError: An error occurred (InvalidAccessKeyId) when calling the PutObject operation: The AWS Access Key Id you provided does not exist in our records.
解決案
原因は、mock_s3を起動する前にboto3のclientを生成していたから。
なので、テストコードを以下のように修正するだけで解決できる。
import pytest
from moto import mock_s3
import boto3
from app.services import S3Service
class TestSample:
@classmethod
def setup_class(cls):
cls.mock_s3 = mock_s3()
cls.mock_s3.start()
cls.client_s3 = boto3.client('s3', region_name="ap-northeast-1")
cls.client_s3.create_bucket(Bucket="my-bucket", CreateBucketConfiguration={ 'LocationConstraint': "my-bucket" })
clse.s3_service = S3Service()
@classmethod
def teardown_class(cls):
cls.mock_s3.stop()
def test_success(self):
self.s3_service.upload("key1", "path_to_test_file")
もしくは、本体のコードが Dependency Injection などの機能を使っていて、
テスト実行時にはどうしてもインスタンスが生成されている場合は、
本体のコードを以下のように変更すると対応可能。
import boto3
class S3Service:
def upload(self, key: str, file_path: str):
with open(file_path, "rb") as file:
self._client().put_object(Body=file.read(),
Key=key,
Bucket="my-bucket")
def _client(self):
return s3.client('s3',
aws_access_key_id="aws_key",
aws_secret_access_key="aws_secret",
region_name="ap-northeast-1")
おわりに
何かありましたらコメント頂けると幸いです。