企业官网建设流程全解析

• Mock
• 测试

本指南以 Moto 仓库中 batch 服务文档 为基础，系统讲解如何用 Moto 在本地测试环境中模拟 AWS Batch 的核心 API：计算环境（Compute Environment）、作业队列（Job Queue）、作业定义（Job Definition）、作业提交与状态流转、调度策略（Scheduling Policy）以及资源标签管理。读完本文，你将掌握@mock_aws装饰器下 Batch 服务的真实行为边界、Docker 与无 Docker（use_docker=False）两种后端模式的区别，以及各 API 当前实现与未实现项，可直接用于编写可复现的 AWS Batch 单元测试。

一、文档概览：batch 服务的实现范围清单

仓库中 batch.rst 是 Moto 各服务“实现清单”（Implementation Coverage）的标准文档：以[X]标记已实现、[ ]标记未实现的 Batch API，并注明每个接口的已知限制。截至当前仓库版本，已实现的核心接口包括：

• 计算环境：create_compute_environment、describe_compute_environments、update_compute_environment、delete_compute_environment
• 作业队列：create_job_queue、describe_job_queues、update_job_queue、delete_job_queue
• 作业定义：register_job_definition、deregister_job_definition、describe_job_definitions
• 作业执行：submit_job、describe_jobs、list_jobs、cancel_job、terminate_job
• 调度策略：create_scheduling_policy、describe_scheduling_policies、list_scheduling_policies、update_scheduling_policy、delete_scheduling_policy
• 标签管理：tag_resource、untag_resource、list_tags_for_resource

尚未实现（[ ]）的接口包括create_consumable_resource、create_quota_share、create_service_environment、describe_consumable_resource、describe_quota_share、describe_service_environments、describe_service_job、get_job_queue_snapshot、list_consumable_resources、list_jobs_by_consumable_resource、list_quota_shares、list_service_jobs、submit_service_job、terminate_service_job、update_consumable_resource、update_quota_share、update_service_environment、update_service_job等一批面向新 AWS 特性的接口，写作测试时应避免依赖这些能力。

文档同时标注了两类已知限制，使用时需特别注意：

1. 分页未实现（Pagination is not yet implemented）：describe_compute_environments、describe_job_definitions、describe_job_queues、list_scheduling_policies等接口均不支持分页参数，会一次性返回全部结果。
2. submit_job的RetryStrategy与Parameters参数未实现：提交作业时传入这两个参数不会按 AWS 语义生效。
3. list_jobs的限制：文档明确说明“根据 Boto3 文档，按数组作业（array job）ID 过滤时不支持 filters”，且当前实现不区分数组作业列表与普通作业列表（对应源码 list_jobs 中的 TODO 注释）。

二、快速上手：第一个@mock_awsBatch 测试

与 Moto 其他服务一样，Batch 的模拟通过moto.mock_aws装饰器（或上下文管理器）开启。最基础的可用性测试如下（参考 tests/test_batch/test_batch.py 的多区域验证方式）：

import boto3 from moto import mock_aws @mock_aws def test_batch_regions(): client = boto3.client("batch", region_name="us-west-2") resp = client.describe_jobs(jobs=[""]) assert resp["ResponseMetadata"]["HTTPStatusCode"] == 200

describe_jobs(jobs=[""])在 Moto 中是合法调用并返回 200，说明后端对空作业过滤集合做了安全处理（源码见 describe_jobs，空列表时返回全部作业）。这样的冒烟测试适合验证 Batch mock 已正确装配。

三、核心后端：BatchBackend 与依赖服务联动

Batch 是 Moto 中耦合度较高的服务之一。从 BatchBackend 的类定义可以看到，它通过property直接引用四个兄弟服务的 backend：

• iam_backend：校验计算环境的serviceRole、instanceRole是否存在（源码）
• ec2_backend：为 MANAGED 计算环境创建/终止 EC2 实例、校验子网与安全组（源码）
• ecs_backend：为每个计算环境创建 ECS 集群（源码）
• logs_backend：把作业容器日志写入 CloudWatch Logs（源码）

