test/functional/zmq_test.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2015-2016 The Bitcoin Core developers
   3 # Distributed under the MIT software license, see the accompanying
   4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5 """Test the ZMQ API."""
   6 import configparser
   7 import os
   8 import struct
   9
  10 from test_framework.test_framework import BitcoinTestFramework, SkipTest
  11 from test_framework.util import (assert_equal,
  12                                  bytes_to_hex_str,
  13                                  hash256,
  14                                 )
  15
  16 class ZMQTest (BitcoinTestFramework):
  17     def set_test_params(self):
  18         self.num_nodes = 2
  19
  20     def setup_nodes(self):
  21         # Try to import python3-zmq. Skip this test if the import fails.
  22         try:
  23             import zmq
  24         except ImportError:
  25             raise SkipTest("python3-zmq module not available.")
  26
  27         # Check that bitcoin has been built with ZMQ enabled
  28         config = configparser.ConfigParser()
  29         if not self.options.configfile:
  30             self.options.configfile = os.path.dirname(__file__) + "/../config.ini"
  31         config.read_file(open(self.options.configfile))
  32
  33         if not config["components"].getboolean("ENABLE_ZMQ"):
  34             raise SkipTest("bitcoind has not been built with zmq enabled.")
  35
  36         self.zmqContext = zmq.Context()
  37         self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
  38         self.zmqSubSocket.set(zmq.RCVTIMEO, 60000)
  39         self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, b"hashblock")
  40         self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, b"hashtx")
  41         self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, b"rawblock")
  42         self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, b"rawtx")
  43         ip_address = "tcp://127.0.0.1:28332"
  44         self.zmqSubSocket.connect(ip_address)
  45         self.extra_args = [['-zmqpubhashblock=%s' % ip_address, '-zmqpubhashtx=%s' % ip_address,
  46                        '-zmqpubrawblock=%s' % ip_address, '-zmqpubrawtx=%s' % ip_address], []]
  47         self.add_nodes(self.num_nodes, self.extra_args)
  48         self.start_nodes()
  49
  50     def run_test(self):
  51         try:
  52             self._zmq_test()
  53         finally:
  54             # Destroy the zmq context
  55             self.log.debug("Destroying zmq context")
  56             self.zmqContext.destroy(linger=None)
  57
  58     def _zmq_test(self):
  59         genhashes = self.nodes[0].generate(1)
  60         self.sync_all()
  61
  62         self.log.info("Wait for tx")
  63         msg = self.zmqSubSocket.recv_multipart()
  64         topic = msg[0]
  65         assert_equal(topic, b"hashtx")
  66         txhash = msg[1]
  67         msgSequence = struct.unpack('<I', msg[-1])[-1]
  68         assert_equal(msgSequence, 0)  # must be sequence 0 on hashtx
  69
  70         # rawtx
  71         msg = self.zmqSubSocket.recv_multipart()
  72         topic = msg[0]
  73         assert_equal(topic, b"rawtx")
  74         body = msg[1]
  75         msgSequence = struct.unpack('<I', msg[-1])[-1]
  76         assert_equal(msgSequence, 0) # must be sequence 0 on rawtx
  77
  78         # Check that the rawtx hashes to the hashtx
  79         assert_equal(hash256(body), txhash)
  80
  81         self.log.info("Wait for block")
  82         msg = self.zmqSubSocket.recv_multipart()
  83         topic = msg[0]
  84         assert_equal(topic, b"hashblock")
  85         body = msg[1]
  86         msgSequence = struct.unpack('<I', msg[-1])[-1]
  87         assert_equal(msgSequence, 0)  # must be sequence 0 on hashblock
  88         blkhash = bytes_to_hex_str(body)
  89         assert_equal(genhashes[0], blkhash)  # blockhash from generate must be equal to the hash received over zmq
  90
  91         # rawblock
  92         msg = self.zmqSubSocket.recv_multipart()
  93         topic = msg[0]
  94         assert_equal(topic, b"rawblock")
  95         body = msg[1]
  96         msgSequence = struct.unpack('<I', msg[-1])[-1]
  97         assert_equal(msgSequence, 0) #must be sequence 0 on rawblock
  98
  99         # Check the hash of the rawblock's header matches generate
 100         assert_equal(genhashes[0], bytes_to_hex_str(hash256(body[:80])))
 101
 102         self.log.info("Generate 10 blocks (and 10 coinbase txes)")
 103         n = 10
 104         genhashes = self.nodes[1].generate(n)
 105         self.sync_all()
 106
 107         zmqHashes = []
 108         zmqRawHashed = []
 109         blockcount = 0
 110         for x in range(n * 4):
 111             msg = self.zmqSubSocket.recv_multipart()
 112             topic = msg[0]
 113             body = msg[1]
 114             if topic == b"hashblock":
 115                 zmqHashes.append(bytes_to_hex_str(body))
 116                 msgSequence = struct.unpack('<I', msg[-1])[-1]
 117                 assert_equal(msgSequence, blockcount + 1)
 118                 blockcount += 1
 119             if topic == b"rawblock":
 120                 zmqRawHashed.append(bytes_to_hex_str(hash256(body[:80])))
 121                 msgSequence = struct.unpack('<I', msg[-1])[-1]
 122                 assert_equal(msgSequence, blockcount)
 123
 124         for x in range(n):
 125             assert_equal(genhashes[x], zmqHashes[x])  # blockhash from generate must be equal to the hash received over zmq
 126             assert_equal(genhashes[x], zmqRawHashed[x])
 127
 128         self.log.info("Wait for tx from second node")
 129         # test tx from a second node
 130         hashRPC = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 1.0)
 131         self.sync_all()
 132
 133         # now we should receive a zmq msg because the tx was broadcast
 134         msg = self.zmqSubSocket.recv_multipart()
 135         topic = msg[0]
 136         assert_equal(topic, b"hashtx")
 137         body = msg[1]
 138         hashZMQ = bytes_to_hex_str(body)
 139         msgSequence = struct.unpack('<I', msg[-1])[-1]
 140         assert_equal(msgSequence, blockcount + 1)
 141
 142         msg = self.zmqSubSocket.recv_multipart()
 143         topic = msg[0]
 144         assert_equal(topic, b"rawtx")
 145         body = msg[1]
 146         hashedZMQ = bytes_to_hex_str(hash256(body))
 147         msgSequence = struct.unpack('<I', msg[-1])[-1]
 148         assert_equal(msgSequence, blockcount+1)
 149         assert_equal(hashRPC, hashZMQ)  # txid from sendtoaddress must be equal to the hash received over zmq
 150         assert_equal(hashRPC, hashedZMQ)
 151
 152 if __name__ == '__main__':
 153     ZMQTest().main()