From 093052bd9be65078aca9c9b57068b5648f9b5f7b Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Fri, 1 Sep 2023 07:07:54 +0000 Subject: [PATCH] Techdebt: Simplify ManagedBlockchain setup & tests (#6749) --- moto/core/models.py | 5 +- moto/managedblockchain/exceptions.py | 17 +- moto/managedblockchain/responses.py | 380 +++++------------- moto/managedblockchain/urls.py | 56 +-- .../test_managedblockchain_invitations.py | 205 +++++----- .../test_managedblockchain_members.py | 7 +- .../test_managedblockchain_networks.py | 7 +- .../test_managedblockchain_nodes.py | 7 +- 8 files changed, 218 insertions(+), 466 deletions(-) diff --git a/moto/core/models.py b/moto/core/models.py index 9f2d2a623..a556423fb 100644 --- a/moto/core/models.py +++ b/moto/core/models.py @@ -392,7 +392,10 @@ class ServerModeMockAWS(BaseMockAWS): region = self._get_region(*args, **kwargs) if region: 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: config = Config(user_agent_extra="region/" + region) kwargs["config"] = config diff --git a/moto/managedblockchain/exceptions.py b/moto/managedblockchain/exceptions.py index 494080c76..f64915b6d 100644 --- a/moto/managedblockchain/exceptions.py +++ b/moto/managedblockchain/exceptions.py @@ -1,21 +1,6 @@ import json -from moto.core.common_types import TYPE_RESPONSE from moto.core.exceptions import JsonRESTError -from functools import wraps -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 +from typing import Any, List, Tuple class ManagedBlockchainClientError(JsonRESTError): diff --git a/moto/managedblockchain/responses.py b/moto/managedblockchain/responses.py index faec856a6..8912f725f 100644 --- a/moto/managedblockchain/responses.py +++ b/moto/managedblockchain/responses.py @@ -1,10 +1,6 @@ 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 .exceptions import exception_handler from .models import managedblockchain_backends, ManagedBlockchainBackend from .utils import ( networkid_from_managedblockchain_url, @@ -23,39 +19,20 @@ class ManagedBlockchainResponse(BaseResponse): def backend(self) -> ManagedBlockchainBackend: return managedblockchain_backends[self.current_account][self.region] - @exception_handler - def network_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - self.setup_class(request, full_url, headers) - return self._network_response(request, headers) + def list_networks(self) -> str: + networks = self.backend.list_networks() + return json.dumps({"Networks": [network.to_dict() for network in networks]}) - def _network_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return] - method = request.method - if method == "GET": - return self._all_networks_response(headers) - elif method == "POST": - json_body = json.loads(self.body) - return self._network_response_post(json_body, headers) - - 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"] + def create_network(self) -> str: + name = self._get_param("Name") + framework = self._get_param("Framework") + frameworkversion = self._get_param("FrameworkVersion") + frameworkconfiguration = self._get_param("FrameworkConfiguration") + voting_policy = self._get_param("VotingPolicy") + member_configuration = self._get_param("MemberConfiguration") # Optional - description = json_body.get("Description", None) + description = self._get_param("Description", None) response = self.backend.create_network( name, @@ -66,257 +43,110 @@ class ManagedBlockchainResponse(BaseResponse): member_configuration, description, ) - return 200, headers, json.dumps(response) + return json.dumps(response) - @exception_handler - def networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - 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: + def get_network(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) mbcnetwork = self.backend.get_network(network_id) - response = json.dumps({"Network": mbcnetwork.get_format()}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Network": mbcnetwork.get_format()}) - @exception_handler - def proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - 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: + def list_proposals(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) proposals = self.backend.list_proposals(network_id) - response = json.dumps( - {"Proposals": [proposal.to_dict() for proposal in proposals]} - ) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Proposals": [proposal.to_dict() for proposal in proposals]}) - def _proposal_response_post( - self, network_id: str, json_body: Dict[str, Any], headers: Any - ) -> TYPE_RESPONSE: - memberid = json_body["MemberId"] - actions = json_body["Actions"] + def create_proposal(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + memberid = self._get_param("MemberId") + actions = self._get_param("Actions") # Optional - description = json_body.get("Description", None) + description = self._get_param("Description", None) response = self.backend.create_proposal( network_id, memberid, actions, description ) - return 200, headers, json.dumps(response) + return json.dumps(response) - @exception_handler - def proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - self.setup_class(request, full_url, headers) - 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: + def get_proposal(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + proposal_id = proposalid_from_managedblockchain_url(self.path) proposal = self.backend.get_proposal(network_id, proposal_id) - response = json.dumps({"Proposal": proposal.get_format()}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Proposal": proposal.get_format()}) - @exception_handler - def proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - self.setup_class(request, full_url, headers) - 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: + def list_proposal_votes(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + proposal_id = proposalid_from_managedblockchain_url(self.path) proposalvotes = self.backend.list_proposal_votes(network_id, proposal_id) - response = json.dumps({"ProposalVotes": proposalvotes}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"ProposalVotes": proposalvotes}) - def _proposal_votes_response_post( - self, network_id: str, proposal_id: str, json_body: Dict[str, Any], headers: Any - ) -> TYPE_RESPONSE: - votermemberid = json_body["VoterMemberId"] - vote = json_body["Vote"] + def vote_on_proposal(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + proposal_id = proposalid_from_managedblockchain_url(self.path) + votermemberid = self._get_param("VoterMemberId") + vote = self._get_param("Vote") self.backend.vote_on_proposal(network_id, proposal_id, votermemberid, vote) - return 200, headers, "" + return "" - @exception_handler - 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: + def list_invitations(self) -> str: invitations = self.backend.list_invitations() - response = json.dumps( + return json.dumps( {"Invitations": [invitation.to_dict() for invitation in invitations]} ) - headers["content-type"] = "application/json" - return 200, headers, response - @exception_handler - def invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - 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: + def reject_invitation(self) -> str: + invitation_id = invitationid_from_managedblockchain_url(self.path) self.backend.reject_invitation(invitation_id) - headers["content-type"] = "application/json" - return 200, headers, "" + return "" - @exception_handler - def member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - 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: + def list_members(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) members = self.backend.list_members(network_id) - response = json.dumps({"Members": [member.to_dict() for member in members]}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Members": [member.to_dict() for member in members]}) - def _member_response_post( - self, network_id: str, json_body: Dict[str, Any], headers: Any - ) -> TYPE_RESPONSE: - invitationid = json_body["InvitationId"] - member_configuration = json_body["MemberConfiguration"] + def create_member(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + invitationid = self._get_param("InvitationId") + member_configuration = self._get_param("MemberConfiguration") response = self.backend.create_member( invitationid, network_id, member_configuration ) - return 200, headers, json.dumps(response) + return json.dumps(response) - @exception_handler - def memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - self.setup_class(request, full_url, headers) - 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: + def get_member(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) member = self.backend.get_member(network_id, member_id) - response = json.dumps({"Member": member.get_format()}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Member": member.get_format()}) - def _memberid_response_patch( - self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any - ) -> TYPE_RESPONSE: - logpublishingconfiguration = json_body["LogPublishingConfiguration"] + def update_member(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) + logpublishingconfiguration = self._get_param("LogPublishingConfiguration") self.backend.update_member(network_id, member_id, logpublishingconfiguration) - return 200, headers, "" + return "" - def _memberid_response_delete( - self, network_id: str, member_id: str, headers: Any - ) -> TYPE_RESPONSE: + def delete_member(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) self.backend.delete_member(network_id, member_id) - headers["content-type"] = "application/json" - return 200, headers, "" + return "" - @exception_handler - def node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - self.setup_class(request, full_url, headers) - return self._node_response(request, full_url, headers) - - 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: + def list_nodes(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) + status = self._get_param("status") nodes = self.backend.list_nodes(network_id, member_id, status) - response = json.dumps({"Nodes": [node.to_dict() for node in nodes]}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Nodes": [node.to_dict() for node in nodes]}) - def _node_response_post( - self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any - ) -> TYPE_RESPONSE: - instancetype = json_body["NodeConfiguration"]["InstanceType"] - availabilityzone = json_body["NodeConfiguration"]["AvailabilityZone"] - logpublishingconfiguration = json_body["NodeConfiguration"][ + def create_node(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) + instancetype = self._get_param("NodeConfiguration")["InstanceType"] + availabilityzone = self._get_param("NodeConfiguration")["AvailabilityZone"] + logpublishingconfiguration = self._get_param("NodeConfiguration")[ "LogPublishingConfiguration" ] @@ -327,53 +157,27 @@ class ManagedBlockchainResponse(BaseResponse): instancetype, logpublishingconfiguration, ) - return 200, headers, json.dumps(response) + return json.dumps(response) - @exception_handler - def nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc] - self.setup_class(request, full_url, headers) - return self._nodeid_response(request, full_url, headers) - - 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: + def get_node(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) + node_id = nodeid_from_managedblockchain_url(self.path) node = self.backend.get_node(network_id, member_id, node_id) - response = json.dumps({"Node": node.get_format()}) - headers["content-type"] = "application/json" - return 200, headers, response + return json.dumps({"Node": node.get_format()}) - def _nodeid_response_patch( - self, - network_id: str, - member_id: str, - node_id: str, - json_body: Dict[str, Any], - headers: Any, - ) -> TYPE_RESPONSE: - logpublishingconfiguration = json_body + def update_node(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + member_id = memberid_from_managedblockchain_request(self.uri, self.body) + node_id = nodeid_from_managedblockchain_url(self.path) 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( - self, network_id: str, member_id: str, node_id: str, headers: Any - ) -> TYPE_RESPONSE: + def delete_node(self) -> str: + network_id = networkid_from_managedblockchain_url(self.path) + 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) - headers["content-type"] = "application/json" - return 200, headers, "" + return "" diff --git a/moto/managedblockchain/urls.py b/moto/managedblockchain/urls.py index f3f54c134..0ac010f55 100644 --- a/moto/managedblockchain/urls.py +++ b/moto/managedblockchain/urls.py @@ -3,47 +3,19 @@ from .responses import ManagedBlockchainResponse url_bases = [r"https?://managedblockchain\.(.+)\.amazonaws.com"] url_paths = { - "{0}/networks$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.network_response - ), - "{0}/networks/(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.networkid_response - ), - "{0}/networks/(?P[^/.]+)/proposals$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.proposal_response - ), - "{0}/networks/(?P[^/.]+)/proposals/(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.proposalid_response - ), - "{0}/networks/(?P[^/.]+)/proposals/(?P[^/.]+)/votes$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.proposal_votes_response - ), - "{0}/invitations$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.invitation_response - ), - "{0}/invitations/(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.invitationid_response - ), - "{0}/networks/(?P[^/.]+)/members$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.member_response - ), - "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.memberid_response - ), - "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)/nodes$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.node_response - ), - "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)/nodes?(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.node_response - ), - "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)/nodes/(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.nodeid_response - ), + "{0}/networks$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/proposals$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/proposals/(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/proposals/(?P[^/.]+)/votes$": ManagedBlockchainResponse.dispatch, + "{0}/invitations$": ManagedBlockchainResponse.dispatch, + "{0}/invitations/(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/members$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)/nodes$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)/nodes?(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/members/(?P[^/.]+)/nodes/(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, # >= botocore 1.19.41 (API change - memberId is now part of query-string or body) - "{0}/networks/(?P[^/.]+)/nodes$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.node_response - ), - "{0}/networks/(?P[^/.]+)/nodes/(?P[^/.]+)$": ManagedBlockchainResponse.method_dispatch( - ManagedBlockchainResponse.nodeid_response - ), + "{0}/networks/(?P[^/.]+)/nodes$": ManagedBlockchainResponse.dispatch, + "{0}/networks/(?P[^/.]+)/nodes/(?P[^/.]+)$": ManagedBlockchainResponse.dispatch, } diff --git a/tests/test_managedblockchain/test_managedblockchain_invitations.py b/tests/test_managedblockchain/test_managedblockchain_invitations.py index 7e0564898..d097f0f32 100644 --- a/tests/test_managedblockchain/test_managedblockchain_invitations.py +++ b/tests/test_managedblockchain/test_managedblockchain_invitations.py @@ -6,129 +6,112 @@ from moto import mock_managedblockchain from . import helpers -@mock_managedblockchain -def test_create_2_invitations(): - conn = boto3.client("managedblockchain", region_name="us-east-1") +class TestManagedBlockchainInvitations: + mock = mock_managedblockchain() - # Create 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"] + @classmethod + def setup_class(cls): + cls.mock.start() + cls.conn = boto3.client("managedblockchain", region_name="us-east-1") + response = cls.conn.create_network( + Name="testnetwork1", + Framework="HYPERLEDGER_FABRIC", + FrameworkVersion="1.2", + FrameworkConfiguration=helpers.default_frameworkconfiguration, + VotingPolicy=helpers.default_votingpolicy, + MemberConfiguration=helpers.default_memberconfiguration, + ) + cls.network_id = response["NetworkId"] + cls.member_id = response["MemberId"] - # Create proposal - proposal_id = conn.create_proposal( - NetworkId=network_id, - MemberId=member_id, - Actions=helpers.multiple_policy_actions, - )["ProposalId"] + @classmethod + def teardown_class(cls): + cls.mock.stop() - # Get proposal details - response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) - assert response["Proposal"]["NetworkId"] == network_id - assert response["Proposal"]["Status"] == "IN_PROGRESS" + def test_create_2_invitations(self): + # Create proposal + proposal_id = self.conn.create_proposal( + NetworkId=self.network_id, + MemberId=self.member_id, + Actions=helpers.multiple_policy_actions, + )["ProposalId"] - # Vote yes - conn.vote_on_proposal( - NetworkId=network_id, - ProposalId=proposal_id, - VoterMemberId=member_id, - Vote="YES", - ) + # Get proposal details + response = self.conn.get_proposal( + NetworkId=self.network_id, ProposalId=proposal_id + ) + assert response["Proposal"]["NetworkId"] == self.network_id + assert response["Proposal"]["Status"] == "IN_PROGRESS" - # Get the invitation - response = conn.list_invitations() - assert len(response["Invitations"]) == 2 - assert response["Invitations"][0]["NetworkSummary"]["Id"] == network_id - assert response["Invitations"][0]["Status"] == "PENDING" - assert response["Invitations"][1]["NetworkSummary"]["Id"] == network_id - assert response["Invitations"][1]["Status"] == "PENDING" + # Vote yes + self.conn.vote_on_proposal( + NetworkId=self.network_id, + ProposalId=proposal_id, + VoterMemberId=self.member_id, + Vote="YES", + ) + # 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(): - conn = boto3.client("managedblockchain", region_name="us-east-1") + def test_reject_invitation(self): + # Create proposal + proposal_id = self.conn.create_proposal( + NetworkId=self.network_id, + MemberId=self.member_id, + Actions=helpers.default_policy_actions, + )["ProposalId"] - # Create 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"] + # Get proposal details + response = self.conn.get_proposal( + NetworkId=self.network_id, ProposalId=proposal_id + ) + assert response["Proposal"]["NetworkId"] == self.network_id + assert response["Proposal"]["Status"] == "IN_PROGRESS" - # Create proposal - proposal_id = conn.create_proposal( - NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions - )["ProposalId"] + # Vote yes + self.conn.vote_on_proposal( + NetworkId=self.network_id, + ProposalId=proposal_id, + VoterMemberId=self.member_id, + Vote="YES", + ) - # Get proposal details - response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) - assert response["Proposal"]["NetworkId"] == network_id - assert response["Proposal"]["Status"] == "IN_PROGRESS" + # Get the invitation + response = self.conn.list_invitations() + assert response["Invitations"][0]["NetworkSummary"]["Id"] == self.network_id + assert response["Invitations"][0]["Status"] == "PENDING" + invitation_id = response["Invitations"][0]["InvitationId"] - # Vote yes - conn.vote_on_proposal( - NetworkId=network_id, - ProposalId=proposal_id, - VoterMemberId=member_id, - Vote="YES", - ) + # Reject - thanks but no thanks + self.conn.reject_invitation(InvitationId=invitation_id) - # Get the invitation - response = conn.list_invitations() - assert response["Invitations"][0]["NetworkSummary"]["Id"] == network_id - assert response["Invitations"][0]["Status"] == "PENDING" - invitation_id = response["Invitations"][0]["InvitationId"] + # Check the invitation status + response = self.conn.list_invitations() + assert response["Invitations"][0]["InvitationId"] == invitation_id + assert response["Invitations"][0]["Status"] == "REJECTED" - # Reject - thanks but no thanks - conn.reject_invitation(InvitationId=invitation_id) + def test_reject_invitation_badinvitation(self): + proposal_id = self.conn.create_proposal( + NetworkId=self.network_id, + MemberId=self.member_id, + Actions=helpers.default_policy_actions, + )["ProposalId"] - # Check the invitation status - response = conn.list_invitations() - assert response["Invitations"][0]["InvitationId"] == invitation_id - assert response["Invitations"][0]["Status"] == "REJECTED" + self.conn.vote_on_proposal( + NetworkId=self.network_id, + ProposalId=proposal_id, + VoterMemberId=self.member_id, + Vote="YES", + ) - -@mock_managedblockchain -def test_reject_invitation_badinvitation(): - conn = boto3.client("managedblockchain", region_name="us-east-1") - - # 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"] + with pytest.raises(ClientError) as ex: + self.conn.reject_invitation(InvitationId="in-ABCDEFGHIJKLMNOP0123456789") + err = ex.value.response["Error"] + assert err["Code"] == "ResourceNotFoundException" + assert "InvitationId in-ABCDEFGHIJKLMNOP0123456789 not found." in err["Message"] diff --git a/tests/test_managedblockchain/test_managedblockchain_members.py b/tests/test_managedblockchain/test_managedblockchain_members.py index 3b41e24c5..40030e5d6 100644 --- a/tests/test_managedblockchain/test_managedblockchain_members.py +++ b/tests/test_managedblockchain/test_managedblockchain_members.py @@ -1,6 +1,7 @@ import boto3 import pytest +from botocore.config import Config from botocore.exceptions import ClientError, ParamValidationError from moto import mock_managedblockchain from . import helpers @@ -280,7 +281,11 @@ def test_invite_and_remove_member(): @mock_managedblockchain 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 response = conn.create_network( diff --git a/tests/test_managedblockchain/test_managedblockchain_networks.py b/tests/test_managedblockchain/test_managedblockchain_networks.py index 38882b85c..4ee6844a5 100644 --- a/tests/test_managedblockchain/test_managedblockchain_networks.py +++ b/tests/test_managedblockchain/test_managedblockchain_networks.py @@ -36,7 +36,7 @@ def test_create_network(): @mock_managedblockchain -def test_create_network_withopts(): +def test_create_network_with_description(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.create_network( @@ -49,11 +49,6 @@ def test_create_network_withopts(): MemberConfiguration=helpers.default_memberconfiguration, ) 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 mbcnetworks = conn.list_networks()["Networks"] diff --git a/tests/test_managedblockchain/test_managedblockchain_nodes.py b/tests/test_managedblockchain/test_managedblockchain_nodes.py index e3ca9b8fb..d53ae4404 100644 --- a/tests/test_managedblockchain/test_managedblockchain_nodes.py +++ b/tests/test_managedblockchain/test_managedblockchain_nodes.py @@ -1,6 +1,7 @@ import boto3 import pytest +from botocore.config import Config from botocore.exceptions import ClientError from moto import mock_managedblockchain from . import helpers @@ -145,7 +146,11 @@ def test_create_node_standard_edition(): @mock_managedblockchain 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 response = conn.create_network(