Mocking boto3 S3 client method Python

Question

I'm trying to mock a singluar method from the boto3 s3 client object to throw an exception. But I need all other methods for this class to work as normal.

This is so I can test a singular Exception test when and error occurs performing a upload_part_copy

1st Attempt

import boto3
from mock import patch

with patch('botocore.client.S3.upload_part_copy', side_effect=Exception('Error Uploading')) as mock:
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

However this gives the following error:

ImportError: No module named S3

2nd Attempt

After looking at the botocore.client.py source code I found that it is doing something clever and the method upload_part_copy does not exist. I found that it seems to call BaseClient._make_api_call instead so I tried to mock that

import boto3
from mock import patch

with patch('botocore.client.BaseClient._make_api_call', side_effect=Exception('Error Uploading')) as mock:
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

This throws an exception... but on the get_object which I want to avoid.

Any ideas about how I can only throw the exception on the upload_part_copy method?

Libido

Answer 1

As soon as I posted on here I managed to come up with a solution. Here it is hope it helps :)

import botocore
from botocore.exceptions import ClientError
from mock import patch
import boto3

orig = botocore.client.BaseClient._make_api_call

def mock_make_api_call(self, operation_name, kwarg):
    if operation_name == 'UploadPartCopy':
        parsed_response = {'Error': {'Code': '500', 'Message': 'Error Uploading'}}
        raise ClientError(parsed_response, operation_name)
    return orig(self, operation_name, kwarg)

with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

Jordan Philips also posted a great solution using the the botocore.stub.Stubber class. Whilst a cleaner solution I was un-able to mock specific operations.

Libido

This is very helpful. It took me a while to realize that a lot of the boto3 clients are effectively generated at runtime, and as such, can't be mocked directly. – Pod

This is the solution that worked for me as Stubber and many other mocking tools are unable to stub boto3 customization functions such as uploading files or generated presigned URLs. – Psycho

This answer was great. I first attempted to use stubber, but it only seemed to work for an immediate call, I couldn't get it working for a call within a sub function for some reason. This on the other hand worked perfectly and is very easy to implement, so thanks! – Parette

How to mock the get_object function, when I tried running the above code, the get object call doesn't go through the mock. – Celeriac

For me it worked using 'from unittest.mock import patch' – Kalb

In my case in python 3, mocking @patch("botocore.client.BaseClient._make_api_call", lambda *args, **kwargs: None) did the trick. Considering 100% code coverage – Dabber

Answer 2

Botocore has a client stubber you can use for just this purpose: docs.

Here's an example of putting an error in:

import boto3
from botocore.stub import Stubber

client = boto3.client('s3')
stubber = Stubber(client)
stubber.add_client_error('upload_part_copy')
stubber.activate()

# Will raise a ClientError
client.upload_part_copy()

Here's an example of putting a normal response in. Additionally, the stubber can now be used in a context. It's important to note that the stubber will verify, so far as it is able, that your provided response matches what the service will actually return. This isn't perfect, but it will protect you from inserting total nonsense responses.

import boto3
from botocore.stub import Stubber

client = boto3.client('s3')
stubber = Stubber(client)
list_buckets_response = {
    "Owner": {
        "DisplayName": "name",
        "ID": "EXAMPLE123"
    },
    "Buckets": [{
        "CreationDate": "2016-05-25T16:55:48.000Z",
        "Name": "foo"
    }]
}
expected_params = {}
stubber.add_response('list_buckets', list_buckets_response, expected_params)

with stubber:
    response = client.list_buckets()

assert response == list_buckets_response

Xiomaraxiong

Well, since it's in botocore you would have had to look in the botocore docs, and not many do. It's also fairly recent. – Xiomaraxiong

Why will client.upload_part_copy() raise a ClientError? – Normy

@AidanMelen, because I explicitly added an error to the response queue. You can also add normal service responses. I'll update to show both. – Xiomaraxiong

Does client need to be injected into the unit under test? My understanding of Pythonic unit testing was that testers use something like unittest.mock to mock imported dependencies. Will this approach mock boto clients that are imported in other files? – Longdrawn

You would use these more or less the same way you would use a mock object. You set them up and then pass them into your code. How you pass them in is up to you. – Xiomaraxiong

I am getting an error saying {NoCredentialsError}Unable to locate credentials – Herpetology

this means the stubber is not intercepting the client call and it is trying to make an actual Boto call – Hardly

Using this stubber is a bad idea because it will fail in CI environments where configuration files in the ~/.aws directory don't exist. It works for S3 without these files, but not for most other AWS services. For this reason I recommend against using the stubber. – Neodarwinism

Answer 3