因此，在你的测试里如果同时使用 IAM、EC2、ECS、Logs 的 mock（例如通过@mock_aws全量开启），Batch 会与它们协同工作；如果只单独 mock Batch，创建 MANAGED 计算环境时依赖的 EC2/ECS 资源可能缺失，这是编写集成测试时需要注意的前提条件。

后端在内存中维护五个主要对象字典（源码）：

self._compute_environments: dict[str, ComputeEnvironment] = {} self._job_queues: dict[str, JobQueue] = {} self._job_definitions: dict[str, JobDefinition] = {} self._jobs: dict[str, Job] = {} self._scheduling_policies: dict[str, SchedulingPolicy] = {}

所有资源均按account_id + region隔离，通过BackendDict(BatchBackend, "batch")注册（moto/batch/models.py），与 Moto 的多账户/多区域机制保持一致。

3.1 资源命名与 ARN 规范

创建资源前，后端会用正则校验名称合法性（moto/batch/models.py）：

• 计算环境名：^[A-Za-z0-9][A-Za-z0-9_-]{1,126}[A-Za-z0-9]$
• 作业名：^[A-Za-z0-9][A-Za-z0-9_-]{1,127}$

各资源的 ARN 由 moto/batch/utils.py 统一生成，格式如下：

partition 依据区域自动推导（如aws、aws-cn），因此cn-northwest-1等区域也能得到合法 ARN。

四、计算环境（Compute Environment）：创建、校验与销毁

4.1 创建参数与校验规则

create_compute_environment的核心参数为：computeEnvironmentName、type（MANAGED/UNMANAGED）、state（ENABLED/DISABLED）、computeResources、serviceRole、tags。后端执行一连串校验（create_compute_environment）：

• 名称必须匹配正则，且不能与已有环境重名；
• serviceRole必填，且必须能在 IAM 后端中找到对应角色（否则抛InvalidParameterValueException）；
• type必须是MANAGED | UNMANAGED，state必须是ENABLED | DISABLED；
• MANAGED 类型必须提供computeResources。

computeResources子字典的校验集中在 _validate_compute_resources，要点如下：

• maxvCpus必须为正数；非 FARGATE 类型还需instanceRole（必须是已存在的 IAM 实例配置文件）、minvCpus（≥ 0）且maxvCpus >= minvCpus；
• instanceTypes至少 1 个，且每个类型必须存在于 EC2 实例类型表EC2_INSTANCE_TYPES/EC2_INSTANCE_FAMILIES中（"optimal"关键字放行）；
• securityGroupIds与subnets至少各 1 个，且必须真实存在于 EC2 后端；
• computeResources.type必须是EC2 | SPOT | FARGATE | FARGATE_SPOT之一。

4.2 MANAGED 环境的实例与集群联动

校验通过后，MANAGED 且非 FARGATE 的环境会触发两件副作用（源码）：

1. 调用find_min_instances_to_meet_vcpus（源码）根据desiredvCpus（缺省回退到minvCpus）贪心挑选最少 EC2 实例，并通过ec2_backend.run_instances真正创建实例；"optimal"会映射为m4.4xlarge；
2. 通过ecs_backend.create_cluster创建一个名为OnDemand_Batch_{uuid}的 ECS 集群并绑定到环境。

因此，一个完整的 MANAGED 计算环境测试会同时在 EC2 与 ECS 后端留下资源痕迹——这也解释了为什么 Batch 测试通常搭配@mock_aws全量 mock。FARGATE 类型不会创建 EC2 实例。

4.3 更新与删除

update_compute_environment（源码）支持更新state与serviceRole（后者同样校验 IAM 角色存在性）；对computeResources中的 vCpus 调整，源码中标注了TODO Implement resizing of instances based on changing vCpus，即当前不会真正扩缩容实例。

delete_compute_environment（源码）会依次删除 ECS 集群、并 terminate 掉 MANAGED 环境创建的 EC2 实例，保持后端状态干净。

五、作业队列（Job Queue）：优先级与计算环境编排

5.1 创建语义

create_job_queue参数为jobQueueName、priority、state、computeEnvironmentOrder、schedulingPolicyArn、tags。后端校验（create_job_queue）：

