- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
本指南围绕 Moto 仓库中对 AWS Application Auto Scaling 服务的模拟支持展开,系统梳理其已实现/未实现的 API 操作、可扩展目标(Scalable Target)、扩缩容策略(Scaling Policy)与定时操作(Scheduled Action)的完整用法,并结合源码与测试给出可直接运行的 boto3 示例,帮助你快速在本地测试、CI 中还原真实的 Auto Scaling 行为。
一、服务概述与实现状态总览
1.1 什么是 Application Auto Scaling
AWS Application Auto Scaling 是一项托管服务,用于自动调整可扩展资源的容量。典型场景包括:
- ECS 服务(
ecs:service:DesiredCount)的动态扩缩容; - DynamoDB 表的读写容量单位(
dynamodb:table:ReadCapacityUnits/WriteCapacityUnits); - DynamoDB 全局二级索引(
dynamodb:index:*); - SageMaker 终端节点变体实例数(
sagemaker:variant:DesiredInstanceCount); - Lambda 预置并发(
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.py | AWSValidationException异常定义 |
| utils.py | 从请求 URL 提取 region 的工具函数 |
2.2 路由与请求分发
在 urls.py 中,URL 基址为:
url_bases = [r"https?://application-autoscaling\.(.+)\.amazonaws.com"]所有 POST 请求统一由ApplicationAutoScalingResponse.dispatch分发。响应类在 responses.py 中,通过applicationautoscaling_backends[self.current_account][self.region]获取当前账号、当前区域的独立后端实例。这意味着不同账号、不同区域之间的数据彼此隔离,与真实 AWS 的多区域行为一致。
2.3 后端数据模型
ApplicationAutoscalingBackend(models.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.3) |
ScalableDimension | 可扩展维度,标识要调整的容量属性 |
MinCapacity/MaxCapacity | 容量下限/上限(整数) |
RoleARN | 扩缩容时使用的 IAM 角色 ARN |
SuspendedState | 暂停状态,可包含DynamicScalingInSuspended、DynamicScalingOutSuspended、ScheduledScalingSuspended |
支持的服务命名空间(定义于 models.py):
appstream, rds, lambda, cassandra, dynamodb, custom-resource, elasticmapreduce, ec2, comprehend, ecs, sagemaker, kafka支持的典型可扩展维度(部分,完整枚举见 models.py):
| 维度值 | 适用服务 |
|---|---|
ecs:service:DesiredCount | ECS 服务期望实例数 |
dynamodb:table:ReadCapacityUnits/WriteCapacityUnits | DynamoDB 表 |
dynamodb:index:ReadCapacityUnits/WriteCapacityUnits | DynamoDB GSI |
rds:cluster:ReadReplicaCount/rds:cluster:Capacity | RDS 集群 |
lambda:function:ProvisionedConcurrency | Lambda 预置并发 |
sagemaker:variant:DesiredInstanceCount | SageMaker 变体实例数 |
ec2:spot-fleet-request:TargetCapacity | EC2 Spot Fleet |
appstream:fleet:DesiredCapacity | AppStream 队列 |
elasticmapreduce:instancefleet:*/instancegroup:InstanceCount | EMR |
comprehend:document-classifier-endpoint:DesiredInferenceUnits | Comprehend |
cassandra:table:ReadCapacityUnits/WriteCapacityUnits | Keyspaces |
kafka:broker-storage:VolumeSize | MSK broker 存储 |
custom-resource:ResourceType:Property | 自定义资源 |
基本示例(ECS)
import boto3 from moto import mock_aws @mock_aws def test_register_ecs_target(): ecs = boto3.client("ecs", region_name="us-east-1") ecs.create_cluster(clusterName="default") ecs.register_task_definition( family="my-task", containerDefinitions=[ {"name": "hello", "image": "docker/hello-world:latest", "cpu": 1024, "memory": 400, "essential": True} ], ) ecs.create_service( cluster="default", serviceName="sample-webapp", taskDefinition="my-task", desiredCount=2, ) client = boto3.client("application-autoscaling", region_name="us-east-1") resp = client.register_scalable_target( ServiceNamespace="ecs", ResourceId="service/default/sample-webapp", ScalableDimension="ecs:service:DesiredCount", MinCapacity=1, MaxCapacity=10, RoleARN="arn: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 doesn't 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)默认MaxResults=50,并支持NextToken分页翻页,返回字段包含CreationTime、MaxCapacity、MinCapacity、ResourceId、RoleARN、ScalableDimension、ServiceNamespace、ScalableTargetARN、SuspendedState。
示例
resp = client.describe_scalable_targets(ServiceNamespace="ecs") for t in resp["ScalableTargets"]: print(t["ResourceId"], t["ScalableDimension"], t["MinCapacity"], t["MaxCapacity"])3.3 ResourceId 的格式差异
Moto 的_get_resource_type_from_resource_id(models.py)与_target_params_are_valid(models.py)共同保证了 resource_id、namespace、dimension 三者的一致性校验。其规则是:dimension形如namespace:resource_type:属性,其中resource_type通常取自resource_id的第一段,但存在例外:
- SageMaker 端点:
endpoint/MyEndPoint/variant/MyVariant,resource_type 为第三段variant; - DynamoDB GSI:
table/my-table/index/my-table-index,resource_type 为第三段index; - Keyspaces 表:
keyspace/mykeyspace/table/mytable,resource_type 为第三段table; - Comprehend:直接使用 ARN,resource_type 取 ARN 的最后一个
:分段; - MSK:
kafka: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( ServiceNamespace="ecs", ResourceId="service/default/sample-webapp", ScalableDimension="ecs: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( PolicyName="dynamodb-scale-out", ServiceNamespace="dynamodb", ResourceId="table/my-table", ScalableDimension="dynamodb:table:ReadCapacityUnits", PolicyType="TargetTrackingScaling", 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:创建定时操作
用于按Schedule(cron 或 rate 表达式)定期调整目标容量。参数:
| 参数 | 说明 |
|---|---|
ServiceNamespace | 服务命名空间 |
ScheduledActionName | 操作名称 |
ResourceId | 目标资源 |
ScalableDimension | 维度 |
Schedule | 如rate(2 minutes)、cron(0 18 * * ? *) |
Timezone | 时区(如UTC) |
StartTime/EndTime | 生效起止时间 |
ScalableTargetAction | 包含MinCapacity/MaxCapacity的容量动作 |
示例(基于 tests/test_applicationautoscaling/test_applicationautoscaling.py):
client.register_scalable_target( ServiceNamespace="dynamodb", ResourceId="table/my-table", ScalableDimension="dynamodb:table:ReadCapacityUnits", MinCapacity=1, MaxCapacity=100, ) client.put_scheduled_action( ServiceNamespace="dynamodb", Schedule="rate(2 minutes)", ScheduledActionName="action_name", ResourceId="table/my-table", ScalableDimension="dynamodb: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: ...的ValidationException(HTTP 400); - 模型层校验(
_target_params_are_valid,见 models.py):注册/更新目标时校验命名空间、维度与 resource_id 的一致性(如ecs命名空间必须配ecs:service:DesiredCount维度和service/...形式的资源 ID)。
6.2 常见错误对照
| 场景 | 错误码 | 说明 |
|---|---|---|
非法ScalableDimension | ValidationException | 如"foo",报 1 个校验错误 |
非法ServiceNamespace | ValidationException | 如"foo",报 1 个校验错误 |
| 两者均非法 | ValidationException | 报 2 个校验错误(2 validation errors detected) |
| ECS 服务不存在 | ClusterNotFoundException | 注册 ECS 目标时底层 ECS 后端抛出的Cluster not found. |
| 删除不存在的目标/策略 | ValidationException | 如No scalable target found for .../No scaling policy found for ... |
| 非法 PolicyType | ValidationException | 枚举限制为[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 doesn't exist); - TargetTracking 策略会自动为 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:响应层默认MaxResults=50,支持NextToken(responses.py),测试见 tests/test_applicationautoscaling/test_applicationautoscaling.py;describe_scaling_policies:后端默认每页 100 条并返回NextToken(models.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_name="us-east-1") client.register_scalable_target(...) client.put_scaling_policy(...) client.put_scheduled_action(...) assert len(client.describe_scalable_targets(ServiceNamespace="ecs")["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_REQUEST=true时直接对真实 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
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
相关推荐
如何快速定位应用卡顿: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 组配置 导
开发工具云原生运维
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考