Differential D9365 Diff 28194 test/functional/abc_p2p_avalanche.py

Changeset View

Standalone View

test/functional/abc_p2p_avalanche.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2020-2021 The Bitcoin developers		# Copyright (c) 2020-2021 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying		# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.		# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the resolution of forks via avalanche."""		"""Test the resolution of forks via avalanche."""
import random		import random

from test_framework.avatools import get_stakes		from test_framework.avatools import get_stakes
from test_framework.key import (		from test_framework.key import (
ECKey,		ECKey,
ECPubKey,		ECPubKey,
		bytes_to_wif,
)		)
from test_framework.mininode import P2PInterface, mininode_lock		from test_framework.mininode import P2PInterface, mininode_lock
from test_framework.messages import (		from test_framework.messages import (
AvalancheResponse,		AvalancheResponse,
AvalancheVote,		AvalancheVote,
CInv,		CInv,
msg_avapoll,		msg_avapoll,
msg_tcpavaresponse,		msg_tcpavaresponse,
▲ Show 20 Lines • Show All 329 Lines • ▼ Show 20 Lines	def run_test(self):
"-avaproof={}".format(proof),		"-avaproof={}".format(proof),
"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN",		"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN",
])		])

assert_equal(		assert_equal(
int(node.getnetworkinfo()['localservices'], 16) & NODE_AVALANCHE,		int(node.getnetworkinfo()['localservices'], 16) & NODE_AVALANCHE,
NODE_AVALANCHE)		NODE_AVALANCHE)

self.log.info("Test the avahello signature")		def check_avahello(args):
		# Restart the node with the given args
		self.restart_node(0, self.extra_args[0] + args)

quorum = get_quorum()		quorum = get_quorum()
poll_node = quorum[0]		poll_node = quorum[0]

avahello = poll_node.wait_for_avahello().hello		avahello = poll_node.wait_for_avahello().hello

avakey.set(bytes.fromhex(node.getavalanchekey()))		avakey.set(bytes.fromhex(node.getavalanchekey()))
assert avakey.verify_schnorr(		assert avakey.verify_schnorr(
avahello.sig, avahello.get_sighash(poll_node))		avahello.sig, avahello.get_sighash(poll_node))

		self.log.info(
		"Test the avahello signature with a generated delegation")
		check_avahello([
		"-avaproof={}".format(proof),
		"-avamasterkey=cND2ZvtabDbJ1gucx9GWH6XT9kgTAqfb6cotPt5Q5CyxVDhid2EN"
		])

		session_key = ECKey()
		session_key.generate()
		delegation = node.delegateavalancheproof(
		proof,
		bytes_to_wif(privkey.get_bytes()),
		session_key.get_pubkey().get_bytes().hex(),
		)

		self.log.info("Test the avahello signature with a supplied delegation")
		check_avahello([
		"-avaproof={}".format(proof),
		"-avadelegation={}".format(delegation),
		"-avamasterkey={}".format(bytes_to_wif(session_key.get_bytes())),
		])


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