• jobQueueName、priority、state、computeEnvironmentOrder均必填；
• state必须是ENABLED | DISABLED，队列名不能重复；
• computeEnvironmentOrder至少包含 1 个环境，并按order字段升序排序后逐一校验环境存在性；
• 队列创建后status固定为VALID（见 JobQueue.describe）。

5.2 更新与删除

update_job_queue（源码）支持更新state、priority、computeEnvironmentOrder（重新排序并校验）、schedulingPolicyArn。delete_job_queue（源码）按名称/ARN 移除队列。

六、作业定义（Job Definition）：类型、默认值与校验

6.1 支持三种属性形态

register_job_definition的type只能是container或multinode（源码）。作业属性可以是：

• containerProperties：标准容器作业；
• eksProperties：EKS 容器作业（当eksProperties非空时优先使用）；
• nodeProperties：多节点作业。

三者至少提供其一，否则抛ClientException。另外文档标注RetryStrategy与Parameters参数尚未实现，但retryStrategy中的evaluateOnExit.action会被规范化为小写存储（源码）。

6.2 自动填充的默认值

创建 container 类型定义时，后端会补齐一组默认值（源码）：

• command、resourceRequirements、secrets、environment、mountPoints、ulimits、volumes缺省为空列表；
• 若platformCapabilities包含FARGATE，自动补充fargatePlatformConfiguration = {"platformVersion": "LATEST"}；
• 空的environment变量（value == ""）会被过滤移除。

对 EKS 作业，每个容器缺省补充command与env空列表（源码）。

6.3 硬性校验

container 定义校验（源码）：

• image必填；
• memory必填且>= 4；
• vcpus必填且> 0。

memory/vcpus既可以放在containerProperties顶层（旧写法），也可以放在resourceRequirements列表中（新写法），_get_resource_requirement（源码）会优先读取resourceRequirements（注意其中VCPUS类型在比较时去掉了末尾的s），再回退到顶层字段。EKS 定义则要求podProperties.containers至少一个容器且每个容器含image（源码）。

6.4 修订版本机制

同名定义重复注册会生成新的 revision：JobDefinition.__init__中self.revision += 1（源码），ARN 形如job-definition/{name}:{revision}。get_job_definition_by_name总是返回最大 revision 的活跃定义，deregister_job_definition则把状态置为INACTIVE（源码），describe_job_definitions支持按status过滤。

七、作业执行：submit → 状态机 → describe/list

7.1 作业状态机

作业状态由 moto/batch/utils.py 中的JobStatus枚举定义，合法值为SUBMITTED → PENDING → RUNNABLE → STARTING → RUNNING → SUCCEEDED/FAILED。status_transitions()给出有序迁移表，is_job_before_starting判断作业是否处于SUBMITTED/PENDING/RUNNABLE（决定cancel_job是否生效），is_job_already_started判断是否进入RUNNING/SUCCEEDED/FAILED（决定describe是否输出startedAt）。

7.2 提交作业与参数校验

submit_job（源码）要求：

• jobName匹配^[A-Za-z0-9][A-Za-z0-9_-]{1,127}$；
• 引用的作业定义与作业队列必须已存在，否则抛ClientException。

提交后返回(jobName, jobId, jobArn)（与 responses 层输出一致，见 responses.py）。作业对象的job_id默认为随机 UUID4；当传入arrayProperties.size时，会为每个数组索引生成子作业，子作业 ID 形如{parentId}:{index}（源码），父作业不执行容器，仅汇总子作业状态（arrayProperties.statusSummary、size）。

7.3 Docker 执行模型：真实容器与日志回传

这是 Batch 后端最独特的部分。作业对象Job同时继承threading.Thread、BaseModel、DockerModel、ManagedState（源码），提交后以后台线程方式运行，run()方法（源码）的流程为：

