☰
moto 中 Amazon Managed Blockchain 的本地模拟实现:API 覆盖、状态机与实战演练
2026/9/26 7:48:01 网站建设 项目流程
  • Mock
  • 测试

【免费下载链接】moto

A 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-ABCDEFGHIJKLMNOP0123456789
Memberm-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.py:

  • Ordering Service:orderer.{networkId}.managedblockchain.{region}.amazonaws.com:30001
  • CA 服务:ca.{memberId}.{networkId}.managedblockchain.{region}.amazonaws.com:30002
  • Peer 服务:{nodeId}.{networkId}.{memberId}.managedblockchain.{region}.amazonaws.com:30003
  • Peer Event 服务:{nodeId}.{networkId}.{memberId}.managedblockchain.{region}.amazonaws.com:30004

get_network返回的FrameworkAttributes.Fabric会携带Edition与OrderingServiceEndpoint,与测试 test_managedblockchain_networks.py 中的断言完全吻合。

Edition 配额与受控取值:create_network 的校验规则

create_network是整棵资源树的根操作。真实 AWS 将 Fabric 网络分为两种 Edition,moto 在 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)依次检查:

  1. Framework必须为HYPERLEDGER_FABRIC,否则抛BadRequestException("Invalid request body");
  2. FrameworkVersion必须为1.2,否则抛出Invalid version {version} requested for framework HYPERLEDGER_FABRIC;
  3. FrameworkConfiguration.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:提案通过所需的最低赞成票百分比(float);
  • ProposalDurationInHours:提案有效期,超期后自动标记为EXPIRED;
  • ThresholdComparator:比较方式,取值GREATER_THAN_OR_EQUAL_TO或GREATER_THAN。

成员管理:邀请制加入与密码强校验

真实 Managed Blockchain 中,新成员必须通过提案邀请、再凭邀请 ID 加入。moto 完整复刻了这一流程:

  1. 提案被投票通过后(APPROVED),Backend 为Actions.Invitations中的每个Principal生成一条PENDING状态的邀请(models.py),邀请有效期固定为 7 天(models.py);
  2. 受邀方调用create_member时传入InvitationId,Backend 校验邀请必须存在且状态为PENDING(models.py);
  3. 校验通过后创建成员,并将邀请状态置为ACCEPTED(models.py)。

create_member还会执行三项额外校验(models.py):

  • 成员名唯一性:同一网络内Name不得重复,否则抛InvalidRequestException;
  • Edition 成员上限:STARTER 网络最多 5 个成员、STANDARD 最多 14 个,超出抛ResourceLimitExceededException;
  • 管理员密码强度:调用 utils.py 的admin_password_ok,必须同时包含小写字母、大写字母与数字,且不能包含' " @ /这四类字符。

update_member允许更新LogPublishingConfiguration,delete_member则执行软删除:成员状态置为DELETED,get_member对已删除成员抛ResourceNotFoundException,但list_members仍会将其列出(models.py),这与真实 API 的"列表可见、详情 404"行为一致。

成员删除的级联效应

delete_member还有两个重要的联动行为(models.py):

  1. 若网络中所有成员均已删除,则整个网络被移除(del self.networks[network_id]),并且所有指向该网络的邀请状态被置为DELETED;
  2. 该成员名下的所有节点被一并删除。

节点管理:实例类型与可用区的精细校验

create_node为成员添加 Peer 节点,其校验逻辑(models.py)体现了 moto 对细节的追求:

  • 节点数不得超过 Edition 的MaxNodesPerMember(STARTER 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_proposal(models.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 行为的刻意模拟。

投票状态机

每个提案(ManagedBlockchainProposal,models.py)维护五个状态字段:

  • status:IN_PROGRESS→APPROVED/REJECTED/EXPIRED;
  • yes_vote_count/no_vote_count:赞成与反对票数;
  • outstanding_vote_count:尚未投票的成员数(初始等于网络成员总数);
  • expirationdate:创建时间 +ProposalDurationInHours。

vote_on_proposal的完整校验链(models.py)依次拒绝:网络/提案/投票成员不存在、投票值不是YES/NO、提案已过期、提案已定案、成员重复投票(抛ResourceAlreadyExistsException,HTTP 409)。

每投一票后set_vote重新计算赞成与反对百分比并判定(models.py):

  • ThresholdComparator == "GREATER_THAN_OR_EQUAL_TO":赞成或反对百分比>= ThresholdPercentage即分别判定APPROVED/REJECTED;
  • ThresholdComparator == "GREATER_THAN":严格大于阈值才定案;
  • 平局处理:当比较器为GREATER_THAN、全部成员投完票且赞成与反对百分比相等时,强制判定REJECTED。

提案一旦APPROVED,vote_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的提案状态刷新为EXPIRED(models.py)。

邀请生命周期与异常模型

list_invitations返回全部邀请(无论属于哪个网络),邀请状态在PENDING/ACCEPTED/REJECTED/DELETED之间流转:接受邀请由create_member触发,reject_invitation将其置为REJECTED,网络整体删除时联动置为DELETED(models.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_name="us-east-1") # 1. 创建 STARTER Edition 网络(同时创建首个成员) create = conn.create_network( Name="testnetwork1", Framework="HYPERLEDGER_FABRIC", FrameworkVersion="1.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( NetworkId=network_id, MemberId=creator_member_id, Actions={"Invitations": [{"Principal": "123456789012"}]}, Description="Adding a new member", )["ProposalId"] # 3. 唯一成员投赞成票,提案立即 APPROVED 并生成邀请 conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=creator_member_id, Vote="YES", ) assert conn.get_proposal(NetworkId=network_id, ProposalId=proposal_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( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration={ "Name": "testmember2", "FrameworkConfiguration": { "Fabric": {"AdminUsername": "admin2", "AdminPassword": "Admin12345"} }, "LogPublishingConfiguration": { "Fabric": {"CaLogs": {"Cloudwatch": {"Enabled": False}}} }, }, )["MemberId"] # 5. 网络内应有 2 个成员,节点/端点可查询 members = conn.list_members(NetworkId=network_id)["Members"] assert len(members) == 2 network = conn.get_network(NetworkId=network_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
  • 测试

【免费下载链接】moto

A library that allows you to easily mock out tests based on AWS infrastructure.

项目地址:https://gitcode.com/gh_mirrors/mo/moto
点击查看免费下载
上一篇:终极跨语言代码依赖分析工具Depends:5步快速掌握源码结构可视化
下一篇:3个关键场景用好d3-org-chart,让组织管理效率提升300%

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询