aboutsummaryrefslogtreecommitdiff
path: root/test/functional/wallet_importmulti.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/functional/wallet_importmulti.py')
-rwxr-xr-xtest/functional/wallet_importmulti.py451
1 files changed, 451 insertions, 0 deletions
diff --git a/test/functional/wallet_importmulti.py b/test/functional/wallet_importmulti.py
new file mode 100755
index 0000000000..56ebc2622a
--- /dev/null
+++ b/test/functional/wallet_importmulti.py
@@ -0,0 +1,451 @@
+#!/usr/bin/env python3
+# Copyright (c) 2014-2017 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Test the importmulti RPC."""
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import *
+
+class ImportMultiTest (BitcoinTestFramework):
+ def set_test_params(self):
+ self.num_nodes = 2
+ self.extra_args = [["-addresstype=legacy"], ["-addresstype=legacy"]]
+ self.setup_clean_chain = True
+
+ def setup_network(self):
+ self.setup_nodes()
+
+ def run_test (self):
+ self.log.info("Mining blocks...")
+ self.nodes[0].generate(1)
+ self.nodes[1].generate(1)
+ timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
+
+ node0_address1 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+
+ #Check only one address
+ assert_equal(node0_address1['ismine'], True)
+
+ #Node 1 sync test
+ assert_equal(self.nodes[1].getblockcount(),1)
+
+ #Address Test - before import
+ address_info = self.nodes[1].getaddressinfo(node0_address1['address'])
+ assert_equal(address_info['iswatchonly'], False)
+ assert_equal(address_info['ismine'], False)
+
+
+ # RPC importmulti -----------------------------------------------
+
+ # Bitcoin Address
+ self.log.info("Should import an address")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal(address_assert['timestamp'], timestamp)
+ watchonly_address = address['address']
+ watchonly_timestamp = timestamp
+
+ self.log.info("Should not import an invalid address")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": "not valid address",
+ },
+ "timestamp": "now",
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -5)
+ assert_equal(result[0]['error']['message'], 'Invalid address')
+
+ # ScriptPubKey + internal
+ self.log.info("Should import a scriptPubKey with internal flag")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "internal": True
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal(address_assert['timestamp'], timestamp)
+
+ # ScriptPubKey + !internal
+ self.log.info("Should not import a scriptPubKey without internal flag")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -8)
+ assert_equal(result[0]['error']['message'], 'Internal must be set for hex scriptPubKey')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+
+ # Address + Public key + !Internal
+ self.log.info("Should import an address with public key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ "pubkeys": [ address['pubkey'] ]
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal(address_assert['timestamp'], timestamp)
+
+
+ # ScriptPubKey + Public key + internal
+ self.log.info("Should import a scriptPubKey with internal and with public key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ request = [{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "pubkeys": [ address['pubkey'] ],
+ "internal": True
+ }]
+ result = self.nodes[1].importmulti(request)
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal(address_assert['timestamp'], timestamp)
+
+ # ScriptPubKey + Public key + !internal
+ self.log.info("Should not import a scriptPubKey without internal and with public key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ request = [{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "pubkeys": [ address['pubkey'] ]
+ }]
+ result = self.nodes[1].importmulti(request)
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -8)
+ assert_equal(result[0]['error']['message'], 'Internal must be set for hex scriptPubKey')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+ # Address + Private key + !watchonly
+ self.log.info("Should import an address with private key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address['address']) ]
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], True)
+ assert_equal(address_assert['timestamp'], timestamp)
+
+ self.log.info("Should not import an address with private key if is already imported")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address['address']) ]
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -4)
+ assert_equal(result[0]['error']['message'], 'The wallet already contains the private key for this address or script')
+
+ # Address + Private key + watchonly
+ self.log.info("Should not import an address with private key and with watchonly")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address['address']) ],
+ "watchonly": True
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -8)
+ assert_equal(result[0]['error']['message'], 'Incompatibility found between watchonly and keys')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+ # ScriptPubKey + Private key + internal
+ self.log.info("Should import a scriptPubKey with internal and with private key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address['address']) ],
+ "internal": True
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], True)
+ assert_equal(address_assert['timestamp'], timestamp)
+
+ # ScriptPubKey + Private key + !internal
+ self.log.info("Should not import a scriptPubKey without internal and with private key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address['address']) ]
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -8)
+ assert_equal(result[0]['error']['message'], 'Internal must be set for hex scriptPubKey')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+
+ # P2SH address
+ sig_address_1 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_3 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['pubkey'], sig_address_2['pubkey'], sig_address_3['pubkey']])
+ self.nodes[1].generate(100)
+ transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
+ self.nodes[1].generate(1)
+ timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
+
+ self.log.info("Should import a p2sh")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": multi_sig_script['address']
+ },
+ "timestamp": "now",
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(multi_sig_script['address'])
+ assert_equal(address_assert['isscript'], True)
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['timestamp'], timestamp)
+ p2shunspent = self.nodes[1].listunspent(0,999999, [multi_sig_script['address']])[0]
+ assert_equal(p2shunspent['spendable'], False)
+ assert_equal(p2shunspent['solvable'], False)
+
+
+ # P2SH + Redeem script
+ sig_address_1 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_3 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['pubkey'], sig_address_2['pubkey'], sig_address_3['pubkey']])
+ self.nodes[1].generate(100)
+ transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
+ self.nodes[1].generate(1)
+ timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
+
+ self.log.info("Should import a p2sh with respective redeem script")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": multi_sig_script['address']
+ },
+ "timestamp": "now",
+ "redeemscript": multi_sig_script['redeemScript']
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(multi_sig_script['address'])
+ assert_equal(address_assert['timestamp'], timestamp)
+
+ p2shunspent = self.nodes[1].listunspent(0,999999, [multi_sig_script['address']])[0]
+ assert_equal(p2shunspent['spendable'], False)
+ assert_equal(p2shunspent['solvable'], True)
+
+
+ # P2SH + Redeem script + Private Keys + !Watchonly
+ sig_address_1 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_3 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['pubkey'], sig_address_2['pubkey'], sig_address_3['pubkey']])
+ self.nodes[1].generate(100)
+ transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
+ self.nodes[1].generate(1)
+ timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
+
+ self.log.info("Should import a p2sh with respective redeem script and private keys")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": multi_sig_script['address']
+ },
+ "timestamp": "now",
+ "redeemscript": multi_sig_script['redeemScript'],
+ "keys": [ self.nodes[0].dumpprivkey(sig_address_1['address']), self.nodes[0].dumpprivkey(sig_address_2['address'])]
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(multi_sig_script['address'])
+ assert_equal(address_assert['timestamp'], timestamp)
+
+ p2shunspent = self.nodes[1].listunspent(0,999999, [multi_sig_script['address']])[0]
+ assert_equal(p2shunspent['spendable'], False)
+ assert_equal(p2shunspent['solvable'], True)
+
+ # P2SH + Redeem script + Private Keys + Watchonly
+ sig_address_1 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ sig_address_3 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ multi_sig_script = self.nodes[0].createmultisig(2, [sig_address_1['pubkey'], sig_address_2['pubkey'], sig_address_3['pubkey']])
+ self.nodes[1].generate(100)
+ transactionid = self.nodes[1].sendtoaddress(multi_sig_script['address'], 10.00)
+ self.nodes[1].generate(1)
+ timestamp = self.nodes[1].getblock(self.nodes[1].getbestblockhash())['mediantime']
+
+ self.log.info("Should import a p2sh with respective redeem script and private keys")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": multi_sig_script['address']
+ },
+ "timestamp": "now",
+ "redeemscript": multi_sig_script['redeemScript'],
+ "keys": [ self.nodes[0].dumpprivkey(sig_address_1['address']), self.nodes[0].dumpprivkey(sig_address_2['address'])],
+ "watchonly": True
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -8)
+ assert_equal(result[0]['error']['message'], 'Incompatibility found between watchonly and keys')
+
+
+ # Address + Public key + !Internal + Wrong pubkey
+ self.log.info("Should not import an address with a wrong public key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ address2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ "pubkeys": [ address2['pubkey'] ]
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -5)
+ assert_equal(result[0]['error']['message'], 'Consistency check failed')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+
+ # ScriptPubKey + Public key + internal + Wrong pubkey
+ self.log.info("Should not import a scriptPubKey with internal and with a wrong public key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ address2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ request = [{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "pubkeys": [ address2['pubkey'] ],
+ "internal": True
+ }]
+ result = self.nodes[1].importmulti(request)
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -5)
+ assert_equal(result[0]['error']['message'], 'Consistency check failed')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+
+ # Address + Private key + !watchonly + Wrong private key
+ self.log.info("Should not import an address with a wrong private key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ address2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": address['address']
+ },
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address2['address']) ]
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -5)
+ assert_equal(result[0]['error']['message'], 'Consistency check failed')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+
+ # ScriptPubKey + Private key + internal + Wrong private key
+ self.log.info("Should not import a scriptPubKey with internal and with a wrong private key")
+ address = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ address2 = self.nodes[0].getaddressinfo(self.nodes[0].getnewaddress())
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "now",
+ "keys": [ self.nodes[0].dumpprivkey(address2['address']) ],
+ "internal": True
+ }])
+ assert_equal(result[0]['success'], False)
+ assert_equal(result[0]['error']['code'], -5)
+ assert_equal(result[0]['error']['message'], 'Consistency check failed')
+ address_assert = self.nodes[1].getaddressinfo(address['address'])
+ assert_equal(address_assert['iswatchonly'], False)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal('timestamp' in address_assert, False)
+
+
+ # Importing existing watch only address with new timestamp should replace saved timestamp.
+ assert_greater_than(timestamp, watchonly_timestamp)
+ self.log.info("Should replace previously saved watch only timestamp.")
+ result = self.nodes[1].importmulti([{
+ "scriptPubKey": {
+ "address": watchonly_address,
+ },
+ "timestamp": "now",
+ }])
+ assert_equal(result[0]['success'], True)
+ address_assert = self.nodes[1].getaddressinfo(watchonly_address)
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal(address_assert['timestamp'], timestamp)
+ watchonly_timestamp = timestamp
+
+
+ # restart nodes to check for proper serialization/deserialization of watch only address
+ self.stop_nodes()
+ self.start_nodes()
+ address_assert = self.nodes[1].getaddressinfo(watchonly_address)
+ assert_equal(address_assert['iswatchonly'], True)
+ assert_equal(address_assert['ismine'], False)
+ assert_equal(address_assert['timestamp'], watchonly_timestamp)
+
+ # Bad or missing timestamps
+ self.log.info("Should throw on invalid or missing timestamp values")
+ assert_raises_rpc_error(-3, 'Missing required timestamp field for key',
+ self.nodes[1].importmulti, [{
+ "scriptPubKey": address['scriptPubKey'],
+ }])
+ assert_raises_rpc_error(-3, 'Expected number or "now" timestamp value for key. got type string',
+ self.nodes[1].importmulti, [{
+ "scriptPubKey": address['scriptPubKey'],
+ "timestamp": "",
+ }])
+
+
+if __name__ == '__main__':
+ ImportMultiTest ().main ()