1. 尝试import docker，若未安装 Docker SDK 则直接把作业标记为失败；
2. 按状态机推进（每个状态转换间sleep(0.5)轮询）；
3. 若配置了dependsOn，等待依赖作业全部成功；任一依赖失败则本作业标记失败且不启动（源码）；
4. 依据作业定义（container / multinode）构造 Docker 容器参数，默认镜像alpine:latest、默认命令为一段打印Hello World的循环脚本；
5. 容器环境注入MOTO_HOST、MOTO_PORT、MOTO_HTTP_ENDPOINT，使容器内的代码可以访问本机其他 Moto 服务（Linux 下额外添加host.docker.internal: host-gateway映射，见 源码）；也可通过 Moto 的moto_network_mode()/moto_network_name()配置指定网络；
6. 轮询容器直到退出，收集 stdout/stderr 日志，写入 CloudWatch Logs（日志组/aws/batch/job，日志流{definition}/default/{jobId}，见 源码）；
7. 依据容器退出码判定作业SUCCEEDED（0）或FAILED（非 0 或主动停止）。

timeout.attemptDurationSeconds会被强制执行：超时抛异常并终止容器（源码）。describe_jobs会输出attempts、container.exitCode、logStreamName、startedAt、stoppedAt等字段（describe）。

环境前提：此模式要求本地可用 Docker 守护进程，且moto.utilities.docker_utilities.DockerModel负责管理 docker client（见 moto/utilities/docker_utilities.py）。若环境无 Docker，应改用下一节的 simple 模式。

7.4 无 Docker 模式：use_docker=False与BatchSimpleBackend

仓库提供独立的 moto/batch_simple/models.py 实现BatchSimpleBackend，不启动任何容器，提交作业后直接标记为SUCCEEDED（除非配置失败）。切换方式有两种：

• 装饰器配置：@mock_aws(config={"batch": {"use_docker": False}})
• 环境变量：MOTO_SIMPLE_BATCH_FAIL_AFTER

其机制见 responses.py：响应层根据default_user_config["batch"]["use_docker"]决定返回batch_simple_backends还是标准batch_backends。BatchSimpleBackend通过__getattribute__拦截submit_job，其余方法透传给标准后端（源码），因此两者 API 行为一致。若设置MOTO_SIMPLE_BATCH_FAIL_AFTER=0则作业立即失败；设为正整数则睡眠对应秒数后失败（源码）。这一模式非常适合 CI 环境（无 Docker）下的快速回归测试。

7.5 作业查询与终止

• describe_jobs：按 jobId / ARN 过滤（空列表返回全部），输出完整详情；
• list_jobs（源码）：支持按jobQueue、jobStatus、arrayJobId过滤；当filters提供时 Boto3 语义为忽略jobStatus；JOB_NAME过滤器支持尾部*通配符前缀匹配，其他未知过滤器一律放行；
• cancel_job：仅对未开始（SUBMITTED/PENDING/RUNNABLE）的作业生效，已开始的作业需用terminate_job（源码）；jobId与reason均不可为空字符串。

八、调度策略（Scheduling Policy）与标签

8.1 Scheduling Policy

SchedulingPolicy（源码）以name创建，fairsharePolicy包含computeReservation、shareDecaySeconds、shareDistribution（缺省分别为0、0、[]），支持create/describe/list/update/delete全流程。队列通过schedulingPolicyArn关联策略。注意list_scheduling_policies同样不支持分页。

8.2 标签系统

Batch 使用 Moto 通用的TaggingService（moto/utilities/tagging_service.py）。计算环境、作业队列、作业定义、作业、调度策略均可打标签。作业标签有数量上限 50（源码），超出抛ValidationError；list_tags_for_resource在 responses 层通过GET /v1/tags/{arn}提供（responses.py）。

九、覆盖检查与常用测试路径

在编写测试前，可通过 IMPLEMENTATION_COVERAGE.md 查看全局覆盖矩阵，或直接对照 batch.rst 的勾选清单。仓库测试目录 tests/test_batch 提供了按功能拆分的参考用例：

