QA: Add ZeroMQ RPC test
[bitcoinplatinum.git] / qa / rpc-tests / zmq_test.py
blobfffaf677d68ad6a00e78651c3c5a65112e8bb59d
1 #!/usr/bin/env python2
2 # Copyright (c) 2015 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.
7 # Test ZMQ interface
10 from test_framework.test_framework import BitcoinTestFramework
11 from test_framework.util import *
12 import zmq
13 import binascii
14 from test_framework.mininode import hash256
16 try:
17 import http.client as httplib
18 except ImportError:
19 import httplib
20 try:
21 import urllib.parse as urlparse
22 except ImportError:
23 import urlparse
25 class ZMQTest (BitcoinTestFramework):
27 port = 28332
29 def setup_nodes(self):
30 self.zmqContext = zmq.Context()
31 self.zmqSubSocket = self.zmqContext.socket(zmq.SUB)
32 self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashblock")
33 self.zmqSubSocket.setsockopt(zmq.SUBSCRIBE, "hashtx")
34 self.zmqSubSocket.connect("tcp://127.0.0.1:%i" % self.port)
35 # Note: proxies are not used to connect to local nodes
36 # this is because the proxy to use is based on CService.GetNetwork(), which return NET_UNROUTABLE for localhost
37 return start_nodes(4, self.options.tmpdir, extra_args=[
38 ['-zmqpubhashtx=tcp://127.0.0.1:'+str(self.port), '-zmqpubhashblock=tcp://127.0.0.1:'+str(self.port)],
39 [],
40 [],
44 def run_test(self):
45 self.sync_all()
47 genhashes = self.nodes[0].generate(1);
48 self.sync_all()
50 print "listen..."
51 msg = self.zmqSubSocket.recv_multipart()
52 topic = str(msg[0])
53 body = msg[1]
55 msg = self.zmqSubSocket.recv_multipart()
56 topic = str(msg[0])
57 body = msg[1]
58 blkhash = binascii.hexlify(body)
60 assert_equal(genhashes[0], blkhash) #blockhash from generate must be equal to the hash received over zmq
62 n = 10
63 genhashes = self.nodes[1].generate(n);
64 self.sync_all()
66 zmqHashes = []
67 for x in range(0,n*2):
68 msg = self.zmqSubSocket.recv_multipart()
69 topic = str(msg[0])
70 body = msg[1]
71 if topic == "hashblock":
72 zmqHashes.append(binascii.hexlify(body))
74 for x in range(0,n):
75 assert_equal(genhashes[x], zmqHashes[x]) #blockhash from generate must be equal to the hash received over zmq
77 #test tx from a second node
78 hashRPC = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 1.0)
79 self.sync_all()
81 #now we should receive a zmq msg because the tx was broadcastet
82 msg = self.zmqSubSocket.recv_multipart()
83 topic = str(msg[0])
84 body = msg[1]
85 hashZMQ = ""
86 if topic == "hashtx":
87 hashZMQ = binascii.hexlify(body)
89 assert_equal(hashRPC, hashZMQ) #blockhash from generate must be equal to the hash received over zmq
92 if __name__ == '__main__':
93 ZMQTest ().main ()