模拟boto3 S3客户端方法Python

Question

我正在尝试从boto3 s3客户端对象模拟一个单一方法来抛出异常.但我需要这个类的所有其他方法正常工作.

这样我可以在执行upload_part_copy时测试单个异常测试并发生错误

第一次尝试

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

但是,这会产生以下错误:

ImportError: No module named S3

第二次尝试

在查看了botocore.client.py源代码后,我发现它正在做一些聪明的事情并且该方法upload_part_copy不存在.我发现它似乎在调用,BaseClient._make_api_call所以我试图嘲笑它

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

这引发了一个例外......但是get_object我要避免它.

关于我如何只能在upload_part_copy方法上抛出异常的任何想法？

Answer 1

Botocore有一个客户端存根,你可以用它来实现这个目的:docs.

以下是将错误放入的示例:

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

这是一个将正常响应放入其中的示例.此外,现在可以在上下文中使用存根.重要的是要注意,存根将尽可能验证您提供的响应是否与服务实际返回的响应相匹配.这并不完美,但它可以保护您不会插入完全无意义的回复.

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

Answer 2

我一发布在这里,我设法找到了解决方案.这是希望它有帮助:)

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还使用botocore.stub.Stubber类发布了一个很棒的解决方案.虽然更清洁的解决方案,我无法模拟特定的操作.

Answer 3

如果您不想使用moto或 botocore stubber（stubber 似乎不会阻止向 AWS API 端点发出 HTTP 请求），您可以使用更详细的 unittest.mock 方式：

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)

Answer 4

这是一个简单的python单元测试的例子,可以用来伪造client = boto3.client('ec2') api调用...

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'
}

希望这会有所帮助.

Answer 5

那么简单地使用moto呢？

它带有一个非常方便的装饰器：

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass