我正在尝试测试一个调用 DynamoDb 并在部署时工作的 Lambda 函数处理程序，但是在测试 Moto 时似乎在设置阶段遇到了问题并产生了一个模糊的错误。测试代码如下：
# -*- coding: utf-8 -*-
from model.terrestrial import Terrestrial
import pytest
import moto
import boto3
import os
import json
REGION = 'ap-southwest-2'
TERRESTRIAL_TABLE_NAME = 'U_terrestrial'
os.environ['TERRESTRIAL_TABLE_NAME'] = TERRESTRIAL_TABLE_NAME
os.environ['PRIMARY_KEY'] = 'id'
from lambdas import terrestrial_persist_function as func
from tests.lambdas import api_gateway_event
@pytest.fixture
def dynamodb_table():
with moto.mock_dynamodb2():
boto3.client('dynamodb', region_name=REGION).create_table(
AttributeDefinitions=[
{'AttributeName': 'id', 'AttributeType': 'S'}
],
TableName=TERRESTRIAL_TABLE_NAME,
KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
'WriteCapacityUnits': 1,
},
)
yield boto3.resource('dynamodb', region_name=REGION).Table(TERRESTRIAL_TABLE_NAME)
@pytest.fixture
def mercury():
file = open('../fixtures/db_2_terrestrial_mercury.json', 'r')
obj: Terrestrial = Terrestrial.dict_to_object(json.loads(file.read().strip()))
file.close()
return obj.dict_dynamodb()
@pytest.fixture
def venus():
file = open('../fixtures/db_3_terrestrial_venus.json', 'r')
obj: Terrestrial = Terrestrial.dict_to_object(json.loads(file.read().strip()))
file.close()
return obj.dict_dynamodb()
@pytest.fixture
def earth():
file = open('../fixtures/db_4_0_terrestrial_earth.json', 'r')
obj: Terrestrial = Terrestrial.dict_to_object(json.loads(file.read().strip()))
file.close()
return obj.dict_dynamodb()
@pytest.fixture
def luna():
file = open('../fixtures/db_4_0_terrestrial_luna.json', 'r')
obj: Terrestrial = Terrestrial.dict_to_object(json.loads(file.read().strip()))
file.close()
return obj.dict_dynamodb()
@pytest.fixture
def mars():
file = open('../fixtures/db_5_0_terrestrial_mars.json', 'r')
obj: Terrestrial = Terrestrial.dict_to_object(json.loads(file.read().strip()))
file.close()
return obj.dict_dynamodb()
@pytest.fixture
def all_terrestrials(mercury, venus, earth, luna, mars) -> []:
return [mercury, venus, earth, luna, mars]
def headers():
return {
"Access-Control-Allow-Origin": "*",
"Access-Control-Allow-Credentials": "True",
"Content-Type": "application/json"
}
def test_get_terrestrials_ok(dynamodb_table, all_terrestrials):
# given
print('table: {}'.format(TERRESTRIAL_TABLE_NAME))
print('os table: {}'.format(os.environ['TERRESTRIAL_TABLE_NAME']))
dirpath = os.getcwd()
print("current directory is : " + dirpath)
with dynamodb_table.batch_writer() as batch:
for i in all_terrestrials:
batch.put_item(Item=i)
# when
resp = func.handler(
api_gateway_event(
'/terrestrial',
'GET',
{}
), None)
# then
assert resp['statusCode'] == 200
assert resp['headers'] == headers()
assert resp['body'] == json.dumps(all_terrestrials())
def test_get_terrestrials_empty(dynamodb_table):
print('table: {}'.format(TERRESTRIAL_TABLE_NAME))
print('os table: {}'.format(os.environ['TERRESTRIAL_TABLE_NAME']))
dirpath = os.getcwd()
print("current directory is : " + dirpath)
# when
resp = func.handler(
api_gateway_event(
'/terrestrial',
'GET',
{}
), None)
# then
assert resp['statusCode'] == 200
assert resp['headers'] == headers()
assert resp['body'] == '[]'
def test_get_terrestrial_ok(dynamodb_table, all_terrestrials):
print('table: {}'.format(TERRESTRIAL_TABLE_NAME))
print('os table: {}'.format(os.environ['TERRESTRIAL_TABLE_NAME']))
dirpath = os.getcwd()
print("current directory is : " + dirpath)
with dynamodb_table.batch_writer() as batch:
for i in all_terrestrials:
batch.put_item(Item=i)
# when
resp = func.handler(
api_gateway_event(
'/terrestrial/40743816-c884-4c79-9ab4-6dd4c740c52c',
'GET',
{}
), None)
# then
assert resp['statusCode'] == 200
assert resp['headers'] == headers()
assert resp['body'] == json.dumps(earth())
def test_get_terrestrial_not_found(dynamodb_table):
print('table: {}'.format(TERRESTRIAL_TABLE_NAME))
print('os table: {}'.format(os.environ['TERRESTRIAL_TABLE_NAME']))
dirpath = os.getcwd()
print("current directory is : " + dirpath)
# when
resp = func.handler(
api_gateway_event(
'/terrestrial/00000000-0000-0000-0000-000000000000',
'GET',
{}
), None)
# then
assert resp['statusCode'] == 200
assert resp['headers'] == headers()
assert resp['body'] == {}
产生的错误（对于每个测试）是：
ERROR [ 25%]
test setup failed
@pytest.fixture
def dynamodb_table():
with moto.mock_dynamodb2():
boto3.client('dynamodb', region_name=REGION).create_table(
AttributeDefinitions=[
{'AttributeName': 'id', 'AttributeType': 'S'}
],
TableName=TERRESTRIAL_TABLE_NAME,
KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
ProvisionedThroughput={
'ReadCapacityUnits': 1,
> 'WriteCapacityUnits': 1,
},
)
../lambdas/test_terrestrial_persist_function.py:37:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/client.py:320: in _api_call
return self._make_api_call(operation_name, kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/client.py:611: in _make_api_call
operation_model, request_dict)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/endpoint.py:102: in make_request
return self._send_request(request_dict, operation_model)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/endpoint.py:136: in _send_request
success_response, exception):
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/endpoint.py:210: in _needs_retry
caught_exception=caught_exception, request_dict=request_dict)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/hooks.py:356: in emit
return self._emitter.emit(aliased_event_name, **kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/hooks.py:228: in emit
return self._emit(event_name, kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/hooks.py:211: in _emit
response = handler(**kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/retryhandler.py:183: in __call__
if self._checker(attempts, response, caught_exception):
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/retryhandler.py:251: in __call__
caught_exception)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/retryhandler.py:269: in _should_retry
return self._checker(attempt_number, response, caught_exception)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/retryhandler.py:317: in __call__
caught_exception)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/retryhandler.py:223: in __call__
attempt_number, caught_exception)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/retryhandler.py:359: in _check_caught_exception
raise caught_exception
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/endpoint.py:176: in _get_response
responses = self._event_emitter.emit(event_name, request=request)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/hooks.py:356: in emit
return self._emitter.emit(aliased_event_name, **kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/hooks.py:228: in emit
return self._emit(event_name, kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/botocore/hooks.py:211: in _emit
response = handler(**kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/core/models.py:292: in __call__
status, headers, body = response_callback(request, request.url, request.headers)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/core/responses.py:117: in dispatch
return cls()._dispatch(*args, **kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/core/responses.py:200: in _dispatch
return self.call_action()
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/core/utils.py:264: in _wrapper
response = f(*args, **kwargs)
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/dynamodb2/responses.py:64: in call_action
response = getattr(self, endpoint)()
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/dynamodb2/responses.py:108: in create_table
table = self.dynamodb_backend.create_table(table_name,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <moto.dynamodb2.responses.DynamoHandler object at 0x122590f50>
@property
def dynamodb_backend(self):
"""
:return: DynamoDB2 Backend
:rtype: moto.dynamodb2.models.DynamoDBBackend
"""
> return dynamodb_backends[self.region]
E KeyError: 'ap-southwest-2'
/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/moto/dynamodb2/responses.py:56: KeyError
为了减少这个问题的冗长，我省略了我正在测试的 lambda，因为我不相信我们能走到这一步。看起来 Moto 设置不正确。但是，如果您需要，我可以在以后的编辑中提供。
如您所见，该错误非常模糊（我希望我的问题不是），而且我不确定从哪里开始解决此问题。任何帮助将不胜感激。
来自 requirements.txt：
pytest==5.1.1
boto3==1.9.66
botocore==1.12.67
moto==1.3.7