test: introduce getnewdestination helper for generating various address types

This serves as a replacement for the getnewaddress RPC if no wallet is
available. In addition to the address, it also returns the corresponding
public key and output script (scriptPubKey).
This commit is contained in:
Sebastian Falbesoner 2021-12-25 21:31:49 +01:00
parent 9bec5b80a0
commit e704d4d26f
2 changed files with 33 additions and 10 deletions

View file

@ -31,7 +31,7 @@ from test_framework.script import MAX_SCRIPT_ELEMENT_SIZE
from test_framework.test_framework import BitcoinTestFramework from test_framework.test_framework import BitcoinTestFramework
from test_framework.wallet import ( from test_framework.wallet import (
MiniWallet, MiniWallet,
random_p2wpkh, getnewdestination,
) )
@ -169,14 +169,14 @@ class FilterTest(BitcoinTestFramework):
self.log.info('Check that we only receive a merkleblock if the filter does not match a tx in a block') self.log.info('Check that we only receive a merkleblock if the filter does not match a tx in a block')
filter_peer.tx_received = False filter_peer.tx_received = False
block_hash = self.generatetoscriptpubkey(random_p2wpkh()) block_hash = self.generatetoscriptpubkey(getnewdestination()[1])
filter_peer.wait_for_merkleblock(block_hash) filter_peer.wait_for_merkleblock(block_hash)
assert not filter_peer.tx_received assert not filter_peer.tx_received
self.log.info('Check that we not receive a tx if the filter does not match a mempool tx') self.log.info('Check that we not receive a tx if the filter does not match a mempool tx')
filter_peer.merkleblock_received = False filter_peer.merkleblock_received = False
filter_peer.tx_received = False filter_peer.tx_received = False
self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=random_p2wpkh(), amount=7 * COIN) self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=getnewdestination()[1], amount=7 * COIN)
filter_peer.sync_send_with_ping() filter_peer.sync_send_with_ping()
assert not filter_peer.merkleblock_received assert not filter_peer.merkleblock_received
assert not filter_peer.tx_received assert not filter_peer.tx_received
@ -190,14 +190,14 @@ class FilterTest(BitcoinTestFramework):
self.log.info('Check that after deleting filter all txs get relayed again') self.log.info('Check that after deleting filter all txs get relayed again')
filter_peer.send_and_ping(msg_filterclear()) filter_peer.send_and_ping(msg_filterclear())
for _ in range(5): for _ in range(5):
txid, _ = self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=random_p2wpkh(), amount=7 * COIN) txid, _ = self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=getnewdestination()[1], amount=7 * COIN)
filter_peer.wait_for_tx(txid) filter_peer.wait_for_tx(txid)
self.log.info('Check that request for filtered blocks is ignored if no filter is set') self.log.info('Check that request for filtered blocks is ignored if no filter is set')
filter_peer.merkleblock_received = False filter_peer.merkleblock_received = False
filter_peer.tx_received = False filter_peer.tx_received = False
with self.nodes[0].assert_debug_log(expected_msgs=['received getdata']): with self.nodes[0].assert_debug_log(expected_msgs=['received getdata']):
block_hash = self.generatetoscriptpubkey(random_p2wpkh()) block_hash = self.generatetoscriptpubkey(getnewdestination()[1])
filter_peer.wait_for_inv([CInv(MSG_BLOCK, int(block_hash, 16))]) filter_peer.wait_for_inv([CInv(MSG_BLOCK, int(block_hash, 16))])
filter_peer.sync_with_ping() filter_peer.sync_with_ping()
assert not filter_peer.merkleblock_received assert not filter_peer.merkleblock_received

View file

@ -9,7 +9,12 @@ from decimal import Decimal
from enum import Enum from enum import Enum
from random import choice from random import choice
from typing import Optional from typing import Optional
from test_framework.address import create_deterministic_address_bcrt1_p2tr_op_true from test_framework.address import (
create_deterministic_address_bcrt1_p2tr_op_true,
key_to_p2pkh,
key_to_p2sh_p2wpkh,
key_to_p2wpkh,
)
from test_framework.descriptors import descsum_create from test_framework.descriptors import descsum_create
from test_framework.key import ECKey from test_framework.key import ECKey
from test_framework.messages import ( from test_framework.messages import (
@ -31,6 +36,8 @@ from test_framework.script import (
) )
from test_framework.script_util import ( from test_framework.script_util import (
key_to_p2pk_script, key_to_p2pk_script,
key_to_p2pkh_script,
key_to_p2sh_p2wpkh_script,
key_to_p2wpkh_script, key_to_p2wpkh_script,
) )
from test_framework.util import ( from test_framework.util import (
@ -209,12 +216,28 @@ class MiniWallet:
return txid return txid
def random_p2wpkh(): def getnewdestination(address_type='bech32'):
"""Generate a random P2WPKH scriptPubKey. Can be used when a random destination is needed, """Generate a random destination of the specified type and return the
but no compiled wallet is available (e.g. as replacement to the getnewaddress RPC).""" corresponding public key, scriptPubKey and address. Supported types are
'legacy', 'p2sh-segwit' and 'bech32'. Can be used when a random
destination is needed, but no compiled wallet is available (e.g. as
replacement to the getnewaddress/getaddressinfo RPCs)."""
key = ECKey() key = ECKey()
key.generate() key.generate()
return key_to_p2wpkh_script(key.get_pubkey().get_bytes()) pubkey = key.get_pubkey().get_bytes()
if address_type == 'legacy':
scriptpubkey = key_to_p2pkh_script(pubkey)
address = key_to_p2pkh(pubkey)
elif address_type == 'p2sh-segwit':
scriptpubkey = key_to_p2sh_p2wpkh_script(pubkey)
address = key_to_p2sh_p2wpkh(pubkey)
elif address_type == 'bech32':
scriptpubkey = key_to_p2wpkh_script(pubkey)
address = key_to_p2wpkh(pubkey)
# TODO: also support bech32m (need to generate x-only-pubkey)
else:
assert False
return pubkey, scriptpubkey, address
def make_chain(node, address, privkeys, parent_txid, parent_value, n=0, parent_locking_script=None, fee=DEFAULT_FEE): def make_chain(node, address, privkeys, parent_txid, parent_value, n=0, parent_locking_script=None, fee=DEFAULT_FEE):