模拟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方法中抛出异常吗？

- ptimson

Answer 1

Botocore有一个客户端存根可供使用，专门用于这个目的：文档。

这里是放置错误的示例：

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

- Jordon Phillips

---

好的，既然它在botocore中，你必须查看botocore文档，而不是许多人所做的。这也是相当近期的事情。 - Jordon Phillips

客户端.upload_part_copy()为什么会引发ClientError？ - aidanmelen

需要将client注入到待测试的单元中吗？我理解的Python单元测试是，测试人员使用类似于unittest.mock的东西来模拟导入的依赖项。这种方法是否会模拟在其他文件中导入的boto客户端？ - Carl G

这意味着存根未拦截客户端调用，而是尝试进行实际的Boto调用。 - nojohnny101

使用这个存根是一个坏主意，因为它会在CI环境中失败，其中~/.aws目录中的配置文件不存在。它适用于没有这些文件的S3，但不适用于大多数其他AWS服务。因此，我建议不要使用存根。 - Asclepius

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 类提供了一个很好的解决方案。虽然这是一种更干净的解决方法，但我无法模拟特定操作。

- ptimson

---

这非常有帮助。我花了一些时间才意识到很多boto3客户端实际上是在运行时生成的，因此不能直接进行虚拟测试。 - rumdrums

这是对我有效的解决方案，因为Stubber和许多其他模拟工具无法存根boto3自定义函数，例如上传文件或生成预签名URL。 - JD D

这个答案非常好。我最初尝试使用stubber，但它似乎只适用于立即调用，我无法让它在子函数内部调用时正常工作。另一方面，这个方法完美地解决了我的问题，并且非常容易实现，所以谢谢！ - Jake Boomgaarden

如何模拟“get_object”函数？当我尝试运行以上代码时，调用“get_object”函数并没有被正确地模拟。 - Bhavani Ravi

对我来说，使用“from unittest.mock import patch”可以正常工作。 - Dre

在我的Python 3中，使用mocking @patch("botocore.client.BaseClient._make_api_call", lambda *args, **kwargs: None) 就可以解决问题了。考虑到100%的代码覆盖率。 - Ryan Aquino

Answer 3

如果您不想使用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)

- c4urself

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

希望那能有所帮助。

- aidanmelen

---

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

'test_open_file_with_existing_file'的第一行不应该是'mock_describe_tags.return_value = mock_get_tags_response'，而应该是'mock_boto_client'吗？ - cloudy_weather

你如何推断 @mock.patch("boto3.client.get_tags") 需要被模拟？ - Shivangi Singh

这篇文章非常好。如果您有进一步的问题，应该参考它。https://www.toptal.com/python/an-introduction-to-mocking-in-python - aidanmelen

Answer 5

何不简单地使用moto呢？

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

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

- wikier

---

如果我的客户在全球各地，那么在文件导入期间，它会调用原始的AWS基础架构，有什么解决方案吗？ - Parvathirajan Natarajan

moto很棒，但需要注意的是，对于一些AWS服务来说，并没有100%的覆盖率。 - Alex Moore-Niemi

Answer 6

我需要模拟boto3客户端进行一些集成测试，但这有点棘手！我遇到的问题是moto对KMS支持不太好，但我不想为S3存储桶重写自己的模拟。因此，我创建了这个答案的混合体。而且它可以全局使用，非常酷！

我已经设置了两个文件。

第一个是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'

我知道这段代码可能不是最漂亮的，但如果你在寻找通用的东西，它应该能很好地工作！

- Barmaley

---

这非常接近我的使用情况 - 我必须处理来自boto3的组织调用，而不是KMS。然而 - 因为现在所有酷炫的孩子都在使用它 - 我正在尝试使用pytest（和pytest-mock），但我无法将您的客户端函数打补丁到MagicMock中。您是否尝试过使用pytest而不是unittest？注意：我自己最近才从unittest切换，因此pytest仍然有点难理解。 - Marakai

更新：刚刚成功让它与Pytest协同工作。当我稳定下来后，可能会发布一个答案。 - Marakai

@Marakai，我实际上正在使用pytest来运行我的测试。我想我对单元测试有点新手，没有意识到pytest有自己的模拟实现。希望这不是一个太难实现的改变！ - Barmaley

如果我理解正确的话（我并不是这方面的专家），pytest中的mocking框架只是unittest mock框架的一个包装器。我发现我可以使用@pytest.fixture和@mock.patch，它也能够工作。我真希望我可以给你的答案点赞不止一次，因为它帮助了我很多，让我能够在那些客户端还不支持stub的情况下一直使用boto3 stubs。 - Marakai

@Marakai，很高兴你觉得我的帖子有用！我也很高兴能为stackoverflow社区做出贡献！ - Barmaley

Answer 7

以下是关于在我的项目中使用pytest fixtures修补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环境变量，否则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入口，以便每次测试运行时都会使用它。

- deargle

---

所以你能够使用补丁，同时保持boto3中 stubber 类的功能吗？我很难让这个工作起来。 - nojohnny101

这是一个棘手的问题，我仍然感到头晕，试图记住我做了什么。但这是我对我所做的猜测：我没有修补Stubber - 我只是在我的包中存根了boto3上的client函数，并且只在其导入位置上进行了存根。pytest中使用的Stubber是在pytest文件中导入的，因此该“版本”永远不会被修补。我可能刚才说错了什么，但希望这有所帮助。 - deargle

谢谢，听起来你做过这个一段时间了？你还有代码吗？我尝试了你建议的方法，但没有成功。这非常令人沮丧，因为我们的整个项目都围绕着在模块内管理客户端展开，所以每个函数不必创建自己的客户端。 - nojohnny101

找到了！https://github.com/NYUCCL/psiTurk/blob/6733724b676b32003be4f727762bc264dd341219/tests/conftest.py#L102-L118MTurkServices有一个类函数setup_mturk_connection，它调用boto3.client()来将返回的客户端设置为自身的属性，即self.mtc。 - deargle

谢谢你的搜索。最终我只是使用了pytest内置的“monkeypatch”来修补lambda_handler外部函数的返回值，这个函数在我的测试中引起了问题。 - nojohnny101

Answer 8

我有一个稍微不同的用例，其中客户端是在类的setup()方法中设置的，因为它会执行一些操作，例如从它正在通信的AWS服务（在我的情况下是Connect）中列出事物。很多上面的方法都不太起作用，所以这里是我为未来的Googlers工作的版本。

为了使一切正常工作，我必须这样做：

在测试类（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 ....

在测试文件中 (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)

我已将代码截取为此答案的相关部分。

- shearn89