• test_batch.py：多区域冒烟与基础行为；
• test_batch_compute_envs.py：计算环境创建/校验/删除；
• test_batch_job_queue.py：队列优先级与计算环境编排；
• test_batch_jobs.py：作业提交、状态、取消/终止；
• test_batch_scheduling_policy.py：调度策略；
• test_batch_tags_*.py 等：各类资源的标签增删查；
• test_batch_eks.py 与 test_batch_task_definition.py：EKS 作业与任务定义细节；
• test_batch_cloudformation.py：CloudFormation 资源（AWS::Batch::ComputeEnvironment、AWS::Batch::JobQueue、AWS::Batch::JobDefinition）的创建路径。

这些测试同时验证了文档清单中的“已实现”项，是理解各接口行为边界的最佳入口。

十、实战：端到端 Batch 测试示例

结合以上内容，一个完整的（无 Docker、使用 simple 后端）端到端流程如下：

import boto3 from moto import mock_aws @mock_aws(config={"batch": {"use_docker": False}}) def test_batch_end_to_end(): iam = boto3.client("iam", region_name="us-east-1") iam.create_role( RoleName="batch-service-role", AssumeRolePolicyDocument='{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"batch.amazonaws.com"},"Action":"sts:AssumeRole"}]}', ) iam.create_instance_profile(InstanceProfileName="batch-instance-profile") iam.add_role_to_instance_profile( InstanceProfileName="batch-instance-profile", RoleName="batch-service-role" ) ec2 = boto3.client("ec2", region_name="us-east-1") vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"] subnet = ec2.create_subnet(VpcId=vpc["VpcId"], CidrBlock="10.0.1.0/24")["Subnet"] sg = ec2.create_security_group( GroupName="batch-sg", Description="batch", VpcId=vpc["VpcId"] )["GroupId"] batch = boto3.client("batch", region_name="us-east-1") env = batch.create_compute_environment( computeEnvironmentName="ce-1", type="MANAGED", state="ENABLED", computeResources={ "type": "EC2", "instanceRole": "batch-instance-profile", "instanceTypes": ["t2.micro"], "minvCpus": 0, "maxvCpus": 8, "subnets": [subnet], "securityGroupIds": [sg], }, serviceRole="arn:aws:iam::123456789012:role/batch-service-role", ) assert env["computeEnvironmentArn"].startswith("arn:aws:batch:") queue = batch.create_job_queue( jobQueueName="q-1", priority=1, state="ENABLED", computeEnvironmentOrder=[ {"order": 1, "computeEnvironment": env["computeEnvironmentArn"]} ], ) job_def = batch.register_job_definition( jobDefinitionName="jd-1", type="container", containerProperties={ "image": "alpine:latest", "vcpus": 1, "memory": 512, "command": ["echo", "hello"], }, ) submitted = batch.submit_job( jobName="job-1", jobQueue=queue["jobQueueArn"], jobDefinition=job_def["jobDefinitionArn"], ) jobs = batch.describe_jobs(jobs=[submitted["jobId"]])["jobs"] assert jobs[0]["status"] in ("SUCCEEDED", "FAILED") assert batch.list_jobs(jobQueue=queue["jobQueueArn"])["jobSummaryList"] batch.terminate_job(jobId=submitted["jobId"], reason="cleanup")

几点实战提示：

• serviceRole与instanceRole都必须在 IAM 后端真实存在（batch-service-role需先创建，instanceRole传实例配置文件名称或 ARN）；
• 子网与安全组必须由 EC2 mock 创建；
• use_docker=False下作业直接到达终态，describe_jobs无需轮询等待；默认 Docker 模式下提交后需轮询状态直至SUCCEEDED/FAILED；
• 结束前用terminate_job或后端reset()（会停止仍在运行的作业线程，见 reset）清理，避免残留线程。

结语

Moto 对 AWS Batch 的模拟覆盖了从资源建模、参数校验到真实容器执行的完整链路：默认模式借助 Docker 实现“真执行 + 日志回传”，simple 模式则提供无 Docker 的快速路径。对照 batch.rst 的勾选清单，可清晰识别describe_*/list_*分页未实现、submit_job部分参数未实现等边界；结合 moto/batch/models.py、moto/batch/responses.py 与 tests/test_batch 中的用例，即可为自己的 Batch 业务代码写出高保真、可复现的本地测试。

• Mock
• 测试