Add more python tests

qa/pull-tester/rpc-tests.py

@@ -122,10 +122,11 @@
     # 'p2p-compactblocks.py',
     # 'segwit.py',
     # vv Tests less than 2m vv
-    # 'wallet.py',
-     'wallet-hd.py',
-     'wallet-dump.py',
-     'walletbackup.py',
+    'wallet.py',
+    'wallet-encryption.py',
+    'wallet-hd.py',
+    'wallet-dump.py',
+    'walletbackup.py',
     # 'wallet-accounts.py',
     # 'p2p-segwit.py',
     'listtransactions.py',
@@ -169,6 +170,7 @@
     # 'rpcnamedargs.py',
     'listsinceblock.py',
     'p2p-leaktests.py',
+    'notifications.py',
 
     # Zcoin-specific tests
     'wallet_dumpnonhd.py',

qa/rpc-tests/notifications.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+# Copyright (c) 2014-2016 The Bitcoin Core developers
+# Copyright (c) 2020 The Zcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test the -alertnotify, -blocknotify and -walletnotify options."""
+import os
+import time
+
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import assert_equal, connect_nodes_bi, start_node, start_nodes, stop_node
+
+def wait_until(predicate, timeout):
+    while timeout > 0:
+        if predicate():
+            return
+        time.sleep(0.1)
+        timeout -= 0.1
+
+    raise Exception("Wait Until Timeout")
+
+class NotificationsTest(BitcoinTestFramework):
+    def set_test_params(self):
+        self.num_nodes = 2
+        self.setup_clean_chain = True
+
+    def setup_network(self):
+        self.alert_filename = os.path.join(self.options.tmpdir, "alert.txt")
+        self.block_filename = os.path.join(self.options.tmpdir, "blocks.txt")
+        self.tx_filename = os.path.join(self.options.tmpdir, "transactions.txt")
+
+        # -alertnotify and -blocknotify on node0, walletnotify on node1
+        self.extra_args = [[ "-blockversion=2",
+                            "-alertnotify=echo %%s >> %s" % self.alert_filename,
+                            "-blocknotify=echo %%s >> %s" % self.block_filename],
+                           [ "-blockversion=211",
+                            "-rescan",
+                            "-walletnotify=echo %%s >> %s" % self.tx_filename]]
+
+        self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, extra_args = self.extra_args)
+        connect_nodes_bi(self.nodes, 0, 1)
+
+        self.is_network_split = False
+        self.sync_all()
+
+    def run_test(self):
+        self.log.info("test -blocknotify")
+        block_count = 10
+        blocks = self.nodes[1].generate(block_count)
+
+        # wait at most 10 seconds for expected file size before reading the content
+        wait_until(lambda: os.path.isfile(self.block_filename) and os.stat(self.block_filename).st_size >= (block_count * 65), timeout=10)
+
+        # file content should equal the generated blocks hashes
+        with open(self.block_filename, 'r') as f:
+            assert_equal(sorted(blocks), sorted(f.read().splitlines()))
+
+        self.log.info("test -walletnotify")
+        # wait at most 10 seconds for expected file size before reading the content
+        wait_until(lambda: os.path.isfile(self.tx_filename) and os.stat(self.tx_filename).st_size >= (block_count * 65), timeout=10)
+
+        # file content should equal the generated transaction hashes
+        txids_rpc = list(map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count)))
+        with open(self.tx_filename, 'r') as f:
+            assert_equal(sorted(txids_rpc), sorted(f.read().splitlines()))
+        os.remove(self.tx_filename)
+
+        self.log.info("test -walletnotify after rescan")
+        # restart node to rescan to force wallet notifications
+        stop_node(self.nodes[1], 1)
+        self.nodes[1] = start_node(1, self.options.tmpdir, self.extra_args[1])
+
+        connect_nodes_bi(self.nodes, 0, 1)
+
+        wait_until(lambda: os.path.isfile(self.tx_filename) and os.stat(self.tx_filename).st_size >= (block_count * 65), timeout=10)
+
+        # file content should equal the generated transaction hashes
+        txids_rpc = list(map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count)))
+        with open(self.tx_filename, 'r') as f:
+            assert_equal(sorted(txids_rpc), sorted(f.read().splitlines()))
+
+        # Mine another 41 up-version blocks. -alertnotify should trigger on the 51st.
+        self.log.info("test -alertnotify")
+        self.nodes[1].generate(41)
+        self.sync_all()
+
+        # Give bitcoind 10 seconds to write the alert notification
+        wait_until(lambda: os.path.isfile(self.alert_filename) and os.path.getsize(self.alert_filename), timeout=10)
+
+        with open(self.alert_filename, 'r', encoding='utf8') as f:
+            alert_text = f.read()
+
+        # Mine more up-version blocks, should not get more alerts:
+        self.nodes[1].generate(2)
+        self.sync_all()
+
+        with open(self.alert_filename, 'r', encoding='utf8') as f:
+            alert_text2 = f.read()
+
+        self.log.info("-alertnotify should not continue notifying for more unknown version blocks")
+        assert_equal(alert_text, alert_text2)
+
+if __name__ == '__main__':
+    NotificationsTest().main()

