from __future__ import unicode_literals import os import boto3 import sure # noqa from freezegun import freeze_time from nose import SkipTest from moto.managedblockchain.exceptions import BadRequestException from moto import mock_managedblockchain, settings from . import helpers @mock_managedblockchain def test_vote_on_proposal_one_member_total_yes(): 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", ) # List proposal votes response = conn.list_proposal_votes(NetworkId=network_id, ProposalId=proposal_id) response["ProposalVotes"][0]["MemberId"].should.equal(member_id) # Get proposal details - should be APPROVED response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response["Proposal"]["Status"].should.equal("APPROVED") response["Proposal"]["YesVoteCount"].should.equal(1) response["Proposal"]["NoVoteCount"].should.equal(0) response["Proposal"]["OutstandingVoteCount"].should.equal(0) @mock_managedblockchain def test_vote_on_proposal_one_member_total_no(): 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 no response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="NO", ) # List proposal votes response = conn.list_proposal_votes(NetworkId=network_id, ProposalId=proposal_id) response["ProposalVotes"][0]["MemberId"].should.equal(member_id) # Get proposal details - should be REJECTED response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response["Proposal"]["Status"].should.equal("REJECTED") response["Proposal"]["YesVoteCount"].should.equal(0) response["Proposal"]["NoVoteCount"].should.equal(1) response["Proposal"]["OutstandingVoteCount"].should.equal(0) @mock_managedblockchain def test_vote_on_proposal_yes_greater_than(): conn = boto3.client("managedblockchain", region_name="us-east-1") votingpolicy = { "ApprovalThresholdPolicy": { "ThresholdPercentage": 50, "ProposalDurationInHours": 24, "ThresholdComparator": "GREATER_THAN", } } # Create network response = conn.create_network( Name="testnetwork1", Framework="HYPERLEDGER_FABRIC", FrameworkVersion="1.2", FrameworkConfiguration=helpers.default_frameworkconfiguration, VotingPolicy=votingpolicy, MemberConfiguration=helpers.default_memberconfiguration, ) network_id = response["NetworkId"] member_id = response["MemberId"] 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"] # Create another proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Vote yes with member 1 response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # 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") @mock_managedblockchain def test_vote_on_proposal_no_greater_than(): conn = boto3.client("managedblockchain", region_name="us-east-1") votingpolicy = { "ApprovalThresholdPolicy": { "ThresholdPercentage": 50, "ProposalDurationInHours": 24, "ThresholdComparator": "GREATER_THAN", } } # Create network response = conn.create_network( Name="testnetwork1", Framework="HYPERLEDGER_FABRIC", FrameworkVersion="1.2", FrameworkConfiguration=helpers.default_frameworkconfiguration, VotingPolicy=votingpolicy, MemberConfiguration=helpers.default_memberconfiguration, ) network_id = response["NetworkId"] member_id = response["MemberId"] 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"] # Create another proposal response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] # Vote no with member 1 response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="NO", ) # Vote no with member 2 response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id2, Vote="NO", ) # 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("REJECTED") @mock_managedblockchain def test_vote_on_proposal_expiredproposal(): if os.environ.get("TEST_SERVER_MODE", "false").lower() == "true": raise SkipTest("Cant manipulate time in server mode") votingpolicy = { "ApprovalThresholdPolicy": { "ThresholdPercentage": 50, "ProposalDurationInHours": 1, "ThresholdComparator": "GREATER_THAN_OR_EQUAL_TO", } } conn = boto3.client("managedblockchain", region_name="us-east-1") with freeze_time("2015-01-01 12:00:00"): # Create network - need a good network response = conn.create_network( Name="testnetwork1", Framework="HYPERLEDGER_FABRIC", FrameworkVersion="1.2", FrameworkConfiguration=helpers.default_frameworkconfiguration, VotingPolicy=votingpolicy, MemberConfiguration=helpers.default_memberconfiguration, ) network_id = response["NetworkId"] member_id = response["MemberId"] response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] with freeze_time("2015-02-01 12:00:00"): # Vote yes - should set status to expired response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Get proposal details - should be EXPIRED response = conn.get_proposal(NetworkId=network_id, ProposalId=proposal_id) response["Proposal"]["Status"].should.equal("EXPIRED") @mock_managedblockchain def test_vote_on_proposal_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.vote_on_proposal.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", ProposalId="p-ABCDEFGHIJKLMNOP0123456789", VoterMemberId="m-ABCDEFGHIJKLMNOP0123456789", Vote="YES", ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_vote_on_proposal_badproposal(): 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.vote_on_proposal.when.called_with( NetworkId=network_id, ProposalId="p-ABCDEFGHIJKLMNOP0123456789", VoterMemberId="m-ABCDEFGHIJKLMNOP0123456789", Vote="YES", ).should.throw(Exception, "Proposal p-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_vote_on_proposal_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"] member_id = response["MemberId"] response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] response = conn.vote_on_proposal.when.called_with( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId="m-ABCDEFGHIJKLMNOP0123456789", Vote="YES", ).should.throw(Exception, "Member m-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_vote_on_proposal_badvote(): 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"] response = conn.create_proposal( NetworkId=network_id, MemberId=member_id, Actions=helpers.default_policy_actions, ) proposal_id = response["ProposalId"] response = conn.vote_on_proposal.when.called_with( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="FOO", ).should.throw(Exception, "Invalid request body") @mock_managedblockchain def test_vote_on_proposal_alreadyvoted(): 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"] 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"] # Create another 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 with member 1 response = conn.vote_on_proposal( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ) # Vote yes with member 1 again response = conn.vote_on_proposal.when.called_with( NetworkId=network_id, ProposalId=proposal_id, VoterMemberId=member_id, Vote="YES", ).should.throw(Exception, "Invalid request body") @mock_managedblockchain def test_list_proposal_votes_badnetwork(): conn = boto3.client("managedblockchain", region_name="us-east-1") response = conn.list_proposal_votes.when.called_with( NetworkId="n-ABCDEFGHIJKLMNOP0123456789", ProposalId="p-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Network n-ABCDEFGHIJKLMNOP0123456789 not found") @mock_managedblockchain def test_list_proposal_votes_badproposal(): 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"] response = conn.list_proposal_votes.when.called_with( NetworkId=network_id, ProposalId="p-ABCDEFGHIJKLMNOP0123456789", ).should.throw(Exception, "Proposal p-ABCDEFGHIJKLMNOP0123456789 not found")