☰
moto 中的 AWS CodeDeploy 模拟:已实现 API 清单、源码实现解析与 boto3 测试实战
2026/9/25 2:41:15 网站建设 项目流程
  • Mock
  • 测试

【免费下载链接】moto

A library that allows you to easily mock out tests based on AWS infrastructure.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载

本文以 moto 官方服务文档codedeploy.rst的核心内容——CodeDeploy 服务已实现/未实现的 API 清单为骨架,结合 moto/codedeploy/models.py、moto/codedeploy/responses.py 等源码与 tests/test_codedeploy/test_codedeploy.py 测试用例,讲解 moto 如何 mock CodeDeploy 的 Application、DeploymentGroup、Deployment 三层资源模型,读者可以据此判断自己的 CodeDeploy 测试代码能否被 moto 覆盖,并直接复用可运行的测试代码与错误码断言。

文档定位:CodeDeploy 的 API 覆盖情况

moto 的服务文档位于 docs/docs/services/codedeploy.rst,它通过autoclass引用moto.codedeploy.models.CodeDeployBackend展示后端接口,并给出该服务的 "Implemented features for this service" 清单。根目录的 IMPLEMENTATION_COVERAGE.md 也给出了同一结论:CodeDeploy 约有 29% 的 API 已实现。

已实现的 16 个 API

以下 API 在当前仓库中均有对应实现,是编写本地测试可以依赖的能力边界:

API作用后端实现位置
create_application创建 CodeDeploy 应用(支持 Server / Lambda / ECS 计算平台与 tags)models.py
create_deployment_group为应用创建部署组(含服务角色、部署配置、蓝绿配置等字段)models.py
create_deployment在部署组下创建一次部署,revision 支持 S3 / GitHub / Stringmodels.py
get_application按名称查询应用详情models.py
batch_get_applications批量查询多个应用models.py
get_deployment按 deploymentId 查询部署详情models.py
batch_get_deployments批量查询部署models.py
get_deployment_group按应用名 + 部署组名查询部署组models.py
list_applications列出全部应用名(源码标注分页尚未实现)models.py
list_deployments按应用/部署组/状态过滤列出部署 IDmodels.py
list_deployment_groups列出指定应用下的部署组名models.py
tag_resource给资源 ARN 打标签models.py
untag_resource按 tag key 移除标签models.py
list_tags_for_resource查询资源 ARN 的标签列表models.py

尚未实现的 API(写测试前需要知道的限制)

以下 API 在文档清单中标记为未勾选(- [ ]),如果你的测试依赖它们,当前版本的 moto 无法模拟:

  • 部署配置管理:create_deployment_config、get_deployment_config、delete_deployment_config、list_deployment_configs;
  • 删除类操作:delete_application、delete_deployment_config、delete_deployment_group、delete_git_hub_account_token、delete_resources_by_external_id;
  • 修订版本管理:register_application_revision、get_application_revision、list_application_revisions、batch_get_application_revisions;
  • 实例级别查询:batch_get_deployment_instances、get_deployment_instance、list_deployment_instances、batch_get_deployment_groups、batch_get_deployment_targets、get_deployment_target、list_deployment_targets;
  • 混合/本地实例管理:add_tags_to_on_premises_instances、batch_get_on_premises_instances、get_on_premises_instance、list_on_premises_instances、register_on_premises_instance、deregister_on_premises_instance、remove_tags_from_on_premises_instances;
  • 部署生命周期控制:continue_deployment、stop_deployment、put_lifecycle_event_hook_execution_status、skip_wait_time_for_instance_termination;
  • 更新类操作:update_application、update_deployment_group;
  • 账户 Token 列表:list_git_hub_account_token_names。

此外,源码中 models.py 的list_applications与 models.py 的list_deployment_groups均带有TODO: implement pagination注释,list_deployments的externalId、createTimeRange参数虽然会被解析(见 responses.py),但从源码结构看,后端过滤逻辑目前主要实现了应用名/部署组名/状态过滤,时间范围过滤尚未生效——使用时应意识到这是与真实 AWS 行为的差异。

