Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载Moto 是一款基于内存的 AWS 基础设施 mock 库项目入口它最大的便利在于资源即建即得——但 AWS 真实环境中的资源如 DAX 集群、ECS 任务、Batch 作业往往是异步创建的状态需要经历creating → available、RUNNING → STOPPED等多个阶段。本文将以 docs/docs/configuration/state_transition/index.rst 为骨架结合 managed_state_model.py、state_manager.py 等源码与真实测试用例系统讲解 Moto 的状态转换State Transition机制三种进度模式immediate / time / manual的配置与底层原理、全部已注册模型的转换表、ServerMode 下的 HTTP 配置方式以及如何通过继承ManagedState为新的 mock 模型接入该机制。读完本文你将能熟练控制 Moto 中任意异步资源的就绪时机精确复现 AWS 的异步行为用于单元测试与超时测试。为什么需要状态转换异步资源测试的痛点在真实 AWS 环境中绝大多数创建类 API 都是异步的create_cluster返回后资源并不会立即可用而是进入initializing/creating等中间状态需要客户端轮询describe_*类接口直到状态变为ready/available。因此面向 AWS 的应用程序几乎都要实现类似下面的等待逻辑def create_and_wait_for_cluster(name): client.create_cluster(ClusterNamename, ...) cluster_status get_cluster_status(name) while cluster_status ! available: sleep() if five_minutes_have_passed(): error() cluster_status get_cluster_status(name)这段逻辑非常值得测试状态为initializing时业务应当如何处理状态迟迟不变成ready时是否要报错然而 Moto 的一切都保存在内存中不会真的去创建任何服务器资源理论上可以瞬间就绪。这固然是使用 Moto 的最大好处却也意味着上述整段代码根本没法测试——因为资源永远立即可用while循环永远不会执行第二次。为此Moto 在moto/moto_api中提供了 State Manager API状态管理器它可以在状态转换之间人为注入延迟让 Moto 按需模拟 AWS 的异步本质。其核心思想是每个支持状态转换的模型如dax::cluster都维护一组状态转换对transition pairsState Manager 决定这些转换何时发生。三种转换模式immediate / time / manual对任意已注册模型均可通过state_manager.set_transition(model_name, transition)覆盖其默认转换配置。转换配置是一个字典根据progression字段取值分为三种模式。模式一immediate —— 立即就绪默认行为这是绝大多数模型的默认配置等价于 Moto 的传统行为资源一经创建即为最终状态方便快速跑单元测试。from moto.moto_api import state_manager state_manager.set_transition( model_namedax::cluster, transition{progression: immediate}, )在源码 managed_state_model.py 中status属性被实现为读取时按需推进当配置为immediate时每次读取status都会直接跳到最后一个状态通过_get_last_status沿转换链迭代至末端。因此创建即available无需任何额外调用。模式二time —— 按秒数延时就绪如果你希望测试创建资源需要 5 秒的场景可以这样配置from moto.moto_api import state_manager state_manager.set_transition( model_namedax::cluster, transition{progression: time, seconds: 5}, ) create_and_wait_for_cluster(my_new_cluster)若要模拟超时场景例如10 分钟后仍未就绪只需把seconds设为一个超过业务等待阈值的数字state_manager.set_transition( model_namedax::cluster, transition{progression: time, seconds: 600}, ) try: create_and_wait_for_cluster(my_new_cluster) except: verify_the_correct_error_was_thrown()源码层面time 模式基于self._time_progressed上一次推进状态的时间戳与当前时间做比较当datetime.now()超过上次推进时间 seconds时才推进到下一个状态并刷新时间戳见 managed_state_model.py。这意味着状态只会在你读取status时按需推进且每次只前进一个状态需要多次读取才能走完整条转换链。模式三manual —— 按读取次数推进前两种模式之外还有第三种每次调用describe类接口推进一次times指定需要调用几次才真正推进状态。state_manager.set_transition( model_namedax::cluster, transition{progression: manual, times: 3}, )它之所以叫manual是因为必须手动调用describe_object方法状态才会推进。下面是一个典型断言序列client.create_cluster(ClusterNamename, ...) # 第一次获取状态 status client.describe_clusters(ClusterNames[name])[Clusters][0][Status] assert status creating # 第二次获取状态 status client.describe_clusters(ClusterNames[name])[Clusters][0][Status] assert status creating # 第三次获取状态 —— 本次将推进到下一状态 status client.describe_clusters(ClusterNames[name])[Clusters][0][Status] assert status available说明上例中的describe_object是一个泛称指代各服务中用于查询资源状态的 boto3 方法例如boto3.client(dax).describe_clusters()、boto3.client(support).describe_cases()、boto3.client(ecs).describe_tasks()等。真实测试中应将其包在while循环里轮询而不是硬编码三次调用。manual 模式的源码实现清晰体现了调用计数逻辑managed_state_model.pyadvance()每次被调用都会使内部计数器_tick加 1当_tick times时status推进到下一个状态并重置计数器。这与immediate的一步到位形成鲜明对比——manual 模式每次只前进一个状态非常适合验证状态确实在变化但又不希望测试耗时太长的场景。State Manager 的底层实现优先级与默认值理解 State Manager 的查找顺序有助于掌握它的全部行为。源码 state_manager.py 维护两张字典_default_transitions由 Moto 各 backend 通过register_default_transition注册的默认配置_transitions用户在测试中通过set_transition设置的自定义配置。get_transition的查找优先级为state_manager.py用户自定义配置_transitions优先其次是模型注册的默认配置_default_transitions兜底为全局常量DEFAULT_TRANSITION {progression: immediate}state_manager.py。也就是说任何set_transition都会覆盖默认值且未注册的模型默认立即就绪。对应的撤销操作是unset_transition(model_name)它是一个幂等的安全操作——只删除自定义配置不影响默认配置见 state_manager.pyfrom moto.moto_api import state_manager state_manager.unset_transition(model_namedax::cluster)已注册模型一览谁支持状态转换以下是 Moto 当前已注册的状态转换模型清单依据 docs/docs/configuration/state_transition/models.rst 与 moto/moto_api/init.py 中的默认注册模型名状态链默认转换推进方式athena::executionQUEUED → RUNNING → SUCCEEDEDimmediateget_query_execution(..)batch::jobSUBMITTED → PENDING → RUNNABLE → STARTING → RUNNING → SUCCEEDED|FAILEDmanual, times1由 Moto 自动推进各阶段cloudfront::distributionInProgress → Deployedmanual, times1get_distribution(..)/list_distributions(..)dax::clustercreating → availabledeleting → deletedmanual, times4describe_clusters(..)dms::connectiontesting → successmanual, times1describe_connections(..)dms::replicationinstancecreating → availablemanual, times1describe_replication_instances(..)dsql::cluster—manual, times1—dsql::stream—manual, times1—ecs::taskRUNNING → DEACTIVATING → STOPPING → DEPROVISIONING → STOPPEDmanual, times1describe_tasks(..)fis::experimentpending → initiating → running → completedstopping → stoppedmanual, times1get_experiment(..)glue::job_runSTARTING → RUNNING → SUCCEEDEDimmediateget_job_run(..)glue::crawlRUNNING → COMPLETEDmanual, times1list_crawls(..)osis::pipeline—immediate—pipes::pipeCREATING/STARTING → RUNNINGSTOPPING → STOPPEDSTOPPED → STARTINGDELETING → DELETEDmanual, times1describe_pipe(..)/list_pipes(..)s3::keyrestoreNone → IN_PROGRESS → RESTOREDimmediate—support::caseopened → pending-customer-action → reopened → resolved → unassigned → work-in-progress → openedmanual, times1describe_cases(..)transcribe::vocabularyNone → PENDING → READYmanual, times1get_vocabulary(..)transcribe::medicalvocabularyNone → PENDING → READYmanual, times1get_medical_vocabulary(..)transcribe::transcriptionjobNone → QUEUED → IN_PROGRESS → COMPLETEDmanual, times1get_transcription_job(..)transcribe::medicaltranscriptionjobNone → QUEUED → IN_PROGRESS → COMPLETEDmanual, times1get_medical_transcription_job(..)注意较老版本的 Moto 可能未包含上表中的全部模型。要查看当前 Moto 版本实际支持哪些模型可调用get_registered_modelsfrom moto.moto_api import state_manager with mock_all(): print(state_manager.get_registered_models())这里必须带上mock_all()或mock_aws装饰器——模型是在对应服务的 mock 启动时才注册到 State Manager 的若在 mock 之外调用很可能得到空列表。表格中的两个细节batch::job 的特殊性batch::job的状态推进与描述资源无关而是由 Moto 内部在提交作业时自动推进各准备阶段SUBMITTED → PENDING → RUNNABLE → STARTING → RUNNING最终在 Docker 容器中执行作业。默认immediate保证作业尽快执行若需要捕获处于某一特定阶段的作业可通过 time 模式人为延长各阶段耗时详见 models.rst。s3::keyrestore对应 S3 Glacier 深度归档对象的恢复流程状态从None → IN_PROGRESS → RESTORED默认立即完成。实战示例一DAX 集群的读取 N 次才就绪测试DAX 集群是 manual 模式的典型代表默认需要4 次describe_clusters调用才会从creating推进到available。我们以仓库中的测试 tests/test_dax/test_dax.py 为参照创建集群后首次读取状态即为creatingmock_aws def test_create_cluster_minimal(): client boto3.client(dax, region_nameus-east-2) created_cluster client.create_cluster( ClusterNamedaxcluster, NodeTypedax.t3.small, ReplicationFactor3, IamRoleArniam_role_arn, )[Cluster] described_cluster client.describe_clusters(ClusterNames[daxcluster])[Clusters][0] assert described_cluster[Status] creating结合前面set_transition(model_namedax::cluster, transition{progression: manual, times: 3})的配置一个完整的三次推进测试逻辑可以写成from moto import mock_aws from moto.moto_api import state_manager mock_aws def test_dax_cluster_manual_transition(): state_manager.set_transition( model_namedax::cluster, transition{progression: manual, times: 3}, ) client boto3.client(dax, region_nameus-east-2) client.create_cluster( ClusterNamedaxcluster, NodeTypedax.t3.small, ReplicationFactor3, IamRoleArniam_role_arn, ) # 前两次读取仍是 creating for _ in range(2): status client.describe_clusters(ClusterNames[daxcluster])[Clusters][0][Status] assert status creating # 第三次读取后推进到 available status client.describe_clusters(ClusterNames[daxcluster])[Clusters][0][Status] assert status available state_manager.unset_transition(model_namedax::cluster)实战示例二ECS 任务的状态机推进ECS 任务定义了完整的停止状态机RUNNING → DEACTIVATING → STOPPING → DEPROVISIONING → STOPPED。仓库测试 tests/test_ecs/test_ecs.py 展示了完整用法——先set_transition为manual, times1停止任务后循环读取describe_tasks并逐状态断言state_manager.set_transition( model_nameecs::task, transition{progression: manual, times: 1}, ) # ... 创建集群、运行任务 ... for status in (DEACTIVATING, STOPPING, DEPROVISIONING, STOPPED): response client.describe_tasks(clustertest_ecs_cluster, tasks[task_arn]) assert response[tasks][0][lastStatus] status state_manager.unset_transition(ecs::task)同理若要在测试中让任务立即停止不关心中间态可配置immediate模式见 tests/test_ecs/test_ecs.pystate_manager.set_transition( model_nameecs::task, transition{progression: immediate}, ) # ... describe_tasks 后 lastStatus 直接为 STOPPED state_manager.unset_transition(ecs::task)值得注意这两个测试都带有if settings.TEST_SERVER_MODE: raise SkipTest(...)的保护——在 ServerMode 下不能直接调用state_manager必须改用下面的 HTTP API见 moto/core/config.py 中的TEST_SERVER_MODE配置。ServerMode 下的配置通过 MotoAPI 的 HTTP 端点当以 ServerMode 运行 Motomotoserver时进程内没有测试上下文state_manager需通过 MotoAPI 的 HTTP 端点来操作。这些端点由 moto/moto_api/_internal/responses.py 实现内部最终仍委托给全局state_manager见 moto/moto_api/_internal/models.py。设置转换例如让dax::cluster延迟 5 秒就绪import json import requests post_body dict( model_namedax::cluster, transition{progression: time, seconds: 5}, ) resp requests.post( http://localhost:5000/moto-api/state-manager/set-transition, datajson.dumps(post_body), )查询某个模型当前生效的转换配置requests.get( http://localhost:5000/moto-api/state-manager/get-transition?model_namedax::cluster )撤销自定义转换恢复默认配置post_body dict(model_namedax::cluster) resp requests.post( http://localhost:5000/moto-api/state-manager/unset-transition, datajson.dumps(post_body), )所有模型与三种转换类型immediate / time / manual在 HTTP 端点上遵循与 Python API 完全相同的格式。注意set-transition与unset-transition均返回 HTTP 201get-transition返回 200 及 JSON 配置体。为新的 mock 模型接入状态转换继承 ManagedState 四步法如果要在自定义或新增的 Moto 模型中启用状态转换支持例如一个新建的NewModel对应文档 docs/docs/contributing/development_tips/new_state_transitions.rst需要依次完成以下四步。第一步继承 ManagedState 并传入转换对ManagedState定义在 managed_state_model.py其构造函数接收model_name全局唯一标识习惯格式为API::type如S3::bucket、APIGateway::Method、DynamoDB::Table与transitions一个状态对列表每个元素是(上一个状态, 下一个状态)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), ], )从源码可见ManagedState.__init__会自动完成三件事managed_state_model.py_status初始化为第一条转换的起点上例即initializing_tick计数器归零供 manual 模式使用_time_progressed记录当前时间供 time 模式使用。第二步在序列化输出中暴露 statusManagedState直接提供一个开箱即用的status属性读取时按当前转换配置自动推进状态。模型只需在to_json或 API 响应中把它包含进去def to_json(self): # ManagedState 开箱即用地提供了 status 属性 # 首次读取时为第一条转换的起始状态 return { name: ..., status: self.status, ... }这一步是关键——状态推进发生在读取self.status的那一刻因此凡是用户能看到状态的接口都应当经由status属性生成响应。第三步在恰当的时机调用 advance()advance()是 manual 模式的计数器触发器源码见 managed_state_model.py。它需要被放入所有用户可查询资源状态的入口list_resources、describe_resource、get_*等。对immediate与time模式调用它不会产生任何副作用但对于配置了 manual 的用户是必需的from moto.moto_api import state_manager class Backend: def list_resources(self): for model in all_resources: # 用户配置为 manual 时如 {progression: manual, times: 3} # 每调用一次 list_resourcesadvance() 使 _tick 1 # 累计 3 次后 status 自动推进——这些都由 ManagedState 完成 model.advance() return all_models def describe_resource(self): resource ... # 同一资源的每种查询入口describe / list / get_*都必须调用 advance() # 否则用户无论查询多少次都无法推进状态 resource.advance() return resource以 moto/dax/models.py 的真实实现为参照DaxCluster(BaseModel, ManagedState)在构造函数中调用super().__init__(model_namedax::cluster, transitions[(creating, available), (deleting, deleted)])与文档中的示例完全一致。更多真实范例见moto/ecs/models.py、moto/fis/models.py、moto/glue/models.py等可用ManagedState关键字在moto/下检索全部 15 处引用。第四步在 State Manager 中注册默认转换最后需要在 moto/moto_api/init.py 中调用register_default_transition将模型的默认转换配置注册进去state_manager.register_default_transition( # 名称必须与 NewModel 中使用的 model_name 完全一致 model_namenew::model, # 任何转换配置均可immediate 是常用的良好默认值 transition{progression: immediate}, )注册完成后用户即可通过set_transition覆盖该默认配置例如改为manual或time并通过get_registered_models()查询到新模型。整个接入流程可概括为继承ManagedState→ 声明转换对 → 在查询入口调用advance()→ 注册默认转换。小结Moto 的状态转换机制为内存中的 AWS补齐了异步资源的行为模拟能力。本文从使用与实现两个视角梳理了其完整图景三种转换模式immediate读取即达终态、time按seconds延时、每段只前进一个状态、manual每累计times次advance()推进一个状态均通过 state_manager.py 与 managed_state_model.py 实现配置方式Python APIset_transition/unset_transition/get_registered_models与 ServerMode 下的 HTTP 端点/moto-api/state-manager/*默认注册表20 余个模型DAX、ECS、Batch、Transcribe、DMS、CloudFront、Athena、Glue 等已内置默认转换完整清单见 models.rst扩展机制任何新模型通过继承ManagedState四步接入相关实现与测试可参考 dax/models.py、tests/test_ecs/test_ecs.py、tests/test_fis/test_fis.py 与 tests/test_glue/test_glue_job_runs.py。掌握这套机制后你可以在单元测试中精确控制任意 AWS 资源的就绪节奏从而完整覆盖异步业务逻辑的每一条分支——包括超时分支这是单纯即建即得的 mock 库难以做到的。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐javascript-state-machine状态与转换机制深度解析javascript state machine状态与转换机制深度解析 本文深入解析了javascript state machine库的核心架构与实现机制重开发工具DeepSeek-V3-Lora-tune镜像部署教程快速搭建开发环境的完整方案DeepSeek V3 Lora tune镜像部署教程快速搭建开发环境的完整方案 DeepSeek V3 Lora tune是基于HuggingFace镜像的Mock测试GetX 状态管理完全指南Simple State Manager、Reactive State Manager 与 Workers 深入解析GetX 状态管理完全指南Simple State Manager、Reactive State Manager 与 Workers 深入解析 本文基于 Ge前端上一篇UKB_RAP完整教程英国生物银行数据分析的终极指南下一篇NVIDIA OpenReasoning-Nemotron-14B多智能体协作引领推理范式革新创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Moto 状态转换机制深度解析：State Manager 使用与扩展指南
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载Moto 是一款基于内存的 AWS 基础设施 mock 库项目入口它最大的便利在于资源即建即得——但 AWS 真实环境中的资源如 DAX 集群、ECS 任务、Batch 作业往往是异步创建的状态需要经历creating → available、RUNNING → STOPPED等多个阶段。本文将以 docs/docs/configuration/state_transition/index.rst 为骨架结合 managed_state_model.py、state_manager.py 等源码与真实测试用例系统讲解 Moto 的状态转换State Transition机制三种进度模式immediate / time / manual的配置与底层原理、全部已注册模型的转换表、ServerMode 下的 HTTP 配置方式以及如何通过继承ManagedState为新的 mock 模型接入该机制。读完本文你将能熟练控制 Moto 中任意异步资源的就绪时机精确复现 AWS 的异步行为用于单元测试与超时测试。为什么需要状态转换异步资源测试的痛点在真实 AWS 环境中绝大多数创建类 API 都是异步的create_cluster返回后资源并不会立即可用而是进入initializing/creating等中间状态需要客户端轮询describe_*类接口直到状态变为ready/available。因此面向 AWS 的应用程序几乎都要实现类似下面的等待逻辑def create_and_wait_for_cluster(name): client.create_cluster(ClusterNamename, ...) cluster_status get_cluster_status(name) while cluster_status ! available: sleep() if five_minutes_have_passed(): error() cluster_status get_cluster_status(name)这段逻辑非常值得测试状态为initializing时业务应当如何处理状态迟迟不变成ready时是否要报错然而 Moto 的一切都保存在内存中不会真的去创建任何服务器资源理论上可以瞬间就绪。这固然是使用 Moto 的最大好处却也意味着上述整段代码根本没法测试——因为资源永远立即可用while循环永远不会执行第二次。为此Moto 在moto/moto_api中提供了 State Manager API状态管理器它可以在状态转换之间人为注入延迟让 Moto 按需模拟 AWS 的异步本质。其核心思想是每个支持状态转换的模型如dax::cluster都维护一组状态转换对transition pairsState Manager 决定这些转换何时发生。三种转换模式immediate / time / manual对任意已注册模型均可通过state_manager.set_transition(model_name, transition)覆盖其默认转换配置。转换配置是一个字典根据progression字段取值分为三种模式。模式一immediate —— 立即就绪默认行为这是绝大多数模型的默认配置等价于 Moto 的传统行为资源一经创建即为最终状态方便快速跑单元测试。from moto.moto_api import state_manager state_manager.set_transition( model_namedax::cluster, transition{progression: immediate}, )在源码 managed_state_model.py 中status属性被实现为读取时按需推进当配置为immediate时每次读取status都会直接跳到最后一个状态通过_get_last_status沿转换链迭代至末端。因此创建即available无需任何额外调用。模式二time —— 按秒数延时就绪如果你希望测试创建资源需要 5 秒的场景可以这样配置from moto.moto_api import state_manager state_manager.set_transition( model_namedax::cluster, transition{progression: time, seconds: 5}, ) create_and_wait_for_cluster(my_new_cluster)若要模拟超时场景例如10 分钟后仍未就绪只需把seconds设为一个超过业务等待阈值的数字state_manager.set_transition( model_namedax::cluster, transition{progression: time, seconds: 600}, ) try: create_and_wait_for_cluster(my_new_cluster) except: verify_the_correct_error_was_thrown()源码层面time 模式基于self._time_progressed上一次推进状态的时间戳与当前时间做比较当datetime.now()超过上次推进时间 seconds时才推进到下一个状态并刷新时间戳见 managed_state_model.py。这意味着状态只会在你读取status时按需推进且每次只前进一个状态需要多次读取才能走完整条转换链。模式三manual —— 按读取次数推进前两种模式之外还有第三种每次调用describe类接口推进一次times指定需要调用几次才真正推进状态。state_manager.set_transition( model_namedax::cluster, transition{progression: manual, times: 3}, )它之所以叫manual是因为必须手动调用describe_object方法状态才会推进。下面是一个典型断言序列client.create_cluster(ClusterNamename, ...) # 第一次获取状态 status client.describe_clusters(ClusterNames[name])[Clusters][0][Status] assert status creating # 第二次获取状态 status client.describe_clusters(ClusterNames[name])[Clusters][0][Status] assert status creating # 第三次获取状态 —— 本次将推进到下一状态 status client.describe_clusters(ClusterNames[name])[Clusters][0][Status] assert status available说明上例中的describe_object是一个泛称指代各服务中用于查询资源状态的 boto3 方法例如boto3.client(dax).describe_clusters()、boto3.client(support).describe_cases()、boto3.client(ecs).describe_tasks()等。真实测试中应将其包在while循环里轮询而不是硬编码三次调用。manual 模式的源码实现清晰体现了调用计数逻辑managed_state_model.pyadvance()每次被调用都会使内部计数器_tick加 1当_tick times时status推进到下一个状态并重置计数器。这与immediate的一步到位形成鲜明对比——manual 模式每次只前进一个状态非常适合验证状态确实在变化但又不希望测试耗时太长的场景。State Manager 的底层实现优先级与默认值理解 State Manager 的查找顺序有助于掌握它的全部行为。源码 state_manager.py 维护两张字典_default_transitions由 Moto 各 backend 通过register_default_transition注册的默认配置_transitions用户在测试中通过set_transition设置的自定义配置。get_transition的查找优先级为state_manager.py用户自定义配置_transitions优先其次是模型注册的默认配置_default_transitions兜底为全局常量DEFAULT_TRANSITION {progression: immediate}state_manager.py。也就是说任何set_transition都会覆盖默认值且未注册的模型默认立即就绪。对应的撤销操作是unset_transition(model_name)它是一个幂等的安全操作——只删除自定义配置不影响默认配置见 state_manager.pyfrom moto.moto_api import state_manager state_manager.unset_transition(model_namedax::cluster)已注册模型一览谁支持状态转换以下是 Moto 当前已注册的状态转换模型清单依据 docs/docs/configuration/state_transition/models.rst 与 moto/moto_api/init.py 中的默认注册模型名状态链默认转换推进方式athena::executionQUEUED → RUNNING → SUCCEEDEDimmediateget_query_execution(..)batch::jobSUBMITTED → PENDING → RUNNABLE → STARTING → RUNNING → SUCCEEDED|FAILEDmanual, times1由 Moto 自动推进各阶段cloudfront::distributionInProgress → Deployedmanual, times1get_distribution(..)/list_distributions(..)dax::clustercreating → availabledeleting → deletedmanual, times4describe_clusters(..)dms::connectiontesting → successmanual, times1describe_connections(..)dms::replicationinstancecreating → availablemanual, times1describe_replication_instances(..)dsql::cluster—manual, times1—dsql::stream—manual, times1—ecs::taskRUNNING → DEACTIVATING → STOPPING → DEPROVISIONING → STOPPEDmanual, times1describe_tasks(..)fis::experimentpending → initiating → running → completedstopping → stoppedmanual, times1get_experiment(..)glue::job_runSTARTING → RUNNING → SUCCEEDEDimmediateget_job_run(..)glue::crawlRUNNING → COMPLETEDmanual, times1list_crawls(..)osis::pipeline—immediate—pipes::pipeCREATING/STARTING → RUNNINGSTOPPING → STOPPEDSTOPPED → STARTINGDELETING → DELETEDmanual, times1describe_pipe(..)/list_pipes(..)s3::keyrestoreNone → IN_PROGRESS → RESTOREDimmediate—support::caseopened → pending-customer-action → reopened → resolved → unassigned → work-in-progress → openedmanual, times1describe_cases(..)transcribe::vocabularyNone → PENDING → READYmanual, times1get_vocabulary(..)transcribe::medicalvocabularyNone → PENDING → READYmanual, times1get_medical_vocabulary(..)transcribe::transcriptionjobNone → QUEUED → IN_PROGRESS → COMPLETEDmanual, times1get_transcription_job(..)transcribe::medicaltranscriptionjobNone → QUEUED → IN_PROGRESS → COMPLETEDmanual, times1get_medical_transcription_job(..)注意较老版本的 Moto 可能未包含上表中的全部模型。要查看当前 Moto 版本实际支持哪些模型可调用get_registered_modelsfrom moto.moto_api import state_manager with mock_all(): print(state_manager.get_registered_models())这里必须带上mock_all()或mock_aws装饰器——模型是在对应服务的 mock 启动时才注册到 State Manager 的若在 mock 之外调用很可能得到空列表。表格中的两个细节batch::job 的特殊性batch::job的状态推进与描述资源无关而是由 Moto 内部在提交作业时自动推进各准备阶段SUBMITTED → PENDING → RUNNABLE → STARTING → RUNNING最终在 Docker 容器中执行作业。默认immediate保证作业尽快执行若需要捕获处于某一特定阶段的作业可通过 time 模式人为延长各阶段耗时详见 models.rst。s3::keyrestore对应 S3 Glacier 深度归档对象的恢复流程状态从None → IN_PROGRESS → RESTORED默认立即完成。实战示例一DAX 集群的读取 N 次才就绪测试DAX 集群是 manual 模式的典型代表默认需要4 次describe_clusters调用才会从creating推进到available。我们以仓库中的测试 tests/test_dax/test_dax.py 为参照创建集群后首次读取状态即为creatingmock_aws def test_create_cluster_minimal(): client boto3.client(dax, region_nameus-east-2) created_cluster client.create_cluster( ClusterNamedaxcluster, NodeTypedax.t3.small, ReplicationFactor3, IamRoleArniam_role_arn, )[Cluster] described_cluster client.describe_clusters(ClusterNames[daxcluster])[Clusters][0] assert described_cluster[Status] creating结合前面set_transition(model_namedax::cluster, transition{progression: manual, times: 3})的配置一个完整的三次推进测试逻辑可以写成from moto import mock_aws from moto.moto_api import state_manager mock_aws def test_dax_cluster_manual_transition(): state_manager.set_transition( model_namedax::cluster, transition{progression: manual, times: 3}, ) client boto3.client(dax, region_nameus-east-2) client.create_cluster( ClusterNamedaxcluster, NodeTypedax.t3.small, ReplicationFactor3, IamRoleArniam_role_arn, ) # 前两次读取仍是 creating for _ in range(2): status client.describe_clusters(ClusterNames[daxcluster])[Clusters][0][Status] assert status creating # 第三次读取后推进到 available status client.describe_clusters(ClusterNames[daxcluster])[Clusters][0][Status] assert status available state_manager.unset_transition(model_namedax::cluster)实战示例二ECS 任务的状态机推进ECS 任务定义了完整的停止状态机RUNNING → DEACTIVATING → STOPPING → DEPROVISIONING → STOPPED。仓库测试 tests/test_ecs/test_ecs.py 展示了完整用法——先set_transition为manual, times1停止任务后循环读取describe_tasks并逐状态断言state_manager.set_transition( model_nameecs::task, transition{progression: manual, times: 1}, ) # ... 创建集群、运行任务 ... for status in (DEACTIVATING, STOPPING, DEPROVISIONING, STOPPED): response client.describe_tasks(clustertest_ecs_cluster, tasks[task_arn]) assert response[tasks][0][lastStatus] status state_manager.unset_transition(ecs::task)同理若要在测试中让任务立即停止不关心中间态可配置immediate模式见 tests/test_ecs/test_ecs.pystate_manager.set_transition( model_nameecs::task, transition{progression: immediate}, ) # ... describe_tasks 后 lastStatus 直接为 STOPPED state_manager.unset_transition(ecs::task)值得注意这两个测试都带有if settings.TEST_SERVER_MODE: raise SkipTest(...)的保护——在 ServerMode 下不能直接调用state_manager必须改用下面的 HTTP API见 moto/core/config.py 中的TEST_SERVER_MODE配置。ServerMode 下的配置通过 MotoAPI 的 HTTP 端点当以 ServerMode 运行 Motomotoserver时进程内没有测试上下文state_manager需通过 MotoAPI 的 HTTP 端点来操作。这些端点由 moto/moto_api/_internal/responses.py 实现内部最终仍委托给全局state_manager见 moto/moto_api/_internal/models.py。设置转换例如让dax::cluster延迟 5 秒就绪import json import requests post_body dict( model_namedax::cluster, transition{progression: time, seconds: 5}, ) resp requests.post( http://localhost:5000/moto-api/state-manager/set-transition, datajson.dumps(post_body), )查询某个模型当前生效的转换配置requests.get( http://localhost:5000/moto-api/state-manager/get-transition?model_namedax::cluster )撤销自定义转换恢复默认配置post_body dict(model_namedax::cluster) resp requests.post( http://localhost:5000/moto-api/state-manager/unset-transition, datajson.dumps(post_body), )所有模型与三种转换类型immediate / time / manual在 HTTP 端点上遵循与 Python API 完全相同的格式。注意set-transition与unset-transition均返回 HTTP 201get-transition返回 200 及 JSON 配置体。为新的 mock 模型接入状态转换继承 ManagedState 四步法如果要在自定义或新增的 Moto 模型中启用状态转换支持例如一个新建的NewModel对应文档 docs/docs/contributing/development_tips/new_state_transitions.rst需要依次完成以下四步。第一步继承 ManagedState 并传入转换对ManagedState定义在 managed_state_model.py其构造函数接收model_name全局唯一标识习惯格式为API::type如S3::bucket、APIGateway::Method、DynamoDB::Table与transitions一个状态对列表每个元素是(上一个状态, 下一个状态)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), ], )从源码可见ManagedState.__init__会自动完成三件事managed_state_model.py_status初始化为第一条转换的起点上例即initializing_tick计数器归零供 manual 模式使用_time_progressed记录当前时间供 time 模式使用。第二步在序列化输出中暴露 statusManagedState直接提供一个开箱即用的status属性读取时按当前转换配置自动推进状态。模型只需在to_json或 API 响应中把它包含进去def to_json(self): # ManagedState 开箱即用地提供了 status 属性 # 首次读取时为第一条转换的起始状态 return { name: ..., status: self.status, ... }这一步是关键——状态推进发生在读取self.status的那一刻因此凡是用户能看到状态的接口都应当经由status属性生成响应。第三步在恰当的时机调用 advance()advance()是 manual 模式的计数器触发器源码见 managed_state_model.py。它需要被放入所有用户可查询资源状态的入口list_resources、describe_resource、get_*等。对immediate与time模式调用它不会产生任何副作用但对于配置了 manual 的用户是必需的from moto.moto_api import state_manager class Backend: def list_resources(self): for model in all_resources: # 用户配置为 manual 时如 {progression: manual, times: 3} # 每调用一次 list_resourcesadvance() 使 _tick 1 # 累计 3 次后 status 自动推进——这些都由 ManagedState 完成 model.advance() return all_models def describe_resource(self): resource ... # 同一资源的每种查询入口describe / list / get_*都必须调用 advance() # 否则用户无论查询多少次都无法推进状态 resource.advance() return resource以 moto/dax/models.py 的真实实现为参照DaxCluster(BaseModel, ManagedState)在构造函数中调用super().__init__(model_namedax::cluster, transitions[(creating, available), (deleting, deleted)])与文档中的示例完全一致。更多真实范例见moto/ecs/models.py、moto/fis/models.py、moto/glue/models.py等可用ManagedState关键字在moto/下检索全部 15 处引用。第四步在 State Manager 中注册默认转换最后需要在 moto/moto_api/init.py 中调用register_default_transition将模型的默认转换配置注册进去state_manager.register_default_transition( # 名称必须与 NewModel 中使用的 model_name 完全一致 model_namenew::model, # 任何转换配置均可immediate 是常用的良好默认值 transition{progression: immediate}, )注册完成后用户即可通过set_transition覆盖该默认配置例如改为manual或time并通过get_registered_models()查询到新模型。整个接入流程可概括为继承ManagedState→ 声明转换对 → 在查询入口调用advance()→ 注册默认转换。小结Moto 的状态转换机制为内存中的 AWS补齐了异步资源的行为模拟能力。本文从使用与实现两个视角梳理了其完整图景三种转换模式immediate读取即达终态、time按seconds延时、每段只前进一个状态、manual每累计times次advance()推进一个状态均通过 state_manager.py 与 managed_state_model.py 实现配置方式Python APIset_transition/unset_transition/get_registered_models与 ServerMode 下的 HTTP 端点/moto-api/state-manager/*默认注册表20 余个模型DAX、ECS、Batch、Transcribe、DMS、CloudFront、Athena、Glue 等已内置默认转换完整清单见 models.rst扩展机制任何新模型通过继承ManagedState四步接入相关实现与测试可参考 dax/models.py、tests/test_ecs/test_ecs.py、tests/test_fis/test_fis.py 与 tests/test_glue/test_glue_job_runs.py。掌握这套机制后你可以在单元测试中精确控制任意 AWS 资源的就绪节奏从而完整覆盖异步业务逻辑的每一条分支——包括超时分支这是单纯即建即得的 mock 库难以做到的。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐javascript-state-machine状态与转换机制深度解析javascript state machine状态与转换机制深度解析 本文深入解析了javascript state machine库的核心架构与实现机制重开发工具DeepSeek-V3-Lora-tune镜像部署教程快速搭建开发环境的完整方案DeepSeek V3 Lora tune镜像部署教程快速搭建开发环境的完整方案 DeepSeek V3 Lora tune是基于HuggingFace镜像的Mock测试GetX 状态管理完全指南Simple State Manager、Reactive State Manager 与 Workers 深入解析GetX 状态管理完全指南Simple State Manager、Reactive State Manager 与 Workers 深入解析 本文基于 Ge前端上一篇UKB_RAP完整教程英国生物银行数据分析的终极指南下一篇NVIDIA OpenReasoning-Nemotron-14B多智能体协作引领推理范式革新创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考