模拟boto3 S3客户端方法Python

Posted 2023-03-04

【中文标题】模拟boto3 S3客户端方法Python

【英文标题】：Mocking boto3 S3 client method Python

【发布时间】：2016-09-05 17:32:33

【问题描述】：

我正在尝试从 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 方法上抛出异常有什么想法吗？

【参考方案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

【讨论】：

好吧，既然它在 botocore 中，您将不得不查看 botocore 文档，而且很少有人这样做。它也是最近的。

为什么 client.upload_part_copy() 会引发 ClientError？

@AidanMelen，因为我在响应队列中明确添加了一个错误。您还可以添加正常的服务响应。我会更新以显示两者。

client 是否需要注入被测单元？我对 Pythonic 单元测试的理解是，测试人员使用 unittest.mock 之类的东西来模拟导入的依赖项。这种方法会模拟在其他文件中导入的 boto 客户端吗？

我收到一条错误消息NoCredentialsErrorUnable to locate credentials

【参考方案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 also posted a great solution 使用 botocore.stub.Stubber 类。虽然更清洁的解决方案我无法模拟特定操作。

【讨论】：

这很有帮助。我花了一段时间才意识到很多 boto3 客户端实际上是generated at runtime，因此不能直接模拟。

这是适合我的解决方案，因为 Stubber 和许多其他模拟工具无法存根 boto3 自定义功能，例如上传文件或生成的预签名 URL。

这个答案很棒。我第一次尝试使用 stubber，但它似乎只适用于立即调用，由于某种原因，我无法让它在子函数中进行调用。另一方面，这工作得很好，很容易实现，所以谢谢！

如何模拟get_object函数，当我尝试运行上面的代码时，get对象调用没有通过模拟。

对我来说，它使用'from unittest.mock import patch'工作

【参考方案3】：

这是一个简单的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_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'

希望对您有所帮助。

【讨论】：

如何管理全局客户端或资源对象？这不能被模拟，因为它的调用发生在模拟设置之前。

'test_open_file_with_existing_file' 的第一行不应该是 'mock_describe_tags.return_value = mock_get_tags_response'？而不是“mock_boto_client”？

你如何推断​​@mock.patch("boto3.client.get_tags") 是要被mock的

这篇文章很好。如果您还有其他问题，请参阅此内容。 toptal.com/python/an-introduction-to-mocking-in-python

【参考方案4】：

如果您不想使用 moto 或 botocore 存根（存根确实不阻止向 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)

【参考方案5】：

简单地使用moto怎么样？

它带有一个非常方便的decorator：

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

【讨论】：

如果我的客户在全球，该怎么办。那时在文件导入过程中，它会调用原始的 AWS Infra 对吗？有什么解决办法吗？

【参考方案6】：

我不得不模拟 boto3 客户端进行一些集成测试，这有点痛苦！我遇到的问题是moto 不太支持KMS，但我不想为S3 存储桶重写我自己的模拟。所以我创造了所有答案的这个变形。它还可以在全球范围内使用，非常酷！

我设置了 2 个文件。

第一个是aws_mock.py。对于KMS 模拟，我得到了一些来自实时boto3 客户端的预定义响应。

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

第二个是实际的测试模块。我们称之为test_my_module.py。我省略了my_module 的代码。以及正在测试的功能。我们将这些函数称为foo、bar。

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...

还有一件事，不确定这是否已解决，但我发现 moto 并不满意，除非您设置一些环境变量，例如凭据和区域。它们不必是实际凭据，但确实需要设置。当您阅读本文时，它可能会修复！但是这里有一些代码以备不时之需，这次是 shell 代码！

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

我知道它可能不是最漂亮的代码，但如果您正在寻找通用的东西，它应该会很好用！

【讨论】：

这与我自己的用例非常接近 - 我必须处理来自 boto3 而不是 KMS 的组织调用。但是 - 因为所有酷孩子现在都在使用它 - 我正在尝试使用 pytest（和 pytest-mock），但我无法将您的客户端功能修补到 MagicMock 中。您是否使用 pytest 而不是 unittest 尝试过这个？注意：我自己最近才从 unittest 切换，所以 pytest 仍然是一个谜。

更新：实际上只是让它与 Pytest 一起玩得很好。当我有这个稳定时，可能会在上面发布答案。

@Marakai，我实际上是在使用 pytest 来运行我的测试。我想我对单元测试有点陌生，并没有意识到 pytest 有自己的模拟实现。希望实施起来不会太难！

如果我理解正确（我不是这方面的专家），pytest 中的模拟框架只是 unittest 模拟框架的包装。我发现我可以使用@pytest.fixture 和@mock.patch 并且它有效。我真希望我能不止一次地支持你的答案，这对我能够持续使用 boto3 存根有很大帮助，即使对于那些存根（尚不）支持的客户也是如此。

@Marakai，很高兴你发现我的帖子很有用！我也很高兴能回馈堆栈溢出社区！

【参考方案7】：

这是我使用pytest 固定装置修补项目内部使用的boto 客户端的解决方案。我只在我的项目中使用“mturk”。

对我来说，诀窍是创建自己的客户端，然后使用返回预先创建的客户端的函数修补 boto3.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()

我还定义了另一个设置虚拟 aws 凭据的装置，这样 boto 就不会意外获取系统上的其他一组凭据。我确实将 'foo' 和 'bar' 设置为我的测试凭据——这不是修订。

取消设置 AWS_PROFILE env 很重要，否则 boto 会寻找该配置文件。

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

然后我将 setup_env 指定为 pytest usefixtures 条目，以便每次测试运行都使用它。

【讨论】：

所以您既可以使用补丁，又可以维护 boto3 中存根类的功能？我正在努力让它发挥作用。

这太棒了，我仍然头晕目眩，试图记住我做了什么。但这是我对我所做的事情的猜测：我没有 修补Stubber——我只在boto3 上对client 函数进行了存根，并且仅在我的包中它的导入位置。 pytest 中使用的Stubber 被导入到 pytest 文件中，因此“版本”永远不会被修补。我可能刚才在某个地方说错了，但希望对您有所帮助。

谢谢，您好像不久前就这样做了？您还没有碰巧有代码吗？我已经尝试了您的建议，但无济于事。这非常令人沮丧，因为我们将整个项目集中在一个模块内管理的客户端上，因此每个函数都不必创建自己的客户端。

找到了！ github.com/NYUCCL/psiTurk/blob/… MTurkServices 有一个类函数 setup_mturk_connection 调用 boto3.client() 将返回的客户端设置为自身的属性为 self.mtc

感谢您搜索。我最终只使用“monkeypatch”中内置的 pytests 来修补导致我的测试出现问题的 lambda_handler 之外的函数的返回