使用botocore存根时的UnrecognizedClientException
我正在使用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()中创建了一个新客户端，它不同于在存根中使用的客户端吗？如果是的话，我如何测试这个？
任何帮助都是非常感谢的。
回答 3
Stack Overflow用户
发布于 2018-04-20 14: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以返回顽固。
Stack Overflow用户
发布于 2021-11-04 00: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 responseunit_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 responseunit_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，这允许我的单元测试使用我创建的同一个测试客户端。
Stack Overflow用户
发布于 2019-10-15 14: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')https://stackoverflow.com/questions/49943938
复制相似问题
腾讯云开发者
