2

使用 moto 中的 botocore.stub 和 mock_s3 我可以像下面那样对 boto3 S3 客户端进行存根，

from moto import mock_s3
from botocore.stub import Stubber

class TestS3(unittest.TestCase):
    @mock_s3
    def setUp(self):
        self.s3_client = boto3.client('s3', region_name='us-west-1')
        self.stubber = Stubber(client=self.s3_client)

然后在测试函数中，我可以像下面一样使用 add_client_error 和 add_response 并根据需要添加断言，

@mock_s3
def test_put_object_exception(self):
    self.stubber.add_client_error('put_object', service_error_code='500')
    with self.stubber:
        self.s3_service.put_object(data='/some/dummy/path', key='/some/dummy/key')

但是在我的 S3 类中，我使用 S3Transfer upload_file 将文件上传到 S3，有没有办法模拟 S3Transfer(self.s3_client).upload_file 方法？

4

2 回答 2

1

Dan Hook 的回答非常好。如果您还想确保请求的上传文本符合您的期望，则需要一个自定义匹配器对象来代替ANY变量。它需要有一个与 boto 读取块类一起使用的自定义__eq__（因此也是__ne__）实现。请注意，下面的代码使用 Python 类型。

class ExpectedBinStream:
    """Expect a binary stream value."""
    def __init__(self, contents: bytes) -> None:
        self.contents = contents

    def __eq__(self, other: Any) -> bool:
        if not hasattr(other, 'read') or not callable(other.read):
            return False
        data = other.read()
        return self.contents == data

    def __ne__(self, other: Any) -> bool:
        return not self.__eq__(other)

    def __repr__(self) -> str:
        return repr(self.contents)

这是有效的，因为存根对传入的参数使用简单的相等性检查。该值还通过始终返回for和for 来ANY利用这一点。True__eq__False__ne__

于 2020-08-04T18:29:32.367 回答
0

我发现存根 put_object 效果很好。在示例中，“test_file”确实需要是一个真实文件，因为 S3Transfer 会检查它是否存在。如果 S3Transfer 传输的文件足够大以证明多部分上传的合理性，S3Transfer 将使用其他服务方法，但模拟它开始感觉更像是为 S3Transfer 编写测试而不是测试我自己的代码。

import botocore.session
from botocore.stub import Stubber, ANY
from boto3.s3.transfer import S3Transfer
s3 = botocore.session.get_session().create_client('s3')
stubber = Stubber(s3)

BUCKET='mock_bucket'
KEY='mock_key'                       
put_object_response = {   'ETag': '"14fe4f49fffffffffff9afbaaaaaaaa9"',
    'ResponseMetadata': {   'HTTPHeaders': {   'content-length': '0',
                                               'date': 'Wed, 08 Apr 2020 '
                                                       '20:35:42 GMT',
                                               'etag': '"14fe4f49fffffffffff9afbaaaaaaaa9"',
                                               'server': 'AmazonS3',
                                               },
                            'HTTPStatusCode': 200,
                            'HostId': 'GEHrJmjk76Ug/clCVUwimbmIjTTb2S4kU0lLg3Ylj8GKrAIsv5+S7AFb2cRkCLd+mpptmxfubLM=',
                            'RequestId': 'A8FFFFFFF84C3A77',
                            'RetryAttempts': 0},
    'VersionId': 'Dbc0gbLVEN4N5F4oz7Hhek0Xd82Mdgyo'}
stubber.add_response('put_object', put_object_response,expected_params= {'Body': ANY, 'Bucket':BUCKET,'Key':KEY})
stubber.activate()
transfer = S3Transfer(s3)
transfer.upload_file('test_file', BUCKET,KEY)
于 2020-04-08T21:29:23.230 回答