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

How to mock list of response objects from 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 类型呢?

谢谢

您可以使用 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 的维护者。