[RPC] Add RPC long poll notifications #7949
| @@ -0,0 +1,64 @@ | ||
| +#!/usr/bin/env python3 | ||
| +# Copyright (c) 2016 The Bitcoin Core developers | ||
| +# Distributed under the MIT software license, see the accompanying | ||
| +# file COPYING or http://www.opensource.org/licenses/mit-license.php. | ||
| + | ||
| +# Exercise the rpc signals API | ||
| + | ||
| +from test_framework.test_framework import BitcoinTestFramework | ||
| +from test_framework.util import * | ||
| +import threading | ||
| +import uuid | ||
| + | ||
| +class pollThread (threading.Thread): | ||
| + def __init__(self, threadID, node, clientUUID): | ||
| + threading.Thread.__init__(self) | ||
| + self.threadID = threadID | ||
| + self.node = node | ||
| + self.uuid = clientUUID | ||
| + def run(self): | ||
| + self.node.pollnotifications(self.uuid, 5) | ||
| + | ||
| +class RpcSignalsTest(BitcoinTestFramework): | ||
| + | ||
| + def pollthread(self, node): | ||
| + print("thread") | ||
| + | ||
| + def setup_nodes(self): | ||
| + return start_nodes(4, self.options.tmpdir) | ||
| + | ||
| + def run_test(self): | ||
| + clientUUID = str(uuid.uuid1()) | ||
| + try: | ||
| + self.nodes[0].getregisterednotifications(clientUUID) | ||
| + raise AssertionError("ClientUUID should not exists") | ||
| + except JSONRPCException as e: | ||
| + assert(e.error['code']==-8) | ||
| + | ||
| + self.nodes[0].setregisterednotifications(clientUUID, ["hashblock", "hashtx"]) | ||
| + data = self.nodes[0].getregisterednotifications(clientUUID) | ||
| + assert("hashblock" in data) | ||
| + assert("hashtx" in data) | ||
| + | ||
| + self.nodes[0].generate(1) | ||
| + data = self.nodes[0].pollnotifications(clientUUID) | ||
| + assert(len(data) == 2) | ||
| + | ||
| + myThread = pollThread(1, self.nodes[0], clientUUID) | ||
| + myThread.start() | ||
| + time.sleep(3) | ||
| + assert(myThread.isAlive() == True) | ||
| + time.sleep(5) | ||
| + assert(myThread.isAlive() == False) | ||
| + myThread.join() | ||
| + | ||
| + self.nodes[0].setregisterednotifications(clientUUID, []) | ||
| + data = self.nodes[0].getregisterednotifications(clientUUID) | ||
| + assert(len(data) == 0) | ||
| + self.nodes[0].generate(1) | ||
| + data = self.nodes[0].pollnotifications(clientUUID, 5) | ||
| + assert(len(data) == 0) | ||
| + | ||
| +if __name__ == '__main__': | ||
| + RpcSignalsTest().main() | ||
| + |
Oops, something went wrong.