Differential D4447 Diff 14198 test/functional/interface_zmq.py

Changeset View

Standalone View

test/functional/interface_zmq.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2015-2019 The Bitcoin Core developers		# Copyright (c) 2015-2019 The Bitcoin Core 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 ZMQ notification interface."""		"""Test the ZMQ notification interface."""
import configparser
import struct		import struct
from io import BytesIO		from io import BytesIO

from test_framework.test_framework import BitcoinTestFramework, SkipTest		from test_framework.test_framework import (
		BitcoinTestFramework,
		skip_if_no_bitcoind_zmq,
		skip_if_no_py3_zmq,
		)
from test_framework.messages import CTransaction		from test_framework.messages import CTransaction
from test_framework.util import (		from test_framework.util import (
assert_equal,		assert_equal,
hash256,		hash256,
)		)


class ZMQSubscriber:		class ZMQSubscriber:
Show All 15 Lines	def receive(self):
return body		return body


class ZMQTest (BitcoinTestFramework):		class ZMQTest (BitcoinTestFramework):
def set_test_params(self):		def set_test_params(self):
self.num_nodes = 2		self.num_nodes = 2

def setup_nodes(self):		def setup_nodes(self):
# Try to import python3-zmq. Skip this test if the import fails.		skip_if_no_py3_zmq()
try:		skip_if_no_bitcoind_zmq(self)
import zmq		import zmq
except ImportError:
raise SkipTest("python3-zmq module not available.")

# Check that bitcoin has been built with ZMQ enabled.
config = configparser.ConfigParser()
config.read_file(open(self.options.configfile, encoding='utf-8'))

if not config["components"].getboolean("ENABLE_ZMQ"):
raise SkipTest("bitcoind has not been built with zmq enabled.")

# Initialize ZMQ context and socket.		# Initialize ZMQ context and socket.
# All messages are received in the same socket which means that this		# All messages are received in the same socket which means that this
# test fails if the publishing order changes.		# test fails if the publishing order changes.
# Note that the publishing order is not defined in the documentation and		# Note that the publishing order is not defined in the documentation and
# is subject to change.		# is subject to change.
address = "tcp://127.0.0.1:28332"		address = "tcp://127.0.0.1:28332"
self.zmq_context = zmq.Context()		self.zmq_context = zmq.Context()
▲ Show 20 Lines • Show All 67 Lines • Show Last 20 Lines