Mocking boto3 S3 client method Python

问题:

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?

回答1:

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

回答2:

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.

回答3:

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.get_tags")
    @mock.patch("boto3.client")
    def test_open_file_with_existing_file(self, mock_boto_client, mock_describe_tags):
        mock_boto_client.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.

回答4:

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