If you don't want to use either moto or the botocore stubber (the stubber does not prevent HTTP requests being made to AWS API endpoints it seems), you can use the more verbose unittest.mock way:

foo/bar.py

import boto3

def my_bar_function():
    client = boto3.client('s3')
    buckets = client.list_buckets()
    ...

bar_test.py

import unittest
from unittest import mock


class MyTest(unittest.TestCase):

     @mock.patch('foo.bar.boto3.client')
     def test_that_bar_works(self, mock_s3_client):
         self.assertTrue(mock_s3_client.return_value.list_buckets.call_count == 1)

Harlan

Answer 4

Here's an example of a simple python unittest that can be used to fake client = boto3.client('ec2') api call...

import boto3 

class MyAWSModule():
    def __init__(self):
        client = boto3.client('ec2')
        tags = client.describe_tags(DryRun=False)


class TestMyAWSModule(unittest.TestCase):
    @mock.patch("boto3.client.describe_tags")
    @mock.patch("boto3.client")
    def test_open_file_with_existing_file(self, mock_boto_client, mock_describe_tags):
        mock_describe_tags.return_value = mock_get_tags_response
        my_aws_module = MyAWSModule()
    
        mock_boto_client.assert_call_once('ec2')
        mock_describe_tags.assert_call_once_with(DryRun=False)

mock_get_tags_response = {
    'Tags': [
        {
            'ResourceId': 'string',
            'ResourceType': 'customer-gateway',
            'Key': 'string',
            'Value': 'string'
        },
    ],
'NextToken': 'string'
}

hopefully that helps.

Normy

how can I manage global client or resource object? this can't be mocked because its invocation happens before mock setup. – Fewness

first line of 'test_open_file_with_existing_file' should not be 'mock_describe_tags.return_value = mock_get_tags_response'? instead of 'mock_boto_client' ? – Pyo

How do you infer @mock.patch("boto3.client.get_tags") is to be mocked – Acrolith

this article is very good. you should refer to this if you have further questions. toptal.com/python/an-introduction-to-mocking-in-python – Normy

Answer 5

What about simply using moto?

It comes with a very handy decorator:

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

Downcast

What to do if my client is at the global. At that time during the file import, it will invoke the original AWS Infra right? any solution? – Erase

moto is great, but want to note that for some AWS Services, does not have 100% coverage. – Wendywendye

Answer 6

I had to mock boto3 client for some integration testing and it was a bit painful! The problem that I had is that moto does not support KMS very well, yet I did not want to rewrite my own mock for the S3 buckets. So I created this morph of all of the answers. Also it works globally which is pretty cool!

I have it setup with 2 files.

First one is aws_mock.py. For the KMS mocking I got some predefined responses that came from live boto3 client.

from unittest.mock import MagicMock

import boto3
from moto import mock_s3

# `create_key` response
create_resp = { ... }

# `generate_data_key` response
generate_resp = { ... }

# `decrypt` response
decrypt_resp = { ... }

def client(*args, **kwargs):
    if args[0] == 's3':
        s3_mock = mock_s3()
        s3_mock.start()
        mock_client = boto3.client(*args, **kwargs)

    else:
        mock_client = boto3.client(*args, **kwargs)

        if args[0] == 'kms':
            mock_client.create_key = MagicMock(return_value=create_resp)
            mock_client.generate_data_key = MagicMock(return_value=generate_resp)
            mock_client.decrypt = MagicMock(return_value=decrypt_resp)

    return mock_client

Second one is the actual test module. Let's call it test_my_module.py. I've omitted the code of my_module. As well as functions that are under the test. Let's call those foo, bar functions.

from unittest.mock import patch

import aws_mock
import my_module

@patch('my_module.boto3')
def test_my_module(boto3):
    # Some prep work for the mock mode
    boto3.client = aws_mock.client

    conn = boto3.client('s3')
    conn.create_bucket(Bucket='my-bucket')

    # Actual testing
    resp = my_module.foo()
    assert(resp == 'Valid')

    resp = my_module.bar()
    assert(resp != 'Not Valid')

    # Etc, etc, etc...

One more thing, not sure if that is fixed but I found out that moto was not happy unless you set some environmental variables like credentials and region. They don't have to be actual credentials but they do need to be set. There is a chance it might be fixed by the time you read this! But here is some code in case you do need it, shell code this time!

export AWS_ACCESS_KEY_ID='foo'
export AWS_SECRET_ACCESS_KEY='bar'
export AWS_DEFAULT_REGION='us-east-1'

I know it is probably not the prettiest piece of code but if you are looking for something universal it should work pretty well!

Regenerative

