使用botocore stubber时无法识别客户端异常

与这里的其他答案类似,我的问题在于,我试图使用一个新客户,而我已经用 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 for 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的解决方案。我的结果文件更改为:

应用程序。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\u测试。py公司

导入boto3
导入moto
导入应用程序

@夹具(name='dynamo\u resource')
def fixture\u dynamo\u resource():
使用mock\u dynamodb2():
资源=boto3。资源('dynamodb')
产量资源

def test\u my\u ddb\u func(发电机资源):
#执行一些基本级别的调用和断言
响应=应用程序。my\u ddb\u func()
断言响应

有一些小细节我没有包括在内,比如我的方法所采用的路径的装饰器,因为在本例中它是一个虚拟方法,可能还有一些导入。 重要的收获是将常量默认为 None 以及编写带有条件检查的getter方法来获取正在使用的客户端。

这使得模拟的Dynamo资源 moto公司 在我的 应用程序。py公司 ,意思是 _DYNAMO_RESOURCE 因此在导入时已定义 应用程序。py公司 所以 应用程序。py公司 没有机会设置自己的Dynamo资源,这允许我的单元测试使用我创建的同一个测试客户端。

目前, 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']

然后通过 斯塔伯 到 my\u函数 。

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

另一种选择是使用 mock 要修补 boto.client 归还你的存根。

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

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