资源模型:Application / DeploymentGroup / Deployment 三层结构

从源码结构看,moto 的 CodeDeploy 模拟完整遵循了 AWS 的资源层级,定义在 moto/codedeploy/models.py 中:

  • Application(models.py#L23-L46):持有applicationId(UUID4 生成)、applicationName、computePlatform、createTime(ISO-8601 毫秒时间戳)。源码注释特别说明 Boto 文档说该字段是 datetime,而 AWS API 返回的是字符串数值格式,moto 按 AWS 行为用iso_8601_datetime_with_milliseconds()生成。
  • DeploymentGroup(models.py#L71-L142):字段覆盖很全,包括ec2TagFilters、onPremisesInstanceTagFilters、autoScalingGroups、serviceRoleArn、triggerConfigurations、alarmConfiguration、autoRollbackConfiguration、deploymentStyle、blueGreenDeploymentConfiguration、loadBalancerInfo、ecsServices、terminationHookEnabled等。其中targetRevision、lastSuccessfulDeployment、lastAttemptedDeployment在to_dict()中是# TODO占位空字典,意味着这些字段目前返回空结构而非真实数据。
  • DeploymentInfo(models.py#L145-L236):新建部署的status固定为"Created",startTime/completeTime初始为None,deploymentOverview各状态计数(Pending / InProgress / Succeeded / Failed / Skipped / Ready)初始全为 0,creator默认为"user"。

默认部署配置的回退逻辑

CodeDeployBackend.create_deployment_group中有一个值得注意的行为:如果调用方没有传deploymentConfigName,会回退到内置默认值CodeDeployDefault.OneAtATime(models.py#L416-L418)。该枚举(models.py#L49-L56)内置了 AWS 官方三种部署配置AllAtOnce、HalfAtATime、OneAtATime。

进一步看create_deployment(models.py#L351-L355):若部署时未显式指定deploymentConfigName,会继承所属部署组的配置名。因此在 mock 环境下get_deployment返回的deploymentConfigName会是字符串形式(枚举的__str__会返回CodeDeployDefault.OneAtATime这样的类限定名,测试中若断言该字段需留意这一点)。

校验规则与错误码

moto 对创建操作做了与 AWS 一致的语义校验,所有异常定义在 moto/codedeploy/exceptions.py,均为 400 状态的JsonRESTError:

触发场景异常类(错误码)
应用/部署/部署组不存在ApplicationDoesNotExistException、DeploymentDoesNotExistException、DeploymentGroupDoesNotExistException
重复创建同名应用/部署组ApplicationAlreadyExistsException、DeploymentGroupAlreadyExistsException
create_deployment缺少deploymentGroupNameDeploymentGroupNameRequiredException(models.py#L346-L349)
list_deployments只传了applicationName未传部署组名DeploymentGroupNameRequiredException(models.py#L467-L470)
list_deployments只传了deploymentGroupName未传应用名ApplicationNameRequiredException(models.py#L472-L475)

这些错误码是测试断言的直接依据,后文测试示例会用到。

请求路由:从 boto3 调用到后端的调用链

理解 mock 如何拦截请求,可以看 moto/codedeploy/urls.py:

  • URL 基础匹配https?://codedeploy\.(.+)\.amazonaws\.com,即 boto3 客户端发往codedeploy.<region>.amazonaws.com的 JSON 请求会被 moto 接管;
  • 大多数 API 走CodeDeployResponse.dispatch通用分发,按操作名路由到对应 handler;
  • list-tags-for-resource、tag-resource、untag-resource三个操作走独立的 URL 路径({0}/tag-resource$等),这是因为它们属于 CodeDeploy 的 REST JSON 协议而非 query 协议。

handler 实现位于 moto/codedeploy/responses.py:CodeDeployResponse的codedeploy_backend属性按current_account+ region 从codedeploy_backends(BackendDict(CodeDeployBackend, "codedeploy"),models.py#L531)取出后端实例。每个 handler 通过self._get_param(...)从 JSON body 中取参(如 create_deployment handler 解析了applicationName、deploymentGroupName、revision、description、autoRollbackConfiguration等全部参数),调用后端方法后以 JSON 字符串返回,Content-Type 固定为application/json。

测试实战:可复制运行的完整示例

以下示例全部取自仓库内 tests/test_codedeploy/test_codedeploy.py,配合@mock_aws装饰器即可本地运行,覆盖了已实现 API 的典型用法。

1. 创建应用与查询

import boto3 from moto import mock_aws @mock_aws def test_create_application(): client = boto3.client("codedeploy", region_name="ap-southeast-1") for platform in ["Server", "Lambda", "ECS"]: name = f"test-application-{platform}" response = client.create_application( applicationName=name, computePlatform=platform, tags=[{"Key": "Name", "Value": "Test"}], ) assert "applicationId" in response resp = client.get_application(applicationName=name) application = resp["application"] assert application["applicationId"] == response["applicationId"] assert application["applicationName"] == name assert application["computePlatform"] == platform assert "createTime" in application

重复创建同名应用会返回ApplicationAlreadyExistsException(对应 test_create_application_existing)。

2. 创建部署组与三种 revision 类型

create_deployment的revision参数是字典,revisionType可以是S3、GitHub、String三种,moto 会原样存储并在get_deployment时返回(对应 test_create_deployment_revision_s3、test_create_deployment_revision_github、test_get_deployment):

@mock_aws def test_create_deployment_revision_s3(): client = boto3.client("codedeploy", region_name="us-west-2") application_name = "mytestapp" deployment_group_name = "test-deployment-group" service_role_arn = "arn:aws:iam::123456789012:role/CodeDeployDemoRole" client.create_application(applicationName=application_name, computePlatform="Server") client.create_deployment_group( applicationName=application_name, deploymentGroupName=deployment_group_name, serviceRoleArn=service_role_arn, ) revision_S3 = { "revisionType": "S3", "s3Location": { "bucket": "my-bucket", "key": "my-key", "bundleType": "zip", "version": "1", "eTag": "my-etag", }, } response = client.create_deployment( applicationName=application_name, deploymentGroupName=deployment_group_name, revision=revision_S3, description="Test deployment", ) deployment_id = response["deploymentId"] response = client.get_deployment(deploymentId=deployment_id) assert response["deploymentInfo"]["revision"]["revisionType"] == "S3" assert ( response["deploymentInfo"]["revision"]["s3Location"] == revision_S3["s3Location"] ) assert response["deploymentInfo"]["status"] == "Created" assert "createTime" in response["deploymentInfo"]

3. 批量查询与列表操作

@mock_aws def test_batch_and_list(): client = boto3.client("codedeploy", region_name="us-east-2") # 批量获取应用 client.create_application(applicationName="sample_app1", computePlatform="Lambda") client.create_application(applicationName="sample_app2", computePlatform="Server") resp = client.batch_get_applications(applicationNames=["sample_app1", "sample_app2"]) assert len(resp["applicationsInfo"]) == 2 # 列出部署 service_role_arn = "arn:aws:iam::123456789012:role/CodeDeployDemoRole" client.create_deployment_group( applicationName="sample_app1", deploymentGroupName="dg-1", serviceRoleArn=service_role_arn, ) client.create_deployment( applicationName="sample_app1", deploymentGroupName="dg-1", revision={"revisionType": "String", "string": {"content": "c", "sha256": "s"}}, ) resp = client.list_deployments() assert len(resp["deployments"]) == 1 # 列出部署组 resp = client.list_deployment_groups(applicationName="sample_app1") assert resp["deploymentGroups"] == ["dg-1"]

list_deployments的参数配对校验同样值得在测试中固化(对应 test_list_deployments_group_required、test_list_deployments_app_required):

@mock_aws def test_list_deployments_validation(): client = boto3.client("codedeploy", region_name="ap-southeast-1") # 只给 applicationName,必须同时给 deploymentGroupName with pytest.raises(ClientError) as exc: client.list_deployments(applicationName="mytestapp") assert exc.value.response["Error"]["Code"] == "DeploymentGroupNameRequiredException" # 只给 deploymentGroupName,必须同时给 applicationName with pytest.raises(ClientError) as exc: client.list_deployments(deploymentGroupName="mygroupname") assert exc.value.response["Error"]["Code"] == "ApplicationNameRequiredException"

4. 标签:应用、部署组与部署的标签继承

标签功能基于通用TaggingService(models.py#L247),资源 ARN 格式为:

  • 应用:arn:aws:codedeploy:<region>:<account>:application:<name>
  • 部署组:arn:aws:codedeploy:<region>:<account>:deploymentgroup:<app>/<group>
  • 部署:arn:aws:codedeploy:<region>:<account>:deployment:<deploymentId>

一个有意思的行为是部署会继承部署组的标签:create_deployment时如果所属部署组带有 tags,会把同样的 tags 打到部署 ARN 上(models.py#L374-L379),该行为由 test_deployment_inherits_tags 验证:

@mock_aws def test_deployment_inherits_tags(): client = boto3.client("codedeploy", region_name="us-west-2") app_name = "test-tag-inherit-app" dg_name = "test-tag-inherit-group" service_role_arn = "arn:aws:iam::123456789012:role/CodeDeployDemoRole" client.create_application(applicationName=app_name, computePlatform="Server") client.create_deployment_group( applicationName=app_name, deploymentGroupName=dg_name, serviceRoleArn=service_role_arn, tags=[{"Key": "Environment", "Value": "Production"}], ) response = client.create_deployment( applicationName=app_name, deploymentGroupName=dg_name, revision={"revisionType": "S3", "s3Location": {"bucket": "b", "key": "k", "bundleType": "zip", "version": "1", "eTag": "e"}}, ) deployment_arn = ( f"arn:aws:codedeploy:us-west-2:123456789012:deployment:{response['deploymentId']}" ) response = client.list_tags_for_resource(ResourceArn=deployment_arn) assert response["Tags"][0]["Key"] == "Environment"

应用与部署组自身的标签查询分别对应 test_application_tagging 与 test_deployment_group_tagging。

使用前提与限制小结

  • 装饰器用法:所有示例均基于from moto import mock_aws+@mock_aws装饰器(context manager 或mock_aws.start()/stop()用法同理),与 docs/docs/getting_started.rst 介绍的一致;安装可用pip install 'moto[codedeploy]'或moto[all]。
  • 能力边界:只能模拟上文"已实现的 16 个 API",部署生命周期状态不会从Created自动流转(startTime、completeTime保持None),也不存在真实的 EC2/ASG 实例部署结果——如果你的测试断言部署最终变为Succeeded,需要自行模拟。
  • 未实现 API 会直接失败:调用清单中未勾选的 API(如stop_deployment、delete_application、create_deployment_config)时,由于 dispatch 找不到对应 handler 会抛异常,编写测试前先对照 docs/docs/services/codedeploy.rst 的清单。
  • 多区域隔离:后端按account + region存储(BackendDict(CodeDeployBackend, "codedeploy")),不同 region 的客户端看到的资源互不干扰,这与 test_list_deployments 等用例中显式指定region_name的写法一致。
  • Mock
  • 测试

【免费下载链接】moto

A library that allows you to easily mock out tests based on AWS infrastructure.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载
上一篇:go2rtc项目中流媒体连接中断问题的分析与解决
下一篇:致命BUG修复实录:Destiny 2 Solo Enabler布尔值解析异常深度排查与架构优化方案

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询