Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Techdebt: Simplify ManagedBlockchain setup & tests (#6749)

Browse Source
This commit is contained in:
Bert Blommers 2023-09-01 07:07:54 +00:00 committed by GitHub
parent 5dd649378c
commit 093052bd9b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 218 additions and 466 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
moto/core/models.py
View File

@ -392,7 +392,10 @@ class ServerModeMockAWS(BaseMockAWS):
region = self._get_region(*args, **kwargs) region = self._get_region(*args, **kwargs)
if region: if region:
if "config" in kwargs: if "config" in kwargs:
kwargs["config"].__dict__["user_agent_extra"] += " region/" + region user_agent = kwargs["config"].__dict__.get("user_agent_extra") or ""
kwargs["config"].__dict__[
"user_agent_extra"
] = f"{user_agent} region/{region}"
else: else:
config = Config(user_agent_extra="region/" + region) config = Config(user_agent_extra="region/" + region)
kwargs["config"] = config kwargs["config"] = config

17
moto/managedblockchain/exceptions.py
View File

@ -1,21 +1,6 @@
import json import json
from moto.core.common_types import TYPE_RESPONSE
from moto.core.exceptions import JsonRESTError from moto.core.exceptions import JsonRESTError
from functools import wraps from typing import Any, List, Tuple
from typing import Any, Callable, List, Tuple
def exception_handler(
f: Callable[[Any, Any, str, Any], TYPE_RESPONSE]
) -> Callable[[Any, Any, str, Any], TYPE_RESPONSE]:
@wraps(f)
def _wrapper(*args: Any, **kwargs: Any) -> TYPE_RESPONSE: # type: ignore[misc]
try:
return f(*args, **kwargs)
except ManagedBlockchainClientError as err:
return err.code, err.get_headers(), err.description # type: ignore
return _wrapper
class ManagedBlockchainClientError(JsonRESTError): class ManagedBlockchainClientError(JsonRESTError):

380
moto/managedblockchain/responses.py
View File

@ -1,10 +1,6 @@
import json import json
from typing import Any, Dict, Optional
from urllib.parse import parse_qs
from moto.core.common_types import TYPE_RESPONSE
from moto.core.responses import BaseResponse from moto.core.responses import BaseResponse
from .exceptions import exception_handler
from .models import managedblockchain_backends, ManagedBlockchainBackend from .models import managedblockchain_backends, ManagedBlockchainBackend
from .utils import ( from .utils import (
networkid_from_managedblockchain_url, networkid_from_managedblockchain_url,
@ -23,39 +19,20 @@ class ManagedBlockchainResponse(BaseResponse):
def backend(self) -> ManagedBlockchainBackend: def backend(self) -> ManagedBlockchainBackend:
return managedblockchain_backends[self.current_account][self.region] return managedblockchain_backends[self.current_account][self.region]
@exception_handler def list_networks(self) -> str:
def network_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] networks = self.backend.list_networks()
self.setup_class(request, full_url, headers) return json.dumps({"Networks": [network.to_dict() for network in networks]})
return self._network_response(request, headers)
def _network_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] def create_network(self) -> str:
method = request.method name = self._get_param("Name")
if method == "GET": framework = self._get_param("Framework")
return self._all_networks_response(headers) frameworkversion = self._get_param("FrameworkVersion")
elif method == "POST": frameworkconfiguration = self._get_param("FrameworkConfiguration")
json_body = json.loads(self.body) voting_policy = self._get_param("VotingPolicy")
return self._network_response_post(json_body, headers) member_configuration = self._get_param("MemberConfiguration")
def _all_networks_response(self, headers: Any) -> TYPE_RESPONSE:
mbcnetworks = self.backend.list_networks()
response = json.dumps(
{"Networks": [mbcnetwork.to_dict() for mbcnetwork in mbcnetworks]}
)
headers["content-type"] = "application/json"
return 200, headers, response
def _network_response_post(
self, json_body: Dict[str, Any], headers: Any
) -> TYPE_RESPONSE:
name = json_body["Name"]
framework = json_body["Framework"]
frameworkversion = json_body["FrameworkVersion"]
frameworkconfiguration = json_body["FrameworkConfiguration"]
voting_policy = json_body["VotingPolicy"]
member_configuration = json_body["MemberConfiguration"]
# Optional # Optional
description = json_body.get("Description", None) description = self._get_param("Description", None)
response = self.backend.create_network( response = self.backend.create_network(
name, name,
@ -66,257 +43,110 @@ class ManagedBlockchainResponse(BaseResponse):
member_configuration, member_configuration,
description, description,
) )
return 200, headers, json.dumps(response) return json.dumps(response)
@exception_handler def get_network(self) -> str:
def networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers)
return self._networkid_response(request, full_url, headers)
def _networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
if method == "GET":
network_id = networkid_from_managedblockchain_url(full_url)
return self._networkid_response_get(network_id, headers)
def _networkid_response_get(self, network_id: str, headers: Any) -> TYPE_RESPONSE:
mbcnetwork = self.backend.get_network(network_id) mbcnetwork = self.backend.get_network(network_id)
response = json.dumps({"Network": mbcnetwork.get_format()}) return json.dumps({"Network": mbcnetwork.get_format()})
headers["content-type"] = "application/json"
return 200, headers, response
@exception_handler def list_proposals(self) -> str:
def proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers)
return self._proposal_response(request, full_url, headers)
def _proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
network_id = networkid_from_managedblockchain_url(full_url)
if method == "GET":
return self._all_proposals_response(network_id, headers)
elif method == "POST":
json_body = json.loads(self.body)
return self._proposal_response_post(network_id, json_body, headers)
def _all_proposals_response(self, network_id: str, headers: Any) -> TYPE_RESPONSE:
proposals = self.backend.list_proposals(network_id) proposals = self.backend.list_proposals(network_id)
response = json.dumps( return json.dumps({"Proposals": [proposal.to_dict() for proposal in proposals]})
{"Proposals": [proposal.to_dict() for proposal in proposals]}
)
headers["content-type"] = "application/json"
return 200, headers, response
def _proposal_response_post( def create_proposal(self) -> str:
self, network_id: str, json_body: Dict[str, Any], headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: memberid = self._get_param("MemberId")
memberid = json_body["MemberId"] actions = self._get_param("Actions")
actions = json_body["Actions"]
# Optional # Optional
description = json_body.get("Description", None) description = self._get_param("Description", None)
response = self.backend.create_proposal( response = self.backend.create_proposal(
network_id, memberid, actions, description network_id, memberid, actions, description
) )
return 200, headers, json.dumps(response) return json.dumps(response)
@exception_handler def get_proposal(self) -> str:
def proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers) proposal_id = proposalid_from_managedblockchain_url(self.path)
return self._proposalid_response(request, full_url, headers)
def _proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
network_id = networkid_from_managedblockchain_url(full_url)
if method == "GET":
proposal_id = proposalid_from_managedblockchain_url(full_url)
return self._proposalid_response_get(network_id, proposal_id, headers)
def _proposalid_response_get(
self, network_id: str, proposal_id: str, headers: Any
) -> TYPE_RESPONSE:
proposal = self.backend.get_proposal(network_id, proposal_id) proposal = self.backend.get_proposal(network_id, proposal_id)
response = json.dumps({"Proposal": proposal.get_format()}) return json.dumps({"Proposal": proposal.get_format()})
headers["content-type"] = "application/json"
return 200, headers, response
@exception_handler def list_proposal_votes(self) -> str:
def proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers) proposal_id = proposalid_from_managedblockchain_url(self.path)
return self._proposal_votes_response(request, full_url, headers)
def _proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
network_id = networkid_from_managedblockchain_url(full_url)
proposal_id = proposalid_from_managedblockchain_url(full_url)
if method == "GET":
return self._all_proposal_votes_response(network_id, proposal_id, headers)
elif method == "POST":
json_body = json.loads(self.body)
return self._proposal_votes_response_post(
network_id, proposal_id, json_body, headers
)
def _all_proposal_votes_response(
self, network_id: str, proposal_id: str, headers: Any
) -> TYPE_RESPONSE:
proposalvotes = self.backend.list_proposal_votes(network_id, proposal_id) proposalvotes = self.backend.list_proposal_votes(network_id, proposal_id)
response = json.dumps({"ProposalVotes": proposalvotes}) return json.dumps({"ProposalVotes": proposalvotes})
headers["content-type"] = "application/json"
return 200, headers, response
def _proposal_votes_response_post( def vote_on_proposal(self) -> str:
self, network_id: str, proposal_id: str, json_body: Dict[str, Any], headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: proposal_id = proposalid_from_managedblockchain_url(self.path)
votermemberid = json_body["VoterMemberId"] votermemberid = self._get_param("VoterMemberId")
vote = json_body["Vote"] vote = self._get_param("Vote")
self.backend.vote_on_proposal(network_id, proposal_id, votermemberid, vote) self.backend.vote_on_proposal(network_id, proposal_id, votermemberid, vote)
return 200, headers, "" return ""
@exception_handler def list_invitations(self) -> str:
def invitation_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers)
return self._invitation_response(request, headers)
def _invitation_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
if method == "GET":
return self._all_invitation_response(headers)
def _all_invitation_response(self, headers: Any) -> TYPE_RESPONSE:
invitations = self.backend.list_invitations() invitations = self.backend.list_invitations()
response = json.dumps( return json.dumps(
{"Invitations": [invitation.to_dict() for invitation in invitations]} {"Invitations": [invitation.to_dict() for invitation in invitations]}
) )
headers["content-type"] = "application/json"
return 200, headers, response
@exception_handler def reject_invitation(self) -> str:
def invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] invitation_id = invitationid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers)
return self._invitationid_response(request, full_url, headers)
def _invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
if method == "DELETE":
invitation_id = invitationid_from_managedblockchain_url(full_url)
return self._invitationid_response_delete(invitation_id, headers)
def _invitationid_response_delete(
self, invitation_id: str, headers: Any
) -> TYPE_RESPONSE:
self.backend.reject_invitation(invitation_id) self.backend.reject_invitation(invitation_id)
headers["content-type"] = "application/json" return ""
return 200, headers, ""
@exception_handler def list_members(self) -> str:
def member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers)
return self._member_response(request, full_url, headers)
def _member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
network_id = networkid_from_managedblockchain_url(full_url)
if method == "GET":
return self._all_members_response(network_id, headers)
elif method == "POST":
json_body = json.loads(self.body)
return self._member_response_post(network_id, json_body, headers)
def _all_members_response(self, network_id: str, headers: Any) -> TYPE_RESPONSE:
members = self.backend.list_members(network_id) members = self.backend.list_members(network_id)
response = json.dumps({"Members": [member.to_dict() for member in members]}) return json.dumps({"Members": [member.to_dict() for member in members]})
headers["content-type"] = "application/json"
return 200, headers, response
def _member_response_post( def create_member(self) -> str:
self, network_id: str, json_body: Dict[str, Any], headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: invitationid = self._get_param("InvitationId")
invitationid = json_body["InvitationId"] member_configuration = self._get_param("MemberConfiguration")
member_configuration = json_body["MemberConfiguration"]
response = self.backend.create_member( response = self.backend.create_member(
invitationid, network_id, member_configuration invitationid, network_id, member_configuration
) )
return 200, headers, json.dumps(response) return json.dumps(response)
@exception_handler def get_member(self) -> str:
def memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers) member_id = memberid_from_managedblockchain_request(self.uri, self.body)
return self._memberid_response(request, full_url, headers)
def _memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
network_id = networkid_from_managedblockchain_url(full_url)
member_id = memberid_from_managedblockchain_request(full_url, self.body)
if method == "GET":
return self._memberid_response_get(network_id, member_id, headers)
elif method == "PATCH":
json_body = json.loads(self.body)
return self._memberid_response_patch(
network_id, member_id, json_body, headers
)
elif method == "DELETE":
return self._memberid_response_delete(network_id, member_id, headers)
def _memberid_response_get(
self, network_id: str, member_id: str, headers: Dict[str, Any]
) -> TYPE_RESPONSE:
member = self.backend.get_member(network_id, member_id) member = self.backend.get_member(network_id, member_id)
response = json.dumps({"Member": member.get_format()}) return json.dumps({"Member": member.get_format()})
headers["content-type"] = "application/json"
return 200, headers, response
def _memberid_response_patch( def update_member(self) -> str:
self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: member_id = memberid_from_managedblockchain_request(self.uri, self.body)
logpublishingconfiguration = json_body["LogPublishingConfiguration"] logpublishingconfiguration = self._get_param("LogPublishingConfiguration")
self.backend.update_member(network_id, member_id, logpublishingconfiguration) self.backend.update_member(network_id, member_id, logpublishingconfiguration)
return 200, headers, "" return ""
def _memberid_response_delete( def delete_member(self) -> str:
self, network_id: str, member_id: str, headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: member_id = memberid_from_managedblockchain_request(self.uri, self.body)
self.backend.delete_member(network_id, member_id) self.backend.delete_member(network_id, member_id)
headers["content-type"] = "application/json" return ""
return 200, headers, ""
@exception_handler def list_nodes(self) -> str:
def node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers) member_id = memberid_from_managedblockchain_request(self.uri, self.body)
return self._node_response(request, full_url, headers) status = self._get_param("status")
def _node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
querystring = parse_qs(self.parsed_url.query, keep_blank_values=True)
network_id = networkid_from_managedblockchain_url(full_url)
member_id = memberid_from_managedblockchain_request(full_url, self.body)
if method == "GET":
status = None
if "status" in querystring:
status = querystring["status"][0]
return self._all_nodes_response(network_id, member_id, status, headers)
elif method == "POST":
json_body = json.loads(self.body)
return self._node_response_post(network_id, member_id, json_body, headers)
def _all_nodes_response(
self, network_id: str, member_id: str, status: Optional[str], headers: Any
) -> TYPE_RESPONSE:
nodes = self.backend.list_nodes(network_id, member_id, status) nodes = self.backend.list_nodes(network_id, member_id, status)
response = json.dumps({"Nodes": [node.to_dict() for node in nodes]}) return json.dumps({"Nodes": [node.to_dict() for node in nodes]})
headers["content-type"] = "application/json"
return 200, headers, response
def _node_response_post( def create_node(self) -> str:
self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: member_id = memberid_from_managedblockchain_request(self.uri, self.body)
instancetype = json_body["NodeConfiguration"]["InstanceType"] instancetype = self._get_param("NodeConfiguration")["InstanceType"]
availabilityzone = json_body["NodeConfiguration"]["AvailabilityZone"] availabilityzone = self._get_param("NodeConfiguration")["AvailabilityZone"]
logpublishingconfiguration = json_body["NodeConfiguration"][ logpublishingconfiguration = self._get_param("NodeConfiguration")[
"LogPublishingConfiguration" "LogPublishingConfiguration"
] ]
@ -327,53 +157,27 @@ class ManagedBlockchainResponse(BaseResponse):
instancetype, instancetype,
logpublishingconfiguration, logpublishingconfiguration,
) )
return 200, headers, json.dumps(response) return json.dumps(response)
@exception_handler def get_node(self) -> str:
def nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] network_id = networkid_from_managedblockchain_url(self.path)
self.setup_class(request, full_url, headers) member_id = memberid_from_managedblockchain_request(self.uri, self.body)
return self._nodeid_response(request, full_url, headers) node_id = nodeid_from_managedblockchain_url(self.path)
def _nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method
network_id = networkid_from_managedblockchain_url(full_url)
member_id = memberid_from_managedblockchain_request(full_url, self.body)
node_id = nodeid_from_managedblockchain_url(full_url)
if method == "GET":
return self._nodeid_response_get(network_id, member_id, node_id, headers)
elif method == "PATCH":
json_body = json.loads(self.body)
return self._nodeid_response_patch(
network_id, member_id, node_id, json_body, headers
)
elif method == "DELETE":
return self._nodeid_response_delete(network_id, member_id, node_id, headers)
def _nodeid_response_get(
self, network_id: str, member_id: str, node_id: str, headers: Any
) -> TYPE_RESPONSE:
node = self.backend.get_node(network_id, member_id, node_id) node = self.backend.get_node(network_id, member_id, node_id)
response = json.dumps({"Node": node.get_format()}) return json.dumps({"Node": node.get_format()})
headers["content-type"] = "application/json"
return 200, headers, response
def _nodeid_response_patch( def update_node(self) -> str:
self, network_id = networkid_from_managedblockchain_url(self.path)
network_id: str, member_id = memberid_from_managedblockchain_request(self.uri, self.body)
member_id: str, node_id = nodeid_from_managedblockchain_url(self.path)
node_id: str,
json_body: Dict[str, Any],
headers: Any,
) -> TYPE_RESPONSE:
logpublishingconfiguration = json_body
self.backend.update_node( self.backend.update_node(
network_id, member_id, node_id, logpublishingconfiguration network_id, member_id, node_id, logpublishingconfiguration=self.body
) )
return 200, headers, "" return ""
def _nodeid_response_delete( def delete_node(self) -> str:
self, network_id: str, member_id: str, node_id: str, headers: Any network_id = networkid_from_managedblockchain_url(self.path)
) -> TYPE_RESPONSE: member_id = memberid_from_managedblockchain_request(self.uri, self.body)
node_id = nodeid_from_managedblockchain_url(self.path)
self.backend.delete_node(network_id, member_id, node_id) self.backend.delete_node(network_id, member_id, node_id)
headers["content-type"] = "application/json" return ""
return 200, headers, ""

