- Mock
- 测试
【免费下载链接】moto
A library that allows you to easily mock out tests based on AWS infrastructure.
本文以 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 / String | models.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 | 按应用/部署组/状态过滤列出部署 ID | models.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缺少deploymentGroupName | DeploymentGroupNameRequiredException(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.
相关推荐
moto 中的 CloudHSM V2 模拟:API 覆盖范围、后端实现与 mock 测试实战
moto 中的 CloudHSM V2 模拟:API 覆盖范围、后端实现与 mock 测试实战 本文基于 moto 仓库的服务文档 cloudhsmv2.rst
Mock测试Moto CodeBuild 模拟实战:在测试中 Mock AWS CodeBuild 项目与构建 API
Moto CodeBuild 模拟实战:在测试中 Mock AWS CodeBuild 项目与构建 API 本篇技术指南围绕 moto 仓库中 CodeBuil
Mock测试boto3单元测试模拟框架:moto vs localstack
boto3单元测试模拟框架:moto vs localstack 你是否还在为AWS服务单元测试发愁?本地环境搭建复杂、测试成本高、依赖外部网络?本文将对比两款
后端云原生
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考