- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
本文以 moto 仓库中的 DynamoDB 服务功能覆盖文档(docs/docs/services/dynamodb.rst)为主体,完整梳理 moto 对 AWS DynamoDB 各 API 的模拟支持现状:哪些操作可直接用于测试、哪些尚未实现、已实现操作存在哪些文档明示的限制。同时结合moto/dynamodb源码与tests/test_dynamodb测试用例,解释这些功能在 moto 内部是如何落地与验证的,帮助你在编写 AWS 基础设施测试时准确判断 mock 能力边界。
1. 功能覆盖清单是如何定义的
docs/docs/services/dynamodb.rst是 moto 为每个 AWS 服务维护的"Implemented features"(已实现功能)页面,DynamoDB 页面以- [X]表示已实现、- [ ]表示未实现的 checkbox 列表形式列出全部 API。该清单与仓库根目录的 IMPLEMENTATION_COVERAGE.md 保持同源关系,并由 scripts/implementation_coverage.py 脚本扫描moto/<service>/models.py(本服务为 moto/dynamodb/models/init.py)中的后端方法自动生成。
从源码结构看,判定"已实现"的依据是后端类 DynamoDBBackend 中是否定义了与 API 同名或可对应的方法。在 DynamoDBBackend 的注释块 中,仓库明确列出了"逻辑完全位于 responses.py、为覆盖脚本而重复注册"的方法:batch_get_item、batch_write_item、transact_get_items、execute_statement、execute_transaction、batch_execute_statement。这说明文档清单、后端方法、HTTP 处理器三层是一一对应的,你在文档页看到的每个[X]都能在源码中找到实现位置,这是后文逐条溯源的基础。
2. 已实现操作全清单(含文档明示的限制)
以下为文档页面标记为[X]的全部操作,按用途分组,并保留文档原文中的每条限制说明。
2.1 表生命周期管理
| 操作 | 说明 |
|---|---|
create_table | 支持计费模式、GSI/LSI、流、TTL、删除保护等参数 |
delete_table | 受删除保护(DeletionProtection)约束 |
describe_table | 支持按表名或 ARN 查询 |
update_table | 支持更新 GSI、流、计费模式、删除保护、WarmThroughput |
list_tables | 支持 Limit 与 ExclusiveStartTableName 分页 |
describe_endpoints | 返回区域端点描述 |
create_table的入口在 DynamoHandler.create_table,其_validate_table_creation(L385-L530)复刻了 AWS 的校验行为:PAY_PER_REQUEST与ProvisionedThroughput互斥、KeySchema 最多两个元素、AttributeDefinitions必须与主键及索引键完全对齐、WarmThroughput 不得低于 ProvisionedThroughput。这些校验错误消息与真实 AWS 保持一致,测试中断言异常文本时可以直接依赖。
2.2 项(Item)读写与查询
put_item:支持Expected、ConditionExpression、ReturnValuesOnConditionCheckFailure。处理逻辑见 DynamoHandler.put_item,并经过三道校验:空字符串主键(validate_put_has_empty_keys)、空集合属性(validate_put_has_empty_attrs)、GSI 键为 NULL(validate_put_has_gsi_keys_set_to_none),均在 responses.py 顶部工具函数 中实现。get_item/delete_item:支持投影表达式(ProjectionExpression)、条件表达式;空字符串主键会抛出KeyIsEmptyStringException(get_item 校验)。update_item:同时支持新式UpdateExpression与旧式AttributeUpdates,二者互斥;ReturnValues的NONE/ALL_OLD/ALL_NEW/UPDATED_OLD/UPDATED_NEW五种语义均在 DynamoHandler.update_item 中逐一对应实现。query/scan:支持KeyConditionExpression、FilterExpression、ProjectionExpression、IndexName、ConsistentRead、Select、分段扫描(Segment/TotalSegments)。scan对 Segment 与 TotalSegments 的成对出现和取值范围做了与 AWS 一致的校验(L1107-L1128)。batch_get_item:单次最多 100 个键(单表与跨表均校验,L835-L851);响应达到 16MB 时未读键进入UnprocessedKeys(L876-L896)。batch_write_item:检测重复键并报错(L754-L759)。transact_get_items:单次最多 25 个项(TRANSACTION_MAX_ITEMS = 25,responses.py#L38)。transact_write_items:单次最多 100 项,且同一事务中不能对同一项执行多次操作;失败时回滚到事务前状态并抛出TransactionCanceledException。核心实现在 DynamoDBBackend.transact_write_items:先对涉及的所有表做deepcopy快照,任何ConditionalCheckFailed、重复项或参数错误都会触发整体回滚。
2.3 备份与时间点恢复
create_backup/describe_backup/delete_backup/list_backups:备份对象由 Backup 类 管理,delete_backup会将状态置为DELETED(models#L789-L795)。restore_table_from_backup:恢复出RestoredTable,目标表名已存在时抛出TableAlreadyExistsException(L803-L816)。restore_table_to_point_in_time:文档明示限制—— "当前仅接受源表与目标表两个参数,会复制源表的全部项,不理会其他参数"(docs/docs/services/dynamodb.rst第 78-83 行)。源码中的 docstring 与之一字不差(L818-L839),恢复结果为RestoredPITTable。describe_continuous_backups/update_continuous_backups:PITR 开启时记录EarliestRestorableDateTime/LatestRestorableDateTime,RecoveryPeriodInDays缺省为 35(L726-L762)。
2.4 导入 / 导出
import_table:文档明示限制—— 仅支持InputFormat=DYNAMODB_JSON;不支持InputCompressionType=ZSTD;不支持InputFormatOptions与CloudWatchLogGroupArn(文档第 61-65 行)。入口在 DynamoHandler.import_table,会复用与create_table相同的_validate_table_creation校验建表参数;实际导入对象为 TableImport。describe_import:按 ImportArn 查询(L999-L1001)。describe_export/list_exports:导出对象为 TableExport;list_exports按 TableArn 过滤(L1044-L1049)。- 关于
export_table_to_point_in_time:文档页面当前标记为[ ](未实现),但从源码结构看,DynamoHandler.export_table_to_point_in_time 与 DynamoDBBackend.export_table 均已存在,且要求表已启用 PITR(否则抛PointInTimeRecoveryUnavailable)。docstring 中记录的当前实现限制为:仅支持ExportFormat=DYNAMODB_JSON、只导出一个 DYNAMODB_JSON 文件、不支持增量导出。可以推断该接口处于"代码已落地、覆盖清单尚未同步更新"的状态,实际使用时建议以当前代码行为为准。
2.5 PartiQL 语句执行(含文档明示限制)
文档对三个 PartiQL 接口给出了一致的限制说明:
execute_statement(文档第 46-50 行):"Pagination is not yet implemented"(分页尚未实现);"Parsing is highly experimental"(解析高度实验性),发现 bug 需提交 issue。execute_transaction(文档第 53-55 行)与batch_execute_statement(文档第 17-19 行):均注明"请参见execute_statement文档了解支持范围的限制"。
底层实现在 DynamoDBBackend.execute_statement:它将当前区域内所有表(含每个索引,索引数据以表名.索引名键组织,仅包含含齐索引键属性的项)作为source_data交给 partiql.query,后者委托py_partiql_parser的DynamoDBStatementParser解析执行;返回的 before/after 项变更再映射回table.put_item/table.delete_item。因此 SELECT 可以查询主表与 GSI(select * from tbl.gsi_name),DML 语句也能真正改写底层表数据;但分页令牌相关行为未实现,这一点与文档声明一致。
2.6 资源策略、标签与 TTL
put_resource_policy/get_resource_policy/delete_resource_policy:带ExpectedRevisionId乐观并发控制,策略相同则幂等返回(L1051-L1100)。tag_resource/untag_resource/list_tags_of_resource:按 ARN 定位表;list_tags_of_resource以 10 个为一页并用NextToken续传(responses#L622-L639)。后端还通过TaggableResourcesMixin暴露给 resourcegroupstaggingapi 服务(L1102-L1122)。update_time_to_live/describe_time_to_live:TTL 规格必须同时包含Enabled与AttributeName(L578-L602)。
3. 未实现操作全清单
文档页面标记为[ ]的操作共 19 个,按主题归类如下。编写测试时若依赖这些 API,moto 目前无法模拟:
| 主题 | 未实现的操作 |
|---|---|
| 全局表 | create_global_table、describe_global_table、update_global_table、list_global_tables、describe_global_table_settings、update_global_table_settings、describe_table_replica_auto_scaling、update_table_replica_auto_scaling |
| Contributor Insights | describe_contributor_insights、list_contributor_insights、update_contributor_insights |
| Kinesis 流式导出目的地 | describe_kinesis_streaming_destination、enable_kinesis_streaming_destination、disable_kinesis_streaming_destination、update_kinesis_streaming_destination |
| 向量检索 | search_vectors |
| 其他 | describe_limits、export_table_to_point_in_time(见 2.4 节说明)、list_imports |
其中describe_limits值得单独说明:文档页面标记为未实现,但从源码结构看,DynamoHandler.describe_limits 已存在并返回一组固定的账户/表容量上限值(如AccountMaxReadCapacityUnits: 20000)。可以推断该接口的处理器已经可用,只是尚未进入覆盖脚本的登记范围。
4. 表达式的解析引擎:理解 query/scan/update 的实现深度
moto 对 DynamoDB 表达式支持之所以接近真实行为,关键在于moto/dynamodb/parsing/目录下的一套多阶段解析器,其开发文档 parsing/README.md 描述了四个阶段:
- 词法分析(tokenize):字符串自左向右转为 token 列表,token 定义见 tokens.py;
- 构建 AST:token 列表解析为抽象语法树,节点类型见 ast_nodes.py;
- 语义校验:对路径与属性做全量校验并解析占位符值,实现于 validators.py;
- 执行:基于校验后的 AST 执行更新表达式。由于校验阶段已完成,执行阶段不会失败,从而保证更新表达式是原子的(要么全部生效,要么全部不生效)。
Query的KeyConditionExpression走 key_condition_expression.py 的parse_expression,将条件拆分为 hash 条件与 range 条件后交给 DynamoDBBackend.query 按主键定位再过滤;FilterExpression与ConditionExpression则统一由 comparisons.py 的create_condition_expression_parser/get_filter_expression编译为可求值操作。这些机制与真实 AWS 校验错误消息对齐,使测试既能验证正常路径,也能精确断言 AWS 风格的错误。
5. 测试体系如何验证这些功能
tests/test_dynamodb/目录包含 37 个测试文件,与文档清单逐项对应:test_dynamodb_create_table.py、test_dynamodb_query.py、test_dynamodb_scan.py、test_dynamodb_statements.py、test_dynamodb_transact.py、test_dynamodb_import_table.py、test_dynamodb_resource_policy.py、test_dynamodb_batch_get_item.py、test_dynamodb_consumedcapacity.py等,另有exceptions/(异常与事务)与models/(项模型与键条件解析器)子目录。
这些测试使用仓库特有的 dynamodb_aws_verified 装饰器:默认在mock_aws上下文中运行;若设置环境变量MOTO_TEST_ALLOW_AWS_REQUEST=true,同一份测试会直接对真实 AWS 建表、执行、删表。装饰器支持按需附加 range 键、GSI(含多属性 GSI)、LSI 等,与文档清单中"已实现"的表结构能力一致。
例如 PartiQL 功能的最小验证用例(摘自 test_dynamodb_statements.py):
@mock_aws def test_execute_statement_select_star(): client = boto3.client("dynamodb", "us-east-1") client.create_table( TableName="my-table", KeySchema=[{"AttributeName": "pk", "KeyType": "HASH"}], AttributeDefinitions=[{"AttributeName": "pk", "AttributeType": "S"}], BillingMode="PAY_PER_REQUEST", ) client.put_item(TableName="my-table", Item={"pk": {"S": "msg1"}, "body": {"S": "some text"}}) items = client.execute_statement(Statement="select * from my-table")["Items"] assert {"pk": {"S": "msg1"}, "body": {"S": "some text"}} in items事务回滚语义同样有测试覆盖(test_dynamodb_transact.py 与 exceptions/test_dynamodb_transactions.py),对应 2.2 节描述的快照-回滚机制。
6. 实践建议
- 可放心 mock 的场景:表 CRUD、含 GSI/LSI 的建表校验、item 级读写与条件表达式、query/scan 的各类表达式与 Select 语义、批量与事务操作、备份/PITR 恢复、TTL、资源策略与标签——这些在文档与源码中双重确认已实现。
- 需要绕开或手工打桩的场景:全局表全家桶、Contributor Insights、Kinesis 流式目的地、
search_vectors,以及execute_statement的分页行为(文档明示"highly experimental",建议对语句解析相关的断言保持宽松并及时向仓库报告 bug)。 - 阅读覆盖清单的方法:以
docs/docs/services/dynamodb.rst与 IMPLEMENTATION_COVERAGE.md 为起点,再沿 DynamoDBBackend → DynamoHandler 的定位路径核对具体实现与 docstring 中的限制说明;两者偶有滞后(如 2.4 节的export_table_to_point_in_time),以代码为准、并可在仓库中查看对应测试确认行为。 - 适用前提:以上结论均基于当前仓库代码;DynamoDB 功能清单随版本演进,升级 moto 后建议重新核对文档页与
IMPLEMENTATION_COVERAGE.md。
- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
相关推荐
moto 中 CodePipeline 服务的 Mock 实现:功能覆盖清单与源码级行为解析
moto 中 CodePipeline 服务的 Mock 实现:功能覆盖清单与源码级行为解析 本篇基于 moto 仓库中的服务文档 codepipeline.r
Mock测试moto 中 cognito-identity 服务 mock 全解:Identity Pool 操作覆盖、实现原理与测试实战
moto 中 cognito identity 服务 mock 全解:Identity Pool 操作覆盖、实现原理与测试实战 本文基于 moto 仓库中 do
Mock测试moto 中 DataBrew 的 Mock 实现:API 覆盖范围、源码结构与测试实战指南
moto 中 DataBrew 的 Mock 实现:API 覆盖范围、源码结构与测试实战指南 本篇以 moto 仓库的 DataBrew 服务文档 https:
Mock测试
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考