56
moto/managedblockchain/urls.py
View File

@ -3,47 +3,19 @@ from .responses import ManagedBlockchainResponse
url_bases = [r"https?://managedblockchain\.(.+)\.amazonaws.com"] url_bases = [r"https?://managedblockchain\.(.+)\.amazonaws.com"]
url_paths = { url_paths = {
"{0}/networks$": ManagedBlockchainResponse.method_dispatch( "{0}/networks$": ManagedBlockchainResponse.dispatch,
ManagedBlockchainResponse.network_response "{0}/networks/(?P<networkid>[^/.]+)$": ManagedBlockchainResponse.dispatch,
), "{0}/networks/(?P<networkid>[^/.]+)/proposals$": ManagedBlockchainResponse.dispatch,
"{0}/networks/(?P<networkid>[^/.]+)$": ManagedBlockchainResponse.method_dispatch( "{0}/networks/(?P<networkid>[^/.]+)/proposals/(?P<proposalid>[^/.]+)$": ManagedBlockchainResponse.dispatch,
ManagedBlockchainResponse.networkid_response "{0}/networks/(?P<networkid>[^/.]+)/proposals/(?P<proposalid>[^/.]+)/votes$": ManagedBlockchainResponse.dispatch,
), "{0}/invitations$": ManagedBlockchainResponse.dispatch,
"{0}/networks/(?P<networkid>[^/.]+)/proposals$": ManagedBlockchainResponse.method_dispatch( "{0}/invitations/(?P<invitationid>[^/.]+)$": ManagedBlockchainResponse.dispatch,
ManagedBlockchainResponse.proposal_response "{0}/networks/(?P<networkid>[^/.]+)/members$": ManagedBlockchainResponse.dispatch,
), "{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)$": ManagedBlockchainResponse.dispatch,
"{0}/networks/(?P<networkid>[^/.]+)/proposals/(?P<proposalid>[^/.]+)$": ManagedBlockchainResponse.method_dispatch( "{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)/nodes$": ManagedBlockchainResponse.dispatch,
ManagedBlockchainResponse.proposalid_response "{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)/nodes?(?P<querys>[^/.]+)$": ManagedBlockchainResponse.dispatch,
), "{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)/nodes/(?P<nodeid>[^/.]+)$": ManagedBlockchainResponse.dispatch,
"{0}/networks/(?P<networkid>[^/.]+)/proposals/(?P<proposalid>[^/.]+)/votes$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.proposal_votes_response
),
"{0}/invitations$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.invitation_response
),
"{0}/invitations/(?P<invitationid>[^/.]+)$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.invitationid_response
),
"{0}/networks/(?P<networkid>[^/.]+)/members$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.member_response
),
"{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.memberid_response
),
"{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)/nodes$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.node_response
),
"{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)/nodes?(?P<querys>[^/.]+)$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.node_response
),
"{0}/networks/(?P<networkid>[^/.]+)/members/(?P<memberid>[^/.]+)/nodes/(?P<nodeid>[^/.]+)$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.nodeid_response
),
# >= botocore 1.19.41 (API change - memberId is now part of query-string or body) # >= botocore 1.19.41 (API change - memberId is now part of query-string or body)
"{0}/networks/(?P<networkid>[^/.]+)/nodes$": ManagedBlockchainResponse.method_dispatch( "{0}/networks/(?P<networkid>[^/.]+)/nodes$": ManagedBlockchainResponse.dispatch,
ManagedBlockchainResponse.node_response "{0}/networks/(?P<networkid>[^/.]+)/nodes/(?P<nodeid>[^/.]+)$": ManagedBlockchainResponse.dispatch,
),
"{0}/networks/(?P<networkid>[^/.]+)/nodes/(?P<nodeid>[^/.]+)$": ManagedBlockchainResponse.method_dispatch(
ManagedBlockchainResponse.nodeid_response
),
} }

