Moto 中的 AWS Resilience Hub 模拟：已实现 API 全景与 list_app_assessments 结果队列机制详解

Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载导读本文以 Moto 仓库中 resiliencehub 服务文档 为骨架全面梳理该开源模拟库对 AWS Resilience HubAWS 韧性评估服务的模拟覆盖范围。你会了解到当前已实现的 17 个 API 操作、App / AppVersion / AppComponent / Resource / ResiliencyPolicy 的底层数据模型以及文档中最具特色的能力——通过 Moto 专有 API 为list_app_assessments预配置评估结果队列的机制。读完本文你可以直接用 boto3 在测试中搭建 Resilience Hub 场景并掌握如何绕过不真正执行评估这一模拟限制。一、模拟覆盖总览哪些 API 已实现AWS Resilience Hub 用于创建韧性Resiliency应用、定义韧性策略、导入云资源并运行评估。Moto 在moto/resiliencehub/目录下实现了该服务的后端模型models.py、HTTP 响应层responses.py、URL 路由urls.py与异常定义exceptions.py并通过resiliencehub_backends BackendDict(ResilienceHubBackend, resiliencehub)注册到全局后端索引中。按文档中的实现清单当前已实现[X]的 17 个操作如下分类已实现操作备注应用管理create_app、describe_app、list_apps、list_app_versionslist_apps的FromAssessmentTime/ToAssessmentTime参数未实现应用版本与组件create_app_version_app_component、list_app_version_app_components、create_app_version_resource、list_app_version_resources、import_resources_to_draft_app_version、publish_app_version均基于 draft / release 版本模型韧性策略create_resiliency_policy、describe_resiliency_policy、list_resiliency_policiescreate_resiliency_policy对 policy 字段有枚举校验评估查询list_app_assessments默认返回空列表可用 Moto 专有 API 注入结果队列标签管理tag_resource、untag_resource、list_tags_for_resource基于 Moto 通用TaggingService未实现[ ]的操作包括delete_app、delete_app_assessment、start_app_assessment、describe_app_assessment、update_app、update_resiliency_policy、create_recommendation_template、各类list_*_recommendations、resolve_app_version_resources、list_metrics、start_metrics_export等——如果你的测试依赖这些删除/更新/评估执行类操作需要注意 Moto 当前并不提供对应行为。两个已实现操作存在明确的参数缺口文档中标注为 not yet implementedcreate_app与create_resiliency_policy的ClientToken 参数未实现即幂等令牌不生效多次调用会创建多个资源。list_apps的FromAssessmentTime / ToAssessmentTime 时间过滤参数未实现。这两个限制在源码中都有对应注释见 models.py 与 models.py属于已知的模拟边界。二、核心数据模型App、AppVersion、AppComponent、Resource 与 Policy要正确使用这些 API先理解 Moto 内部如何组织 Resilience Hub 的资源层次。从 models.py 可以梳理出如下对象关系1. App韧性应用App是顶层实体创建后自动生成一个 ARN格式为arn:{partition}:resiliencehub:{region}:{account_id}:app/{uuid}。Moto 对create_app的默认行为models.py包括assessmentSchedule未提供时默认DisabledcomplianceStatus固定为NotAssessed、driftStatus固定为NotChecked模拟数据resilienceScore初始为0.0status初始为Active创建 App 时会自动附带一个draft草稿版本AppVersionversion_nameNoneidentifier0响应 JSON 中的description、eventSubscriptions、permissionModel、policyArn仅在提供时才返回。describe_app会对不存在的 ARN 抛出ResourceNotFoundException错误信息形如App not found for appArn ...该行为在 exceptions.py 中定义并有对应测试验证见 tests/test_resiliencehub/test_resiliencehub.py 中test_describe_unknown_app。2. AppVersion应用版本版本模型models.py区分两种version_name为 None 时app_version标记为draft草稿版提供version_name时标记为release发布版。publish_app_versionmodels.py实现了发布逻辑调用时生成一个新的 release 版本并把旧 release 版本重命名为其数字标识符如1因此一个 App 的历史版本会累积为draft、release、1、2……测试test_publishtests/test_resiliencehub/test_resiliencyhub_resources.py完整验证了这一行为。3. AppComponent 与 Resource组件与物理资源AppComponent只有id、name、type三个字段通常以AWS::ResilienceHub::*AppComponent命名如DatabaseAppComponent、AppCommonAppComponent。Resource挂载在 AppVersion 下包含logicalResourceId、physicalResourceId响应中包装为{identifier: ...}、resourceType以及关联的组件列表。create_app_version_resource内部会按传入的appComponents名称过滤该版本的既有组件并关联到新资源传入未知组件名时不会报错而是返回空组件列表源码注释明确说明这是有意保持灵活见 tests/test_resiliencehub/test_resiliencyhub_resources.py 中test_create_app_version_resource_with_unknown_component。4. ResiliencyPolicy韧性策略create_resiliency_policy的policy参数是本次模拟中校验最严格的部分。从 responses.py 可见其校验逻辑合法的失败类型枚举为Software、Hardware、Region、AZ四种必须包含Software、Hardware、AZ三种缺任一都会抛出ValidationException错误信息形如FailureType HARDWARE does not exist每种失败类型下是{rpoInSecs: int, rtoInSecs: int}结构RPO/RTO 以秒为单位。对应测试 tests/test_resiliencehub/test_resiliencehub.py 中的valid_resiliency_policy常量给出了标准写法valid_resiliency_policy { Software: {rpoInSecs: 1, rtoInSecs: 1}, Hardware: {rpoInSecs: 2, rtoInSecs: 2}, AZ: {rpoInSecs: 3, rtoInSecs: 3}, }Policy模型还会生成独立 ARNarn:{partition}:resiliencehub:{region}:{account_id}:resiliency-policy/{uuid}并支持tier如NonCritical、dataLocationConstraint如AnyLocation、policyDescription等字段。三、list_app_assessments 的特殊机制Moto 专有结果队列 API这是 resiliencehub 文档中最有特色的内容也是 Moto 对无法真正执行评估这一模拟短板的补偿设计。1. 默认行为永远返回空列表由于 Moto 不会实际运行任何韧性评估list_app_assessments在未做任何配置时固定返回空列表。注意其参数处理逻辑responses.py请求中出现的appArn、assessmentName、assessmentStatus、complianceStatus、invoker、maxResults、nextToken、reverseOrder参数会被提取并按字典序序列化作为请求标识符request_identifier。也就是说不同的参数组合会被视为不同的请求用于区分结果队列中的不同条目。2. 通过 /moto-api/static 端点配置结果队列在进程内使用 Motomock_aws装饰器时可以通过 Moto 内置的motoapi.amazonaws.com主机发出 HTTP 请求向 Resilience Hub 后端写入一条评估结果队列。文档给出了完整的 Python 调用示例summary1 {appArn: app_arn1, appVersion: some version, ...} summary2 {appArn: app_arn2, ...} results {results: [[summary1, summary2], [summary2]], region: us-east-1} resp requests.post( http://motoapi.amazonaws.com/moto-api/static/resilience-hub-assessments/response, jsonresults, ) assert resp.status_code 201 client boto3.client(lambda, region_nameus-east-1) # First result resp client.list_app_assessments() # [summary1, summary2] # Second result resp client.list_app_assessments(assessmentStatusPending) # [summary2]请求体结构为{results: [[...], [...], ...], region: ...}其中results是多个结果批次组成的列表region指定该队列生效的区域。发送后返回 HTTP 201。从底层实现看该请求在 moto_api/_internal/urls.py 中被路由到response_instance.set_resilience_result最终由 moto_api/_internal/models.py 把结果追加到对应 region 的ResilienceHubBackend.app_assessments_queue中。3. 队列消费规则后端list_app_assessments的实现models.py遵循三条规则与文档描述完全一致同一参数组合的结果会被缓存第一次以某参数组合调用后结果存入app_assessments_results字典后续以完全相同参数调用永远返回同一个结果不会再次消耗队列。不同参数组合依次取队头换一组参数调用时从队列中弹出下一个结果批次返回依此类推。队列耗尽后返回空列表所有批次消费完后任何新参数组合都返回[]。以文档示例解释results [[summary1, summary2], [summary2]]表示预置了两个批次。第一次调用无参数取出[summary1, summary2]之后即使再次无参调用仍返回同一结果换用assessmentStatusPending调用取出第二个批次[summary2]第三次使用新参数组合时队列已空返回[]。这一点在测试 tests/test_resiliencehub/test_resiliencehub.py 的test_list_app_assessments中得到了完整验证包括相同参数重复调用结果不变不同参数依次消耗队列队列耗尽返回空结果队列按 region 隔离——只在ap-southeast-1配置的队列对us-east-1的客户端不生效返回空列表。4. MotoServer 模式下的端点地址如果使用 MotoServer 独立运行模式文档明确要求将请求发送到 MotoServer 实际监听的地址例如http://localhost:5000/moto-api/static/resilience-hub-assessments/response测试代码中对应逻辑是TEST_SERVER_MODE开启时使用settings.test_server_mode_endpoint()否则回退到http://motoapi.amazonaws.com见 tests/test_resiliencehub/test_resiliencehub.py。四、分页行为list_apps 与 list_resiliency_policieslist_apps与list_resiliency_policies通过 Moto 通用分页工具paginate实现分页分页模型定义在 models.py输入令牌参数nextToken限制参数maxResults每页默认上限 100 条以资源 ARN 作为唯一排序属性。list_apps还支持按name过滤、按appArn精确查询以及reverseOrderTrue倒序返回list_resiliency_policies支持按policyName过滤。这些行为均有测试覆盖见 tests/test_resiliencehub/test_resiliencehub.py 中test_list_apps与test_list_resiliency_policies测试还验证了maxResults/nextToken组合翻页以及末页不再返回nextToken的细节。五、标签管理与 Moto 通用 TaggingService 的集成三个标签操作全部基于 Moto 通用TaggingServicemodels.pytag_resource通过 HTTP POST/tags/{resourceArn}添加标签ARN 取自 URL 路径untag_resource通过 URL 查询参数tagKeys移除标签list_tags_for_resource返回该资源的全部标签字典。App 与 Policy 的to_json()都会实时调用list_tags_for_resource把当前标签写入响应。测试 tests/test_resiliencehub/test_resiliencyhub_tagging.py 验证了创建时打标签 → 追加标签 → 删除标签 → 列表查询的完整闭环且describe_app/describe_resiliency_policy返回的对象与创建时返回的对象完全一致。六、import_resources_to_draft_app_version 与组件自动创建import_resources_to_draft_app_versionmodels.py支持三类资源来源eksSourcesEKS 集群与命名空间列表sourceArns普通资源 ARN 列表terraformSourcesTerraform 状态文件s3StateFileUrl列表。导入的资源会追加到当前 App 的draft 版本上并且 Moto 会自动创建一个 id/name 均为appcommon、类型为AWS::ResilienceHub::AppCommonAppComponent的默认组件源码注释指出真实 AWS 会根据提供的来源创建更多组件Moto 仅模拟了这一默认组件。响应中的status固定为Pending。对应测试见 tests/test_resiliencehub/test_resiliencyhub_resources.py 中test_import_resources_to_draft_app_version。七、快速上手一份可运行的端到端示例结合上文下面是一段覆盖创建 App → 创建策略 → 导入资源 → 查询 → 配置评估队列完整流程的测试代码骨架基于 boto3 与 Motoimport boto3 import requests from moto import mock_aws mock_aws def test_resiliencehub_flow(): client boto3.client(resiliencehub, region_nameus-east-1) # 1. 创建韧性应用自动生成 draft 版本 app client.create_app( namemyapp, assessmentScheduleDaily, tags{env: test}, )[app] app_arn app[appArn] assert app[complianceStatus] NotAssessed # 2. 创建韧性策略必须包含 Software/Hardware/AZ policy client.create_resiliency_policy( policy{ Software: {rpoInSecs: 3600, rtoInSecs: 3600}, Hardware: {rpoInSecs: 7200, rtoInSecs: 7200}, AZ: {rpoInSecs: 10800, rtoInSecs: 10800}, }, policyNamepolname, tierNonCritical, )[policy] # 3. 向 draft 版本导入资源自动获得 appcommon 组件 resp client.import_resources_to_draft_app_version( appArnapp_arn, sourceArns[arn:aws:lambda:us-east-1:123456789012:function:my-fn], ) assert resp[status] Pending # 4. 为 list_app_assessments 预置结果队列Moto 专有 API summary1 {appArn: app_arn, appVersion: draft} resp requests.post( http://motoapi.amazonaws.com/moto-api/static/resilience-hub-assessments/response, json{results: [[summary1]], region: us-east-1}, ) assert resp.status_code 201 # 5. 查询评估结果默认空列表被队列结果替换 assert client.list_app_assessments()[assessmentSummaries] [summary1]运行前提安装当前仓库pip install -e .以及boto3、requests示例中创建策略时请确保policy完整包含Software、Hardware、AZ三类否则会收到ValidationException。八、总结与已知限制Moto 的 Resilience Hub 模拟覆盖了应用、版本、组件、资源、策略、标签这条主干链路足够支撑基于 boto3 的单元测试与本地集成测试。需要特别留意以下边界不执行真实评估list_app_assessments默认返回空评估结果完全依赖 Moto 专有队列 API 注入且队列按 region 隔离、按参数组合区分幂等令牌缺失create_app/create_resiliency_policy的 ClientToken 参数未实现时间过滤缺失list_apps的FromAssessmentTime/ToAssessmentTime未实现删除/更新类操作未实现delete_app、update_app、update_resiliency_policy、start_app_assessment等评估执行链路操作均不在当前支持范围内校验规则有选择唯一严格的入参校验是create_resiliency_policy的失败类型枚举与必填项检查源码位于 responses.py。以上行为均可通过阅读 models.py、responses.py 以及 tests/test_resiliencehub/ 下的三组测试用例进一步验证测试文件同时也是编写新用例时最直接的参考样例。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Moto 模拟 AWS Athena 服务指南查询执行、结果队列与已实现 API 全解析Moto 模拟 AWS Athena 服务指南查询执行、结果队列与已实现 API 全解析 导读 本文围绕 docs/docs/services/athena.Mock测试moto 中的 AWS FSx 模拟已实现 API 全景、源码结构与测试实战指南moto 中的 AWS FSx 模拟已实现 API 全景、源码结构与测试实战指南 本文基于 moto 仓库中 docs/docs/services/fsx.rMock测试moto 中的 AWS Inspector2 模拟接口覆盖范围、list_findings 结果队列与实现细节moto 中的 AWS Inspector2 模拟接口覆盖范围、list_findings 结果队列与实现细节 本文基于 moto 仓库中 docs/docsMock测试上一篇Requestly API Client终极指南隐私优先的Postman替代方案下一篇如何用 GitDagBundle 让 Airflow DAG 运行记录代码版本并支持重跑复用同一版本创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考