This comes so very close my to own use case - I have to deal with Organizations calls from boto3 instead of KMS. However - because all the cool kids are using it now - I'm trying to use pytest (and pytest-mock) and I can't get it to patch your client function into the MagicMock. Have you tried this with pytest instead of unittest? NOTE: I myself have only recently switched from unittest so pytest is still a bit of a puzzle. – Ockeghem

UPDATE: literally just got it to play nice with Pytest. Might post an answer on it, when I have this stable. – Ockeghem

@Marakai, I am actually using pytest to run my tests. I guess I am somewhat newb with unit-testing and did not realize that pytest had its own mock implementation. Hopefully it is not too hard of a change to implement! – Regenerative

If I understand it correctly (and I'm no expert on this), the mocking framework in pytest is merely a wrapper around the unittest mock framework. I found that I can use @pytest.fixture and @mock.patch and it works. I do wish I could upvote your answer more than once, it has helped me tremendously for being able to consistently use boto3 stubs even for those clients that stub doesn't (yet) support. – Ockeghem

@Marakai, glad you found my post useful! Also I am glad I can contribute back to the stack overflow community! – Regenerative

Answer 7

Here is my solution for patching a boto client used in the bowels of my project, with pytest fixtures. I'm only using 'mturk' in my project.

The trick for me was to create my own client, and then patch boto3.client with a function that returns that pre-created client.

@pytest.fixture(scope='session')
def patched_boto_client():
    my_client = boto3.client('mturk')

    def my_client_func(*args, **kwargs):
        return my_client

    with patch('bowels.of.project.other_module.boto3.client', my_client_func):
        yield my_client_func


def test_create_hit(patched_boto_client):    
    client = patched_boto_client()
    stubber = Stubber(client)
    stubber.add_response('create_hit_type', {'my_response':'is_great'})
    stubber.add_response('create_hit_with_hit_type', {'my_other_response':'is_greater'})
    stubber.activate()

    import bowels.of.project # this module imports `other_module`
    bowels.of.project.create_hit_function_that_calls_a_function_in_other_module_which_invokes_boto3_dot_client_at_some_point()

I also define another fixture that sets up dummy aws creds so that boto doesn't accidentally pick up some other set of credentials on the system. I literally set 'foo' and 'bar' as my creds for testing -- that's not a redaction.

It's important that AWS_PROFILE env be unset because otherwise boto will go looking for that profile.

@pytest.fixture(scope='session')
def setup_env():
    os.environ['AWS_ACCESS_KEY_ID'] = 'foo'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'bar'
    os.environ.pop('AWS_PROFILE', None)

And then I specify setup_env as a pytest usefixtures entry so that it gets used for every test run.

Desiredesirea

So you are able to use patching but also maintain the functionality of the stubber class within boto3? I'm struggling to get this working. – Hardly

It's a doozy and I still get dizzy trying to remember what I did. But here's my guess of what I did: I did not patch Stubber -- I only stubbed the client function on boto3, and only on its import location within my package. The Stubber used in pytest is imported within the pytest file, so that "version" never gets patched. I probably misspoke somewhere just now, but hope that helps. – Desiredesirea

Thanks, sounds like you did this a while ago? You don't still happen to have the code around do you? I've tried what you suggested but to no avail. This is very frustrating because we have centered our whole project around clients that are managed within a module so each function doesn't have to create their own clients. – Hardly

Found it! github.com/NYUCCL/psiTurk/blob/… MTurkServices has a class function setup_mturk_connection which calls boto3.client() to set the returned client as a property on itself as self.mtc – Desiredesirea

Thank you for searching for that. I ended up just using pytests built in "monkeypatch" to patch the return of the function outside the lambda_handler that was causing problems in my test – Hardly

Answer 8

I had a slightly different use case where the client is set up during a setup() method in a Class, as it does a few things such as listing things from the AWS service it's talking to (Connect, in my case). Lots of the above approaches weren't quite working, so here's my working version for future Googlers.

In order to get everything to work properly, I had to do this:

In the class under test (src/flow_manager.py):

class FlowManager:
    client: botocore.client.BaseClient
    
    def setup(self):
        self.client = boto3.client('connect')
    
    def set_instance(self):
        response = self.client.list_instances()
        ... do stuff ....

In the test file (tests/unit/test_flow_manager.py):

@mock.patch('src.flow_manager.boto3.client')
def test_set_instance(self, mock_client):
    expected = 'bar'
    instance_list = {'alias': 'foo', 'id': 'bar'}
    mock_client.list_instances.return_value = instance_list
    actual = flow_manager.FlowManager("", "", "", "", 'foo')
    actual.client = mock_client
    actual.set_instance()
    self.assertEqual(expected, actual.instance_id)

I've truncated the code to the relevant bits for this answer.

Futures