from __future__ import unicode_literals import boto3 import sure # noqa from moto import mock_managedblockchain from . import helpers @mock_managedblockchain def test_create_another_member(): conn = boto3.client("managedblockchain", region_name="us-east-1") # Create network response = conn.create_network( Name="testnetwork1", Description="Test Network 1", 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"] # Create proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Get proposal details response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response["Proposal"]["NetworkId"].should.equal(network_id) response["Proposal"]["Status"].should.equal("IN_PROGRESS") # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Get the invitation response = conn.list_invitations() response["Invitations"][0]["NetworkSummary"]["Id"].should.equal(network_id) response["Invitations"][0]["Status"].should.equal("PENDING") invitation_id = response["Invitations"][0]["InvitationId"] # Create the member response = conn.create_member( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration=helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False ), ) member_id2 = response["MemberId"] # Check the invitation status response = conn.list_invitations() response["Invitations"][0]["InvitationId"].should.equal(invitation_id) response["Invitations"][0]["Status"].should.equal("ACCEPTED") # Find member in full list response = conn.list_members(NetworkId=network_id) members = response["Members"] members.should.have.length_of(2) helpers.member_id_exist_in_list(members, member_id2).should.equal(True) # Get member 2 details response = conn.get_member(NetworkId=network_id, MemberId=member_id2) response["Member"]["Name"].should.equal("testmember2") # Update member logconfignewenabled = not helpers.default_memberconfiguration[ "LogPublishingConfiguration" ]["Fabric"]["CaLogs"]["Cloudwatch"]["Enabled"] logconfignew = { "Fabric": {"CaLogs": {"Cloudwatch": {"Enabled": logconfignewenabled}}} } conn.update_member( NetworkId=network_id, MemberId=member_id2, LogPublishingConfiguration=logconfignew, ) # Get member 2 details response = conn.get_member(NetworkId=network_id, MemberId=member_id2) response["Member"]["LogPublishingConfiguration"]["Fabric"]["CaLogs"]["Cloudwatch"][ "Enabled" ].should.equal(logconfignewenabled) @mock_managedblockchain def test_create_another_member_withopts(): conn = boto3.client("managedblockchain", region_name="us-east-1") # 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"] # Create proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Get proposal details response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response["Proposal"]["NetworkId"].should.equal(network_id) response["Proposal"]["Status"].should.equal("IN_PROGRESS") # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Get the invitation response = conn.list_invitations() response["Invitations"][0]["NetworkSummary"]["Id"].should.equal(network_id) response["Invitations"][0]["Status"].should.equal("PENDING") invitation_id = response["Invitations"][0]["InvitationId"] # Create the member response = conn.create_member( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration=helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False, "Test Member 2" ), ) member_id2 = response["MemberId"] # Check the invitation status response = conn.list_invitations() response["Invitations"][0]["InvitationId"].should.equal(invitation_id) response["Invitations"][0]["Status"].should.equal("ACCEPTED") # Find member in full list response = conn.list_members(NetworkId=network_id) members = response["Members"] members.should.have.length_of(2) helpers.member_id_exist_in_list(members, member_id2).should.equal(True) # Get member 2 details response = conn.get_member(NetworkId=network_id, MemberId=member_id2) response["Member"]["Description"].should.equal("Test Member 2") # Try to create member with already used invitation response = conn.create_member.when.called_with( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration=helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False, "Test Member 2 Duplicate" ), ).should.throw(Exception, "Invitation {0} not valid".format(invitation_id)) # Delete member 2 conn.delete_member(NetworkId=network_id, MemberId=member_id2) # Member is still in the list response = conn.list_members(NetworkId=network_id) members = response["Members"] members.should.have.length_of(2) # But cannot get response = conn.get_member.when.called_with( NetworkId=network_id, MemberId=member_id2, ).should.throw(Exception, "Member {0} not found".format(member_id2)) # Delete member 1 conn.delete_member(NetworkId=network_id, MemberId=member_id) # Network should be gone response = conn.list_networks() mbcnetworks = response["Networks"] mbcnetworks.should.have.length_of(0) # Verify the invitation network status is DELETED # Get the invitation response = conn.list_invitations() response["Invitations"].should.have.length_of(1) response["Invitations"][0]["NetworkSummary"]["Id"].should.equal(network_id) response["Invitations"][0]["NetworkSummary"]["Status"].should.equal("DELETED") @mock_managedblockchain def test_invite_and_remove_member(): conn = boto3.client("managedblockchain", region_name="us-east-1") # 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"] # Create proposal (create additional member) response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Get the invitation response = conn.list_invitations() invitation_id = response["Invitations"][0]["InvitationId"] # Create the member response = conn.create_member( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration=helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False, "Test Member 2" ), ) member_id2 = response["MemberId"] both_policy_actions = { "Invitations": [{"Principal": "123456789012"}], "Removals": [{"MemberId": member_id2}], } # Create proposal (invite and remove member) response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=both_policy_actions, ) proposal_id2 = response["ProposalId"] # Get proposal details response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id2) response["Proposal"]["NetworkId"].should.equal(network_id) response["Proposal"]["Status"].should.equal("IN_PROGRESS") # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id2, VoterMemberId=member_id, Vote="YES", ) # Check the invitation status response = conn.list_invitations() invitations = helpers.select_invitation_id_for_network( response["Invitations"], network_id, "PENDING" ) invitations.should.have.length_of(1) # Member is still in the list response = conn.list_members(NetworkId=network_id) members = response["Members"] members.should.have.length_of(2) foundmember2 = False for member in members: if member["Id"] == member_id2 and member["Status"] == "DELETED": foundmember2 = True foundmember2.should.equal(True) @mock_managedblockchain def test_create_too_many_members(): conn = boto3.client("managedblockchain", region_name="us-east-1") # 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"] # Create 4 more members - create invitations for 5 for counter in range(2, 7): # Create proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) for counter in range(2, 6): # Get the invitation response = conn.list_invitations() invitation_id = helpers.select_invitation_id_for_network( response["Invitations"], network_id, "PENDING" )[0] # Create the member response = conn.create_member( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration=helpers.create_member_configuration( "testmember" + str(counter), "admin", "Admin12345", False, "Test Member " + str(counter), ), ) member_id = response["MemberId"] # Find member in full list response = conn.list_members(NetworkId=network_id) members = response["Members"] members.should.have.length_of(counter) helpers.member_id_exist_in_list(members, member_id).should.equal(True) # Get member details response = conn.get_member(NetworkId=network_id, MemberId=member_id) response["Member"]["Description"].should.equal("Test Member " + str(counter)) # Try to create the sixth response = conn.list_invitations() invitation_id = helpers.select_invitation_id_for_network( response["Invitations"], network_id, "PENDING" )[0] # Try to create one too many members response = conn.create_member.when.called_with( InvitationId=invitation_id, NetworkId=network_id, MemberConfiguration=helpers.create_member_configuration( "testmember6", "admin", "Admin12345", False, "Test Member 6" ), ).should.throw(Exception, "is the maximum number of members allowed in a",) @mock_managedblockchain def test_create_another_member_alreadyhave(): conn = boto3.client("managedblockchain", region_name="us-east-1") # Create network response = conn.create_network( Name="testnetwork1", Description="Test Network 1", 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"] # Create proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Get the invitation response = conn.list_invitations() invitation_id = response["Invitations"][0]["InvitationId"] # Should fail trying to create with same name response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId=invitation_id, MemberConfiguration=helpers.create_member_configuration( "testmember1", "admin", "Admin12345", False ), ).should.throw( Exception, "Member name {0} already exists in network {1}".format( "testmember1", network_id ), ) @mock_managedblockchain def test_create_another_member_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.create_member.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", InvitationId="id-ABCDEFGHIJKLMNOP0123456789", MemberConfiguration=helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False ), ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_create_another_member_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"] response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId="in-ABCDEFGHIJKLMNOP0123456789", MemberConfiguration=helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False ), ).should.throw(Exception, "Invitation in-ABCDEFGHIJKLMNOP0123456789 not valid") @mock_managedblockchain def test_create_another_member_adminpassword(): 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"] # Create proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Get proposal details response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response["Proposal"]["NetworkId"].should.equal(network_id) response["Proposal"]["Status"].should.equal("IN_PROGRESS") # Vote yes response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Get the invitation response = conn.list_invitations() invitation_id = response["Invitations"][0]["InvitationId"] badadminpassmemberconf = helpers.create_member_configuration( "testmember2", "admin", "Admin12345", False ) # Too short badadminpassmemberconf["FrameworkConfiguration"]["Fabric"][ "AdminPassword" ] = "badap" response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId=invitation_id, MemberConfiguration=badadminpassmemberconf, ).should.throw( Exception, "Invalid length for parameter MemberConfiguration.FrameworkConfiguration.Fabric.AdminPassword", ) # No uppercase or numbers badadminpassmemberconf["FrameworkConfiguration"]["Fabric"][ "AdminPassword" ] = "badadminpwd" response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId=invitation_id, MemberConfiguration=badadminpassmemberconf, ).should.throw(Exception, "Invalid request body") # No lowercase or numbers badadminpassmemberconf["FrameworkConfiguration"]["Fabric"][ "AdminPassword" ] = "BADADMINPWD" response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId=invitation_id, MemberConfiguration=badadminpassmemberconf, ).should.throw(Exception, "Invalid request body") # No numbers badadminpassmemberconf["FrameworkConfiguration"]["Fabric"][ "AdminPassword" ] = "badAdminpwd" response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId=invitation_id, MemberConfiguration=badadminpassmemberconf, ).should.throw(Exception, "Invalid request body") # Invalid character badadminpassmemberconf["FrameworkConfiguration"]["Fabric"][ "AdminPassword" ] = "badAdmin@pwd1" response = conn.create_member.when.called_with( NetworkId=network_id, InvitationId=invitation_id, MemberConfiguration=badadminpassmemberconf, ).should.throw(Exception, "Invalid request body") @mock_managedblockchain def test_list_members_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.list_members.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_get_member_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.get_member.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", MemberId="m-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_get_member_badmember(): 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"] response = conn.get_member.when.called_with( NetworkId=network_id, MemberId="m-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_delete_member_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.delete_member.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", MemberId="m-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_delete_member_badmember(): 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"] response = conn.delete_member.when.called_with( NetworkId=network_id, MemberId="m-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_update_member_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.update_member.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", MemberId="m-ABCDEFGHIJKLMNOP0123456789", LogPublishingConfiguration=helpers.default_memberconfiguration[ "LogPublishingConfiguration" ], ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_update_member_badmember(): 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"] response = conn.update_member.when.called_with( NetworkId=network_id, MemberId="m-ABCDEFGHIJKLMNOP0123456789", LogPublishingConfiguration=helpers.default_memberconfiguration[ "LogPublishingConfiguration" ], ).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found")