python - 使用 botocore stub 时出现 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 stub 在我的单元测试中 stub 此请求，如下所示:

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() 中创建的新客户端与 stub 中使用的客户端不同？如果是这样，我该如何测试呢？

非常感谢任何帮助。

Answer 1

与此处的其他答案类似，我的问题在于，当我已经用 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/

在标题为“使用适用于 Python 的 AWS 开发工具包进行测试”的部分中，它有一个指定 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 资源，这允许我的单元测试使用相同的测试客户端我创建了。