205
tests/test_managedblockchain/test_managedblockchain_invitations.py
View File

@ -6,129 +6,112 @@ from moto import mock_managedblockchain
from . import helpers from . import helpers
@mock_managedblockchain class TestManagedBlockchainInvitations:
def test_create_2_invitations(): mock = mock_managedblockchain()
conn = boto3.client("managedblockchain", region_name="us-east-1")
# Create network @classmethod
response = conn.create_network( def setup_class(cls):
Name="testnetwork1", cls.mock.start()
Framework="HYPERLEDGER_FABRIC", cls.conn = boto3.client("managedblockchain", region_name="us-east-1")
FrameworkVersion="1.2", response = cls.conn.create_network(
FrameworkConfiguration=helpers.default_frameworkconfiguration, Name="testnetwork1",
VotingPolicy=helpers.default_votingpolicy, Framework="HYPERLEDGER_FABRIC",
MemberConfiguration=helpers.default_memberconfiguration, FrameworkVersion="1.2",
) FrameworkConfiguration=helpers.default_frameworkconfiguration,
network_id = response["NetworkId"] VotingPolicy=helpers.default_votingpolicy,
member_id = response["MemberId"] MemberConfiguration=helpers.default_memberconfiguration,
)
cls.network_id = response["NetworkId"]
cls.member_id = response["MemberId"]
# Create proposal @classmethod
proposal_id = conn.create_proposal( def teardown_class(cls):
NetworkId=network_id, cls.mock.stop()
MemberId=member_id,
Actions=helpers.multiple_policy_actions,
)["ProposalId"]
# Get proposal details def test_create_2_invitations(self):
response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) # Create proposal
assert response["Proposal"]["NetworkId"] == network_id proposal_id = self.conn.create_proposal(
assert response["Proposal"]["Status"] == "IN_PROGRESS" NetworkId=self.network_id,
MemberId=self.member_id,
Actions=helpers.multiple_policy_actions,
)["ProposalId"]
# Vote yes # Get proposal details
conn.vote_on_proposal( response = self.conn.get_proposal(
NetworkId=network_id, NetworkId=self.network_id, ProposalId=proposal_id
ProposalId=proposal_id, )
VoterMemberId=member_id, assert response["Proposal"]["NetworkId"] == self.network_id
Vote="YES", assert response["Proposal"]["Status"] == "IN_PROGRESS"
)
# Get the invitation # Vote yes
response = conn.list_invitations() self.conn.vote_on_proposal(
assert len(response["Invitations"]) == 2 NetworkId=self.network_id,
assert response["Invitations"][0]["NetworkSummary"]["Id"] == network_id ProposalId=proposal_id,
assert response["Invitations"][0]["Status"] == "PENDING" VoterMemberId=self.member_id,
assert response["Invitations"][1]["NetworkSummary"]["Id"] == network_id Vote="YES",
assert response["Invitations"][1]["Status"] == "PENDING" )
# Get the invitation
response = self.conn.list_invitations()
assert len(response["Invitations"]) == 2
assert response["Invitations"][0]["NetworkSummary"]["Id"] == self.network_id
assert response["Invitations"][0]["Status"] == "PENDING"
assert response["Invitations"][1]["NetworkSummary"]["Id"] == self.network_id
assert response["Invitations"][1]["Status"] == "PENDING"
@mock_managedblockchain def test_reject_invitation(self):
def test_reject_invitation(): # Create proposal
conn = boto3.client("managedblockchain", region_name="us-east-1") proposal_id = self.conn.create_proposal(
NetworkId=self.network_id,
MemberId=self.member_id,
Actions=helpers.default_policy_actions,
)["ProposalId"]
# Create network # Get proposal details
response = conn.create_network( response = self.conn.get_proposal(
Name="testnetwork1", NetworkId=self.network_id, ProposalId=proposal_id
Framework="HYPERLEDGER_FABRIC", )
FrameworkVersion="1.2", assert response["Proposal"]["NetworkId"] == self.network_id
FrameworkConfiguration=helpers.default_frameworkconfiguration, assert response["Proposal"]["Status"] == "IN_PROGRESS"
VotingPolicy=helpers.default_votingpolicy,
MemberConfiguration=helpers.default_memberconfiguration,
)
network_id = response["NetworkId"]
member_id = response["MemberId"]
# Create proposal # Vote yes
proposal_id = conn.create_proposal( self.conn.vote_on_proposal(
NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions NetworkId=self.network_id,
)["ProposalId"] ProposalId=proposal_id,
VoterMemberId=self.member_id,
Vote="YES",
)
# Get proposal details # Get the invitation
response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response = self.conn.list_invitations()
assert response["Proposal"]["NetworkId"] == network_id assert response["Invitations"][0]["NetworkSummary"]["Id"] == self.network_id
assert response["Proposal"]["Status"] == "IN_PROGRESS" assert response["Invitations"][0]["Status"] == "PENDING"
invitation_id = response["Invitations"][0]["InvitationId"]
# Vote yes # Reject - thanks but no thanks
conn.vote_on_proposal( self.conn.reject_invitation(InvitationId=invitation_id)
NetworkId=network_id,
ProposalId=proposal_id,
VoterMemberId=member_id,
Vote="YES",
)
# Get the invitation # Check the invitation status
response = conn.list_invitations() response = self.conn.list_invitations()
assert response["Invitations"][0]["NetworkSummary"]["Id"] == network_id assert response["Invitations"][0]["InvitationId"] == invitation_id
assert response["Invitations"][0]["Status"] == "PENDING" assert response["Invitations"][0]["Status"] == "REJECTED"
invitation_id = response["Invitations"][0]["InvitationId"]
# Reject - thanks but no thanks def test_reject_invitation_badinvitation(self):
conn.reject_invitation(InvitationId=invitation_id) proposal_id = self.conn.create_proposal(
NetworkId=self.network_id,
MemberId=self.member_id,
Actions=helpers.default_policy_actions,
)["ProposalId"]
# Check the invitation status self.conn.vote_on_proposal(
response = conn.list_invitations() NetworkId=self.network_id,
assert response["Invitations"][0]["InvitationId"] == invitation_id ProposalId=proposal_id,
assert response["Invitations"][0]["Status"] == "REJECTED" VoterMemberId=self.member_id,
Vote="YES",
)
with pytest.raises(ClientError) as ex:
@mock_managedblockchain self.conn.reject_invitation(InvitationId="in-ABCDEFGHIJKLMNOP0123456789")
def test_reject_invitation_badinvitation(): err = ex.value.response["Error"]
conn = boto3.client("managedblockchain", region_name="us-east-1") assert err["Code"] == "ResourceNotFoundException"
assert "InvitationId in-ABCDEFGHIJKLMNOP0123456789 not found." in err["Message"]
# Create network - need a good network
response = conn.create_network(
Name="testnetwork1",
Framework="HYPERLEDGER_FABRIC",
FrameworkVersion="1.2",
FrameworkConfiguration=helpers.default_frameworkconfiguration,
VotingPolicy=helpers.default_votingpolicy,
MemberConfiguration=helpers.default_memberconfiguration,
)
network_id = response["NetworkId"]
member_id = response["MemberId"]
proposal_id = conn.create_proposal(
NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions
)["ProposalId"]
conn.vote_on_proposal(
NetworkId=network_id,
ProposalId=proposal_id,
VoterMemberId=member_id,
Vote="YES",
)
with pytest.raises(ClientError) as ex:
conn.reject_invitation(InvitationId="in-ABCDEFGHIJKLMNOP0123456789")
err = ex.value.response["Error"]
assert err["Code"] == "ResourceNotFoundException"
assert "InvitationId in-ABCDEFGHIJKLMNOP0123456789 not found." in err["Message"]

