如何模拟boto3中的响应对象列表?

2022-03-12 00:00:00 python amazon-s3 mocking unit-testing boto3

问题描述

我想获取S3存储桶上特定目录中的所有档案,如下所示:

def get_files_from_s3(bucket_name, s3_prefix):
    files = []
    s3_resource = boto3.resource("s3")
    bucket = s3_resource.Bucket(bucket_name)
    response = bucket.objects.filter(Prefix=s3_prefix)
    for obj in response:
        if obj.key.endswidth('.zip'):
            # get all archives
            files.append(obj.key)
    return files

我的问题是关于测试它;因为我想模拟response中的对象列表,以便能够对其进行迭代。以下是我尝试过的内容:

from unittest.mock import patch
from dataclasses import dataclass

@dataclass
class MockZip:
    key = 'file.zip'

@patch('module.boto3')
def test_get_files_from_s3(self, mock_boto3):
    bucket = mock_boto3.resource('s3').Bucket(self.bucket_name)
    response = bucket.objects.filter(Prefix=S3_PREFIX)
    response.return_value = [MockZip()]
    files = module.get_files_from_s3(BUCKET_NAME, S3_PREFIX)
    self.assertEqual(['file.zip'], files)

我收到如下断言错误:E AssertionError: ['file.zip'] != []

有没有更好的方法?我使用了struct,但我认为这不是问题所在,我想我得到了一个空列表,因为response是不可迭代的。那么,我如何才能将其模拟为模拟对象列表,而不仅仅是MockMagick类型呢?

谢谢

libray

您可以使用moto,它是专门用于模拟boto3调用的开源推荐答案构建。它允许您直接使用boto3,而不必担心手动设置模拟。

您当前使用的测试函数如下所示:

from moto import mock_s3

@pytest.fixture(scope='function')
def aws_credentials():
    """Mocked AWS Credentials, to ensure we're not touching AWS directly"""
    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
    os.environ['AWS_SECURITY_TOKEN'] = 'testing'
    os.environ['AWS_SESSION_TOKEN'] = 'testing'


@mock_s3
def test_get_files_from_s3(self, aws_credentials):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(self.bucket_name)
    # Create the bucket first, as we're interacting with an empty mocked 'AWS account'
    bucket.create()

    # Create some example files that are representative of what the S3 bucket would look like in production
    client = boto3.client('s3', region_name='us-east-1')
    client.put_object(Bucket=self.bucket_name, Key="file.zip", Body="...")
    client.put_object(Bucket=self.bucket_name, Key="file.nonzip", Body="...")

    # Retrieve the files again using whatever logic
    files = module.get_files_from_s3(BUCKET_NAME, S3_PREFIX)
    self.assertEqual(['file.zip'], files)
可以在以下位置找到Moto的完整文档: http://docs.getmoto.org/en/latest/index.html

免责声明:我是Moto的维护者。

相关文章