多次调用botocore.stub.Stubber会导致异常

问题描述 投票:0回答:1

我使用botocore.stub.Stubber来模拟kmsclient。我正在使用的代码是

with botocore.stub.Stubber(s3) as stubber:
        with botocore.stub.Stubber(kms) as stubber2:
            stubber.add_response('copy_object', response, expectedParams)
            stubber.activate()
            stubber2.add_response('decrypt', response2, expectedParams2)
            stubber2.activate()
            handleCore(__makeValidEvent(), None, s3, kms)
            stubber.assert_no_pending_responses()
            stubber2.assert_no_pending_responses()

在实际实现中,kmsclient调用发生两次,导致以下异常

params = {'CiphertextBlob': b"\x01\x02\x02\x00x#\xc1\xdbp6\xe1Y\x0fS\x15\x80<\x86\xb5\xb2\x86\x9f\xaf\xa2Z\x07\xfef\x8d\xb2\xd7...\t'\xe2\xb9\x10w0\x83\xcb\xe1\xcb`\xd1\xc2\x8c\xe4\x82Q/*\xb3]\xcfZ\xb9\xbd\x1c\x9c\x96(e\x94j\x1a\x91\xba\xaeO[>\x97"}

    def _assert_expected_call_order(self, model, params):
        if not self._queue:
            raise UnStubbedResponseError(
                operation_name=model.name,
                reason=(
>                       'Unexpected API Call: A call was made but no additional calls expected.'
                        'Either the API Call was not stubbed or it was called multiple times.'
                        )
            )
E           botocore.exceptions.UnStubbedResponseError: Error getting response stub for operation Decrypt: Unexpected API Call: A call was made but no additional calls expected.Either the API Call was not stubbed or it was called multiple times.

有人可以让我知道如何可以在同一个对象上使用多个调用(在这种情况下为kmsclient)

python boto3 aws-kms
1个回答
1
投票

我通过为同一方法添加多个响应来解决这个问题。

示例测试代码:

############## Test Code
client = botocore.session.get_session().create_client("cloudformation", region_name="us-east-1")
stubber = Stubber(client)

# Generate multiple responses to be returned by boto
responses = [
    {
        "StackId": "stack-a",
    },
    {
        "StackId": "stack-b",
    },
    {
        "StackId": "stack-c",
    },
]

# Add each response to stubber for the same method - "update_termination_protection"
for response in responses:
    stubber.add_response(
        "update_termination_protection",
        response,
    )
stubber.activate()
actual = method_to_test(client, data)
stubber.deactivate()
assert actual == True

示例测试方法:

############## Real method
def method_to_test(self, client, data):
    for item in data:
        client.update_termination_protection(
           EnableTerminationProtection=True,
           StackName=item,
        )
    return True

这有效并且没有例外。

© www.soinside.com 2019 - 2024. All rights reserved.