python-3.x - 如何为 AWS Athena 服务编写 Pytest
问题描述
我正在为使用 AThena 的 lambda 编写 pytest,使用以下代码创建 pytest.fixture:
@pytest.fixture()
def moto_boto_athena():
@mock_athena
def boto_resource_athena():
res = boto3.client('athena')
return res
return boto_resource_athena
我的 Lambda 有以下代码:
response = get_athena_client().start_query_execution(
QueryString=query_string,
QueryExecutionContext={
'Database': athena_database
},
ResultConfiguration={
'OutputLocation': athena_output_bucket,
}
)
query_status = None
#Check the status of QueryExecution
while query_status == 'QUEUED' or query_status == 'RUNNING' or query_status is None:
query_status = get_athena_client(). \
get_query_execution(QueryExecutionId=response['QueryExecutionId'])['QueryExecution']['Status']['State']
if query_status == 'FAILED' or query_status == 'CANCELLED':
raise Exception
但是,一旦我开始执行测试,就会出现以下异常:
raise NotImplementedError(
> "The {0} action has not been implemented".format(action)
)
E NotImplementedError: The start_query_execution action has not been implemented
..\..\..\..\..\AppData\Roaming\Python\Python36\site-packages\moto\core\responses.py:393: NotImplementedError
请指导。
解决方案
推荐阅读
- javascript - React 组件更新时刷新页面
- php - 对数组进行数字排序
- java - 为什么在 void 方法中不允许 this()?
- c# - 如果我更改 Identity 主键,我应该在 ASP.NET Core 2.1 中更改哪些依赖项注册?
- php - PHP API 允许从反应应用程序访问
- javascript - 我正在使用事件处理程序探索 JavaScript Promises
- sql-server-2008 - 将 varchar 值 'Debit>0' 转换为数据类型 int 时转换失败
- django - 我是否需要在 Django 的整体布局和标题上使用两次“加载静态”?
- django - Django:添加多对多字段
- android - 当查看寻呼机跨越阈值到新页面时如何收听?