7
tests/test_managedblockchain/test_managedblockchain_members.py
View File

@ -1,6 +1,7 @@
import boto3 import boto3
import pytest import pytest
from botocore.config import Config
from botocore.exceptions import ClientError, ParamValidationError from botocore.exceptions import ClientError, ParamValidationError
from moto import mock_managedblockchain from moto import mock_managedblockchain
from . import helpers from . import helpers
@ -280,7 +281,11 @@ def test_invite_and_remove_member():
@mock_managedblockchain @mock_managedblockchain
def test_create_too_many_members(): def test_create_too_many_members():
conn = boto3.client("managedblockchain", region_name="us-east-1") # This test throws a ResourceLimitException, with HTTP status code 429
# Boto3 automatically retries a request with that status code up to 5 times
# Retrying is not going to make a difference to the output though...
config = Config(retries={"max_attempts": 1, "mode": "standard"})
conn = boto3.client("managedblockchain", region_name="us-east-1", config=config)
# Create network # Create network
response = conn.create_network( response = conn.create_network(

7
tests/test_managedblockchain/test_managedblockchain_networks.py
View File

@ -36,7 +36,7 @@ def test_create_network():
@mock_managedblockchain @mock_managedblockchain
def test_create_network_withopts(): def test_create_network_with_description():
conn = boto3.client("managedblockchain", region_name="us-east-1") conn = boto3.client("managedblockchain", region_name="us-east-1")
response = conn.create_network( response = conn.create_network(
@ -49,11 +49,6 @@ def test_create_network_withopts():
MemberConfiguration=helpers.default_memberconfiguration, MemberConfiguration=helpers.default_memberconfiguration,
) )
network_id = response["NetworkId"] network_id = response["NetworkId"]
member_id = response["MemberId"]
assert network_id.startswith("n-")
assert len(network_id) == 28
assert member_id.startswith("m-")
assert len(member_id) == 28
# Find in full list # Find in full list
mbcnetworks = conn.list_networks()["Networks"] mbcnetworks = conn.list_networks()["Networks"]

7
tests/test_managedblockchain/test_managedblockchain_nodes.py
View File

@ -1,6 +1,7 @@
import boto3 import boto3
import pytest import pytest
from botocore.config import Config
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_managedblockchain from moto import mock_managedblockchain
from . import helpers from . import helpers
@ -145,7 +146,11 @@ def test_create_node_standard_edition():
@mock_managedblockchain @mock_managedblockchain
def test_create_too_many_nodes(): def test_create_too_many_nodes():
conn = boto3.client("managedblockchain", region_name="us-east-1") # This test throws a ResourceLimitException, with HTTP status code 429
# Boto3 automatically retries a request with that status code up to 5 times
# Retrying is not going to make a difference to the output though...
config = Config(retries={"max_attempts": 1, "mode": "standard"})
conn = boto3.client("managedblockchain", region_name="us-east-1", config=config)
# Create network # Create network
response = conn.create_network( response = conn.create_network(