使用 botocore 存根时出现 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 存根来存根这个请求，如下所示：

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：发生错误 (UnrecognizedClientException) 调用 StartQueryExecution 时 operation: 请求中包含的安全令牌无效。

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

非常感谢任何帮助。

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 开发工具包进行测试”的部分中，它有一个代码 sn-p，它指定了一个 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 资源，这允许我的单元测试使用我创建的相同测试客户端。

Answer 2

目前，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 以返回您的存根。

Answer 3

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

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']

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

testmymodule.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')