腾讯云

开发者社区

文档建议反馈控制台

首页

文章/答案/技术大牛

发布

社区首页 >问答首页 >使用botocore存根时的UnrecognizedClientException

问使用botocore存根时的UnrecognizedClientException
EN

Stack Overflow用户

提问于 2018-04-20 06:25:09

回答 3查看 3K关注 0票数 2

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

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

任何帮助都是非常感谢的。

python

boto3

python-unittest

botocore

媒体处理1元起

智能、强大、全面的多媒体数据处理服务，助您提升媒体质量、降低成本，媒体处理套餐低至1元

回答 3

Stack Overflow用户

回答已采纳

发布于 2018-04-20 06:38:39

目前，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)

另一种选择是使用模拟修补boto.client以返回顽固。

票数 2

Stack Overflow用户

发布于 2021-11-03 16:35:31

与这里的其他答案类似，我的问题在于我试图使用一个新客户端，而我已经使用了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应用程序进行单元测试的博客(这就是我的应用程序，但仍然应该适用于不使用AWS的人)：https://aws.amazon.com/blogs/developer/introducing-the-new-test-client-for-aws-chalice/

在题为“用进行测试”一节中，它有一个指定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点击。生成的文件更改为：

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之前实例化，这意味着在导入app.py时已经定义了_DYNAMO_RESOURCE，因此app.py没有机会设置自己的Dynamo，这允许我的单元测试使用我创建的同一个测试客户端。

票数 2

Stack Overflow用户

发布于 2019-10-15 06:57:00

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

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