使用 botocore stubber 时发生 UnrecognizedClientException

Question

我正在使用 unittest 来测试使用 boto3 调用 AWS 的函数。

函数如下所示：

import boto3


def my_function():
    client = boto3.client('athena')
    res = client.start_query_exeuction(
        QueryString='SELECT * FROM logs',
        ResultConfiguration={'OutputLocation': 's3://mybucket'}
    )

    return res['QueryExecutionId']

我正在使用 botocore stubber 在我的单元测试中存根这个请求，如下所示：

from botocore.stub import Stubber
import botocore.session

def test_my_function():    
    client = botocore.session.get_session().create_client('athena')
    client_res = {'QueryExecutionId': 'testid'}
    exp_params = {
        'QueryString': 'SELECT * FROM logs',
        'ResultConfiguration': {
            'OutputLocation': 's3://mybucket'
        }
    }
    with Stubber(client) as stubber:
        stubber.add_response('start_query_execution', client_res, exp_params)
        res = my_function()

    self.assertEqual(res, 'testid')

此测试失败

botocore.exceptions.ClientError: An error occurred (UnrecognizedClientException) when calling the StartQueryExecution operation: The security token included in the request is invalid.

为什么会失败？是因为我在 my_function() 中创建了一个新客户端，它与 stubber 中使用的客户端不同吗？如果是这样，我该如何测试？

非常感谢任何帮助。

Answer 1

目前，my_function() 正在创建一个新客户端，并使用它代替 stubber。

一个选择是改变 my_function 以将 _client 作为参数。

def my_function(_client=None):
    if _client is not None:
        client = _client
    else:
        client = boto3.client('athena')
    res = client.start_query_exeuction(
        QueryString='SELECT * FROM logs',
        ResultConfiguration={'OutputLocation': 's3://mybucket'}
    )

    return res['QueryExecutionId']

然后将stubber传递给my_function。

with Stubber(client) as stubber:
    stubber.add_response('start_query_execution', client_res, exp_params)
    res = my_function(_client=stubber)

另一种选择是使用 mock 将 boto.client 修补为 return 你的存根。

Answer 2

您还可以为客户端命名空间并执行以下操作：

mymodule.py

import boto3

class Amazon
    client = boto3.client('athena') # giving easy namespace access
    @classmethod
    def my_function(cls):
        res = cls.client.start_query_exeuction(
                QueryString='SELECT * FROM logs',
                ResultConfiguration={'OutputLocation': 's3://mybucket'}
            )

        return res['QueryExecutionId']

然后在你的测试中做一个：

测试mymodule.py

from botocore.stub import Stubber
from mymodule import Amazon

def test_my_function():
    client_res = {'QueryExecutionId': 'testid'}
    exp_params = {
        'QueryString': 'SELECT * FROM logs',
        'ResultConfiguration': {
            'OutputLocation': 's3://mybucket'
        }
    }
    with Stubber(Amazon.client) as stubber:
        stubber.add_response('start_query_execution', client_res, exp_params)
        res = Amazon.my_function()

    self.assertEqual(res, 'testid')

Answer 3

与这里的其他答案类似，我的问题在于我已经用 moto 模拟了一个客户端，但我试图使用一个新客户端。

我的错误设置：

app.py

import boto3

_DYNAMO_RESOURCE = boto3.resource('dynamodb')
_METADATA_TABLE_NAME = 'metadata'

def my_ddb_func():
  table = _DYNAMO_RESOURCE.Table(_METADATA_TABLE_NAME)
  # perform some Dynamo call
  response = table.scan(...)
  return response

unit_test.py

import boto3
import moto
import app

@fixture(name='dynamo_resource')
def fixture_dynamo_resource():
  with mock_dynamodb2():
    resource = boto3.resource('dynamodb')
    yield resource

def test_my_ddb_func(dynamo_resource):
  # perform some base level call and assert
  response = app.my_ddb_func()
  assert response

这将导致 UnrecognizedClientException。经过几个小时搜索我的问题后，我找不到任何适合我的修复程序，因此我把它放在这里以供将来使用。

修复了这个关于如何对 AWS Chalice 应用程序进行单元测试的博客（这是我的应用程序，但仍然适用于任何不使用 AWS Chalice 的人）：https://aws.amazon.com/blogs/developer/introducing-the-new-test-client-for-aws-chalice/

在标题为“使用 AWS SDK 测试 Python”的部分中，它有一个指定 S3 常量和 getter 的代码片段，如下所示：

_S3 = None


def get_s3_client():
    global _S3
    if _S3 is None:
        _S3 = boto3.client('s3')
    return _S3


@app.route('/resources/{name}', methods=['PUT'])
def send_to_s3(name):
    s3 = get_s3_client()
    s3.put_object(
        Bucket='mybucket',
        Key=name,
        Body=app.current_request.raw_body
    )
    return Response(status_code=204, body='')

这有助于@Alasdair click 的解决方案。我生成的文件更改为：

app.py

import boto3

_DYNAMO_RESOURCE = None
#                  ^^^^ Note the new default of None
_METADATA_TABLE_NAME = 'metadata'

# New getter method
def get_dynamo_resource():
  global _DYNAMO_RESOURCE
  if _DYNAMO_RESOURCE is None:
    _DYNAMO_RESOURCE = boto3.resource('dynamodb')
  return _DYNAMO_RESOURCE

def my_ddb_func():
  table = get_dynamo_resource().Table(_METADATA_TABLE_NAME)
  #       ^^^^^^^^^^^^^^^^^^^^^ Note the change to calling getter method
  # perform some Dynamo call
  response = table.scan(...)
  return response

unit_test.py

import boto3
import moto
import app

@fixture(name='dynamo_resource')
def fixture_dynamo_resource():
  with mock_dynamodb2():
    resource = boto3.resource('dynamodb')
    yield resource

def test_my_ddb_func(dynamo_resource):
  # perform some base level call and assert
  response = app.my_ddb_func()
  assert response

有一些我没有包括的小细节，比如我的方法所采用的路由的装饰器，因为它是这个例子的虚拟方法，可能还有一些导入。 重要的一点是将常量默认为 None 并编写一个 getter 方法并进行条件检查以获取正在使用的客户端。

这允许 moto 的模拟 Dynamo 资源通过在我的 app.py 之前实例化而实际使用，这意味着 _DYNAMO_RESOURCE 因此在导入 app.py 时已经定义，所以 app.py 没有机会设置自己的 Dynamo 资源，这允许我的单元测试使用我创建的相同测试客户端。

使用 botocore stubber 时发生 UnrecognizedClientException

UnrecognizedClientException when using botocore stubber

python

boto3

python-unittest

botocore