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

Techdebt: MyPy ManagedBlockchain (#6134)

Browse Source
This commit is contained in:
Bert Blommers 2023-03-27 18:00:24 +01:00 committed by GitHub
parent 94d35af520
commit 14d3c5a76a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 360 additions and 305 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
moto/managedblockchain/exceptions.py
View File

@ -1,38 +1,46 @@
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 functools import wraps
from typing import Any, Callable, List, Tuple
def exception_handler(f): def exception_handler(
f: Callable[[Any, Any, Any, Any], TYPE_RESPONSE]
) -> Callable[[Any], TYPE_RESPONSE]:
@wraps(f) @wraps(f)
def _wrapper(*args, **kwargs): def _wrapper(*args: Any, **kwargs: Any) -> TYPE_RESPONSE: # type: ignore[misc]
try: try:
return f(*args, **kwargs) return f(*args, **kwargs)
except ManagedBlockchainClientError as err: except ManagedBlockchainClientError as err:
return err.code, err.get_headers(), err.description return err.code, err.get_headers(), err.description # type: ignore
return _wrapper return _wrapper
class ManagedBlockchainClientError(JsonRESTError): class ManagedBlockchainClientError(JsonRESTError):
def __init__(self, error_type, message): def __init__(self, error_type: str, message: str):
super().__init__(error_type=error_type, message=message) super().__init__(error_type=error_type, message=message)
self.error_type = error_type self.error_type = error_type
self.message = message self.message = message
self.description = json.dumps({"message": self.message}) self.description = json.dumps({"message": self.message})
def get_headers(self, *args, **kwargs): # pylint: disable=unused-argument def get_headers(
self, *args: Any, **kwargs: Any
) -> List[Tuple[str, str]]: # pylint: disable=unused-argument
return [ return [
("Content-Type", "application/json"), ("Content-Type", "application/json"),
("x-amzn-ErrorType", self.error_type), ("x-amzn-ErrorType", self.error_type),
] ]
def get_body(self, *args, **kwargs): # pylint: disable=unused-argument def get_body(
self, *args: Any, **kwargs: Any
) -> str: # pylint: disable=unused-argument
return self.description return self.description
class BadRequestException(ManagedBlockchainClientError): class BadRequestException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error): def __init__(self, pretty_called_method: str, operation_error: str):
super().__init__( super().__init__(
"BadRequestException", "BadRequestException",
f"An error occurred (BadRequestException) when calling the {pretty_called_method} operation: {operation_error}", f"An error occurred (BadRequestException) when calling the {pretty_called_method} operation: {operation_error}",
@ -40,7 +48,7 @@ class BadRequestException(ManagedBlockchainClientError):
class InvalidRequestException(ManagedBlockchainClientError): class InvalidRequestException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error): def __init__(self, pretty_called_method: str, operation_error: str):
super().__init__( super().__init__(
"InvalidRequestException", "InvalidRequestException",
f"An error occurred (InvalidRequestException) when calling the {pretty_called_method} operation: {operation_error}", f"An error occurred (InvalidRequestException) when calling the {pretty_called_method} operation: {operation_error}",
@ -48,7 +56,7 @@ class InvalidRequestException(ManagedBlockchainClientError):
class ResourceNotFoundException(ManagedBlockchainClientError): class ResourceNotFoundException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error): def __init__(self, pretty_called_method: str, operation_error: str):
self.code = 404 self.code = 404
super().__init__( super().__init__(
"ResourceNotFoundException", "ResourceNotFoundException",
@ -57,7 +65,7 @@ class ResourceNotFoundException(ManagedBlockchainClientError):
class ResourceAlreadyExistsException(ManagedBlockchainClientError): class ResourceAlreadyExistsException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error): def __init__(self, pretty_called_method: str, operation_error: str):
self.code = 409 self.code = 409
super().__init__( super().__init__(
"ResourceAlreadyExistsException", "ResourceAlreadyExistsException",
@ -66,7 +74,7 @@ class ResourceAlreadyExistsException(ManagedBlockchainClientError):
class ResourceLimitExceededException(ManagedBlockchainClientError): class ResourceLimitExceededException(ManagedBlockchainClientError):
def __init__(self, pretty_called_method, operation_error): def __init__(self, pretty_called_method: str, operation_error: str):
self.code = 429 self.code = 429
super().__init__( super().__init__(
"ResourceLimitExceededException", "ResourceLimitExceededException",

422
moto/managedblockchain/models.py
View File

@ -1,5 +1,6 @@
import datetime import datetime
import re import re
from typing import Any, Dict, List, Optional
from moto.core import BaseBackend, BackendDict, BaseModel from moto.core import BaseBackend, BackendDict, BaseModel
@ -32,7 +33,7 @@ FRAMEWORKVERSIONS = [
"1.2", "1.2",
] ]
EDITIONS = { EDITIONS: Dict[str, Any] = {
"STARTER": { "STARTER": {
"MaxMembers": 5, "MaxMembers": 5,
"MaxNodesPerMember": 2, "MaxNodesPerMember": 2,
@ -51,15 +52,15 @@ VOTEVALUES = ["YES", "NO"]
class ManagedBlockchainNetwork(BaseModel): class ManagedBlockchainNetwork(BaseModel):
def __init__( def __init__(
self, self,
network_id, network_id: str,
name, name: str,
framework, framework: str,
frameworkversion, frameworkversion: str,
frameworkconfiguration, frameworkconfiguration: Dict[str, Any],
voting_policy, voting_policy: Dict[str, Any],
member_configuration, member_configuration: Dict[str, Any],
region, region: str,
description=None, description: Optional[str] = None,
): ):
self.creationdate = datetime.datetime.utcnow() self.creationdate = datetime.datetime.utcnow()
self.id = network_id self.id = network_id
@ -73,42 +74,42 @@ class ManagedBlockchainNetwork(BaseModel):
self.region = region self.region = region
@property @property
def network_name(self): def network_name(self) -> str:
return self.name return self.name
@property @property
def network_framework(self): def network_framework(self) -> str:
return self.framework return self.framework
@property @property
def network_framework_version(self): def network_framework_version(self) -> str:
return self.frameworkversion return self.frameworkversion
@property @property
def network_creationdate(self): def network_creationdate(self) -> str:
return self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z") return self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z")
@property @property
def network_description(self): def network_description(self) -> Optional[str]:
return self.description return self.description
@property @property
def network_edition(self): def network_edition(self) -> str:
return self.frameworkconfiguration["Fabric"]["Edition"] return self.frameworkconfiguration["Fabric"]["Edition"]
@property @property
def vote_pol_proposal_duration(self): def vote_pol_proposal_duration(self) -> float:
return self.voting_policy["ApprovalThresholdPolicy"]["ProposalDurationInHours"] return self.voting_policy["ApprovalThresholdPolicy"]["ProposalDurationInHours"]
@property @property
def vote_pol_threshold_percentage(self): def vote_pol_threshold_percentage(self) -> float:
return self.voting_policy["ApprovalThresholdPolicy"]["ThresholdPercentage"] return self.voting_policy["ApprovalThresholdPolicy"]["ThresholdPercentage"]
@property @property
def vote_pol_threshold_comparator(self): def vote_pol_threshold_comparator(self) -> str:
return self.voting_policy["ApprovalThresholdPolicy"]["ThresholdComparator"] return self.voting_policy["ApprovalThresholdPolicy"]["ThresholdComparator"]
def to_dict(self): def to_dict(self) -> Dict[str, Any]:
# Format for list_networks # Format for list_networks
d = { d = {
"Id": self.id, "Id": self.id,
@ -122,7 +123,7 @@ class ManagedBlockchainNetwork(BaseModel):
d["Description"] = self.description d["Description"] = self.description
return d return d
def get_format(self): def get_format(self) -> Dict[str, Any]:
# Format for get_network # Format for get_network
frameworkattributes = { frameworkattributes = {
"Fabric": { "Fabric": {
@ -154,16 +155,16 @@ class ManagedBlockchainNetwork(BaseModel):
class ManagedBlockchainProposal(BaseModel): class ManagedBlockchainProposal(BaseModel):
def __init__( def __init__(
self, self,
proposal_id, proposal_id: str,
networkid, networkid: str,
memberid, memberid: str,
membername, membername: str,
numofmembers, numofmembers: int,
actions, actions: Dict[str, Any],
network_expirtation, network_expiration: float,
network_threshold, network_threshold: float,
network_threshold_comp, network_threshold_comp: str,
description=None, description: Optional[str] = None,
): ):
# In general, passing all values instead of creating # In general, passing all values instead of creating
# an apparatus to look them up # an apparatus to look them up
@ -173,60 +174,58 @@ class ManagedBlockchainProposal(BaseModel):
self.membername = membername self.membername = membername
self.numofmembers = numofmembers self.numofmembers = numofmembers
self.actions = actions self.actions = actions
self.network_expirtation = network_expirtation self.network_expiration = network_expiration
self.network_threshold = network_threshold self.network_threshold = network_threshold
self.network_threshold_comp = network_threshold_comp self.network_threshold_comp = network_threshold_comp
self.description = description self.description = description
self.creationdate = datetime.datetime.utcnow() self.creationdate = datetime.datetime.utcnow()
self.expirtationdate = self.creationdate + datetime.timedelta( self.expirationdate = self.creationdate + datetime.timedelta(
hours=network_expirtation hours=network_expiration
) )
self.yes_vote_count = 0 self.yes_vote_count = 0
self.no_vote_count = 0 self.no_vote_count = 0
self.outstanding_vote_count = self.numofmembers self.outstanding_vote_count = self.numofmembers
self.status = "IN_PROGRESS" self.status = "IN_PROGRESS"
self.votes = {} self.votes: Dict[str, Dict[str, str]] = {}
@property @property
def network_id(self): def network_id(self) -> str:
return self.networkid return self.networkid
@property @property
def proposal_status(self): def proposal_status(self) -> str:
return self.status return self.status
@property @property
def proposal_votes(self): def proposal_votes(self) -> Dict[str, Any]: # type: ignore[misc]
return self.votes return self.votes
def proposal_actions(self, action_type): def proposal_actions(self, action_type: str) -> List[Dict[str, Any]]:
default_return = []
if action_type.lower() == "invitations": if action_type.lower() == "invitations":
if "Invitations" in self.actions: if "Invitations" in self.actions:
return self.actions["Invitations"] return self.actions["Invitations"]
elif action_type.lower() == "removals": elif action_type.lower() == "removals":
if "Removals" in self.actions: if "Removals" in self.actions:
return self.actions["Removals"] return self.actions["Removals"]
return default_return return []
def check_to_expire_proposal(self): def check_to_expire_proposal(self) -> None:
if datetime.datetime.utcnow() > self.expirtationdate: if datetime.datetime.utcnow() > self.expirationdate:
self.status = "EXPIRED" self.status = "EXPIRED"
def to_dict(self): def to_dict(self) -> Dict[str, Any]:
# Format for list_proposals # Format for list_proposals
d = { return {
"ProposalId": self.id, "ProposalId": self.id,
"ProposedByMemberId": self.memberid, "ProposedByMemberId": self.memberid,
"ProposedByMemberName": self.membername, "ProposedByMemberName": self.membername,
"Status": self.status, "Status": self.status,
"CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
"ExpirationDate": self.expirtationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "ExpirationDate": self.expirationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
} }
return d
def get_format(self): def get_format(self) -> Dict[str, Any]:
# Format for get_proposal # Format for get_proposal
d = { d = {
"ProposalId": self.id, "ProposalId": self.id,
@ -236,7 +235,7 @@ class ManagedBlockchainProposal(BaseModel):
"ProposedByMemberName": self.membername, "ProposedByMemberName": self.membername,
"Status": self.status, "Status": self.status,
"CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
"ExpirationDate": self.expirtationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "ExpirationDate": self.expirationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
"YesVoteCount": self.yes_vote_count, "YesVoteCount": self.yes_vote_count,
"NoVoteCount": self.no_vote_count, "NoVoteCount": self.no_vote_count,
"OutstandingVoteCount": self.outstanding_vote_count, "OutstandingVoteCount": self.outstanding_vote_count,
@ -245,7 +244,7 @@ class ManagedBlockchainProposal(BaseModel):
d["Description"] = self.description d["Description"] = self.description
return d return d
def set_vote(self, votermemberid, votermembername, vote): def set_vote(self, votermemberid: str, votermembername: str, vote: str) -> None:
if vote.upper() == "YES": if vote.upper() == "YES":
self.yes_vote_count += 1 self.yes_vote_count += 1
else: else:
@ -284,14 +283,14 @@ class ManagedBlockchainProposal(BaseModel):
class ManagedBlockchainInvitation(BaseModel): class ManagedBlockchainInvitation(BaseModel):
def __init__( def __init__(
self, self,
invitation_id, invitation_id: str,
networkid, networkid: str,
networkname, networkname: str,
networkframework, networkframework: str,
networkframeworkversion, networkframeworkversion: str,
networkcreationdate, networkcreationdate: str,
region, region: str,
networkdescription=None, networkdescription: Optional[str] = None,
): ):
self.id = invitation_id self.id = invitation_id
self.networkid = networkid self.networkid = networkid
@ -305,21 +304,21 @@ class ManagedBlockchainInvitation(BaseModel):
self.region = region self.region = region
self.creationdate = datetime.datetime.utcnow() self.creationdate = datetime.datetime.utcnow()
self.expirtationdate = self.creationdate + datetime.timedelta(days=7) self.expirationdate = self.creationdate + datetime.timedelta(days=7)
@property @property
def invitation_status(self): def invitation_status(self) -> str:
return self.status return self.status
@property @property
def invitation_networkid(self): def invitation_networkid(self) -> str:
return self.networkid return self.networkid
def to_dict(self): def to_dict(self) -> Dict[str, Any]:
d = { d: Dict[str, Any] = {
"InvitationId": self.id, "InvitationId": self.id,
"CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
"ExpirationDate": self.expirtationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "ExpirationDate": self.expirationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
"Status": self.status, "Status": self.status,
"NetworkSummary": { "NetworkSummary": {
"Id": self.networkid, "Id": self.networkid,
@ -334,18 +333,24 @@ class ManagedBlockchainInvitation(BaseModel):
d["NetworkSummary"]["Description"] = self.networkdescription d["NetworkSummary"]["Description"] = self.networkdescription
return d return d
def accept_invitation(self): def accept_invitation(self) -> None:
self.status = "ACCEPTED" self.status = "ACCEPTED"
def reject_invitation(self): def reject_invitation(self) -> None:
self.status = "REJECTED" self.status = "REJECTED"
def set_network_status(self, network_status): def set_network_status(self, network_status: str) -> None:
self.networkstatus = network_status self.networkstatus = network_status
class ManagedBlockchainMember(BaseModel): class ManagedBlockchainMember(BaseModel):
def __init__(self, member_id, networkid, member_configuration, region): def __init__(
self,
member_id: str,
networkid: str,
member_configuration: Dict[str, Any],
region: str,
):
self.creationdate = datetime.datetime.utcnow() self.creationdate = datetime.datetime.utcnow()
self.id = member_id self.id = member_id
self.networkid = networkid self.networkid = networkid
@ -355,18 +360,18 @@ class ManagedBlockchainMember(BaseModel):
self.description = None self.description = None
@property @property
def network_id(self): def network_id(self) -> str:
return self.networkid return self.networkid
@property @property
def name(self): def name(self) -> str:
return self.member_configuration["Name"] return self.member_configuration["Name"]
@property @property
def member_status(self): def member_status(self) -> str:
return self.status return self.status
def to_dict(self): def to_dict(self) -> Dict[str, Any]:
# Format for list_members # Format for list_members
d = { d = {
"Id": self.id, "Id": self.id,
@ -379,7 +384,7 @@ class ManagedBlockchainMember(BaseModel):
self.description = self.member_configuration["Description"] self.description = self.member_configuration["Description"]
return d return d
def get_format(self): def get_format(self) -> Dict[str, Any]:
# Format for get_member # Format for get_member
frameworkattributes = { frameworkattributes = {
"Fabric": { "Fabric": {
@ -405,10 +410,10 @@ class ManagedBlockchainMember(BaseModel):
d["Description"] = self.description d["Description"] = self.description
return d return d
def delete(self): def delete(self) -> None:
self.status = "DELETED" self.status = "DELETED"
def update(self, logpublishingconfiguration): def update(self, logpublishingconfiguration: Dict[str, Any]) -> None:
self.member_configuration[ self.member_configuration[
"LogPublishingConfiguration" "LogPublishingConfiguration"
] = logpublishingconfiguration ] = logpublishingconfiguration
@ -417,13 +422,13 @@ class ManagedBlockchainMember(BaseModel):
class ManagedBlockchainNode(BaseModel): class ManagedBlockchainNode(BaseModel):
def __init__( def __init__(
self, self,
node_id, node_id: str,
networkid, networkid: str,
memberid, memberid: str,
availabilityzone, availabilityzone: str,
instancetype, instancetype: str,
logpublishingconfiguration, logpublishingconfiguration: Dict[str, Any],
region, region: str,
): ):
self.creationdate = datetime.datetime.utcnow() self.creationdate = datetime.datetime.utcnow()
self.id = node_id self.id = node_id
@ -436,25 +441,24 @@ class ManagedBlockchainNode(BaseModel):
self.availabilityzone = availabilityzone self.availabilityzone = availabilityzone
@property @property
def member_id(self): def member_id(self) -> str:
return self.memberid return self.memberid
@property @property
def node_status(self): def node_status(self) -> str:
return self.status return self.status
def to_dict(self): def to_dict(self) -> Dict[str, Any]:
# Format for list_nodes # Format for list_nodes
d = { return {
"Id": self.id, "Id": self.id,
"Status": self.status, "Status": self.status,
"CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
"AvailabilityZone": self.availabilityzone, "AvailabilityZone": self.availabilityzone,
"InstanceType": self.instancetype, "InstanceType": self.instancetype,
} }
return d
def get_format(self): def get_format(self) -> Dict[str, Any]:
# Format for get_node # Format for get_node
frameworkattributes = { frameworkattributes = {
"Fabric": { "Fabric": {
@ -463,7 +467,7 @@ class ManagedBlockchainNode(BaseModel):
} }
} }
d = { return {
"NetworkId": self.networkid, "NetworkId": self.networkid,
"MemberId": self.memberid, "MemberId": self.memberid,
"Id": self.id, "Id": self.id,
@ -474,34 +478,33 @@ class ManagedBlockchainNode(BaseModel):
"Status": self.status, "Status": self.status,
"CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"), "CreationDate": self.creationdate.strftime("%Y-%m-%dT%H:%M:%S.%f%z"),
} }
return d
def delete(self): def delete(self) -> None:
self.status = "DELETED" self.status = "DELETED"
def update(self, logpublishingconfiguration): def update(self, logpublishingconfiguration: Dict[str, Any]) -> None:
self.logpublishingconfiguration = logpublishingconfiguration self.logpublishingconfiguration = logpublishingconfiguration
class ManagedBlockchainBackend(BaseBackend): class ManagedBlockchainBackend(BaseBackend):
def __init__(self, region_name, account_id): def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id) super().__init__(region_name, account_id)
self.networks = {} self.networks: Dict[str, ManagedBlockchainNetwork] = {}
self.members = {} self.members: Dict[str, ManagedBlockchainMember] = {}
self.proposals = {} self.proposals: Dict[str, ManagedBlockchainProposal] = {}
self.invitations = {} self.invitations: Dict[str, ManagedBlockchainInvitation] = {}
self.nodes = {} self.nodes: Dict[str, ManagedBlockchainNode] = {}
def create_network( def create_network(
self, self,
name, name: str,
framework, framework: str,
frameworkversion, frameworkversion: str,
frameworkconfiguration, frameworkconfiguration: Dict[str, Any],
voting_policy, voting_policy: Dict[str, Any],
member_configuration, member_configuration: Dict[str, Any],
description=None, description: Optional[str] = None,
): ) -> Dict[str, str]:
# Check framework # Check framework
if framework not in FRAMEWORKS: if framework not in FRAMEWORKS:
raise BadRequestException("CreateNetwork", "Invalid request body") raise BadRequestException("CreateNetwork", "Invalid request body")
@ -542,20 +545,25 @@ class ManagedBlockchainBackend(BaseBackend):
) )
# Return the network and member ID # Return the network and member ID
d = {"NetworkId": network_id, "MemberId": member_id} return {"NetworkId": network_id, "MemberId": member_id}
return d
def list_networks(self): def list_networks(self) -> List[ManagedBlockchainNetwork]:
return self.networks.values() return list(self.networks.values())
def get_network(self, network_id): def get_network(self, network_id: str) -> ManagedBlockchainNetwork:
if network_id not in self.networks: if network_id not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
"GetNetwork", f"Network {network_id} not found." "GetNetwork", f"Network {network_id} not found."
) )
return self.networks.get(network_id) return self.networks[network_id]
def create_proposal(self, networkid, memberid, actions, description=None): def create_proposal(
self,
networkid: str,
memberid: str,
actions: Dict[str, Any],
description: Optional[str] = None,
) -> Dict[str, str]:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -593,24 +601,21 @@ class ManagedBlockchainBackend(BaseBackend):
proposal_id=proposal_id, proposal_id=proposal_id,
networkid=networkid, networkid=networkid,
memberid=memberid, memberid=memberid,
membername=self.members.get(memberid).name, membername=self.members[memberid].name,
numofmembers=number_of_members_in_network(self.members, networkid), numofmembers=number_of_members_in_network(self.members, networkid),
actions=actions, actions=actions,
network_expirtation=self.networks.get(networkid).vote_pol_proposal_duration, network_expiration=self.networks[networkid].vote_pol_proposal_duration,
network_threshold=self.networks.get( network_threshold=self.networks[networkid].vote_pol_threshold_percentage,
network_threshold_comp=self.networks[
networkid networkid
).vote_pol_threshold_percentage, ].vote_pol_threshold_comparator,
network_threshold_comp=self.networks.get(
networkid
).vote_pol_threshold_comparator,
description=description, description=description,
) )
# Return the proposal ID # Return the proposal ID
d = {"ProposalId": proposal_id} return {"ProposalId": proposal_id}
return d
def list_proposals(self, networkid): def list_proposals(self, networkid: str) -> List[ManagedBlockchainProposal]:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -619,13 +624,15 @@ class ManagedBlockchainBackend(BaseBackend):
proposalsfornetwork = [] proposalsfornetwork = []
for proposal_id in self.proposals: for proposal_id in self.proposals:
if self.proposals.get(proposal_id).network_id == networkid: if self.proposals[proposal_id].network_id == networkid:
# See if any are expired # See if any are expired
self.proposals.get(proposal_id).check_to_expire_proposal() self.proposals[proposal_id].check_to_expire_proposal()
proposalsfornetwork.append(self.proposals[proposal_id]) proposalsfornetwork.append(self.proposals[proposal_id])
return proposalsfornetwork return proposalsfornetwork
def get_proposal(self, networkid, proposalid): def get_proposal(
self, networkid: str, proposalid: str
) -> ManagedBlockchainProposal:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -638,10 +645,12 @@ class ManagedBlockchainBackend(BaseBackend):
) )
# See if it needs to be set to expipred # See if it needs to be set to expipred
self.proposals.get(proposalid).check_to_expire_proposal() self.proposals[proposalid].check_to_expire_proposal()
return self.proposals.get(proposalid) return self.proposals[proposalid]
def vote_on_proposal(self, networkid, proposalid, votermemberid, vote): def vote_on_proposal(
self, networkid: str, proposalid: str, votermemberid: str, vote: str
) -> None:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -662,60 +671,56 @@ class ManagedBlockchainBackend(BaseBackend):
raise BadRequestException("VoteOnProposal", "Invalid request body") raise BadRequestException("VoteOnProposal", "Invalid request body")
# See if it needs to be set to expipred # See if it needs to be set to expipred
self.proposals.get(proposalid).check_to_expire_proposal() self.proposals[proposalid].check_to_expire_proposal()
# Exception if EXPIRED # Exception if EXPIRED
if self.proposals.get(proposalid).proposal_status == "EXPIRED": if self.proposals[proposalid].proposal_status == "EXPIRED":
raise InvalidRequestException( raise InvalidRequestException(
"VoteOnProposal", "VoteOnProposal",
f"Proposal {proposalid} is expired and you cannot vote on it.", f"Proposal {proposalid} is expired and you cannot vote on it.",
) )
# Check if IN_PROGRESS # Check if IN_PROGRESS
if self.proposals.get(proposalid).proposal_status != "IN_PROGRESS": if self.proposals[proposalid].proposal_status != "IN_PROGRESS":
raise InvalidRequestException( raise InvalidRequestException(
"VoteOnProposal", "VoteOnProposal",
f"Proposal {proposalid} has status {self.proposals.get(proposalid).proposal_status} and you cannot vote on it.", f"Proposal {proposalid} has status {self.proposals[proposalid].proposal_status} and you cannot vote on it.",
) )
# Check to see if this member already voted # Check to see if this member already voted
if votermemberid in self.proposals.get(proposalid).proposal_votes: if votermemberid in self.proposals[proposalid].proposal_votes:
raise ResourceAlreadyExistsException( raise ResourceAlreadyExistsException(
"VoteOnProposal", "VoteOnProposal",
f"Member {votermemberid} has already voted on proposal {proposalid}.", f"Member {votermemberid} has already voted on proposal {proposalid}.",
) )
# Cast vote # Cast vote
self.proposals.get(proposalid).set_vote( self.proposals[proposalid].set_vote(
votermemberid, self.members.get(votermemberid).name, vote.upper() votermemberid, self.members[votermemberid].name, vote.upper()
) )
if self.proposals.get(proposalid).proposal_status == "APPROVED": if self.proposals[proposalid].proposal_status == "APPROVED":
# Generate invitations # Generate invitations
for _ in self.proposals.get(proposalid).proposal_actions("Invitations"): for _ in self.proposals[proposalid].proposal_actions("Invitations"):
invitation_id = get_invitation_id() invitation_id = get_invitation_id()
self.invitations[invitation_id] = ManagedBlockchainInvitation( self.invitations[invitation_id] = ManagedBlockchainInvitation(
invitation_id=invitation_id, invitation_id=invitation_id,
networkid=networkid, networkid=networkid,
networkname=self.networks.get(networkid).network_name, networkname=self.networks[networkid].network_name,
networkframework=self.networks.get(networkid).network_framework, networkframework=self.networks[networkid].network_framework,
networkframeworkversion=self.networks.get( networkframeworkversion=self.networks[
networkid networkid
).network_framework_version, ].network_framework_version,
networkcreationdate=self.networks.get( networkcreationdate=self.networks[networkid].network_creationdate,
networkid
).network_creationdate,
region=self.region_name, region=self.region_name,
networkdescription=self.networks.get(networkid).network_description, networkdescription=self.networks[networkid].network_description,
) )
# Delete members # Delete members
for propmember in self.proposals.get(proposalid).proposal_actions( for propmember in self.proposals[proposalid].proposal_actions("Removals"):
"Removals"
):
self.delete_member(networkid, propmember["MemberId"]) self.delete_member(networkid, propmember["MemberId"])
def list_proposal_votes(self, networkid, proposalid): def list_proposal_votes(self, networkid: str, proposalid: str) -> List[str]:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -729,25 +734,25 @@ class ManagedBlockchainBackend(BaseBackend):
# Output the vote summaries # Output the vote summaries
proposalvotesfornetwork = [] proposalvotesfornetwork = []
for proposal_id in self.proposals: for proposal in self.proposals.values():
if self.proposals.get(proposal_id).network_id == networkid: if proposal.network_id == networkid:
for pvmemberid in self.proposals.get(proposal_id).proposal_votes: for proposal_vote in proposal.proposal_votes.values():
proposalvotesfornetwork.append( proposalvotesfornetwork.append(proposal_vote)
self.proposals.get(proposal_id).proposal_votes[pvmemberid]
)
return proposalvotesfornetwork return proposalvotesfornetwork
def list_invitations(self): def list_invitations(self) -> List[ManagedBlockchainInvitation]:
return self.invitations.values() return list(self.invitations.values())
def reject_invitation(self, invitationid): def reject_invitation(self, invitationid: str) -> None:
if invitationid not in self.invitations: if invitationid not in self.invitations:
raise ResourceNotFoundException( raise ResourceNotFoundException(
"RejectInvitation", f"InvitationId {invitationid} not found." "RejectInvitation", f"InvitationId {invitationid} not found."
) )
self.invitations.get(invitationid).reject_invitation() self.invitations[invitationid].reject_invitation()
def create_member(self, invitationid, networkid, member_configuration): def create_member(
self, invitationid: str, networkid: str, member_configuration: Dict[str, Any]
) -> Dict[str, str]:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -759,7 +764,7 @@ class ManagedBlockchainBackend(BaseBackend):
"CreateMember", f"Invitation {invitationid} not valid" "CreateMember", f"Invitation {invitationid} not valid"
) )
if self.invitations.get(invitationid).invitation_status != "PENDING": if self.invitations[invitationid].invitation_status != "PENDING":
raise InvalidRequestException( raise InvalidRequestException(
"CreateMember", f"Invitation {invitationid} not valid" "CreateMember", f"Invitation {invitationid} not valid"
) )
@ -772,7 +777,7 @@ class ManagedBlockchainBackend(BaseBackend):
f"Member name {member_configuration['Name']} already exists in network {networkid}.", f"Member name {member_configuration['Name']} already exists in network {networkid}.",
) )
networkedition = self.networks.get(networkid).network_edition networkedition = self.networks[networkid].network_edition
if ( if (
number_of_members_in_network(self.members, networkid) number_of_members_in_network(self.members, networkid)
>= EDITIONS[networkedition]["MaxMembers"] >= EDITIONS[networkedition]["MaxMembers"]
@ -797,13 +802,12 @@ class ManagedBlockchainBackend(BaseBackend):
) )
# Accept the invitaiton # Accept the invitaiton
self.invitations.get(invitationid).accept_invitation() self.invitations[invitationid].accept_invitation()
# Return the member ID # Return the member ID
d = {"MemberId": member_id} return {"MemberId": member_id}
return d
def list_members(self, networkid): def list_members(self, networkid: str) -> List[ManagedBlockchainMember]:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -811,12 +815,12 @@ class ManagedBlockchainBackend(BaseBackend):
) )
membersfornetwork = [] membersfornetwork = []
for member_id in self.members: for member in self.members.values():
if self.members.get(member_id).network_id == networkid: if member.network_id == networkid:
membersfornetwork.append(self.members[member_id]) membersfornetwork.append(member)
return membersfornetwork return membersfornetwork
def get_member(self, networkid, memberid): def get_member(self, networkid: str, memberid: str) -> ManagedBlockchainMember:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -829,14 +833,14 @@ class ManagedBlockchainBackend(BaseBackend):
) )
# Cannot get a member than has been deleted (it does show up in the list) # Cannot get a member than has been deleted (it does show up in the list)
if self.members.get(memberid).member_status == "DELETED": if self.members[memberid].member_status == "DELETED":
raise ResourceNotFoundException( raise ResourceNotFoundException(
"GetMember", f"Member {memberid} not found." "GetMember", f"Member {memberid} not found."
) )
return self.members.get(memberid) return self.members[memberid]
def delete_member(self, networkid, memberid): def delete_member(self, networkid: str, memberid: str) -> None:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -848,19 +852,16 @@ class ManagedBlockchainBackend(BaseBackend):
"DeleteMember", f"Member {memberid} not found." "DeleteMember", f"Member {memberid} not found."
) )
self.members.get(memberid).delete() self.members[memberid].delete()
# Is this the last member in the network? (all set to DELETED) # Is this the last member in the network? (all set to DELETED)
if number_of_members_in_network( if number_of_members_in_network(
self.members, networkid, member_status="DELETED" self.members, networkid, member_status="DELETED"
) == len(self.members): ) == len(self.members):
# Set network status to DELETED for all invitations # Set network status to DELETED for all invitations
for invitation_id in self.invitations: for invitation in self.invitations.values():
if ( if invitation.invitation_networkid == networkid:
self.invitations.get(invitation_id).invitation_networkid invitation.set_network_status("DELETED")
== networkid
):
self.invitations.get(invitation_id).set_network_status("DELETED")
# Remove network # Remove network
del self.networks[networkid] del self.networks[networkid]
@ -869,7 +870,9 @@ class ManagedBlockchainBackend(BaseBackend):
for nodeid in nodes_in_member(self.nodes, memberid): for nodeid in nodes_in_member(self.nodes, memberid):
del self.nodes[nodeid] del self.nodes[nodeid]
def update_member(self, networkid, memberid, logpublishingconfiguration): def update_member(
self, networkid: str, memberid: str, logpublishingconfiguration: Dict[str, Any]
) -> None:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -881,16 +884,16 @@ class ManagedBlockchainBackend(BaseBackend):
"UpdateMember", f"Member {memberid} not found." "UpdateMember", f"Member {memberid} not found."
) )
self.members.get(memberid).update(logpublishingconfiguration) self.members[memberid].update(logpublishingconfiguration)
def create_node( def create_node(
self, self,
networkid, networkid: str,
memberid, memberid: str,
availabilityzone, availabilityzone: str,
instancetype, instancetype: str,
logpublishingconfiguration, logpublishingconfiguration: Dict[str, Any],
): ) -> Dict[str, str]:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -902,7 +905,7 @@ class ManagedBlockchainBackend(BaseBackend):
"CreateNode", f"Member {memberid} not found." "CreateNode", f"Member {memberid} not found."
) )
networkedition = self.networks.get(networkid).network_edition networkedition = self.networks[networkid].network_edition
if ( if (
number_of_nodes_in_member(self.nodes, memberid) number_of_nodes_in_member(self.nodes, memberid)
>= EDITIONS[networkedition]["MaxNodesPerMember"] >= EDITIONS[networkedition]["MaxNodesPerMember"]
@ -953,10 +956,11 @@ class ManagedBlockchainBackend(BaseBackend):
) )
# Return the node ID # Return the node ID
d = {"NodeId": node_id} return {"NodeId": node_id}
return d
def list_nodes(self, networkid, memberid, status=None): def list_nodes(
self, networkid: str, memberid: str, status: Optional[str] = None
) -> List[ManagedBlockchainNode]:
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
"ListNodes", f"Network {networkid} not found." "ListNodes", f"Network {networkid} not found."
@ -968,20 +972,22 @@ class ManagedBlockchainBackend(BaseBackend):
) )
# If member is deleted, cannot list nodes # If member is deleted, cannot list nodes
if self.members.get(memberid).member_status == "DELETED": if self.members[memberid].member_status == "DELETED":
raise ResourceNotFoundException( raise ResourceNotFoundException(
"ListNodes", f"Member {memberid} not found." "ListNodes", f"Member {memberid} not found."
) )
nodesformember = [] nodesformember = []
for node_id in self.nodes: for node in self.nodes.values():
if self.nodes.get(node_id).member_id == memberid and ( if node.member_id == memberid and (
status is None or self.nodes.get(node_id).node_status == status status is None or node.node_status == status
): ):
nodesformember.append(self.nodes[node_id]) nodesformember.append(node)
return nodesformember return nodesformember
def get_node(self, networkid, memberid, nodeid): def get_node(
self, networkid: str, memberid: str, nodeid: str
) -> ManagedBlockchainNode:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -995,12 +1001,12 @@ class ManagedBlockchainBackend(BaseBackend):
raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.") raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.")
# Cannot get a node than has been deleted (it does show up in the list) # Cannot get a node than has been deleted (it does show up in the list)
if self.nodes.get(nodeid).node_status == "DELETED": if self.nodes[nodeid].node_status == "DELETED":
raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.") raise ResourceNotFoundException("GetNode", f"Node {nodeid} not found.")
return self.nodes.get(nodeid) return self.nodes[nodeid]
def delete_node(self, networkid, memberid, nodeid): def delete_node(self, networkid: str, memberid: str, nodeid: str) -> None:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -1015,9 +1021,15 @@ class ManagedBlockchainBackend(BaseBackend):
if nodeid not in self.nodes: if nodeid not in self.nodes:
raise ResourceNotFoundException("DeleteNode", f"Node {nodeid} not found.") raise ResourceNotFoundException("DeleteNode", f"Node {nodeid} not found.")
self.nodes.get(nodeid).delete() self.nodes[nodeid].delete()
def update_node(self, networkid, memberid, nodeid, logpublishingconfiguration): def update_node(
self,
networkid: str,
memberid: str,
nodeid: str,
logpublishingconfiguration: Dict[str, Any],
) -> None:
# Check if network exists # Check if network exists
if networkid not in self.networks: if networkid not in self.networks:
raise ResourceNotFoundException( raise ResourceNotFoundException(
@ -1032,7 +1044,7 @@ class ManagedBlockchainBackend(BaseBackend):
if nodeid not in self.nodes: if nodeid not in self.nodes:
raise ResourceNotFoundException("UpdateNode", f"Node {nodeid} not found.") raise ResourceNotFoundException("UpdateNode", f"Node {nodeid} not found.")
self.nodes.get(nodeid).update(logpublishingconfiguration) self.nodes[nodeid].update(logpublishingconfiguration)
managedblockchain_backends = BackendDict(ManagedBlockchainBackend, "managedblockchain") managedblockchain_backends = BackendDict(ManagedBlockchainBackend, "managedblockchain")

127
moto/managedblockchain/responses.py
View File

@ -1,9 +1,11 @@
import json import json
from typing import Any, Dict, Optional
from urllib.parse import urlparse, parse_qs from urllib.parse import urlparse, 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 .exceptions import exception_handler
from .models import managedblockchain_backends from .models import managedblockchain_backends, ManagedBlockchainBackend
from .utils import ( from .utils import (
networkid_from_managedblockchain_url, networkid_from_managedblockchain_url,
proposalid_from_managedblockchain_url, proposalid_from_managedblockchain_url,
@ -14,19 +16,19 @@ from .utils import (
class ManagedBlockchainResponse(BaseResponse): class ManagedBlockchainResponse(BaseResponse):
def __init__(self): def __init__(self) -> None:
super().__init__(service_name="managedblockchain") super().__init__(service_name="managedblockchain")
@property @property
def backend(self): def backend(self) -> ManagedBlockchainBackend:
return managedblockchain_backends[self.current_account][self.region] return managedblockchain_backends[self.current_account][self.region]
@exception_handler @exception_handler
def network_response(self, request, full_url, headers): def network_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._network_response(request, headers) return self._network_response(request, headers)
def _network_response(self, request, headers): def _network_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
if method == "GET": if method == "GET":
return self._all_networks_response(headers) return self._all_networks_response(headers)
@ -34,7 +36,7 @@ class ManagedBlockchainResponse(BaseResponse):
json_body = json.loads(self.body) json_body = json.loads(self.body)
return self._network_response_post(json_body, headers) return self._network_response_post(json_body, headers)
def _all_networks_response(self, headers): def _all_networks_response(self, headers: Any) -> TYPE_RESPONSE:
mbcnetworks = self.backend.list_networks() mbcnetworks = self.backend.list_networks()
response = json.dumps( response = json.dumps(
{"Networks": [mbcnetwork.to_dict() for mbcnetwork in mbcnetworks]} {"Networks": [mbcnetwork.to_dict() for mbcnetwork in mbcnetworks]}
@ -42,7 +44,9 @@ class ManagedBlockchainResponse(BaseResponse):
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _network_response_post(self, json_body, headers): def _network_response_post(
self, json_body: Dict[str, Any], headers: Any
) -> TYPE_RESPONSE:
name = json_body["Name"] name = json_body["Name"]
framework = json_body["Framework"] framework = json_body["Framework"]
frameworkversion = json_body["FrameworkVersion"] frameworkversion = json_body["FrameworkVersion"]
@ -65,29 +69,29 @@ class ManagedBlockchainResponse(BaseResponse):
return 200, headers, json.dumps(response) return 200, headers, json.dumps(response)
@exception_handler @exception_handler
def networkid_response(self, request, full_url, headers): def networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._networkid_response(request, full_url, headers) return self._networkid_response(request, full_url, headers)
def _networkid_response(self, request, full_url, headers): def _networkid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
if method == "GET": if method == "GET":
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
return self._networkid_response_get(network_id, headers) return self._networkid_response_get(network_id, headers)
def _networkid_response_get(self, 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()}) response = json.dumps({"Network": mbcnetwork.get_format()})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
@exception_handler @exception_handler
def proposal_response(self, request, full_url, headers): def proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._proposal_response(request, full_url, headers) return self._proposal_response(request, full_url, headers)
def _proposal_response(self, request, full_url, headers): def _proposal_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
if method == "GET": if method == "GET":
@ -96,7 +100,7 @@ class ManagedBlockchainResponse(BaseResponse):
json_body = json.loads(self.body) json_body = json.loads(self.body)
return self._proposal_response_post(network_id, json_body, headers) return self._proposal_response_post(network_id, json_body, headers)
def _all_proposals_response(self, network_id, 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( response = json.dumps(
{"Proposals": [proposal.to_dict() for proposal in proposals]} {"Proposals": [proposal.to_dict() for proposal in proposals]}
@ -104,7 +108,9 @@ class ManagedBlockchainResponse(BaseResponse):
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _proposal_response_post(self, network_id, json_body, headers): def _proposal_response_post(
self, network_id: str, json_body: Dict[str, Any], headers: Any
) -> TYPE_RESPONSE:
memberid = json_body["MemberId"] memberid = json_body["MemberId"]
actions = json_body["Actions"] actions = json_body["Actions"]
@ -117,29 +123,31 @@ class ManagedBlockchainResponse(BaseResponse):
return 200, headers, json.dumps(response) return 200, headers, json.dumps(response)
@exception_handler @exception_handler
def proposalid_response(self, request, full_url, headers): def proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._proposalid_response(request, full_url, headers) return self._proposalid_response(request, full_url, headers)
def _proposalid_response(self, request, full_url, headers): def _proposalid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
if method == "GET": if method == "GET":
proposal_id = proposalid_from_managedblockchain_url(full_url) proposal_id = proposalid_from_managedblockchain_url(full_url)
return self._proposalid_response_get(network_id, proposal_id, headers) return self._proposalid_response_get(network_id, proposal_id, headers)
def _proposalid_response_get(self, 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()}) response = json.dumps({"Proposal": proposal.get_format()})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
@exception_handler @exception_handler
def proposal_votes_response(self, request, full_url, headers): def proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._proposal_votes_response(request, full_url, headers) return self._proposal_votes_response(request, full_url, headers)
def _proposal_votes_response(self, request, full_url, headers): def _proposal_votes_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
proposal_id = proposalid_from_managedblockchain_url(full_url) proposal_id = proposalid_from_managedblockchain_url(full_url)
@ -151,15 +159,17 @@ class ManagedBlockchainResponse(BaseResponse):
network_id, proposal_id, json_body, headers network_id, proposal_id, json_body, headers
) )
def _all_proposal_votes_response(self, network_id, proposal_id, 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}) response = json.dumps({"ProposalVotes": proposalvotes})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _proposal_votes_response_post( def _proposal_votes_response_post(
self, network_id, proposal_id, json_body, headers self, network_id: str, proposal_id: str, json_body: Dict[str, Any], headers: Any
): ) -> TYPE_RESPONSE:
votermemberid = json_body["VoterMemberId"] votermemberid = json_body["VoterMemberId"]
vote = json_body["Vote"] vote = json_body["Vote"]
@ -167,16 +177,16 @@ class ManagedBlockchainResponse(BaseResponse):
return 200, headers, "" return 200, headers, ""
@exception_handler @exception_handler
def invitation_response(self, request, full_url, headers): def invitation_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._invitation_response(request, headers) return self._invitation_response(request, headers)
def _invitation_response(self, request, headers): def _invitation_response(self, request: Any, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
if method == "GET": if method == "GET":
return self._all_invitation_response(headers) return self._all_invitation_response(headers)
def _all_invitation_response(self, headers): def _all_invitation_response(self, headers: Any) -> TYPE_RESPONSE:
invitations = self.backend.list_invitations() invitations = self.backend.list_invitations()
response = json.dumps( response = json.dumps(
{"Invitations": [invitation.to_dict() for invitation in invitations]} {"Invitations": [invitation.to_dict() for invitation in invitations]}
@ -185,27 +195,29 @@ class ManagedBlockchainResponse(BaseResponse):
return 200, headers, response return 200, headers, response
@exception_handler @exception_handler
def invitationid_response(self, request, full_url, headers): def invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._invitationid_response(request, full_url, headers) return self._invitationid_response(request, full_url, headers)
def _invitationid_response(self, request, full_url, headers): def _invitationid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
if method == "DELETE": if method == "DELETE":
invitation_id = invitationid_from_managedblockchain_url(full_url) invitation_id = invitationid_from_managedblockchain_url(full_url)
return self._invitationid_response_delete(invitation_id, headers) return self._invitationid_response_delete(invitation_id, headers)
def _invitationid_response_delete(self, 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" headers["content-type"] = "application/json"
return 200, headers, "" return 200, headers, ""
@exception_handler @exception_handler
def member_response(self, request, full_url, headers): def member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._member_response(request, full_url, headers) return self._member_response(request, full_url, headers)
def _member_response(self, request, full_url, headers): def _member_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
if method == "GET": if method == "GET":
@ -214,13 +226,15 @@ class ManagedBlockchainResponse(BaseResponse):
json_body = json.loads(self.body) json_body = json.loads(self.body)
return self._member_response_post(network_id, json_body, headers) return self._member_response_post(network_id, json_body, headers)
def _all_members_response(self, network_id, 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]}) response = json.dumps({"Members": [member.to_dict() for member in members]})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _member_response_post(self, network_id, json_body, headers): def _member_response_post(
self, network_id: str, json_body: Dict[str, Any], headers: Any
) -> TYPE_RESPONSE:
invitationid = json_body["InvitationId"] invitationid = json_body["InvitationId"]
member_configuration = json_body["MemberConfiguration"] member_configuration = json_body["MemberConfiguration"]
@ -230,11 +244,11 @@ class ManagedBlockchainResponse(BaseResponse):
return 200, headers, json.dumps(response) return 200, headers, json.dumps(response)
@exception_handler @exception_handler
def memberid_response(self, request, full_url, headers): def memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._memberid_response(request, full_url, headers) return self._memberid_response(request, full_url, headers)
def _memberid_response(self, request, full_url, headers): def _memberid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
member_id = memberid_from_managedblockchain_request(full_url, self.body) member_id = memberid_from_managedblockchain_request(full_url, self.body)
@ -248,28 +262,34 @@ class ManagedBlockchainResponse(BaseResponse):
elif method == "DELETE": elif method == "DELETE":
return self._memberid_response_delete(network_id, member_id, headers) return self._memberid_response_delete(network_id, member_id, headers)
def _memberid_response_get(self, 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()}) response = json.dumps({"Member": member.get_format()})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _memberid_response_patch(self, network_id, member_id, json_body, headers): def _memberid_response_patch(
self, network_id: str, member_id: str, json_body: Dict[str, Any], headers: Any
) -> TYPE_RESPONSE:
logpublishingconfiguration = json_body["LogPublishingConfiguration"] logpublishingconfiguration = json_body["LogPublishingConfiguration"]
self.backend.update_member(network_id, member_id, logpublishingconfiguration) self.backend.update_member(network_id, member_id, logpublishingconfiguration)
return 200, headers, "" return 200, headers, ""
def _memberid_response_delete(self, network_id, member_id, headers): def _memberid_response_delete(
self, network_id: str, member_id: str, headers: Any
) -> TYPE_RESPONSE:
self.backend.delete_member(network_id, member_id) self.backend.delete_member(network_id, member_id)
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, "" return 200, headers, ""
@exception_handler @exception_handler
def node_response(self, request, full_url, headers): def node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._node_response(request, full_url, headers) return self._node_response(request, full_url, headers)
def _node_response(self, request, full_url, headers): def _node_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
parsed_url = urlparse(full_url) parsed_url = urlparse(full_url)
querystring = parse_qs(parsed_url.query, keep_blank_values=True) querystring = parse_qs(parsed_url.query, keep_blank_values=True)
@ -284,13 +304,17 @@ class ManagedBlockchainResponse(BaseResponse):
json_body = json.loads(self.body) json_body = json.loads(self.body)
return self._node_response_post(network_id, member_id, json_body, headers) return self._node_response_post(network_id, member_id, json_body, headers)
def _all_nodes_response(self, network_id, member_id, status, 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]}) response = json.dumps({"Nodes": [node.to_dict() for node in nodes]})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _node_response_post(self, network_id, member_id, json_body, headers): 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"] instancetype = json_body["NodeConfiguration"]["InstanceType"]
availabilityzone = json_body["NodeConfiguration"]["AvailabilityZone"] availabilityzone = json_body["NodeConfiguration"]["AvailabilityZone"]
logpublishingconfiguration = json_body["NodeConfiguration"][ logpublishingconfiguration = json_body["NodeConfiguration"][
@ -307,11 +331,11 @@ class ManagedBlockchainResponse(BaseResponse):
return 200, headers, json.dumps(response) return 200, headers, json.dumps(response)
@exception_handler @exception_handler
def nodeid_response(self, request, full_url, headers): def nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[misc]
self.setup_class(request, full_url, headers) self.setup_class(request, full_url, headers)
return self._nodeid_response(request, full_url, headers) return self._nodeid_response(request, full_url, headers)
def _nodeid_response(self, request, full_url, headers): def _nodeid_response(self, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE: # type: ignore[return]
method = request.method method = request.method
network_id = networkid_from_managedblockchain_url(full_url) network_id = networkid_from_managedblockchain_url(full_url)
member_id = memberid_from_managedblockchain_request(full_url, self.body) member_id = memberid_from_managedblockchain_request(full_url, self.body)
@ -326,22 +350,31 @@ class ManagedBlockchainResponse(BaseResponse):
elif method == "DELETE": elif method == "DELETE":
return self._nodeid_response_delete(network_id, member_id, node_id, headers) return self._nodeid_response_delete(network_id, member_id, node_id, headers)
def _nodeid_response_get(self, 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()}) response = json.dumps({"Node": node.get_format()})
headers["content-type"] = "application/json" headers["content-type"] = "application/json"
return 200, headers, response return 200, headers, response
def _nodeid_response_patch( def _nodeid_response_patch(
self, network_id, member_id, node_id, json_body, headers self,
): network_id: str,
member_id: str,
node_id: str,
json_body: Dict[str, Any],
headers: Any,
) -> TYPE_RESPONSE:
logpublishingconfiguration = json_body 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
) )
return 200, headers, "" return 200, headers, ""
def _nodeid_response_delete(self, network_id, member_id, node_id, headers): def _nodeid_response_delete(
self, network_id: str, member_id: str, node_id: str, headers: Any
) -> TYPE_RESPONSE:
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" headers["content-type"] = "application/json"
return 200, headers, "" return 200, headers, ""

84
moto/managedblockchain/utils.py
View File

@ -2,24 +2,25 @@ import json
import re import re
import string import string
from moto.moto_api._internal import mock_random as random from moto.moto_api._internal import mock_random as random
from typing import Any, Dict, List, Optional
from urllib.parse import parse_qs, urlparse from urllib.parse import parse_qs, urlparse
def networkid_from_managedblockchain_url(full_url): def networkid_from_managedblockchain_url(full_url: str) -> str:
id_search = re.search(r"\/n-[A-Z0-9]{26}", full_url, re.IGNORECASE) id_search = re.search(r"\/n-[A-Z0-9]{26}", full_url, re.IGNORECASE)
return_id = None return_id = None
if id_search: if id_search:
return_id = id_search.group(0).replace("/", "") return_id = id_search.group(0).replace("/", "")
return return_id return return_id # type: ignore[return-value]
def get_network_id(): def get_network_id() -> str:
return "n-" + "".join( return "n-" + "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(26) random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
) )
def memberid_from_managedblockchain_request(full_url, body): def memberid_from_managedblockchain_request(full_url: str, body: Dict[str, Any]) -> str:
id_search = re.search(r"\/m-[A-Z0-9]{26}", full_url, re.IGNORECASE) id_search = re.search(r"\/m-[A-Z0-9]{26}", full_url, re.IGNORECASE)
return_id = None return_id = None
if id_search: if id_search:
@ -29,72 +30,71 @@ def memberid_from_managedblockchain_request(full_url, body):
parsed_url = urlparse(full_url) parsed_url = urlparse(full_url)
qs = parse_qs(parsed_url.query) qs = parse_qs(parsed_url.query)
if "memberId" in qs: if "memberId" in qs:
return_id = qs.get("memberId")[0] return_id = qs.get("memberId")[0] # type: ignore
elif body: elif body:
body = json.loads(body) body = json.loads(body) # type: ignore
return_id = body["MemberId"] return_id = body["MemberId"]
return return_id return return_id # type: ignore[return-value]
def get_member_id(): def get_member_id() -> str:
return "m-" + "".join( return "m-" + "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(26) random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
) )
def proposalid_from_managedblockchain_url(full_url): def proposalid_from_managedblockchain_url(full_url: str) -> str:
id_search = re.search(r"\/p-[A-Z0-9]{26}", full_url, re.IGNORECASE) id_search = re.search(r"\/p-[A-Z0-9]{26}", full_url, re.IGNORECASE)
return_id = None return_id = None
if id_search: if id_search:
return_id = id_search.group(0).replace("/", "") return_id = id_search.group(0).replace("/", "")
return return_id return return_id # type: ignore[return-value]
def get_proposal_id(): def get_proposal_id() -> str:
return "p-" + "".join( return "p-" + "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(26) random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
) )
def invitationid_from_managedblockchain_url(full_url): def invitationid_from_managedblockchain_url(full_url: str) -> str:
id_search = re.search(r"\/in-[A-Z0-9]{26}", full_url, re.IGNORECASE) id_search = re.search(r"\/in-[A-Z0-9]{26}", full_url, re.IGNORECASE)
return_id = None return_id = None
if id_search: if id_search:
return_id = id_search.group(0).replace("/", "") return_id = id_search.group(0).replace("/", "")
return return_id return return_id # type: ignore[return-value]
def get_invitation_id(): def get_invitation_id() -> str:
return "in-" + "".join( return "in-" + "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(26) random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
) )
def member_name_exist_in_network(members, networkid, membername): def member_name_exist_in_network(
membernamexists = False members: Dict[str, Any], networkid: str, membername: str
for member_id in members: ) -> bool:
if members.get(member_id).network_id == networkid: for member in members.values():
if members.get(member_id).name == membername: if member.network_id == networkid:
membernamexists = True if member.name == membername:
break return True
return membernamexists return False
def number_of_members_in_network(members, networkid, member_status=None): def number_of_members_in_network(
members: Dict[str, Any], networkid: str, member_status: Optional[str] = None
) -> int:
return len( return len(
[ [
membid member
for membid in members for member in members.values()
if members.get(membid).network_id == networkid if member.network_id == networkid
and ( and (member_status is None or member.member_status == member_status)
member_status is None
or members.get(membid).member_status == member_status
)
] ]
) )
def admin_password_ok(password): def admin_password_ok(password: str) -> bool:
if not re.search("[a-z]", password): if not re.search("[a-z]", password):
return False return False
elif not re.search("[A-Z]", password): elif not re.search("[A-Z]", password):
@ -107,30 +107,32 @@ def admin_password_ok(password):
return True return True
def nodeid_from_managedblockchain_url(full_url): def nodeid_from_managedblockchain_url(full_url: str) -> str:
id_search = re.search(r"\/nd-[A-Z0-9]{26}", full_url, re.IGNORECASE) id_search = re.search(r"\/nd-[A-Z0-9]{26}", full_url, re.IGNORECASE)
return_id = None return_id = None
if id_search: if id_search:
return_id = id_search.group(0).replace("/", "") return_id = id_search.group(0).replace("/", "")
return return_id return return_id # type: ignore[return-value]
def get_node_id(): def get_node_id() -> str:
return "nd-" + "".join( return "nd-" + "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(26) random.choice(string.ascii_uppercase + string.digits) for _ in range(26)
) )
def number_of_nodes_in_member(nodes, memberid, node_status=None): def number_of_nodes_in_member(
nodes: Dict[str, Any], memberid: str, node_status: Optional[str] = None
) -> int:
return len( return len(
[ [
nodid node
for nodid in nodes for node in nodes.values()
if nodes.get(nodid).member_id == memberid if node.member_id == memberid
and (node_status is None or nodes.get(nodid).node_status == node_status) and (node_status is None or node.node_status == node_status)
] ]
) )
def nodes_in_member(nodes, memberid): def nodes_in_member(nodes: Dict[str, Any], memberid: str) -> List[str]:
return [nodid for nodid in nodes if nodes.get(nodid).member_id == memberid] return [nodid for nodid in nodes if nodes[nodid].member_id == memberid]

2
setup.cfg
View File

@ -235,7 +235,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy] [mypy]
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/moto_api,moto/neptune,moto/opensearch files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/managedblockchain,moto/moto_api,moto/neptune,moto/opensearch
show_column_numbers=True show_column_numbers=True
show_error_codes = True show_error_codes = True
disable_error_code=abstract disable_error_code=abstract