尧图网络科技YAOTU DIGITAL 获取报价
获取报价
首页 / 资讯中心 / 文章详情

moto 中 Amazon Managed Blockchain 的本地模拟实现:API 覆盖、状态机与实战演练

发布时间:2026/9/26 7:48:03

资讯中心
01
ARTICLE

moto 中 Amazon Managed Blockchain 的本地模拟实现:API 覆盖、状态机与实战演练

moto 中 Amazon Managed Blockchain 的本地模拟实现:API 覆盖、状态机与实战演练
Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载Amazon Managed Blockchain 是 AWS 提供的托管式区块链服务支持创建 Hyperledger Fabric 网络、成员、节点并通过提案投票机制治理网络。moto 在 docs/docs/services/managedblockchain.rst 中给出了该服务的能力清单在已列出的 25 个 API 操作中moto 完整模拟了 19 个覆盖网络、成员、节点、提案、投票与邀请的全生命周期。本文以该清单为骨架结合 moto/managedblockchain 模块的源码与 tests/test_managedblockchain 测试套件逐层拆解 moto 的实现机制并给出可直接运行的 boto3 演练脚本帮助你无需真实 AWS 账号即可开发、调试与测试依赖 Managed Blockchain 的代码。功能覆盖总览哪些 API 已实现原文档通过[X]/[ ]标记逐一列出了每个 API 的实现状态。moto 当前支持以下 19 个操作覆盖了区块链网络治理的主流程类别已实现[X]未实现[ ]网络create_network、get_network、list_networks—成员create_member、get_member、list_members、update_member、delete_member—节点create_node、get_node、list_nodes、update_node、delete_node—提案与投票create_proposal、get_proposal、list_proposals、list_proposal_votes、vote_on_proposal—邀请list_invitations、reject_invitation—Accessor 与标签—create_accessor、delete_accessor、get_accessor、list_accessors、list_tags_for_resource、tag_resource、untag_resource未实现的 7 个操作主要涉及 Amazon Managed Blockchain Accessor以太坊网络访问器与资源标签管理调用时会落入未注册的 URL 路由。需要这些能力时当前版本无法模拟代码中应做好相应的降级处理或跳过测试。模块架构从请求路由到内存状态机整个模拟实现采用 moto 标准的「URL 路由 → Response → Backend → Model」四层结构全部代码位于 moto/managedblockchain 目录urls.py 定义 REST 路径到处理器的映射responses.py 解析请求体、调用 Backend 并序列化响应models.py 承载全部业务逻辑与内存状态utils.py 提供 ID 生成、参数校验等辅助函数exceptions.py 定义与真实 AWS 对齐的异常类型。Backend 在 models.py 中以五个内存字典保存全部状态networks、members、proposals、invitations、nodes并通过managedblockchain_backends BackendDict(ManagedBlockchainBackend, managedblockchain)按账号与区域隔离状态。这意味着不同 region如us-east-1与eu-west-1下创建的网络互不可见与真实 AWS 的多区域隔离语义一致。URL 路由集中在 urls.py可归纳为以下 REST 资源树/networks /networks/{networkId} /networks/{networkId}/proposals /networks/{networkId}/proposals/{proposalId} /networks/{networkId}/proposals/{proposalId}/votes /networks/{networkId}/members /networks/{networkId}/members/{memberId} /networks/{networkId}/members/{memberId}/nodes /networks/{networkId}/nodes # botocore 1.19.41 的 API 变体 /invitations /invitations/{invitationId}注意注释里标注的兼容性细节botocore 1.19.41 之后memberId从 URL 路径迁移到了查询字符串或请求体中因此 utils.py 的memberid_from_managedblockchain_request会依次从 URL、query string、body 三处提取urls.py 也额外注册了不带 memberId 的节点路由。资源 ID 与 Fabric 端点模拟的真实性来源为了让模拟数据看起来和真实 AWS 返回一致moto 严格按照 AWS 的格式生成资源 ID见 utils.py资源前缀长度示例Networkn-28 字符n-ABCDEFGHIJKLMNOP0123456789Memberm-28 字符m-...Proposalp-28 字符p-...Invitationin-28 字符in-...Nodend-28 字符nd-...测试 test_managedblockchain_networks.py 明确断言network_id.startswith(n-)且长度为 28与真实 AWS 的 ID 规范完全一致。模拟的另一个关键点是 Fabric 端点。模型层在创建资源时会生成结构化的 DNS 端点见 models.py、models.py 与 models.pyOrdering Serviceorderer.{networkId}.managedblockchain.{region}.amazonaws.com:30001CA 服务ca.{memberId}.{networkId}.managedblockchain.{region}.amazonaws.com:30002Peer 服务{nodeId}.{networkId}.{memberId}.managedblockchain.{region}.amazonaws.com:30003Peer Event 服务{nodeId}.{networkId}.{memberId}.managedblockchain.{region}.amazonaws.com:30004get_network返回的FrameworkAttributes.Fabric会携带Edition与OrderingServiceEndpoint与测试 test_managedblockchain_networks.py 中的断言完全吻合。Edition 配额与受控取值create_network 的校验规则create_network是整棵资源树的根操作。真实 AWS 将 Fabric 网络分为两种 Editionmoto 在 models.py 中以常量精确复刻了它们的配额与实例类型限制FRAMEWORKS [HYPERLEDGER_FABRIC] FRAMEWORKVERSIONS [1.2] EDITIONS { STARTER: { MaxMembers: 5, MaxNodesPerMember: 2, AllowedNodeInstanceTypes: [bc.t3.small, bc.t3.medium], }, STANDARD: { MaxMembers: 14, MaxNodesPerMember: 3, AllowedNodeInstanceTypes: [bc.t3, bc.m5, bc.c5], }, }create_network的校验链models.py依次检查Framework必须为HYPERLEDGER_FABRIC否则抛BadRequestException(Invalid request body)FrameworkVersion必须为1.2否则抛出Invalid version {version} requested for framework HYPERLEDGER_FABRICFrameworkConfiguration.Fabric.Edition必须为STARTER或STANDARD。对应测试 test_managedblockchain_networks.py 覆盖了这三种非法输入的BadRequestException分支。创建成功后Backend 会同时生成 Network 与首个 Member即创建者成员一次性返回{NetworkId: ..., MemberId: ...}。可直接复制的完整入参测试套件把默认参数沉淀在 helpers.py是编写真实调用时最可靠的参考frameworkconfiguration {Fabric: {Edition: STARTER}} votingpolicy { ApprovalThresholdPolicy: { ThresholdPercentage: 50, ProposalDurationInHours: 24, ThresholdComparator: GREATER_THAN_OR_EQUAL_TO, } } memberconfiguration { Name: testmember1, Description: Test Member 1, FrameworkConfiguration: { Fabric: {AdminUsername: admin, AdminPassword: Admin12345} }, LogPublishingConfiguration: { Fabric: {CaLogs: {Cloudwatch: {Enabled: False}}} }, }其中VotingPolicy是网络治理的核心配置ThresholdPercentage提案通过所需的最低赞成票百分比floatProposalDurationInHours提案有效期超期后自动标记为EXPIREDThresholdComparator比较方式取值GREATER_THAN_OR_EQUAL_TO或GREATER_THAN。成员管理邀请制加入与密码强校验真实 Managed Blockchain 中新成员必须通过提案邀请、再凭邀请 ID 加入。moto 完整复刻了这一流程提案被投票通过后APPROVEDBackend 为Actions.Invitations中的每个Principal生成一条PENDING状态的邀请models.py邀请有效期固定为 7 天models.py受邀方调用create_member时传入InvitationIdBackend 校验邀请必须存在且状态为PENDINGmodels.py校验通过后创建成员并将邀请状态置为ACCEPTEDmodels.py。create_member还会执行三项额外校验models.py成员名唯一性同一网络内Name不得重复否则抛InvalidRequestExceptionEdition 成员上限STARTER 网络最多 5 个成员、STANDARD 最多 14 个超出抛ResourceLimitExceededException管理员密码强度调用 utils.py 的admin_password_ok必须同时包含小写字母、大写字母与数字且不能包含 /这四类字符。update_member允许更新LogPublishingConfigurationdelete_member则执行软删除成员状态置为DELETEDget_member对已删除成员抛ResourceNotFoundException但list_members仍会将其列出models.py这与真实 API 的列表可见、详情 404行为一致。成员删除的级联效应delete_member还有两个重要的联动行为models.py若网络中所有成员均已删除则整个网络被移除del self.networks[network_id]并且所有指向该网络的邀请状态被置为DELETED该成员名下的所有节点被一并删除。节点管理实例类型与可用区的精细校验create_node为成员添加 Peer 节点其校验逻辑models.py体现了 moto 对细节的追求节点数不得超过 Edition 的MaxNodesPerMemberSTARTER 2 个、STANDARD 3 个实例类型必须匹配AllowedNodeInstanceTypes的前缀正则不区分大小写例如 STANDARD 网络接受bc.t3.large、bc.m5.xlarge、bc.c5.large等衍生型号STARTER 网络则只能使用bc.t3.small或bc.t3.medium这两种精确型号可用区必须匹配{region}{字母}的正则如us-east-1a否则抛InvalidRequestException(Availability Zone is not valid)。测试文件 test_managedblockchain_nodes.py 对实例类型与可用区校验均有专门用例。节点同样支持update_node更新日志发布配置与delete_node软删除删除后get_node返回 404。提案与投票阈值判定、过期与平局处理提案是网络的治理中枢。create_proposalmodels.py要求传入Actions其结构支持两类治理动作# 邀请新成员加入 actions_invite {Invitations: [{Principal: 123456789012}]} # 移除现有成员 actions_remove {Removals: [{MemberId: m-XXXXXXXXXXXXXXXXXXXXXXXXXX}]}校验规则包括Invitations中的Principal必须是 12 位数字的账号 ID正则[0-9]{12}Removals中的MemberId必须存在于当前网络。值得注意的兼容性细节源码注释明确说明CLI 文档声称 Invitations 与 Removals 不能同时传入但实际实现允许同时传入且不报错models.py这是 moto 对真实 AWS 行为的刻意模拟。投票状态机每个提案ManagedBlockchainProposalmodels.py维护五个状态字段statusIN_PROGRESS→APPROVED/REJECTED/EXPIREDyes_vote_count/no_vote_count赞成与反对票数outstanding_vote_count尚未投票的成员数初始等于网络成员总数expirationdate创建时间 ProposalDurationInHours。vote_on_proposal的完整校验链models.py依次拒绝网络/提案/投票成员不存在、投票值不是YES/NO、提案已过期、提案已定案、成员重复投票抛ResourceAlreadyExistsExceptionHTTP 409。每投一票后set_vote重新计算赞成与反对百分比并判定models.pyThresholdComparator GREATER_THAN_OR_EQUAL_TO赞成或反对百分比 ThresholdPercentage即分别判定APPROVED/REJECTEDThresholdComparator GREATER_THAN严格大于阈值才定案平局处理当比较器为GREATER_THAN、全部成员投完票且赞成与反对百分比相等时强制判定REJECTED。提案一旦APPROVEDvote_on_proposal会立即执行治理动作models.py为Invitations生成PENDING邀请、为Removals执行delete_member。也就是说模拟环境中提案通过 → 邀请/移除生效是同步完成的调用vote_on_proposal返回后即可用list_invitations或list_members观察到结果。list_proposals与get_proposal在读取前会先调用check_to_expire_proposal将超过expirationdate的提案状态刷新为EXPIREDmodels.py。邀请生命周期与异常模型list_invitations返回全部邀请无论属于哪个网络邀请状态在PENDING/ACCEPTED/REJECTED/DELETED之间流转接受邀请由create_member触发reject_invitation将其置为REJECTED网络整体删除时联动置为DELETEDmodels.py。对应测试见 test_managedblockchain_invitations.py。异常体系定义在 exceptions.py所有异常继承自JsonRESTError并设置与真实 AWS 一致的 HTTP 状态码与x-amzn-ErrorType响应头异常类HTTP 状态码典型触发场景BadRequestException400非法的 Framework / Edition / 投票值 / 弱密码InvalidRequestException400无效邀请 ID、重复成员名、非法账号 ID、非法可用区ResourceNotFoundException404网络 / 成员 / 节点 / 提案 / 邀请不存在或已删除ResourceAlreadyExistsException409成员对同一提案重复投票ResourceLimitExceededException429成员数或节点数超过 Edition 上限这些异常通过 botocore 抛给调用方的是ClientError测试用例中统一使用pytest.raises(ClientError)并断言err[Code]与err[Message]来验证行为。端到端演练用 mock_aws 跑通完整治理流程以下脚本将完整演示「创建网络 → 创建提案邀请新成员 → 投票通过 → 受邀方加入 → 查看邀请与成员列表」的整条链路可直接在本地运行import boto3 from moto import mock_aws mock_aws def test_managedblockchain_full_flow(): conn boto3.client(managedblockchain, region_nameus-east-1) # 1. 创建 STARTER Edition 网络同时创建首个成员 create conn.create_network( Nametestnetwork1, FrameworkHYPERLEDGER_FABRIC, FrameworkVersion1.2, FrameworkConfiguration{Fabric: {Edition: STARTER}}, VotingPolicy{ ApprovalThresholdPolicy: { ThresholdPercentage: 50, ProposalDurationInHours: 24, ThresholdComparator: GREATER_THAN_OR_EQUAL_TO, } }, MemberConfiguration{ Name: testmember1, FrameworkConfiguration: { Fabric: {AdminUsername: admin, AdminPassword: Admin12345} }, LogPublishingConfiguration: { Fabric: {CaLogs: {Cloudwatch: {Enabled: False}}} }, }, ) network_id, creator_member_id create[NetworkId], create[MemberId] assert network_id.startswith(n-) and len(network_id) 28 # 2. 创建提案邀请账号 123456789012 加入 proposal_id conn.create_proposal( NetworkIdnetwork_id, MemberIdcreator_member_id, Actions{Invitations: [{Principal: 123456789012}]}, DescriptionAdding a new member, )[ProposalId] # 3. 唯一成员投赞成票提案立即 APPROVED 并生成邀请 conn.vote_on_proposal( NetworkIdnetwork_id, ProposalIdproposal_id, VoterMemberIdcreator_member_id, VoteYES, ) assert conn.get_proposal(NetworkIdnetwork_id, ProposalIdproposal_id)[Proposal][ Status ] APPROVED # 4. 受邀方凭邀请 ID 加入网络 invitations conn.list_invitations()[Invitations] invitation_id next( inv[InvitationId] for inv in invitations if inv[NetworkSummary][Id] network_id and inv[Status] PENDING ) new_member_id conn.create_member( InvitationIdinvitation_id, NetworkIdnetwork_id, MemberConfiguration{ Name: testmember2, FrameworkConfiguration: { Fabric: {AdminUsername: admin2, AdminPassword: Admin12345} }, LogPublishingConfiguration: { Fabric: {CaLogs: {Cloudwatch: {Enabled: False}}} }, }, )[MemberId] # 5. 网络内应有 2 个成员节点/端点可查询 members conn.list_members(NetworkIdnetwork_id)[Members] assert len(members) 2 network conn.get_network(NetworkIdnetwork_id)[Network] assert network[FrameworkAttributes][Fabric][Edition] STARTER assert orderer. in network[FrameworkAttributes][Fabric][OrderingServiceEndpoint] test_managedblockchain_full_flow()运行方式与 moto 其他服务完全一致pip install moto[managedblockchain]后用mock_aws装饰器包裹测试或用moto_server启动独立模拟服务供任意 HTTP 客户端访问。所有状态均保存在进程内存中测试结束后自动清空无需清理任何资源。局限与注意事项Accessor 相关 API 未实现create_accessor/get_accessor/list_accessors/delete_accessor以及全部标签 API 不在模拟范围内涉及以太坊网络场景的代码无法用当前版本验证模拟的非持久化BackendDict按进程隔离重启即丢失全部网络状态多进程场景需借助 moto 的 Server 模式集中管理状态软删除语义delete_member/delete_node只是标记DELETED而非真正移除list_members/list_nodes仍能看到已删除资源这与真实 AWS 的列表行为一致但若你的业务代码依赖列表不出现已删除资源需要在测试断言中特别留意区域隔离网络、成员等状态按 region 隔离跨区域操作前必须先切换到对应区域的 client。以上行为均有明确的源码与测试依据配额与实例类型见 models.py投票判定见 models.py默认参数与完整调用示例见 helpers.py 与 test_managedblockchain_networks.py。借助这套实现你可以在 CI 中无成本地构建 Managed Blockchain 相关的集成测试覆盖网络创建、成员邀请、节点配置、提案投票等真实业务路径。赞分享Mock测试【免费下载链接】motoA library that allows you to easily mock out tests based on AWS infrastructure.项目地址https://gitcode.com/gh_mirrors/mo/moto点击查看免费下载相关推荐Moto 中 Amazon Managed Prometheusamp服务的模拟实现与实战指南Moto 中 Amazon Managed Prometheusamp服务的模拟实现与实战指南 Amazon Managed PrometheusAMPMock测试moto AWS DataSync 模拟服务详解功能覆盖、状态机实现与测试实战moto AWS DataSync 模拟服务详解功能覆盖、状态机实现与测试实战 本文围绕 moto 仓库中 AWS DataSync 服务的模拟mock实Mock测试如何用 self-improvement-loops 技能防止自我修改的智能体循环发生奖励作弊如何用 self improvement loops 技能防止自我修改的智能体循环发生奖励作弊 当你的智能体循环开始修改自己——agent 挖掘自己的失败轨迹Mock测试上一篇终极跨语言代码依赖分析工具Depends5步快速掌握源码结构可视化下一篇3个关键场景用好d3-org-chart让组织管理效率提升300%创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
02
RELATED NEWS

相关资讯

更多网站建设与数字化升级内容

03
WHY YAOTU

想打造同款高转化官网?

懂行业、懂生意,从建站到增长一站式陪跑

◈

场景化定制

不做模板站,围绕你的业务场景量身设计,小众不撞款。

◐

营销型架构

以转化目标组织内容与路径,让官网真正带来询盘。

▲

全周期服务

设计、开发、运营、运维一体,上线只是开始。

免费获取你的建站方案

留下需求,专属顾问 24 小时内为你输出方案建议。