尧图网站建设 
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本篇指南面向所有想为 MotoAWS 基础设施模拟测试库贡献代码的开发者系统梳理官方开发建议中的核心技巧responses.py与models.py的分工与命名约定、部分实现如何用warnings提示用户、测试的编写规范与 ServerMode 验证方式、URL 拦截路径的确定方法以及 TaggingService、paginate分页装饰器和ManagedState状态管理器等内置工具的正确用法。读完你不仅能写出风格统一、易于合入的 Moto 代码还能借助 scripts/scaffold.py 脚手架快速起步并对照真实源码理解每个技巧背后的实现原理。一、开发前先建立整体认知Moto 的三层架构在动手实现任何功能前需要先理解 Moto 每个服务模块的代码布局。以acm服务为例其目录结构为moto/acm/responses.py负责请求解析、输入/输出格式化与参数校验moto/acm/models.py负责真正的业务逻辑与资源状态存储moto/acm/urls.py定义需要拦截的 URL 与方法的映射关系。这一“Responses 做编排、Models 做实现”的分层是所有 Moto 服务共通的骨架也是后续所有命名规范与实现技巧的出发点。二、命名规范让每个方法各归其位2.1 responses.py 与 models.py 的分工官方开发建议以一个例子说明假如你要为 ACM 服务实现import_certificate导入证书功能做法是在 moto/acm/responses.py 中创建import_certificate方法负责处理输入/输出格式化和校验在 moto/acm/models.py 中再创建一个同名import_certificate方法承载实际的证书导入逻辑。这一点在真实源码中得到了完整印证。在 moto/acm/responses.py 中import_certificate先读取请求参数再调用后端方法def import_certificate(self) - ActionResult: # 从请求体中解析 certificate / private_key / chain / arn / tags 等参数 ... arn self.acm_backend.import_certificate( # 委托给 backend certificate, private_key, chainchain, arnarn, tagstags ) return self.import_certificate_response(arn)而 moto/acm/models.py 中的同名方法则负责核心业务逻辑包括校验 ARN 是否存在、构建CertBundle对象并存入证书仓库def import_certificate(self, certificate, private_key, chain, arn, tags) - str: if arn is not None: if arn not in self._certificates: raise CertificateNotFound(arnarn, account_idself.account_id) else: # 复用用户提供的 ARN bundle CertBundle(self.account_id, certificate, private_key, chainchain, regionself.region_name, arnarn) else: # 自动生成随机 ARN bundle CertBundle(self.account_id, certificate, private_key, ...) ...由此可见Responses 方法名与 Models 方法名保持完全一致是该项目的既有惯例便于读者在两层代码间快速跳转对照。输入/输出的“长相”问题XML/JSON 的字段结构、错误响应的格式一律留在 responses 层处理业务状态则一律沉淀在 models 层。2.2 测试的命名约定编写测试时为新增功能添加的测试方法直接以功能名命名例如在 tests/test_acm/test_acm.py 中添加def test_import_certificate(): ...负向测试失败场景的命名则要能直白地表达“发生了什么、为何失败”例如test_import_certificate_fails_without_name—— 缺少名称时导入失败test_import_existing_certificate—— 重复导入已存在的证书。这样的命名让测试失败时的输出信息自解释也方便后续维护者通过名字快速定位行为边界。三、部分实现用 warnings 主动告知用户当某个服务只有部分实现时官方建议使用warnings模块向用户发出显式提醒而不是静默忽略未支持的参数import warnings warnings.warn(The Filters-parameter is not yet implemented for client.method())这一做法在代码库中被广泛采用。例如 moto/awslambda/models.py 与 moto/s3/models.py 中都存在类似调用用于提示某些请求参数或行为尚未被模拟。这样使用者在测试中就能立刻感知“当前 mock 行为与真实 AWS 存在差异”避免将未实现的参数误当作已支持。四、编写测试的实用规范4.1 一个测试只验证一个功能官方强调一个测试应当只验证单一的功能/方法——例如create_resource()一个测试、update_resource()另一个测试不要在一个测试里串联多个 API 调用。这与命名约定相辅相成保证了失败时能精确定位到具体的实现环节。4.2 负向测试的标准写法对于期望抛错的负向测试官方给出了统一格式这是确保异常被 Moto 正确处理的最佳方式with pytest.raises(botocore.exceptions.ClientError) as exc: client.failing_call(..) err exc.value.response[Error] # 使用 pytest 的 assert 方法 assert err[Code] .. assert err[Message] ..要点有两个捕获类型必须是botocore.exceptions.ClientError这与真实 AWS 客户端抛出的异常类型一致从exc.value.response[Error]中取出Code与Message分别断言从而验证 Moto 返回的错误码与错误文案符合预期。4.3 ServerMode 下的双重验证Moto 的 CI 会把全部测试跑两遍一遍是常规的进程内 mock 模式另一遍是ServerMode——Moto 作为独立的 Flask 服务启动所有测试都针对这个 Flask 实例运行。在本地验证测试能否通过 ServerMode只需两条命令python moto/server.py TEST_SERVER_MODEtrue pytest -sv tests/test_service/..其中 moto/server.py 是 ServerMode 的启动入口。凡是新增测试都应该在两种模式下都跑通因为 ServerMode 下的 URL 路由、请求转发链路与进程内 mock 并不完全相同。4.4 并行测试的注意事项为加速 CIawslambda、batch、ec2、sqs四个服务的测试会并行执行这意味着必须为函数、队列等资源使用唯一命名避免不同测试相互覆盖describe_reservations()、list_queues()等查询类调用可能返回其他测试创建的资源断言时不要假设查询结果集是“干净”的。五、拦截 URL确定 Moto 要拦截哪些地址Moto 通过 URL 分发决定拦截哪些 AWS 请求因此新增功能时必须先确定正确的 URL 路径。官方给出了三种方法照搬既有功能对于已有服务直接复制某个既有特性的 url-path再祈祷它能匹配适合快速起步但需要验证查阅 botocore 服务模型botocore 维护着 AWS 服务的 API 定义数据在对应服务的services.json中查找requestUri字段即可获得精确的请求路径模板代理真实 AWS 请求自己向 AWS 发一次调用并用代理截获可以得到最完整的信息——URL、参数、请求与响应格式一应俱全。5.1 用 MITM 代理截获真实 AWS 请求官方推荐的做法是下载并安装一个代理如 MITMProxy然后通过 CLI 或 boto3 把流量指向本地代理export HTTP_PROXYhttp://localhost:8080 export HTTPS_PROXYhttp://localhost:8080 aws ses describe-rule-set --no-verify-ssl等价地也可以在 Python 中使用 boto3 的代理配置from botocore.config import Config proxy_config Config(proxies{http: localhost:8080, https: localhost:8080}) boto3.client(ses, configproxy_config, use_sslFalse, verifyFalse)注意示例中关闭了 SSL 校验--no-verify-ssl/use_sslFalse, verifyFalse因为代理截获的是明文/自签名的 HTTPS 流量。截获到的请求报文可以直接反推 Moto 的urls.py映射与 responses 层解析逻辑。5.2 修改 URL 后别忘了重建索引另外需要特别提醒为了提升 MotoServer 的启动性能所有需要拦截的 AWS URL 都被集中索引。每次在{service}/urls.py中新增或替换任何 URL 后都必须运行python scripts/update_backend_index.py来更新索引否则 ServerMode 下新增的 URL 可能不会被正确拦截。新增整个服务或功能后还应运行 scripts/implementation_coverage.py 来刷新支持的服务列表。六、复用 UtilitiesTaggingService 与分页器6.1 TaggingService统一的标签存储moto/utilities 中提供了一个专用的TaggingService用于帮助各服务存储/检索资源的标签。虽然目前并非所有服务都启用了它但官方明确鼓励所有新功能都应该优先使用 TaggingService而非自行维护标签逻辑以保证标签的增删改查行为在各服务间保持一致。6.2 分页器paginate装饰器几乎所有的 AWS 列表类 API 都使用分页来分批返回结果。Moto 提供了一个工具类让后端无需手写分页逻辑即可自动完成分页。它由三个组件协同工作Responses 方法调用后端时传入max_results/next_token参数后端方法是一个普通方法但被paginate装饰配置模型PAGINATION_MODEL提供给装饰器描述分页所需的全部细节。完整示例摘自官方开发建议如下class MyResponse(BaseResponse): # Response 类看起来和其他类一样——读取输入参数调用后端获取资源 def list_resources(self): max_results 100 next_token self._get_param(NextToken) # 注意这里同时拿到结果和 next_token # 后端的装饰器返回的就是这个二元组 paged_results, next_token self.backend.list_resources( max_resultsmax_results, next_tokennext_token ) ... from moto.utilities.paginator import paginate class MyBackend(BaseBackend): # 包含分页器所需配置的模型 PAGINATION_MODEL { # key 后端方法名 list_resources: { # 包含 next token 的 kwarg 名称会被传给后端 # backend.list_resources(next_token..) input_token: next_token, # 包含最大结果数的 kwarg 名称 limit_key: max_results, # 上述参数的默认值 limit_default: 100, # 一个或多个能保证资源唯一性的属性 # 大多数资源用 ID 或 ARN 即可它们总是唯一的 # 这些属性经过编码后用作 NextToken unique_attribute: arn, # 如果只有多个属性的组合才能保证唯一则传一个列表 unique_attribute: [start_date, execution_arn], # 默认情况下用户传入无效 next_token 时会抛异常 fail_on_invalid_token: True # 默认值无需显式指定 # 也可以自定义为 # - 静默失败直接返回空列表 fail_on_invalid_token: False, # - 抛出自定义异常传入异常类即可 # 分页器会执行 raise CustomException() 或 raise CustomException(invalid_token) fail_on_invalid_token: CustomException }, # 另一个使用不同配置的方法 list_other_things: { ... }, } # 携带分页逻辑的装饰器 paginate(pagination_modelPAGINATION_MODEL) # 注意此方法没有 next_token / max_results 参数 def list_resources(self): # 只需返回全部资源——分页魔法全部由装饰器完成 return self.full_list_of_resources paginate(pagination_modelPAGINATION_MODEL) # 如果确实需要 next_token / max_results 参数直接加在函数上即可 # 装饰器只在需要时才会把它们传进来 def list_other_things(self, max_resultsNone): if max_results 42: # 自定义校验逻辑 pass return self.full_list_of_resources6.3 分页器源码解析paginate的实现位于 moto/utilities/paginator.py。从源码可以看出几个关键设计装饰器通过func.__name__从pagination_model中取出对应方法的配置若缺失直接抛出ValueError它从 kwargs 中剥离input_token与limit_key后将剩余参数深拷贝input_kwargs deepcopy(kwargs)交给分页器比对——这正是防止“携带 A 参数生成的 token 却被用于 B 参数的查询”这类误用的机制fail_on_invalid_token默认值为True对应 moto/core/exceptions.py 中的InvalidToken异常设为False时静默返回空列表传入异常类时则抛出该自定义异常。换句话说后端方法只需“朴素地返回全部资源”token 的编码/解码、切片、合法性校验与默认 limit 均由装饰器统一完成这大大降低了各服务分页实现不一致的风险。七、状态转换管理让资源“慢慢就绪”很多 AWS 资源如 EC2 实例创建后并非立即可用而是经历若干状态转换。Moto 为此提供了配置选项默认可以立即就绪加速单元测试也允许人为延迟以更贴近真实 AWS 的行为。7.1 让模型继承 ManagedState要让一个新模型开箱即用地支持这种“状态推进”行为需要按以下四步操作让新模型继承ManagedState类在ManagedState构造函数中声明支持哪些状态转换决定在何时推进状态调用advance()把模型注册到StateManager。官方给出的示例模型如下from moto.moto_api._internal.managed_state_model import ManagedState class NewModel(ManagedState): def __init__(self): ManagedState.__init__(self, # 唯一名称用于标识该模型 # 任何名字都可以——典型格式为 API:type # 例如 S3::bucket、APIGateway::Method、DynamoDB::Table model_namenew::model, # 在此列出所有可能的状态转换 transitions[(initializing, starting), (starting, ready)]) def to_json(self): # ManagedState 开箱即用地提供 status 属性 # 第一次迭代时它会被置为第一个转换的初始状态 return { name: ..., status: self.status, ... }7.2 在合适的位置调用 advance()advance()决定状态推进的时机。官方建议在用户获取资源的每一种途径describe、list、get 等中都调用它确保资源能够真正推进到下一状态from moto.moto_api import state_manager class Backend(): def list_resources(self): for resource in all_resources: # 如果用户为该类型模型配置了手动推进例如 {progression: manual, times: 3} # 那么用户每调用 3 次 list_resourcesadvance 被调用 3 次 # 第 3 次调用后状态管理器才会推进状态——这一切都在开箱即用中完成 # 只要确保在恰当的时机调用 advance() 即可 # # 如果配置为立即推进此方法不做任何事 # 如果用户改为按时间推进每隔 y 秒变一次状态此方法同样不做任何事 # 但仍必须被调用因为配置了手动推进的用户依赖它 resource.advance() return all_models def describe_resource(self): resource ... # 不同 API 可能有不同的方式让用户获取同一信息 # 确保每种方式describe、list、get_ 等都调用 advance() resource.advance() return resource7.3 向 StateManager 注册默认转换最后必须把模型注册到StateManager这一步在 moto/moto_api/init.py 中完成state_manager.register_default_transition( # 该名称必须与 NewModel 中使用的名称一致 model_namenew::model, # 任何转换配置都可以——不过下面这个是不错的默认选择 transition{progression: immediate}, )注册之后用户便可以通过 Moto 的 state_manager 配置如手动推进、按次推进或按时间推进来控制该模型的状态演进而模型本身无需再关心用户侧的配置细节。八、进阶用脚手架脚本加速开发除了上述技巧Moto 还提供了 scripts/scaffold.py 脚手架脚本可以自动为“新服务”或“既有服务的新功能”生成模板代码。它会读取某个 boto3 方法的 API 定义并自动生成models.py、responses.py、urls.py、__init__.py以及对应的测试文件骨架。使用前请先设置区域export AWS_DEFAULT_REGIONus-east-2 python scripts/scaffold.py脚本基于click模块提供交互式补全用 Tab 补全服务名、方向键在下拉列表中切换、回车确认。例如选择codedeploy服务后脚本会列出该服务的当前实现状态勾选create_deployment操作即可自动生成目录与代码骨架。开发完成后记得按前文所述运行 scripts/implementation_coverage.py 与 scripts/update_backend_index.py 完成收尾。结语本文从 Moto 官方开发建议出发覆盖了从命名规范、部分实现提示、测试编写、URL 拦截到 TaggingService、分页器与状态管理器的一整套实用技巧。这些建议并非孤立规则而是与代码库中真实的实现方式一一对应——moto/acm/responses.py、moto/acm/models.py、moto/utilities/paginator.py、moto/moto_api/init.py 等都是对照学习的最佳范本。遵循这些约定你写出的代码将更容易被维护者接受也能让 Moto 的整体行为在各服务之间保持高度一致。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Codon 贡献指南开发工作流、编码规范与测试编写实战Codon 贡献指南开发工作流、编码规范与测试编写实战 本文是 Codon 编译器项目A high performance, zero overhead,编译器编程语言语言运行时终极LOLBAS开发指南贡献者必备的YAML编写规范与实战技巧终极LOLBAS开发指南贡献者必备的YAML编写规范与实战技巧 LOLBASLiving Off The Land Binaries And Scripts应用安全渗透测试红蓝对抗Infer 静态分析器贡献指南从开发环境搭建到测试编写与代码规范Infer 静态分析器贡献指南从开发环境搭建到测试编写与代码规范 Infer 是 Facebook/Meta 开源的 Java、C、C 与 Objecti静态分析代码质量开发工具上一篇突破静态限制用ShapeShifter打造会呼吸的界面图标下一篇为什么说KeyClu是Mac用户的终极快捷键记忆神器创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Moto 贡献者开发实战指南：命名规范、测试编写与分页器、状态机等核心开发技巧
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载本篇指南面向所有想为 MotoAWS 基础设施模拟测试库贡献代码的开发者系统梳理官方开发建议中的核心技巧responses.py与models.py的分工与命名约定、部分实现如何用warnings提示用户、测试的编写规范与 ServerMode 验证方式、URL 拦截路径的确定方法以及 TaggingService、paginate分页装饰器和ManagedState状态管理器等内置工具的正确用法。读完你不仅能写出风格统一、易于合入的 Moto 代码还能借助 scripts/scaffold.py 脚手架快速起步并对照真实源码理解每个技巧背后的实现原理。一、开发前先建立整体认知Moto 的三层架构在动手实现任何功能前需要先理解 Moto 每个服务模块的代码布局。以acm服务为例其目录结构为moto/acm/responses.py负责请求解析、输入/输出格式化与参数校验moto/acm/models.py负责真正的业务逻辑与资源状态存储moto/acm/urls.py定义需要拦截的 URL 与方法的映射关系。这一“Responses 做编排、Models 做实现”的分层是所有 Moto 服务共通的骨架也是后续所有命名规范与实现技巧的出发点。二、命名规范让每个方法各归其位2.1 responses.py 与 models.py 的分工官方开发建议以一个例子说明假如你要为 ACM 服务实现import_certificate导入证书功能做法是在 moto/acm/responses.py 中创建import_certificate方法负责处理输入/输出格式化和校验在 moto/acm/models.py 中再创建一个同名import_certificate方法承载实际的证书导入逻辑。这一点在真实源码中得到了完整印证。在 moto/acm/responses.py 中import_certificate先读取请求参数再调用后端方法def import_certificate(self) - ActionResult: # 从请求体中解析 certificate / private_key / chain / arn / tags 等参数 ... arn self.acm_backend.import_certificate( # 委托给 backend certificate, private_key, chainchain, arnarn, tagstags ) return self.import_certificate_response(arn)而 moto/acm/models.py 中的同名方法则负责核心业务逻辑包括校验 ARN 是否存在、构建CertBundle对象并存入证书仓库def import_certificate(self, certificate, private_key, chain, arn, tags) - str: if arn is not None: if arn not in self._certificates: raise CertificateNotFound(arnarn, account_idself.account_id) else: # 复用用户提供的 ARN bundle CertBundle(self.account_id, certificate, private_key, chainchain, regionself.region_name, arnarn) else: # 自动生成随机 ARN bundle CertBundle(self.account_id, certificate, private_key, ...) ...由此可见Responses 方法名与 Models 方法名保持完全一致是该项目的既有惯例便于读者在两层代码间快速跳转对照。输入/输出的“长相”问题XML/JSON 的字段结构、错误响应的格式一律留在 responses 层处理业务状态则一律沉淀在 models 层。2.2 测试的命名约定编写测试时为新增功能添加的测试方法直接以功能名命名例如在 tests/test_acm/test_acm.py 中添加def test_import_certificate(): ...负向测试失败场景的命名则要能直白地表达“发生了什么、为何失败”例如test_import_certificate_fails_without_name—— 缺少名称时导入失败test_import_existing_certificate—— 重复导入已存在的证书。这样的命名让测试失败时的输出信息自解释也方便后续维护者通过名字快速定位行为边界。三、部分实现用 warnings 主动告知用户当某个服务只有部分实现时官方建议使用warnings模块向用户发出显式提醒而不是静默忽略未支持的参数import warnings warnings.warn(The Filters-parameter is not yet implemented for client.method())这一做法在代码库中被广泛采用。例如 moto/awslambda/models.py 与 moto/s3/models.py 中都存在类似调用用于提示某些请求参数或行为尚未被模拟。这样使用者在测试中就能立刻感知“当前 mock 行为与真实 AWS 存在差异”避免将未实现的参数误当作已支持。四、编写测试的实用规范4.1 一个测试只验证一个功能官方强调一个测试应当只验证单一的功能/方法——例如create_resource()一个测试、update_resource()另一个测试不要在一个测试里串联多个 API 调用。这与命名约定相辅相成保证了失败时能精确定位到具体的实现环节。4.2 负向测试的标准写法对于期望抛错的负向测试官方给出了统一格式这是确保异常被 Moto 正确处理的最佳方式with pytest.raises(botocore.exceptions.ClientError) as exc: client.failing_call(..) err exc.value.response[Error] # 使用 pytest 的 assert 方法 assert err[Code] .. assert err[Message] ..要点有两个捕获类型必须是botocore.exceptions.ClientError这与真实 AWS 客户端抛出的异常类型一致从exc.value.response[Error]中取出Code与Message分别断言从而验证 Moto 返回的错误码与错误文案符合预期。4.3 ServerMode 下的双重验证Moto 的 CI 会把全部测试跑两遍一遍是常规的进程内 mock 模式另一遍是ServerMode——Moto 作为独立的 Flask 服务启动所有测试都针对这个 Flask 实例运行。在本地验证测试能否通过 ServerMode只需两条命令python moto/server.py TEST_SERVER_MODEtrue pytest -sv tests/test_service/..其中 moto/server.py 是 ServerMode 的启动入口。凡是新增测试都应该在两种模式下都跑通因为 ServerMode 下的 URL 路由、请求转发链路与进程内 mock 并不完全相同。4.4 并行测试的注意事项为加速 CIawslambda、batch、ec2、sqs四个服务的测试会并行执行这意味着必须为函数、队列等资源使用唯一命名避免不同测试相互覆盖describe_reservations()、list_queues()等查询类调用可能返回其他测试创建的资源断言时不要假设查询结果集是“干净”的。五、拦截 URL确定 Moto 要拦截哪些地址Moto 通过 URL 分发决定拦截哪些 AWS 请求因此新增功能时必须先确定正确的 URL 路径。官方给出了三种方法照搬既有功能对于已有服务直接复制某个既有特性的 url-path再祈祷它能匹配适合快速起步但需要验证查阅 botocore 服务模型botocore 维护着 AWS 服务的 API 定义数据在对应服务的services.json中查找requestUri字段即可获得精确的请求路径模板代理真实 AWS 请求自己向 AWS 发一次调用并用代理截获可以得到最完整的信息——URL、参数、请求与响应格式一应俱全。5.1 用 MITM 代理截获真实 AWS 请求官方推荐的做法是下载并安装一个代理如 MITMProxy然后通过 CLI 或 boto3 把流量指向本地代理export HTTP_PROXYhttp://localhost:8080 export HTTPS_PROXYhttp://localhost:8080 aws ses describe-rule-set --no-verify-ssl等价地也可以在 Python 中使用 boto3 的代理配置from botocore.config import Config proxy_config Config(proxies{http: localhost:8080, https: localhost:8080}) boto3.client(ses, configproxy_config, use_sslFalse, verifyFalse)注意示例中关闭了 SSL 校验--no-verify-ssl/use_sslFalse, verifyFalse因为代理截获的是明文/自签名的 HTTPS 流量。截获到的请求报文可以直接反推 Moto 的urls.py映射与 responses 层解析逻辑。5.2 修改 URL 后别忘了重建索引另外需要特别提醒为了提升 MotoServer 的启动性能所有需要拦截的 AWS URL 都被集中索引。每次在{service}/urls.py中新增或替换任何 URL 后都必须运行python scripts/update_backend_index.py来更新索引否则 ServerMode 下新增的 URL 可能不会被正确拦截。新增整个服务或功能后还应运行 scripts/implementation_coverage.py 来刷新支持的服务列表。六、复用 UtilitiesTaggingService 与分页器6.1 TaggingService统一的标签存储moto/utilities 中提供了一个专用的TaggingService用于帮助各服务存储/检索资源的标签。虽然目前并非所有服务都启用了它但官方明确鼓励所有新功能都应该优先使用 TaggingService而非自行维护标签逻辑以保证标签的增删改查行为在各服务间保持一致。6.2 分页器paginate装饰器几乎所有的 AWS 列表类 API 都使用分页来分批返回结果。Moto 提供了一个工具类让后端无需手写分页逻辑即可自动完成分页。它由三个组件协同工作Responses 方法调用后端时传入max_results/next_token参数后端方法是一个普通方法但被paginate装饰配置模型PAGINATION_MODEL提供给装饰器描述分页所需的全部细节。完整示例摘自官方开发建议如下class MyResponse(BaseResponse): # Response 类看起来和其他类一样——读取输入参数调用后端获取资源 def list_resources(self): max_results 100 next_token self._get_param(NextToken) # 注意这里同时拿到结果和 next_token # 后端的装饰器返回的就是这个二元组 paged_results, next_token self.backend.list_resources( max_resultsmax_results, next_tokennext_token ) ... from moto.utilities.paginator import paginate class MyBackend(BaseBackend): # 包含分页器所需配置的模型 PAGINATION_MODEL { # key 后端方法名 list_resources: { # 包含 next token 的 kwarg 名称会被传给后端 # backend.list_resources(next_token..) input_token: next_token, # 包含最大结果数的 kwarg 名称 limit_key: max_results, # 上述参数的默认值 limit_default: 100, # 一个或多个能保证资源唯一性的属性 # 大多数资源用 ID 或 ARN 即可它们总是唯一的 # 这些属性经过编码后用作 NextToken unique_attribute: arn, # 如果只有多个属性的组合才能保证唯一则传一个列表 unique_attribute: [start_date, execution_arn], # 默认情况下用户传入无效 next_token 时会抛异常 fail_on_invalid_token: True # 默认值无需显式指定 # 也可以自定义为 # - 静默失败直接返回空列表 fail_on_invalid_token: False, # - 抛出自定义异常传入异常类即可 # 分页器会执行 raise CustomException() 或 raise CustomException(invalid_token) fail_on_invalid_token: CustomException }, # 另一个使用不同配置的方法 list_other_things: { ... }, } # 携带分页逻辑的装饰器 paginate(pagination_modelPAGINATION_MODEL) # 注意此方法没有 next_token / max_results 参数 def list_resources(self): # 只需返回全部资源——分页魔法全部由装饰器完成 return self.full_list_of_resources paginate(pagination_modelPAGINATION_MODEL) # 如果确实需要 next_token / max_results 参数直接加在函数上即可 # 装饰器只在需要时才会把它们传进来 def list_other_things(self, max_resultsNone): if max_results 42: # 自定义校验逻辑 pass return self.full_list_of_resources6.3 分页器源码解析paginate的实现位于 moto/utilities/paginator.py。从源码可以看出几个关键设计装饰器通过func.__name__从pagination_model中取出对应方法的配置若缺失直接抛出ValueError它从 kwargs 中剥离input_token与limit_key后将剩余参数深拷贝input_kwargs deepcopy(kwargs)交给分页器比对——这正是防止“携带 A 参数生成的 token 却被用于 B 参数的查询”这类误用的机制fail_on_invalid_token默认值为True对应 moto/core/exceptions.py 中的InvalidToken异常设为False时静默返回空列表传入异常类时则抛出该自定义异常。换句话说后端方法只需“朴素地返回全部资源”token 的编码/解码、切片、合法性校验与默认 limit 均由装饰器统一完成这大大降低了各服务分页实现不一致的风险。七、状态转换管理让资源“慢慢就绪”很多 AWS 资源如 EC2 实例创建后并非立即可用而是经历若干状态转换。Moto 为此提供了配置选项默认可以立即就绪加速单元测试也允许人为延迟以更贴近真实 AWS 的行为。7.1 让模型继承 ManagedState要让一个新模型开箱即用地支持这种“状态推进”行为需要按以下四步操作让新模型继承ManagedState类在ManagedState构造函数中声明支持哪些状态转换决定在何时推进状态调用advance()把模型注册到StateManager。官方给出的示例模型如下from moto.moto_api._internal.managed_state_model import ManagedState class NewModel(ManagedState): def __init__(self): ManagedState.__init__(self, # 唯一名称用于标识该模型 # 任何名字都可以——典型格式为 API:type # 例如 S3::bucket、APIGateway::Method、DynamoDB::Table model_namenew::model, # 在此列出所有可能的状态转换 transitions[(initializing, starting), (starting, ready)]) def to_json(self): # ManagedState 开箱即用地提供 status 属性 # 第一次迭代时它会被置为第一个转换的初始状态 return { name: ..., status: self.status, ... }7.2 在合适的位置调用 advance()advance()决定状态推进的时机。官方建议在用户获取资源的每一种途径describe、list、get 等中都调用它确保资源能够真正推进到下一状态from moto.moto_api import state_manager class Backend(): def list_resources(self): for resource in all_resources: # 如果用户为该类型模型配置了手动推进例如 {progression: manual, times: 3} # 那么用户每调用 3 次 list_resourcesadvance 被调用 3 次 # 第 3 次调用后状态管理器才会推进状态——这一切都在开箱即用中完成 # 只要确保在恰当的时机调用 advance() 即可 # # 如果配置为立即推进此方法不做任何事 # 如果用户改为按时间推进每隔 y 秒变一次状态此方法同样不做任何事 # 但仍必须被调用因为配置了手动推进的用户依赖它 resource.advance() return all_models def describe_resource(self): resource ... # 不同 API 可能有不同的方式让用户获取同一信息 # 确保每种方式describe、list、get_ 等都调用 advance() resource.advance() return resource7.3 向 StateManager 注册默认转换最后必须把模型注册到StateManager这一步在 moto/moto_api/init.py 中完成state_manager.register_default_transition( # 该名称必须与 NewModel 中使用的名称一致 model_namenew::model, # 任何转换配置都可以——不过下面这个是不错的默认选择 transition{progression: immediate}, )注册之后用户便可以通过 Moto 的 state_manager 配置如手动推进、按次推进或按时间推进来控制该模型的状态演进而模型本身无需再关心用户侧的配置细节。八、进阶用脚手架脚本加速开发除了上述技巧Moto 还提供了 scripts/scaffold.py 脚手架脚本可以自动为“新服务”或“既有服务的新功能”生成模板代码。它会读取某个 boto3 方法的 API 定义并自动生成models.py、responses.py、urls.py、__init__.py以及对应的测试文件骨架。使用前请先设置区域export AWS_DEFAULT_REGIONus-east-2 python scripts/scaffold.py脚本基于click模块提供交互式补全用 Tab 补全服务名、方向键在下拉列表中切换、回车确认。例如选择codedeploy服务后脚本会列出该服务的当前实现状态勾选create_deployment操作即可自动生成目录与代码骨架。开发完成后记得按前文所述运行 scripts/implementation_coverage.py 与 scripts/update_backend_index.py 完成收尾。结语本文从 Moto 官方开发建议出发覆盖了从命名规范、部分实现提示、测试编写、URL 拦截到 TaggingService、分页器与状态管理器的一整套实用技巧。这些建议并非孤立规则而是与代码库中真实的实现方式一一对应——moto/acm/responses.py、moto/acm/models.py、moto/utilities/paginator.py、moto/moto_api/init.py 等都是对照学习的最佳范本。遵循这些约定你写出的代码将更容易被维护者接受也能让 Moto 的整体行为在各服务之间保持高度一致。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Codon 贡献指南开发工作流、编码规范与测试编写实战Codon 贡献指南开发工作流、编码规范与测试编写实战 本文是 Codon 编译器项目A high performance, zero overhead,编译器编程语言语言运行时终极LOLBAS开发指南贡献者必备的YAML编写规范与实战技巧终极LOLBAS开发指南贡献者必备的YAML编写规范与实战技巧 LOLBASLiving Off The Land Binaries And Scripts应用安全渗透测试红蓝对抗Infer 静态分析器贡献指南从开发环境搭建到测试编写与代码规范Infer 静态分析器贡献指南从开发环境搭建到测试编写与代码规范 Infer 是 Facebook/Meta 开源的 Java、C、C 与 Objecti静态分析代码质量开发工具上一篇突破静态限制用ShapeShifter打造会呼吸的界面图标下一篇为什么说KeyClu是Mac用户的终极快捷键记忆神器创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考