Changeset View
Changeset View
Standalone View
Standalone View
test/functional/wallet_import_rescan.py
| Show All 27 Lines | from test_framework.util import ( | ||||
| assert_equal, | assert_equal, | ||||
| assert_raises_rpc_error, | assert_raises_rpc_error, | ||||
| connect_nodes, | connect_nodes, | ||||
| set_node_times, | set_node_times, | ||||
| sync_blocks, | sync_blocks, | ||||
| ) | ) | ||||
| Call = enum.Enum("Call", "single multi") | Call = enum.Enum("Call", "single multiaddress multiscript") | ||||
| Data = enum.Enum("Data", "address pub priv") | Data = enum.Enum("Data", "address pub priv") | ||||
| Rescan = enum.Enum("Rescan", "no yes late_timestamp") | Rescan = enum.Enum("Rescan", "no yes late_timestamp") | ||||
| class Variant(collections.namedtuple("Variant", "call data rescan prune")): | class Variant(collections.namedtuple("Variant", "call data rescan prune")): | ||||
| """Helper for importing one key and verifying scanned transactions.""" | """Helper for importing one key and verifying scanned transactions.""" | ||||
| Show All 24 Lines | def do_import(self, timestamp): | ||||
| elif self.data == Data.priv: | elif self.data == Data.priv: | ||||
| response = self.try_rpc( | response = self.try_rpc( | ||||
| self.node.importprivkey, | self.node.importprivkey, | ||||
| privkey=self.key, | privkey=self.key, | ||||
| label=self.label, | label=self.label, | ||||
| rescan=rescan) | rescan=rescan) | ||||
| assert_equal(response, None) | assert_equal(response, None) | ||||
| elif self.call == Call.multi: | elif self.call in (Call.multiaddress, Call.multiscript): | ||||
| response = self.node.importmulti([{ | response = self.node.importmulti([{ | ||||
| "scriptPubKey": { | "scriptPubKey": { | ||||
| "address": self.address["address"] | "address": self.address["address"] | ||||
| }, | } if self.call == Call.multiaddress else self.address["scriptPubKey"], | ||||
| "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, | "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}]) | ||||
| ▲ Show 20 Lines • Show All 84 Lines • ▼ Show 20 Lines | class ImportRescanTest(BitcoinTestFramework): | ||||
| 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.label = "label {} {}".format(i, variant) | ||||
| variant.address = self.nodes[1].getaddressinfo( | variant.address = self.nodes[1].getaddressinfo( | ||||
| self.nodes[1].getnewaddress(variant.label)) | 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 = 1 - (i + 1) / 64 | ||||
| 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) | ||||
| assert_equal(self.nodes[0].getrawmempool(), []) | assert_equal(self.nodes[0].getrawmempool(), []) | ||||
| timestamp = self.nodes[0].getblockheader( | timestamp = self.nodes[0].getblockheader( | ||||
| Show All 16 Lines | def run_test(self): | ||||
| variant.check(variant.initial_txid, variant.initial_amount, 2) | variant.check(variant.initial_txid, variant.initial_amount, 2) | ||||
| else: | else: | ||||
| variant.expected_balance = 0 | variant.expected_balance = 0 | ||||
| variant.expected_txs = 0 | variant.expected_txs = 0 | ||||
| variant.check() | variant.check() | ||||
| # Create new transactions sending to each address. | # Create new transactions sending to each address. | ||||
| for i, variant in enumerate(IMPORT_VARIANTS): | for i, variant in enumerate(IMPORT_VARIANTS): | ||||
| variant.sent_amount = 10 - (2 * i + 1) / 8.0 | variant.sent_amount = 1 - (2 * i + 1) / 128 | ||||
| variant.sent_txid = self.nodes[0].sendtoaddress( | variant.sent_txid = self.nodes[0].sendtoaddress( | ||||
| variant.address["address"], variant.sent_amount) | variant.address["address"], variant.sent_amount) | ||||
| # Generate a block containing the new transactions. | # Generate a block containing the new transactions. | ||||
| self.nodes[0].generate(1) | self.nodes[0].generate(1) | ||||
| assert_equal(self.nodes[0].getrawmempool(), []) | assert_equal(self.nodes[0].getrawmempool(), []) | ||||
| sync_blocks(self.nodes) | sync_blocks(self.nodes) | ||||
| Show All 12 Lines | |||||