Differential D9360 Diff 28065 test/functional/test_framework/messages.py

Changeset View

Standalone View

test/functional/test_framework/messages.py

Show All 20 Lines
from codecs import encode		from codecs import encode
import copy		import copy
import hashlib		import hashlib
from io import BytesIO		from io import BytesIO
import random		import random
import socket		import socket
import struct		import struct
import time		import time
		import unittest

from typing import List		from typing import List

from test_framework.siphash import siphash256		from test_framework.siphash import siphash256
from test_framework.util import hex_str_to_bytes, assert_equal		from test_framework.util import hex_str_to_bytes, assert_equal


MIN_VERSION_SUPPORTED = 60001		MIN_VERSION_SUPPORTED = 60001
Show All 21 Lines
NODE_COMPACT_FILTERS = (1 << 6)		NODE_COMPACT_FILTERS = (1 << 6)
NODE_NETWORK_LIMITED = (1 << 10)		NODE_NETWORK_LIMITED = (1 << 10)
NODE_AVALANCHE = (1 << 24)		NODE_AVALANCHE = (1 << 24)

MSG_TX = 1		MSG_TX = 1
MSG_BLOCK = 2		MSG_BLOCK = 2
MSG_FILTERED_BLOCK = 3		MSG_FILTERED_BLOCK = 3
MSG_CMPCT_BLOCK = 4		MSG_CMPCT_BLOCK = 4
		MSG_AVA_PROOF = 0x1f000001
MSG_TYPE_MASK = 0xffffffff >> 2		MSG_TYPE_MASK = 0xffffffff >> 2

FILTER_TYPE_BASIC = 0		FILTER_TYPE_BASIC = 0

# Serialization/deserialization tools		# Serialization/deserialization tools


def sha256(s):		def sha256(s):
▲ Show 20 Lines • Show All 223 Lines • ▼ Show 20 Lines
class CInv:		class CInv:
__slots__ = ("hash", "type")		__slots__ = ("hash", "type")

typemap = {		typemap = {
0: "Error",		0: "Error",
MSG_TX: "TX",		MSG_TX: "TX",
MSG_BLOCK: "Block",		MSG_BLOCK: "Block",
MSG_FILTERED_BLOCK: "filtered Block",		MSG_FILTERED_BLOCK: "filtered Block",
MSG_CMPCT_BLOCK: "CompactBlock"		MSG_CMPCT_BLOCK: "CompactBlock",
		MSG_AVA_PROOF: "avalanche proof",
}		}

def __init__(self, t=0, h=0):		def __init__(self, t=0, h=0):
self.type = t		self.type = t
self.hash = h		self.hash = h

def deserialize(self, f):		def deserialize(self, f):
self.type = struct.unpack("<i", f.read(4))[0]		self.type = struct.unpack("<i", f.read(4))[0]
▲ Show 20 Lines • Show All 511 Lines • ▼ Show 20 Lines	def serialize(self):
r += ser_vector(self.transactions)		r += ser_vector(self.transactions)
return r		return r

def __repr__(self):		def __repr__(self):
return "BlockTransactions(hash={:064x} transactions={})".format(		return "BlockTransactions(hash={:064x} transactions={})".format(
self.blockhash, repr(self.transactions))		self.blockhash, repr(self.transactions))


		class AvalancheProof:
		__slots__ = ("blob")

		def __init__(self, blob: bytes = b""):
		self.blob: bytes = blob

		def deserialize(self, f):
		self.blob = f.read()

		def serialize(self):
		r = b""
		r += self.blob
		return r

		def __repr__(self):
		return "AvalancheProof({})".format(self.serialize().hex())


class AvalanchePoll():		class AvalanchePoll():
__slots__ = ("round", "invs")		__slots__ = ("round", "invs")

def __init__(self, round=0, invs=None):		def __init__(self, round=0, invs=None):
self.round = round		self.round = round
self.invs = invs if invs is not None else []		self.invs = invs if invs is not None else []

def deserialize(self, f):		def deserialize(self, f):
▲ Show 20 Lines • Show All 954 Lines • ▼ Show 20 Lines	def serialize(self):
r += ser_uint256_vector(self.headers)		r += ser_uint256_vector(self.headers)
return r		return r

def __repr__(self):		def __repr__(self):
return "msg_cfcheckpt(filter_type={:#x}, stop_hash={:x})".format(		return "msg_cfcheckpt(filter_type={:#x}, stop_hash={:x})".format(
self.filter_type, self.stop_hash)		self.filter_type, self.stop_hash)


		class msg_avaproof():
		__slots__ = ("proof",)
		msgtype = b"avaproof"

		def __init__(self):
		self.proof = AvalancheProof()

		def deserialize(self, f):
		self.proof.deserialize(f)

		def serialize(self):
		r = b""
		r += self.proof.serialize()
		return r

		def __repr__(self):
		return "msg_avaproof(proof={})".format(repr(self.proof))


class msg_avapoll():		class msg_avapoll():
__slots__ = ("poll",)		__slots__ = ("poll",)
msgtype = b"avapoll"		msgtype = b"avapoll"

def __init__(self):		def __init__(self):
self.poll = AvalanchePoll()		self.poll = AvalanchePoll()

def deserialize(self, f):		def deserialize(self, f):
▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines	class msg_avahello():

def serialize(self):		def serialize(self):
r = b""		r = b""
r += self.hello.serialize()		r += self.hello.serialize()
return r		return r

def __repr__(self):		def __repr__(self):
return "msg_avahello(response={})".format(repr(self.hello))		return "msg_avahello(response={})".format(repr(self.hello))


		class TestFrameworkMessages(unittest.TestCase):
		def test_serialization_round_trip(self):
		"""Verify that messages and serialized objects are unchanged after
		a round-trip of deserialization-serialization.
		"""
		avaproof = AvalancheProof()
		proof_hex = (
		"2a00000000000000fff053650000000021030b4c866585dd868a9d62348a9cd00"
		"8d6a312937048fff31670e7e920cfc7a74401b7fc19792583e9cb39843fc5e22a"
		"4e3648ab1cb18a70290b341ee8d4f550ae2400000000102700000000000078881"
		"4004104d0de0aaeaefad02b8bdc8a01a1b8b11c696bd3d66a2c5f10780d95b7df"
		"42645cd85228a6fb29940e858e7e55842ae2bd115d1ed7cc0e82d934e929c9764"
		"8cb0ac3052d58da74de7404e84ebe2940ed2b0fe85578d8230788d8387aeaa618"
		"274b0f2edc73679fd398f60e6315258c9ec348df7fcc09340ae1af37d009719b0"
		"665"
		)
		avaproof.deserialize(BytesIO(bytes.fromhex(proof_hex)))
		self.assertEqual(avaproof.serialize().hex(), proof_hex)

		msg_proof = msg_avaproof()
		msg_proof.proof = avaproof
		self.assertEqual(msg_proof.serialize().hex(), proof_hex)