Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本指南以 Moto 仓库中 batch 服务文档 为基础系统讲解如何用 Moto 在本地测试环境中模拟 AWS Batch 的核心 API计算环境Compute Environment、作业队列Job Queue、作业定义Job Definition、作业提交与状态流转、调度策略Scheduling Policy以及资源标签管理。读完本文你将掌握mock_aws装饰器下 Batch 服务的真实行为边界、Docker 与无 Dockeruse_dockerFalse两种后端模式的区别以及各 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 特性的接口写作测试时应避免依赖这些能力。文档同时标注了两类已知限制使用时需特别注意分页未实现Pagination is not yet implementeddescribe_compute_environments、describe_job_definitions、describe_job_queues、list_scheduling_policies等接口均不支持分页参数会一次性返回全部结果。submit_job的RetryStrategy与Parameters参数未实现提交作业时传入这两个参数不会按 AWS 语义生效。list_jobs的限制文档明确说明“根据 Boto3 文档按数组作业array jobID 过滤时不支持 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_nameus-west-2) resp client.describe_jobs(jobs[]) assert resp[ResponseMetadata][HTTPStatusCode] 200describe_jobs(jobs[])在 Moto 中是合法调用并返回 200说明后端对空作业过滤集合做了安全处理源码见 describe_jobs空列表时返回全部作业。这样的冒烟测试适合验证 Batch mock 已正确装配。三、核心后端BatchBackend 与依赖服务联动Batch 是 Moto 中耦合度较高的服务之一。从 BatchBackend 的类定义可以看到它通过property直接引用四个兄弟服务的 backendiam_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 统一生成格式如下资源ARN 格式计算环境arn:{partition}:batch:{region}:{account}:compute-environment/{name}作业队列arn:{partition}:batch:{region}:{account}:job-queue/{name}作业定义arn:{partition}:batch:{region}:{account}:job-definition/{name}:{revision}作业arn:{partition}:batch:{region}:{account}:job/{job_id}调度策略arn:{partition}:batch:{region}:{account}:scheduling-policy/{name}partition 依据区域自动推导如aws、aws-cn因此cn-northwest-1等区域也能得到合法 ARN。四、计算环境Compute Environment创建、校验与销毁4.1 创建参数与校验规则create_compute_environment的核心参数为computeEnvironmentName、typeMANAGED/UNMANAGED、stateENABLED/DISABLED、computeResources、serviceRole、tags。后端执行一连串校验create_compute_environment名称必须匹配正则且不能与已有环境重名serviceRole必填且必须能在 IAM 后端中找到对应角色否则抛InvalidParameterValueExceptiontype必须是MANAGED | UNMANAGEDstate必须是ENABLED | DISABLEDMANAGED 类型必须提供computeResources。computeResources子字典的校验集中在 _validate_compute_resources要点如下maxvCpus必须为正数非 FARGATE 类型还需instanceRole必须是已存在的 IAM 实例配置文件、minvCpus≥ 0且maxvCpus minvCpusinstanceTypes至少 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 的环境会触发两件副作用源码调用find_min_instances_to_meet_vcpus源码根据desiredvCpus缺省回退到minvCpus贪心挑选最少 EC2 实例并通过ec2_backend.run_instances真正创建实例optimal会映射为m4.4xlarge通过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_queuejobQueueName、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标准容器作业eksPropertiesEKS 容器作业当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必填且 4vcpus必填且 0。memory/vcpus既可以放在containerProperties顶层旧写法也可以放在resourceRequirements列表中新写法_get_resource_requirement源码会优先读取resourceRequirements注意其中VCPUS类型在比较时去掉了末尾的s再回退到顶层字段。EKS 定义则要求podProperties.containers至少一个容器且每个容器含image源码。6.4 修订版本机制同名定义重复注册会生成新的 revisionJobDefinition.__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/list7.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()方法源码的流程为尝试import docker若未安装 Docker SDK 则直接把作业标记为失败按状态机推进每个状态转换间sleep(0.5)轮询若配置了dependsOn等待依赖作业全部成功任一依赖失败则本作业标记失败且不启动源码依据作业定义container / multinode构造 Docker 容器参数默认镜像alpine:latest、默认命令为一段打印Hello World的循环脚本容器环境注入MOTO_HOST、MOTO_PORT、MOTO_HTTP_ENDPOINT使容器内的代码可以访问本机其他 Moto 服务Linux 下额外添加host.docker.internal: host-gateway映射见 源码也可通过 Moto 的moto_network_mode()/moto_network_name()配置指定网络轮询容器直到退出收集 stdout/stderr 日志写入 CloudWatch Logs日志组/aws/batch/job日志流{definition}/default/{jobId}见 源码依据容器退出码判定作业SUCCEEDED0或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_dockerFalse与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_AFTER0则作业立即失败设为正整数则睡眠对应秒数后失败源码。这一模式非常适合 CI 环境无 Docker下的快速回归测试。7.5 作业查询与终止describe_jobs按 jobId / ARN 过滤空列表返回全部输出完整详情list_jobs源码支持按jobQueue、jobStatus、arrayJobId过滤当filters提供时 Boto3 语义为忽略jobStatusJOB_NAME过滤器支持尾部*通配符前缀匹配其他未知过滤器一律放行cancel_job仅对未开始SUBMITTED/PENDING/RUNNABLE的作业生效已开始的作业需用terminate_job源码jobId与reason均不可为空字符串。八、调度策略Scheduling Policy与标签8.1 Scheduling PolicySchedulingPolicy源码以name创建fairsharePolicy包含computeReservation、shareDecaySeconds、shareDistribution缺省分别为0、0、[]支持create/describe/list/update/delete全流程。队列通过schedulingPolicyArn关联策略。注意list_scheduling_policies同样不支持分页。8.2 标签系统Batch 使用 Moto 通用的TaggingServicemoto/utilities/tagging_service.py。计算环境、作业队列、作业定义、作业、调度策略均可打标签。作业标签有数量上限 50源码超出抛ValidationErrorlist_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.pyEKS 作业与任务定义细节test_batch_cloudformation.pyCloudFormation 资源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_nameus-east-1) iam.create_role( RoleNamebatch-service-role, AssumeRolePolicyDocument{Version:2012-10-17,Statement:[{Effect:Allow,Principal:{Service:batch.amazonaws.com},Action:sts:AssumeRole}]}, ) iam.create_instance_profile(InstanceProfileNamebatch-instance-profile) iam.add_role_to_instance_profile( InstanceProfileNamebatch-instance-profile, RoleNamebatch-service-role ) ec2 boto3.client(ec2, region_nameus-east-1) vpc ec2.create_vpc(CidrBlock10.0.0.0/16)[Vpc] subnet ec2.create_subnet(VpcIdvpc[VpcId], CidrBlock10.0.1.0/24)[Subnet] sg ec2.create_security_group( GroupNamebatch-sg, Descriptionbatch, VpcIdvpc[VpcId] )[GroupId] batch boto3.client(batch, region_nameus-east-1) env batch.create_compute_environment( computeEnvironmentNamece-1, typeMANAGED, stateENABLED, computeResources{ type: EC2, instanceRole: batch-instance-profile, instanceTypes: [t2.micro], minvCpus: 0, maxvCpus: 8, subnets: [subnet], securityGroupIds: [sg], }, serviceRolearn:aws:iam::123456789012:role/batch-service-role, ) assert env[computeEnvironmentArn].startswith(arn:aws:batch:) queue batch.create_job_queue( jobQueueNameq-1, priority1, stateENABLED, computeEnvironmentOrder[ {order: 1, computeEnvironment: env[computeEnvironmentArn]} ], ) job_def batch.register_job_definition( jobDefinitionNamejd-1, typecontainer, containerProperties{ image: alpine:latest, vcpus: 1, memory: 512, command: [echo, hello], }, ) submitted batch.submit_job( jobNamejob-1, jobQueuequeue[jobQueueArn], jobDefinitionjob_def[jobDefinitionArn], ) jobs batch.describe_jobs(jobs[submitted[jobId]])[jobs] assert jobs[0][status] in (SUCCEEDED, FAILED) assert batch.list_jobs(jobQueuequeue[jobQueueArn])[jobSummaryList] batch.terminate_job(jobIdsubmitted[jobId], reasoncleanup)几点实战提示serviceRole与instanceRole都必须在 IAM 后端真实存在batch-service-role需先创建instanceRole传实例配置文件名称或 ARN子网与安全组必须由 EC2 mock 创建use_dockerFalse下作业直接到达终态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测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Floci 本地模拟 AWS Batch从任务编排到 Docker 执行的控制面实现解析Floci 本地模拟 AWS Batch从任务编排到 Docker 执行的控制面实现解析 FlociLight, fluffy, and always fr如何在Docker容器中高效运行Android模拟器完整实践指南在移动应用开发和测试过程中搭建和维护Android模拟器环境往往是一项耗时且复杂的工作。传统的Android Studio模拟器需要大量的系统资源而且难以在移动开发测试使用 Docker 容器运行 mypy 与 mypyc 测试从环境构建到命令执行完整指南使用 Docker 容器运行 mypy 与 mypyc 测试从环境构建到命令执行完整指南 在非 Linux 主机上开发 mypy/mypyc 时跨平台测试往开发工具静态分析代码质量上一篇教育资源数字化获取的技术解决方案tchMaterial-parser工具深度解析下一篇3分钟掌握Tantivy全文搜索引擎Python绑定终极实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
