Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本指南围绕 Moto 仓库中对 AWS Application Auto Scaling 服务的模拟支持展开系统梳理其已实现/未实现的 API 操作、可扩展目标Scalable Target、扩缩容策略Scaling Policy与定时操作Scheduled Action的完整用法并结合源码与测试给出可直接运行的 boto3 示例帮助你快速在本地测试、CI 中还原真实的 Auto Scaling 行为。一、服务概述与实现状态总览1.1 什么是 Application Auto ScalingAWS Application Auto Scaling 是一项托管服务用于自动调整可扩展资源的容量。典型场景包括ECS 服务ecs:service:DesiredCount的动态扩缩容DynamoDB 表的读写容量单位dynamodb:table:ReadCapacityUnits/WriteCapacityUnitsDynamoDB 全局二级索引dynamodb:index:*SageMaker 终端节点变体实例数sagemaker:variant:DesiredInstanceCountLambda 预置并发lambda:function:ProvisionedConcurrency等。Moto 对该服务进行模拟使开发者可以在不触碰真实 AWS 资源的前提下测试依赖 Auto Scaling 的应用逻辑。1.2 官方文档中的实现清单根据 docs/docs/services/application-autoscaling.rst 中列出的特性清单Moto 当前实现状态如下操作状态delete_scaling_policy✅ 已实现delete_scheduled_action✅ 已实现deregister_scalable_target✅ 已实现describe_scalable_targets✅ 已实现describe_scaling_activities❌ 未实现describe_scaling_policies✅ 已实现describe_scheduled_actions✅ 已实现get_predictive_scaling_forecast❌ 未实现list_tags_for_resource❌ 未实现put_scaling_policy✅ 已实现put_scheduled_action✅ 已实现register_scalable_target✅ 已实现tag_resource❌ 未实现untag_resource❌ 未实现文档同时明确指出Pagination is not yet implemented即分页尚未实现不过从源码看describe_scalable_targets与describe_scaling_policies已具备 NextToken 处理逻辑具体见下文分析。1.3 未实现操作的说明describe_scaling_activities无法查询扩容/缩容活动历史get_predictive_scaling_forecast预测式扩缩容Predictive Scaling的预测能力未模拟tag_resource/untag_resource/list_tags_for_resource标签管理操作暂不支持。同时put_scaling_policy仅支持StepScaling与TargetTrackingScaling两种策略类型源码中FakeApplicationAutoscalingPolicy对PredictiveScaling会直接抛出ValidationException见 moto/applicationautoscaling/models.py。二、源码结构与请求处理链路2.1 模块文件组成Moto 的 application-autoscaling 模拟模块位于 moto/applicationautoscaling/由以下文件组成文件职责models.py后端核心模型可扩展目标、扩缩容策略、定时操作及参数校验responses.py接收 HTTP 请求、解析参数、调用后端并序列化 JSON 响应urls.py路由注册匹配application-autoscaling.{region}.amazonaws.com端点exceptions.pyAWSValidationException异常定义utils.py从请求 URL 提取 region 的工具函数2.2 路由与请求分发在 urls.py 中URL 基址为url_bases [rhttps?://application-autoscaling\.(.)\.amazonaws.com]所有 POST 请求统一由ApplicationAutoScalingResponse.dispatch分发。响应类在 responses.py 中通过applicationautoscaling_backends[self.current_account][self.region]获取当前账号、当前区域的独立后端实例。这意味着不同账号、不同区域之间的数据彼此隔离与真实 AWS 的多区域行为一致。2.3 后端数据模型ApplicationAutoscalingBackendmodels.py内部维护三类内存数据self.targets: dict[str, dict[str, FakeScalableTarget]] OrderedDict() self.policies: dict[str, FakeApplicationAutoscalingPolicy] {} self.scheduled_actions: list[FakeScheduledAction] []targets按scalable_dimension分组、再按resource_id索引的嵌套字典因此同一个资源可以针对不同维度分别注册policies以服务命名空间\t资源ID\t可扩展维度\t策略名组合键存储formulate_key见 models.py确保策略定位唯一scheduled_actions按创建顺序追加的列表。每个FakeScalableTarget在创建时即生成形如arn:{partition}:application-autoscaling:{region}:{account_id}:scalable-target/{36位随机串}的 ARN见 models.py其中 partition 根据区域自动推导如aws/aws-cn/aws-us-gov。三、可扩展目标Scalable Target的注册与管理3.1 register_scalable_target注册或更新目标这是使用该服务的第一步将某个资源注册为可扩展目标并设定容量上下限。核心参数如下参数说明ServiceNamespace服务命名空间取值见下方枚举ResourceId资源标识格式因服务而异详见 3.3ScalableDimension可扩展维度标识要调整的容量属性MinCapacity/MaxCapacity容量下限/上限整数RoleARN扩缩容时使用的 IAM 角色 ARNSuspendedState暂停状态可包含DynamicScalingInSuspended、DynamicScalingOutSuspended、ScheduledScalingSuspended支持的服务命名空间定义于 models.pyappstream, rds, lambda, cassandra, dynamodb, custom-resource, elasticmapreduce, ec2, comprehend, ecs, sagemaker, kafka支持的典型可扩展维度部分完整枚举见 models.py维度值适用服务ecs:service:DesiredCountECS 服务期望实例数dynamodb:table:ReadCapacityUnits/WriteCapacityUnitsDynamoDB 表dynamodb:index:ReadCapacityUnits/WriteCapacityUnitsDynamoDB GSIrds:cluster:ReadReplicaCount/rds:cluster:CapacityRDS 集群lambda:function:ProvisionedConcurrencyLambda 预置并发sagemaker:variant:DesiredInstanceCountSageMaker 变体实例数ec2:spot-fleet-request:TargetCapacityEC2 Spot Fleetappstream:fleet:DesiredCapacityAppStream 队列elasticmapreduce:instancefleet:*/instancegroup:InstanceCountEMRcomprehend:document-classifier-endpoint:DesiredInferenceUnitsComprehendcassandra:table:ReadCapacityUnits/WriteCapacityUnitsKeyspaceskafka:broker-storage:VolumeSizeMSK broker 存储custom-resource:ResourceType:Property自定义资源基本示例ECSimport boto3 from moto import mock_aws mock_aws def test_register_ecs_target(): ecs boto3.client(ecs, region_nameus-east-1) ecs.create_cluster(clusterNamedefault) ecs.register_task_definition( familymy-task, containerDefinitions[ {name: hello, image: docker/hello-world:latest, cpu: 1024, memory: 400, essential: True} ], ) ecs.create_service( clusterdefault, serviceNamesample-webapp, taskDefinitionmy-task, desiredCount2, ) client boto3.client(application-autoscaling, region_nameus-east-1) resp client.register_scalable_target( ServiceNamespaceecs, ResourceIdservice/default/sample-webapp, ScalableDimensionecs:service:DesiredCount, MinCapacity1, MaxCapacity10, RoleARNarn:aws:iam::123456789012:role/ecs-autoscale, SuspendedState{ DynamicScalingInSuspended: True, DynamicScalingOutSuspended: False, ScheduledScalingSuspended: False, }, ) assert resp[ScalableTargetARN].startswith( arn:aws:application-autoscaling:us-east-1: )注意在 Moto 中注册 ECS 目标前必须先创建对应的 ECS 服务。后端会调用_ecs_service_exists_for_target检查服务是否存在否则抛出ValidationException: ECS service doesnt exist: ...见 models.py。对应的测试用例位于 tests/test_applicationautoscaling/test_validation.py。重复注册同一目标时行为是更新register_scalable_target会先调用_scalable_target_exists判断是否已存在若存在则仅更新min_capacity/max_capacity/suspended_state见 models.py。验证该行为的测试见 tests/test_applicationautoscaling/test_applicationautoscaling.py。3.2 describe_scalable_targets查询可扩展目标支持按ServiceNamespace、ResourceIds列表与ScalableDimension过滤。从源码看models.py先按命名空间展平全部目标_flatten_scalable_targets若传了ScalableDimension只保留维度匹配者若传了ResourceIds只保留 ID 命中者。响应层responses.py默认MaxResults50并支持NextToken分页翻页返回字段包含CreationTime、MaxCapacity、MinCapacity、ResourceId、RoleARN、ScalableDimension、ServiceNamespace、ScalableTargetARN、SuspendedState。示例resp client.describe_scalable_targets(ServiceNamespaceecs) for t in resp[ScalableTargets]: print(t[ResourceId], t[ScalableDimension], t[MinCapacity], t[MaxCapacity])3.3 ResourceId 的格式差异Moto 的_get_resource_type_from_resource_idmodels.py与_target_params_are_validmodels.py共同保证了 resource_id、namespace、dimension 三者的一致性校验。其规则是dimension形如namespace:resource_type:属性其中resource_type通常取自resource_id的第一段但存在例外SageMaker 端点endpoint/MyEndPoint/variant/MyVariantresource_type 为第三段variantDynamoDB GSItable/my-table/index/my-table-indexresource_type 为第三段indexKeyspaces 表keyspace/mykeyspace/table/mytableresource_type 为第三段tableComprehend直接使用 ARNresource_type 取 ARN 的最后一个:分段MSKkafka:broker-storage属于例外resource_type 不取自 resource_id。以下组合在测试中被验证有效见 tests/test_applicationautoscaling/test_applicationautoscaling.py# (namespace, resource_id, scalable_dimension) (ecs, service/default/sample-webapp, ecs:service:DesiredCount) (ec2, spot-fleet-request/sfr-73fbd2ce-aa30-494c-8788-1cee4EXAMPLE, ec2:spot-fleet-request:TargetCapacity) (elasticmapreduce, instancegroup/j-2EEZNYKUA1NTV/ig-1791Y4E1L8YI0, elasticmapreduce:instancegroup:InstanceCount) (appstream, fleet/sample-fleet, appstream:fleet:DesiredCapacity) (dynamodb, table/my-table, dynamodb:table:ReadCapacityUnits) (dynamodb, table/my-table/index/my-table-index, dynamodb:index:ReadCapacityUnits) (rds, cluster:my-db-cluster, rds:cluster:ReadReplicaCount) (sagemaker, endpoint/MyEndPoint/variant/MyVariant, sagemaker:variant:DesiredInstanceCount) (comprehend, arn:aws:comprehend:us-west-2:123456789012:document-classifier-endpoint/EXAMPLE, comprehend:document-classifier-endpoint:DesiredInferenceUnits) (lambda, function:my-function:prod, lambda:function:ProvisionedConcurrency) (cassandra, keyspace/mykeyspace/table/mytable, cassandra:table:ReadCapacityUnits) (custom-resource, https://test-endpoint.amazon.com/ScalableDimension/test-resource, custom-resource:ResourceType:Property)3.4 deregister_scalable_target注销目标删除指定目标若目标不存在则抛出ValidationException: No scalable target found for service namespace: ..., resource ID: ..., scalable dimension: ...见 models.py。client.deregister_scalable_target( ServiceNamespaceecs, ResourceIdservice/default/sample-webapp, ScalableDimensionecs:service:DesiredCount, )四、扩缩容策略Scaling Policy的完整生命周期4.1 put_scaling_policy创建或更新策略支持StepScaling与TargetTrackingScaling两种策略类型分别通过StepScalingPolicyConfiguration与TargetTrackingScalingPolicyConfiguration传入策略体。策略类型配置结构示例TargetTrackingScaling{TargetValue: 70.0, PredefinedMetricSpecification: {PredefinedMetricType: DynamoDBReadCapacityUtilization}}StepScaling{AdjustmentType: ChangeInCapacity, StepAdjustments: [{ScalingAdjustment: 10}], MinAdjustmentMagnitude: 2}在 models.py 中FakeApplicationAutoscalingPolicy会根据policy_type将策略体分别存入step_scaling_policy_configuration或target_tracking_scaling_policy_configuration传入其他类型如PredictiveScaling或非法字符串会抛出1 validation error detected: Value {policy_type} at policyType failed to satisfy constraint: Member must satisfy enum value set: [PredictiveScaling, StepScaling, TargetTrackingScaling]该行为由 tests/test_applicationautoscaling/test_applicationautoscaling_policies.py 验证。关键联动自动创建 CloudWatch 告警。对于TargetTrackingScaling策略Moto 会自动在 CloudWatch 后端创建告警create_alarms见 models.py命名空间dynamodb创建 4 个告警AlarmHigh/AlarmLow/ProvisionedCapacityHigh/ProvisionedCapacityLow监控ConsumedReadCapacityUnits与ProvisionedReadCapacityUnits见 models.py命名空间ecs创建 2 个告警依据PredefinedMetricType是否包含Memory选择MemoryUtilization或CPUUtilization见 models.py。这些告警会出现在put_scaling_policy的响应Alarms字段中也可通过describe_scaling_policies查询到并能通过 CloudWatch 的describe_alarms直接读取——测试用例 tests/test_applicationautoscaling/test_applicationautoscaling_policies.py 对此做了交叉验证。client.put_scaling_policy( PolicyNamedynamodb-scale-out, ServiceNamespacedynamodb, ResourceIdtable/my-table, ScalableDimensiondynamodb:table:ReadCapacityUnits, PolicyTypeTargetTrackingScaling, TargetTrackingScalingPolicyConfiguration{ TargetValue: 70.0, PredefinedMetricSpecification: { PredefinedMetricType: DynamoDBReadCapacityUtilization }, }, )4.2 describe_scaling_policies查询策略支持按ServiceNamespace必填、PolicyNames、ResourceId、ScalableDimension过滤并支持MaxResults与NextToken分页默认每页 100见 models.py。返回字段包括PolicyARN、PolicyName、ServiceNamespace、ResourceId、ScalableDimension、PolicyType、CreationTime、Alarms并依据策略类型附带StepScalingPolicyConfiguration或TargetTrackingScalingPolicyConfiguration见 responses.py。策略 ARN 格式为arn:aws:autoscaling:{region}:{account_id}:scalingPolicy:{uuid}:resource/{namespace}/{resource_id}:policyName/{policy_name}4.3 delete_scaling_policy删除策略按PolicyName、ServiceNamespace、ResourceId、ScalableDimension定位并删除删除时会同步清理自动创建的 CloudWatch 告警delete_alarms见 models.py。若策略不存在抛出ValidationException: No scaling policy found for ...。五、定时操作Scheduled Action的使用5.1 put_scheduled_action创建定时操作用于按Schedulecron 或 rate 表达式定期调整目标容量。参数参数说明ServiceNamespace服务命名空间ScheduledActionName操作名称ResourceId目标资源ScalableDimension维度Schedule如rate(2 minutes)、cron(0 18 * * ? *)Timezone时区如UTCStartTime/EndTime生效起止时间ScalableTargetAction包含MinCapacity/MaxCapacity的容量动作示例基于 tests/test_applicationautoscaling/test_applicationautoscaling.pyclient.register_scalable_target( ServiceNamespacedynamodb, ResourceIdtable/my-table, ScalableDimensiondynamodb:table:ReadCapacityUnits, MinCapacity1, MaxCapacity100, ) client.put_scheduled_action( ServiceNamespacedynamodb, Schedulerate(2 minutes), ScheduledActionNameaction_name, ResourceIdtable/my-table, ScalableDimensiondynamodb:table:ReadCapacityUnits, ScalableTargetAction{MinCapacity: 1, MaxCapacity: 5}, )重复对相同(namespace, action_name, resource_id, dimension)调用put_scheduled_action会更新既有操作而非新增见 models.py 及测试 tests/test_applicationautoscaling/test_applicationautoscaling.py。生成的操作 ARN 格式为arn:aws:autoscaling:{region}:{account_id}:scheduledAction:{namespace}/{resource_id}:scheduledActionName/{action_name}注意Moto 仅记录定时操作元数据并不会真的在调度时刻触发扩缩容动作。5.2 describe_scheduled_actions查询定时操作按ServiceNamespace必填过滤可选ScheduledActionNames、ResourceId、ScalableDimension进一步筛选。源码中明确注释Pagination is not yet implemented见 models.py。返回字段ScheduledActionName、ScheduledActionARN、ServiceNamespace、Schedule、Timezone、ResourceId、ScalableDimension、StartTime、EndTime、CreationTime、ScalableTargetAction。5.3 delete_scheduled_action删除定时操作按四个维度定位并删除若不存在则静默成功不抛错见 models.py。六、参数校验规则与常见报错6.1 双层级联校验Moto 对该服务实现了两道校验响应层校验_validate_params见 responses.py在分发前校验ServiceNamespace与ScalableDimension是否属于合法枚举非法时抛出形如1 validation error detected: Value foo at serviceNamespace failed to satisfy constraint: ...的ValidationExceptionHTTP 400模型层校验_target_params_are_valid见 models.py注册/更新目标时校验命名空间、维度与 resource_id 的一致性如ecs命名空间必须配ecs:service:DesiredCount维度和service/...形式的资源 ID。6.2 常见错误对照场景错误码说明非法ScalableDimensionValidationException如foo报 1 个校验错误非法ServiceNamespaceValidationException如foo报 1 个校验错误两者均非法ValidationException报 2 个校验错误2 validation errors detectedECS 服务不存在ClusterNotFoundException注册 ECS 目标时底层 ECS 后端抛出的Cluster not found.删除不存在的目标/策略ValidationException如No scalable target found for .../No scaling policy found for ...非法 PolicyTypeValidationException枚举限制为[PredictiveScaling, StepScaling, TargetTrackingScaling]相关验证测试集中在 tests/test_applicationautoscaling/test_validation.py。七、与其他服务的联动与注意事项7.1 与 ECS 的联动后端初始化时即持有同账号同区域的 ECS 后端引用self.ecs_backend ecs_backends[account_id][region_name]见 models.py。这意味着注册 ECS 目标前必须先存在对应的 ECS 服务否则报ECS service doesnt existTargetTracking 策略会自动为 ECS 服务创建基于CPUUtilization/MemoryUtilization的 CloudWatch 告警告警维度为ClusterName与ServiceName。7.2 与 CloudWatch 的联动FakeApplicationAutoscalingPolicy通过cloudwatch_backends[account_id][region_name]直接操作 CloudWatch 后端因此在mock_aws上下文中创建策略后可以立刻通过 CloudWatch 客户端查询到对应告警删除策略时告警也随之删除。这在测试真实告警依赖逻辑时非常有用。7.3 分页实现的现状文档声明分页未实现但源码表明两个查询接口已具备分页能力describe_scalable_targets响应层默认MaxResults50支持NextTokenresponses.py测试见 tests/test_applicationautoscaling/test_applicationautoscaling.pydescribe_scaling_policies后端默认每页 100 条并返回NextTokenmodels.py。而describe_scheduled_actions确实未实现分页。7.4 区域与账号隔离后端通过BackendDict(ApplicationAutoscalingBackend, application-autoscaling)注册models.py数据按account_id region隔离。使用不同 region 的客户端如eu-west-1、ap-southeast-1看到的将是彼此独立的数据集合测试中对此有覆盖如 tests/test_applicationautoscaling/test_applicationautoscaling.py。八、测试环境快速上手8.1 启用方式在测试中使用mock_aws装饰器即可无需任何额外配置from moto import mock_aws import boto3 mock_aws def test_auto_scaling_flow(): # ... 创建 ECS 服务 / DynamoDB 表 ... client boto3.client(application-autoscaling, region_nameus-east-1) client.register_scalable_target(...) client.put_scaling_policy(...) client.put_scheduled_action(...) assert len(client.describe_scalable_targets(ServiceNamespaceecs)[ScalableTargets]) 18.2 既有测试参考仓库中已有完整测试可作参考tests/test_applicationautoscaling/test_applicationautoscaling.py目标注册/查询/更新、策略、定时操作及分页tests/test_applicationautoscaling/test_applicationautoscaling_policies.py策略类型校验与 CloudWatch 告警联动tests/test_applicationautoscaling/test_validation.py参数校验行为。其中application_autoscaling_aws_verified装饰器tests/test_applicationautoscaling/init.py允许在设置MOTO_TEST_ALLOW_AWS_REQUESTtrue时直接对真实 AWS 运行验证性测试未设置时则在mock_aws上下文中运行可作为与真实行为对照的依据。8.3 局限提醒在 Moto 中使用该服务时请留意以下限制describe_scaling_activities、get_predictive_scaling_forecast、标签相关操作未实现put_scaling_policy不支持PredictiveScaling类型定时操作只记录元数据不会在计划时间点真实触发容量变更TargetTracking 策略创建的 CloudWatch 告警为 Moto 自动生成的模拟告警其阈值如 DynamoDB 的 42.0/30.0为内部实现值不应将其当作真实 AWS 的告警阈值。九、小结Moto 对 AWS Application Auto Scaling 的模拟覆盖了核心闭环注册可扩展目标 → 配置扩缩容策略 → 创建定时操作 → 查询与清理并在 ECS、DynamoDB 场景下与 CloudWatch 告警产生真实联动足以支撑大多数本地与 CI 场景下的扩缩容逻辑测试。尚未实现的接口活动历史、预测式扩缩容、标签管理在使用时应提前规避。结合本文给出的参数表格、资源 ID 格式与测试用例你可以快速在自己的项目中复现并验证 Auto Scaling 相关行为。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐如何快速定位应用卡顿Perfetto 性能追踪与 trace 分析入门如何快速定位应用卡顿Perfetto 性能追踪与 trace 分析入门 App 偶尔卡一下你复现不了一抓日志现象就没了。Perfetto 给到的是时间轴可观测性后端开发工具前端数据可视化tchMaterial-parser 电子课本下载tchMaterial parser 电子课本下载 tchMaterial parser 是一个基于 Python Tkinter 的桌面小工具把国家中小学智网页爬虫教育aws-cli 实操指南用 aws autoscaling update-auto-scaling-group 动态调整 Auto Scaling 组配置aws cli 实操指南用 aws autoscaling update auto scaling group 动态调整 Auto Scaling 组配置 导开发工具云原生运维上一篇SVGnest终极指南材料切割优化利器完全解析下一篇terraform-provider-aws 数据源实战用 aws_connect_vocabulary 查询 Amazon Connect 自定义词汇表创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
