如何从boto3资源中模拟bucket.objects.filter()

Question

我在模拟bucket.objects.filter((方法时遇到了一些问题，但我能够模拟大多数其他boto3调用。

我在一个文件中有一个类似于此代码的类，路径为my_project.utils.s3_api:

from boto3.session import Session, Config

class S3Resource:
def __init__(self, kwargs):
session = Session()
self.client = session.resource(
's3',
aws_access_key_id=kwargs['access_key'],
aws_secret_access_key=kwargs['secret_key'],
endpoint_url=kwargs['s3_url'],
region_name=kwargs['region'],
config=Config(signature_version='s3v4'
)
self.kwargs = kwargs
<many methods that are already successfully tested here>
# this is the method that I cannot test correctly:
def list_bucket_contents(self):
bucket = self.client.Bucket(self.kwargs['bucket'])
return [summary.key for summary in bucket.objects.filter()]

然后在我的测试文件中，我有这样的东西：

@mock.patch('my_project.utils.s3_api.Session.resource')
def test_list_bucket_contents(self, mock_connection):
ObjectSummary = namedtuple('ObjectSummary', 'bucket_name key')
obj_collection = (
ObjectSummary(bucket_name='mybucket', key='file1.txt'),
ObjectSummary(bucket_name='mybucket', key='file2.txt'),
ObjectSummary(bucket_name='mybucket', key='file3.txt')
)
mock_client = mock.MagicMock()
mock_client.filter.return_value = obj_collection
mock_connection.return_value = mock_client
s3_client = S3Resource(**self.init_args)
s3_client.list_bucket_contents()
print(result)

返回的列表始终为空。

命名的元组部分只是试图模仿bucket.objects。

我对使用botocore Stub的解决方案持开放态度，但我不能使用像moto这样的第三方库。我只需要模拟对bucket.objects.filter((的调用。提前谢谢。

Answer 1

简单回答：在测试中用mock_connection.return_value.Bucket.return_value.objects = mock_client替换mock_connection.return_value = mock_client，它就会起作用。

原因是self.client是您模拟的session.resource，然后创建一个Bucket(将Bucket.return_value添加到模拟路径(，然后执行.objects，然后才应用filter()(添加您已经拥有的filter.return_value(。

在未来的情况下，一种可以帮助您的简单方法是使用我编写的帮助程序库为您生成断言：mock生成器。

要在您的情况下使用它，请在呼叫list_bucket_contents后立即添加mock_autogen.generate_asserts(mock_connection)，如下所示：

import mock_autogen

@mock.patch('my_project.utils.s3_api.Session.resource')
def test_list_bucket_contents(self, mock_connection):
ObjectSummary = namedtuple('ObjectSummary', 'bucket_name key')
obj_collection = (
ObjectSummary(bucket_name='mybucket', key='file1.txt'),
ObjectSummary(bucket_name='mybucket', key='file2.txt'),
ObjectSummary(bucket_name='mybucket', key='file3.txt')
)
mock_client = mock.MagicMock()
mock_client.filter.return_value = obj_collection
mock_connection.return_value = mock_client
s3_client = S3Resource(**self.init_args)
s3_client.list_bucket_contents()
mock_autogen.generate_asserts(mock_connection)

它将打印mock_connection所需的所有断言，这将告诉您所使用的确切方法。在您的情况下，输入将包含以下行：

mock_connection.return_value.Bucket.return_value.objects.filter.assert_called_once_with()

从这一行中，您可以派生出需要模拟并替换为过滤器对象的路径。