Differential D12370 Diff 36407 test/functional/abc_p2p_avalanche_policy_early_blocks.py

Changeset View

Standalone View

test/functional/abc_p2p_avalanche_policy_early_blocks.py

This file was added.

Property	Old Value	New Value
File Mode	null	100755

				#!/usr/bin/env python3
				# Copyright (c) 2022 The Bitcoin developers
				# Distributed under the MIT software license, see the accompanying
				# file COPYING or http://www.opensource.org/licenses/mit-license.php.
				"""Test avalanche post-consensus policy: difficulty penalty for early blocks"""
				import random
				import time

				from test_framework.avatools import get_ava_p2p_interface
				from test_framework.blocktools import create_block, create_coinbase
				from test_framework.messages import (
				AvalancheVote,
				AvalancheVoteError,
				msg_block,
				uint256_from_compact,
				)
				from test_framework.test_framework import BitcoinTestFramework
				from test_framework.util import assert_equal

				QUORUM_NODE_COUNT = 8
				NUM_BLOCKS_TO_CHECK = 20


				class AvalanchePolicyEarlyBlocksTest(BitcoinTestFramework):
				def set_test_params(self):
				self.setup_clean_chain = True
				self.num_nodes = 1
				self.penaltyfactor = 10
				self.penaltywindow = 100
				self.extra_args = [
				[
				'-avalanche=1',
				'-avapolicyearlyblockpenalty=1',
				f'-avapolicyearlyblockpenaltyfactor={self.penaltyfactor}',
				f'-avapolicyearlyblockpenaltywindow={self.penaltywindow}',
				'-avaproofstakeutxodustthreshold=1000000',
				'-avaproofstakeutxoconfirmations=1',
				'-avacooldown=0',
				'-avaminquorumstake=0',
				'-avaminavaproofsnodecount=0',
				'-whitelist=noban@127.0.0.1',
				],
				]

				def run_test(self):
				node = self.nodes[0]

				mocktime = int(time.time())
				node.setmocktime(mocktime)

				# Build a fake quorum of nodes.
				def get_quorum():
				nonlocal mocktime

				quorum = []
				for _ in range(0, QUORUM_NODE_COUNT):
				# Set mocktime far beyond the penalty window so that the block
				# generated as part of proof creation is always accepted.
				mocktime += self.penaltywindow * 10
				node.setmocktime(mocktime)
				quorum.append(get_ava_p2p_interface(self, node))

				return quorum

				# Pick a node from the quorum for polling.
				quorum = get_quorum()
				poll_node = quorum[0]

				assert node.getavalancheinfo()['ready_to_poll'] is True

				def assert_response(expected):
				r = poll_node.wait_for_avaresponse().response

				votes = r.votes
				assert_equal(len(votes), len(expected))
				for i in range(0, len(votes)):
				assert_equal(repr(votes[i]), repr(expected[i]))

				# Generate some blocks outside of the penalty window and check them
				# against the policy.
				bestblockhash = node.getbestblockhash()
				tip = int(bestblockhash, 16)
				height = node.getblockcount() + 1
				blocktime = node.getblock(bestblockhash)['time'] + 1
				for _ in range(0, NUM_BLOCKS_TO_CHECK):
				block = create_block(
				tip,
				create_coinbase(height),
				blocktime,
				version=4)
				block.solve()
				tip = block.sha256
				height += 1
				blocktime += 1

				# Make block received time just outside of the penalty window.
				mocktime += self.penaltywindow
				node.setmocktime(mocktime)
				poll_node.send_and_ping(msg_block(block))

				# Blocks outside of the penalty window are accepted.
				poll_node.send_poll([tip])
				assert_response([AvalancheVote(AvalancheVoteError.ACCEPTED, tip)])

				# Generate some test cases and always include the penalty window
				# boundaries.
				time_deltas = sorted(
				random.sample(
				range(
				0,
				self.penaltywindow),
				NUM_BLOCKS_TO_CHECK))
				time_deltas[0] = 0
				time_deltas[-1] = self.penaltywindow - 1

				# Generate some blocks that may be rejected inside of the penalty
				# window.
				for blockindex in range(0, NUM_BLOCKS_TO_CHECK):
				block = create_block(
				tip,
				create_coinbase(height),
				blocktime,
				version=4)
				block.solve()
				target = uint256_from_compact(block.nBits)
				expect_parked = block.sha256
				assert expect_parked <= target

				if expect_parked <= target / self.penaltyfactor:
				# Block difficulty exceeds highest possible difficulty penalty,
				# so just ignore it
				continue

				# Set block received time such that the block is rejected.
				node.setmocktime(mocktime + time_deltas[blockindex])
				poll_node.send_and_ping(msg_block(block))

				poll_node.send_poll([expect_parked])
				assert_response(
				[AvalancheVote(AvalancheVoteError.PARKED, expect_parked)])


				if __name__ == '__main__':
				AvalanchePolicyEarlyBlocksTest().main()