Differential D13878 Diff 40252 test/functional/chronik_ws.py

Changeset View

Standalone View

test/functional/chronik_ws.py

	Show All 11 Lines


	class ChronikWsTest(BitcoinTestFramework):			class ChronikWsTest(BitcoinTestFramework):
	def set_test_params(self):			def set_test_params(self):
	self.setup_clean_chain = True			self.setup_clean_chain = True
	self.num_nodes = 1			self.num_nodes = 1
	self.extra_args = [			self.extra_args = [
	[			[
	'-avaproofstakeutxodustthreshold=1000000',			"-avaproofstakeutxodustthreshold=1000000",
	'-avaproofstakeutxoconfirmations=1',			"-avaproofstakeutxoconfirmations=1",
	'-avacooldown=0',			"-avacooldown=0",
	'-avaminquorumstake=0',			"-avaminquorumstake=0",
	'-avaminavaproofsnodecount=0',			"-avaminavaproofsnodecount=0",
	'-chronik',			"-chronik",
	'-whitelist=noban@127.0.0.1',			"-whitelist=noban@127.0.0.1",
	],			],
	]			]
	self.supports_cli = False			self.supports_cli = False

	def skip_test_if_missing_module(self):			def skip_test_if_missing_module(self):
	self.skip_if_no_chronik()			self.skip_if_no_chronik()

	def run_test(self):			def run_test(self):
	from test_framework.chronik.client import ChronikClient, pb			from test_framework.chronik.client import ChronikClient, pb

	node = self.nodes[0]			node = self.nodes[0]
	chronik = ChronikClient('127.0.0.1', node.chronik_port)			chronik = ChronikClient("127.0.0.1", node.chronik_port)

	# Build a fake quorum of nodes.			# Build a fake quorum of nodes.
	def get_quorum():			def get_quorum():
	return [get_ava_p2p_interface(self, node)			return [
	for _ in range(0, QUORUM_NODE_COUNT)]			get_ava_p2p_interface(self, node) for _ in range(0, QUORUM_NODE_COUNT)
				]

	def has_finalized_tip(tip_expected):			def has_finalized_tip(tip_expected):
	hash_tip_final = int(tip_expected, 16)			hash_tip_final = int(tip_expected, 16)
	can_find_inv_in_poll(quorum, hash_tip_final)			can_find_inv_in_poll(quorum, hash_tip_final)
	return node.isfinalblock(tip_expected)			return node.isfinalblock(tip_expected)

	# Connect, but don't subscribe yet			# Connect, but don't subscribe yet
	ws = chronik.ws(timeout=30)			ws = chronik.ws(timeout=30)

	# Pick one node from the quorum for polling.			# Pick one node from the quorum for polling.
	# ws will not receive msgs because it's not subscribed to blocks yet.			# ws will not receive msgs because it's not subscribed to blocks yet.
	quorum = get_quorum()			quorum = get_quorum()

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

	tip = node.getbestblockhash()			tip = node.getbestblockhash()
	self.wait_until(lambda: has_finalized_tip(tip))			self.wait_until(lambda: has_finalized_tip(tip))

	# Now subscribe to blocks, we'll get block updates from now on			# Now subscribe to blocks, we'll get block updates from now on
	ws.sub_to_blocks()			ws.sub_to_blocks()

	# Mine block			# Mine block
	tip = self.generate(node, 1)[-1]			tip = self.generate(node, 1)[-1]
	height = node.getblockcount()			height = node.getblockcount()

	# We get a CONNECTED msg			# We get a CONNECTED msg
	assert_equal(ws.recv(), pb.WsMsg(block=pb.MsgBlock(			assert_equal(
				ws.recv(),
				pb.WsMsg(
				block=pb.MsgBlock(
	msg_type=pb.BLK_CONNECTED,			msg_type=pb.BLK_CONNECTED,
	block_hash=bytes.fromhex(tip)[::-1],			block_hash=bytes.fromhex(tip)[::-1],
	block_height=height,			block_height=height,
	)))			)
				),
				)

	# After we wait, we get a FINALIZED msg			# After we wait, we get a FINALIZED msg
	self.wait_until(lambda: has_finalized_tip(tip))			self.wait_until(lambda: has_finalized_tip(tip))
	assert_equal(ws.recv(), pb.WsMsg(block=pb.MsgBlock(			assert_equal(
				ws.recv(),
				pb.WsMsg(
				block=pb.MsgBlock(
	msg_type=pb.BLK_FINALIZED,			msg_type=pb.BLK_FINALIZED,
	block_hash=bytes.fromhex(tip)[::-1],			block_hash=bytes.fromhex(tip)[::-1],
	block_height=height,			block_height=height,
	)))			)
				),
				)

	# When we invalidate, we get a DISCONNECTED msg			# When we invalidate, we get a DISCONNECTED msg
	node.invalidateblock(tip)			node.invalidateblock(tip)
	assert_equal(ws.recv(), pb.WsMsg(block=pb.MsgBlock(			assert_equal(
				ws.recv(),
				pb.WsMsg(
				block=pb.MsgBlock(
	msg_type=pb.BLK_DISCONNECTED,			msg_type=pb.BLK_DISCONNECTED,
	block_hash=bytes.fromhex(tip)[::-1],			block_hash=bytes.fromhex(tip)[::-1],
	block_height=height,			block_height=height,
	)))			)
				),
				)


	if __name__ == '__main__':			if __name__ == "__main__":
	ChronikWsTest().main()			ChronikWsTest().main()