qa/rpc-tests/wallet-encryption.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+# Copyright (c) 2016 The Bitcoin Core developers
+# Copyright (c) 2020 The Zcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test Wallet encryption"""
+
+import re
+import time
+
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import (
+    assert_equal,
+    assert_raises_jsonrpc,
+    bitcoind_processes,
+    start_nodes,
+)
+
+class WalletEncryptionTest(BitcoinTestFramework):
+
+    def __init__(self):
+        super().__init__()
+        self.setup_clean_chain = True # lgtm [py/overwritten-inherited-attribute]
+        self.num_nodes = 1 # lgtm [py/overwritten-inherited-attribute]
+
+    def get_otp(self, address):
+        # assume test on only 1 node
+        try:
+            self.nodes[0].dumpprivkey(address)
+        except Exception as ex:
+            found = re.search(
+                'WARNING! Your one time authorization code is: (.+?)\n',
+                ex.error['message'])
+            if found:
+                return found.group(1)
+
+        raise Exception("Fail to get OTP")
+
+    def run_test(self):
+        passphrase = "WalletPassphrase"
+        passphrase2 = "SecondWalletPassphrase"
+
+        # Make sure the wallet isn't encrypted first
+        address = self.nodes[0].getnewaddress()
+        privkey = self.nodes[0].dumpprivkey(address, self.get_otp(address))
+
+        assert_equal(privkey[:1], "c")
+        assert_equal(len(privkey), 52)
+        assert_raises_jsonrpc(-15, "Error: running with an unencrypted wallet, but walletpassphrase was called", self.nodes[0].walletpassphrase, 'ff', 1)
+        assert_raises_jsonrpc(-15, "Error: running with an unencrypted wallet, but walletpassphrasechange was called.", self.nodes[0].walletpassphrasechange, 'ff', 'ff')
+
+        # Encrypt the wallet
+        assert_raises_jsonrpc(-1, "encryptwallet <passphrase>\nEncrypts the wallet with <passphrase>.", self.nodes[0].encryptwallet, '')
+        self.nodes[0].encryptwallet(passphrase)
+        bitcoind_processes[0].wait()
+        self.nodes = start_nodes(self.num_nodes, self.options.tmpdir)
+
+        # Test that the wallet is encrypted
+        otp = self.get_otp(address)
+        assert_raises_jsonrpc(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].dumpprivkey, address, otp)
+        assert_raises_jsonrpc(-15, "Error: running with an encrypted wallet, but encryptwallet was called.", self.nodes[0].encryptwallet, 'ff')
+        assert_raises_jsonrpc(-1, "walletpassphrase <passphrase> <timeout>\nStores the wallet decryption key in memory for <timeout> seconds.",
+            self.nodes[0].walletpassphrase, '', 1)
+        assert_raises_jsonrpc(-1, "walletpassphrasechange <oldpassphrase> <newpassphrase>\nChanges the wallet passphrase from <oldpassphrase> to <newpassphrase>.",
+            self.nodes[0].walletpassphrasechange, '', 'ff')
+
+        # Check that walletpassphrase works
+        self.nodes[0].walletpassphrase(passphrase, 2)
+        otp = self.get_otp(address)
+        assert_equal(privkey, self.nodes[0].dumpprivkey(address, otp))
+
+        # Check that the timeout is right
+        time.sleep(3)
+        otp = self.get_otp(address)
+        assert_raises_jsonrpc(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].dumpprivkey, address, otp)
+
+        # Test wrong passphrase
+        assert_raises_jsonrpc(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase + "wrong", 10)
+
+        # Test walletlock
+        self.nodes[0].walletpassphrase(passphrase, 84600)
+        otp = self.get_otp(address)
+        assert_equal(privkey, self.nodes[0].dumpprivkey(address, otp))
+        self.nodes[0].walletlock()
+        otp = self.get_otp(address)
+        assert_raises_jsonrpc(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].dumpprivkey, address, otp)
+
+        # Test passphrase changes
+        self.nodes[0].walletpassphrasechange(passphrase, passphrase2)
+        assert_raises_jsonrpc(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase, 10)
+        self.nodes[0].walletpassphrase(passphrase2, 10)
+        otp = self.get_otp(address)
+        assert_equal(privkey, self.nodes[0].dumpprivkey(address, otp))
+
+if __name__ == '__main__':
+    WalletEncryptionTest().main()