Differential D11766 Diff 34421 test/functional/abc_p2p_avalanche_voting.py

Changeset View

Standalone View

test/functional/abc_p2p_avalanche_voting.py

Show First 20 Lines • Show All 44 Lines • ▼ Show 20 Lines	def run_test(self):
poll_node = quorum[0]		poll_node = quorum[0]

# Generate many block and poll for them.		# Generate many block and poll for them.
addrkey0 = node.get_deterministic_priv_key()		addrkey0 = node.get_deterministic_priv_key()
blockhashes = node.generatetoaddress(100, addrkey0.address)		blockhashes = node.generatetoaddress(100, addrkey0.address)
# Use the first coinbase to create a stake		# Use the first coinbase to create a stake
stakes = create_coinbase_stakes(node, [blockhashes[0]], addrkey0.key)		stakes = create_coinbase_stakes(node, [blockhashes[0]], addrkey0.key)

		# duplicate the deterministic sig test from src/test/key_tests.cpp
		privkey = ECKey()
		privkey.set(bytes.fromhex(
		"12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747"), True)

		proof_sequence = 11
		proof_expiration = 12
		proof = node.buildavalancheproof(
		proof_sequence, proof_expiration, bytes_to_wif(
		privkey.get_bytes()),
		stakes)

		# Activate the quorum.
		for n in quorum:
		success = node.addavalanchenode(
		n.nodeid, privkey.get_pubkey().get_bytes().hex(), proof)
		assert success is True

fork_node = self.nodes[1]		fork_node = self.nodes[1]
# Make sure the fork node has synced the blocks		# Make sure the fork node has synced the blocks
self.sync_blocks([node, fork_node])		self.sync_blocks([node, fork_node])

# Get the key so we can verify signatures.		# Get the key so we can verify signatures.
avakey = ECPubKey()		avakey = ECPubKey()
avakey.set(bytes.fromhex(node.getavalanchekey()))		avakey.set(bytes.fromhex(node.getavalanchekey()))

▲ Show 20 Lines • Show All 60 Lines • ▼ Show 20 Lines	def run_test(self):
random.randrange(1 << 255, (1 << 256) - 1),		random.randrange(1 << 255, (1 << 256) - 1),
]		]
poll_node.send_poll(various_block_hashes)		poll_node.send_poll(various_block_hashes)
assert_response([AvalancheVote(AvalancheVoteError.ACCEPTED, h) for h in various_block_hashes[:3]] +		assert_response([AvalancheVote(AvalancheVoteError.ACCEPTED, h) for h in various_block_hashes[:3]] +
[AvalancheVote(AvalancheVoteError.FORK, h) for h in various_block_hashes[3:6]] +		[AvalancheVote(AvalancheVoteError.FORK, h) for h in various_block_hashes[3:6]] +
[AvalancheVote(AvalancheVoteError.UNKNOWN, h) for h in various_block_hashes[-3:]])		[AvalancheVote(AvalancheVoteError.UNKNOWN, h) for h in various_block_hashes[-3:]])

self.log.info("Trigger polling from the node...")		self.log.info("Trigger polling from the node...")
# duplicate the deterministic sig test from src/test/key_tests.cpp
privkey = ECKey()
privkey.set(bytes.fromhex(
"12b004fff7f4b69ef8650e767f18f11ede158148b425660723b9f9a66e61f747"), True)

proof_sequence = 11
proof_expiration = 12
proof = node.buildavalancheproof(
proof_sequence, proof_expiration, bytes_to_wif(
privkey.get_bytes()),
stakes)

# Activate the quorum.
for n in quorum:
success = node.addavalanchenode(
n.nodeid, privkey.get_pubkey().get_bytes().hex(), proof)
assert success is True

def can_find_block_in_poll(hash, resp=AvalancheVoteError.ACCEPTED):		def can_find_block_in_poll(hash, resp=AvalancheVoteError.ACCEPTED):
found_hash = False		found_hash = False
for n in quorum:		for n in quorum:
poll = n.get_avapoll_if_available()		poll = n.get_avapoll_if_available()

# That node has not received a poll		# That node has not received a poll
if poll is None:		if poll is None:
▲ Show 20 Lines • Show All 82 Lines • Show Last 20 Lines