Changeset View
Changeset View
Standalone View
Standalone View
test/functional/wallet_import_rescan.py
Show First 20 Lines • Show All 50 Lines • ▼ Show 20 Lines | class Variant(collections.namedtuple("Variant", "call data rescan prune")): | ||||
def do_import(self, timestamp): | def do_import(self, timestamp): | ||||
"""Call one key import RPC.""" | """Call one key import RPC.""" | ||||
rescan = self.rescan == Rescan.yes | rescan = self.rescan == Rescan.yes | ||||
if self.call == Call.single: | if self.call == Call.single: | ||||
if self.data == Data.address: | if self.data == Data.address: | ||||
response = self.try_rpc( | response = self.try_rpc( | ||||
self.node.importaddress, address=self.address["address"], rescan=rescan) | self.node.importaddress, | ||||
address=self.address["address"], | |||||
label=self.label, | |||||
rescan=rescan) | |||||
elif self.data == Data.pub: | elif self.data == Data.pub: | ||||
response = self.try_rpc( | response = self.try_rpc( | ||||
self.node.importpubkey, pubkey=self.address["pubkey"], rescan=rescan) | self.node.importpubkey, | ||||
pubkey=self.address["pubkey"], | |||||
label=self.label, | |||||
rescan=rescan) | |||||
elif self.data == Data.priv: | elif self.data == Data.priv: | ||||
response = self.try_rpc( | response = self.try_rpc( | ||||
self.node.importprivkey, privkey=self.key, rescan=rescan) | self.node.importprivkey, | ||||
privkey=self.key, | |||||
label=self.label, | |||||
rescan=rescan) | |||||
assert_equal(response, None) | assert_equal(response, None) | ||||
elif self.call == Call.multi: | elif self.call == Call.multi: | ||||
response = self.node.importmulti([{ | response = self.node.importmulti([{ | ||||
"scriptPubKey": { | "scriptPubKey": { | ||||
"address": self.address["address"] | "address": self.address["address"] | ||||
}, | }, | ||||
"timestamp": timestamp + TIMESTAMP_WINDOW + (1 if self.rescan == Rescan.late_timestamp else 0), | "timestamp": timestamp + TIMESTAMP_WINDOW + (1 if self.rescan == Rescan.late_timestamp else 0), | ||||
"pubkeys": [self.address["pubkey"]] if self.data == Data.pub else [], | "pubkeys": [self.address["pubkey"]] if self.data == Data.pub else [], | ||||
"keys": [self.key] if self.data == Data.priv else [], | "keys": [self.key] if self.data == Data.priv else [], | ||||
"label": self.label, | |||||
"watchonly": self.data != Data.priv | "watchonly": self.data != Data.priv | ||||
}], {"rescan": self.rescan in (Rescan.yes, Rescan.late_timestamp)}) | }], {"rescan": self.rescan in (Rescan.yes, Rescan.late_timestamp)}) | ||||
assert_equal(response, [{"success": True}]) | assert_equal(response, [{"success": True}]) | ||||
def check(self, txid=None, amount=None, confirmations=None): | def check(self, txid=None, amount=None, confirmations=None): | ||||
"""Verify that listreceivedbyaddress returns expected values.""" | """Verify that listtransactions/listreceivedbyaddress return expected values.""" | ||||
txs = self.node.listtransactions( | |||||
label=self.label, count=10000, include_watchonly=True) | |||||
assert_equal(len(txs), self.expected_txs) | |||||
addresses = self.node.listreceivedbyaddress( | addresses = self.node.listreceivedbyaddress( | ||||
minconf=0, include_watchonly=True, address_filter=self.address['address']) | minconf=0, include_watchonly=True, address_filter=self.address['address']) | ||||
if self.expected_txs: | if self.expected_txs: | ||||
assert_equal(len(addresses[0]["txids"]), self.expected_txs) | assert_equal(len(addresses[0]["txids"]), self.expected_txs) | ||||
if txid is not None: | if txid is not None: | ||||
tx, = [tx for tx in txs if tx["txid"] == txid] | |||||
assert_equal(tx["label"], self.label) | |||||
assert_equal(tx["address"], self.address["address"]) | |||||
assert_equal(tx["amount"], amount) | |||||
assert_equal(tx["category"], "receive") | |||||
assert_equal(tx["label"], self.label) | |||||
assert_equal(tx["txid"], txid) | |||||
assert_equal(tx["confirmations"], confirmations) | |||||
assert_equal("trusted" not in tx, True) | |||||
address, = [ad for ad in addresses if txid in ad["txids"]] | address, = [ad for ad in addresses if txid in ad["txids"]] | ||||
assert_equal(address["address"], self.address["address"]) | assert_equal(address["address"], self.address["address"]) | ||||
assert_equal(address["amount"], self.expected_balance) | assert_equal(address["amount"], self.expected_balance) | ||||
assert_equal(address["confirmations"], confirmations) | assert_equal(address["confirmations"], confirmations) | ||||
# Verify the transaction is correctly marked watchonly depending on | # Verify the transaction is correctly marked watchonly depending on | ||||
# whether the transaction pays to an imported public key or | # whether the transaction pays to an imported public key or | ||||
# imported private key. The test setup ensures that transaction | # imported private key. The test setup ensures that transaction | ||||
# inputs will not be from watchonly keys (important because | # inputs will not be from watchonly keys (important because | ||||
▲ Show 20 Lines • Show All 49 Lines • ▼ Show 20 Lines | class ImportRescanTest(BitcoinTestFramework): | ||||
def import_deterministic_coinbase_privkeys(self): | def import_deterministic_coinbase_privkeys(self): | ||||
pass | pass | ||||
def run_test(self): | def run_test(self): | ||||
# Create one transaction on node 0 with a unique amount for | # Create one transaction on node 0 with a unique amount for | ||||
# each possible type of wallet import RPC. | # each possible type of wallet import RPC. | ||||
for i, variant in enumerate(IMPORT_VARIANTS): | for i, variant in enumerate(IMPORT_VARIANTS): | ||||
variant.label = "label {} {}".format(i, variant) | |||||
variant.address = self.nodes[1].getaddressinfo( | variant.address = self.nodes[1].getaddressinfo( | ||||
self.nodes[1].getnewaddress()) | self.nodes[1].getnewaddress(variant.label)) | ||||
variant.key = self.nodes[1].dumpprivkey(variant.address["address"]) | variant.key = self.nodes[1].dumpprivkey(variant.address["address"]) | ||||
variant.initial_amount = 10 - (i + 1) / 4.0 | variant.initial_amount = 10 - (i + 1) / 4.0 | ||||
variant.initial_txid = self.nodes[0].sendtoaddress( | variant.initial_txid = self.nodes[0].sendtoaddress( | ||||
variant.address["address"], variant.initial_amount) | variant.address["address"], variant.initial_amount) | ||||
# Generate a block containing the initial transactions, then another | # Generate a block containing the initial transactions, then another | ||||
# block further in the future (past the rescan window). | # block further in the future (past the rescan window). | ||||
self.nodes[0].generate(1) | self.nodes[0].generate(1) | ||||
▲ Show 20 Lines • Show All 47 Lines • Show Last 20 Lines |