How to mock bucket.objects.filter() from boto3 resource?

Question

I am having some problems mocking the bucket.objects.filter() method, but I am able to mock most other boto3 calls.

I have a class that is similar to this code in a file with the path 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

    

    # 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()]

Then in my tests file I have something like this:

    @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)

The returned list is always empty.

The namedtuple part is just an attempt to mimic the bucket.objects.

I would be open to solutions using botocore Stub, but I cannot use a third party library like moto. I just need to mock the call to bucket.objects.filter(). Thank you in advance.

userQIWf · Answer

Short answer: replace mock_connection.return_value = mock_client with mock_connection.return_value.Bucket.return_value.objects = mock_client in your test and it would work.

The reason is that self.client is session.resource that you mock, then you create a Bucket (which adds the Bucket.return_value to the mocked path), then you do .objects and only then do you apply the filter() (adds filter.return_value which you already had).

A simple approach that can help you in future cases is to use a helper library I wrote to generate the asserts for you: the mock-generator.

To use it in your case, add mock_autogen.generate_asserts(mock_connection) right after your call to list_bucket_contents, like so:

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)

It would print all the needed asserts for mock_connection, which would tell you the exact methods used. In your case, the input would contain the following line:

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

From this line you can derive the path you need to mock and replace with the filter objects.

How to mock bucket.objects.filter() from boto3 resource?

1 Answers