feature_p2spkh.py raw

   1  #!/usr/bin/env python3
   2  """Test P2SPKH: address generation, send, spend, migration, and decodescript.
   3  
   4  End-to-end test of Pay to Schnorr Public Key Hash.
   5  """
   6  
   7  from test_framework.descriptors import descsum_create
   8  from test_framework.key import compute_xonly_pubkey
   9  from test_framework.test_framework import LimenkaTestFramework
  10  from test_framework.util import assert_equal, assert_greater_than, assert_raises_rpc_error
  11  from test_framework.wallet_util import generate_keypair
  12  from test_framework.messages import hash256
  13  
  14  
  15  class P2SPKHTest(LimenkaTestFramework):
  16      def set_test_params(self):
  17          self.num_nodes = 1
  18          self.setup_clean_chain = True
  19  
  20      def skip_test_if_missing_module(self):
  21          self.skip_if_no_wallet()
  22  
  23      def run_test(self):
  24          node = self.nodes[0]
  25          default_w = node.get_wallet_rpc(self.default_wallet_name)
  26  
  27          # Fund the default wallet with mature coinbase outputs
  28          self.generate(node, 101)
  29          assert_equal(default_w.getbalance(), 50)
  30  
  31          self._test_address_generation(node, default_w)
  32          self._test_send_and_spend(node, default_w)
  33          self._test_decodescript(node, default_w)
  34          self._test_migration(node, default_w)
  35          self._test_fundrawtransaction(node, default_w)
  36          self._test_multiple_spends(node, default_w)
  37  
  38      def _create_spk_wallet(self, node, wallet_name, use_xpub=True):
  39          """Create a wallet with a spk() descriptor and return (wallet_rpc, address)."""
  40          node.createwallet(wallet_name=wallet_name, blank=True)
  41          wallet = node.get_wallet_rpc(wallet_name)
  42  
  43          if use_xpub:
  44              from test_framework.extendedkey import ExtendedPrivateKey
  45              xprv = ExtendedPrivateKey.generate()
  46              desc_pay = descsum_create("spk(" + xprv.to_string() + "/87h/1h/0h/0/*)")
  47              desc_change = descsum_create("spk(" + xprv.to_string() + "/87h/1h/0h/1/*)")
  48          else:
  49              privkey_ec, pubkey = generate_keypair(compressed=True, wif=False)
  50              privkey_b = privkey_ec.get_bytes()
  51              xonly_pub, _ = compute_xonly_pubkey(privkey_b)
  52              desc_pay = descsum_create("spk(" + xonly_pub.hex() + ")")
  53              desc_change = descsum_create("spk(" + xonly_pub.hex() + ")")
  54  
  55          wallet.importdescriptors([
  56              {"desc": desc_pay, "active": True, "timestamp": "now"},
  57              {"desc": desc_change, "active": True, "timestamp": "now", "internal": True},
  58          ])
  59  
  60          if use_xpub:
  61              addr = wallet.getnewaddress(address_type="p2spkh")
  62          else:
  63              addrs = wallet.deriveaddresses(desc_pay)
  64              addr = addrs[0]
  65  
  66          info = wallet.getaddressinfo(addr)
  67          assert info['iswitness'], "P2SPKH should be witness"
  68          assert_equal(info.get('witness_version'), 3)
  69  
  70          return wallet, addr
  71  
  72      def _test_address_generation(self, node, default_w):
  73          """Test that spk() descriptor produces valid P2SPKH addresses."""
  74          wallet, addr = self._create_spk_wallet(node, "spk_addr", use_xpub=True)
  75          self.log.info(f"P2SPKH address: {addr}")
  76  
  77          # Bech32m v3 address prefix. v0='q', v1='p', v2='z', v3='r'
  78          assert addr.startswith(("bcrt1r", "tb1r", "bc1r")), f"Expected v3 bech32m prefix, got {addr}"
  79  
  80          # Verify we can decode and re-derive
  81          info = wallet.getaddressinfo(addr)
  82          spk_hex = info['scriptPubKey']
  83          assert len(spk_hex) == 68, f"ScriptPubKey should be 34 bytes (68 hex), got {len(spk_hex)}"
  84          # OP_3 (0x53) + OP_PUSH32 (0x20) + 32-byte hash
  85          assert spk_hex[:4] == "5320", f"Expected 0x53 0x20, got {spk_hex[:4]}"
  86  
  87          # Verify ranged wallets produce unique addresses
  88          addr2 = wallet.getnewaddress(address_type="p2spkh")
  89          assert addr != addr2
  90  
  91          self.log.info("Address generation: PASS")
  92  
  93      def _test_send_and_spend(self, node, default_w):
  94          """Test sending to a P2SPKH address and spending from it."""
  95          wallet, addr = self._create_spk_wallet(node, "spk_spend", use_xpub=True)
  96          self.log.info(f"P2SPKH spend test address: {addr}")
  97  
  98          # Send 10 BTC to P2SPKH address
  99          txid = default_w.sendtoaddress(addr, 10)
 100          self.generate(node, 1)
 101  
 102          # Verify the P2SPKH wallet sees the balance
 103          assert_equal(wallet.getbalance(), 10)
 104          utxos = wallet.listunspent()
 105          assert_equal(len(utxos), 1)
 106          assert_equal(utxos[0]['amount'], 10)
 107  
 108          # Verify output type is recognized
 109          tx = default_w.gettransaction(txid)
 110          assert tx['confirmations'] > 0
 111  
 112          # Spend 5 BTC from P2SPKH to a new default-wallet address
 113          dest = default_w.getnewaddress()
 114          spend_txid = wallet.sendtoaddress(dest, 5)
 115          self.generate(node, 1)
 116  
 117          assert_greater_than(wallet.getbalance(), 4.9)  # accounts for fee
 118  
 119          # Verify the spend transaction confirms
 120          spend_tx = wallet.gettransaction(spend_txid)
 121          assert_greater_than(spend_tx['confirmations'], 0)
 122  
 123          self.log.info("Send and spend: PASS")
 124  
 125      def _test_decodescript(self, node, default_w):
 126          """Test that decodescript recognizes P2SPKH scriptPubKeys."""
 127          wallet, addr = self._create_spk_wallet(node, "spk_decode", use_xpub=False)
 128          info = wallet.getaddressinfo(addr)
 129          spk_hex = info['scriptPubKey']
 130  
 131          decoded = node.decodescript(spk_hex)
 132          assert_equal(decoded['type'], 'witness_v3_spkhash')
 133          assert 'address' in decoded
 134  
 135          self.log.info("Decodescript: PASS")
 136  
 137      def _test_migration(self, node, default_w):
 138          """Test sweeping taproot UTXOs to P2SPKH (migration path)."""
 139          wallet, addr = self._create_spk_wallet(node, "spk_migrate", use_xpub=True)
 140  
 141          # Create a P2TR (taproot) address in the default wallet and fund it
 142          taproot_addr = default_w.getnewaddress(address_type="bech32m")
 143          txid = default_w.sendtoaddress(taproot_addr, 1)
 144          self.generate(node, 1)
 145  
 146          # Now sweep ALL funds from default wallet to P2SPKH address
 147          sweep_result = default_w.sendall(recipients=[addr], fee_rate=1)
 148          self.generate(node, 1)
 149          assert_equal(sweep_result['complete'], True)
 150  
 151          # The P2SPKH wallet should have received funds
 152          spk_balance = wallet.getbalance()
 153          assert_greater_than(spk_balance, 0)
 154  
 155          # The sweep moved the 1 BTC taproot UTXO plus any remaining coinbase
 156          # Construct a raw P2SPKH descriptor send (without using wallet descriptor)
 157          dest = wallet.getnewaddress(address_type="p2spkh")
 158          txid2 = default_w.sendtoaddress(dest, 1)
 159          assert txid2 is not None
 160  
 161          self.log.info("Migration sweep: PASS")
 162  
 163      def _test_fundrawtransaction(self, node, default_w):
 164          """Test fundrawtransaction with P2SPKH outputs."""
 165          wallet, addr = self._create_spk_wallet(node, "spk_fund", use_xpub=True)
 166  
 167          # Fund the wallet with some BTC
 168          default_w.sendtoaddress(addr, 5)
 169          self.generate(node, 1)
 170          assert_equal(wallet.getbalance(), 5)
 171  
 172          # Create a raw transaction sending from P2SPKH
 173          dest = default_w.getnewaddress()
 174          raw = wallet.createrawtransaction([], [{dest: 2}])
 175          funded = wallet.fundrawtransaction(raw)
 176          signed = wallet.signrawtransactionwithwallet(funded['hex'])
 177          assert signed['complete']
 178  
 179          txid = wallet.sendrawtransaction(signed['hex'])
 180          self.generate(node, 1)
 181          tx = wallet.gettransaction(txid)
 182          assert_greater_than(tx['confirmations'], 0)
 183  
 184          self.log.info("fundrawtransaction: PASS")
 185  
 186      def _test_multiple_spends(self, node, default_w):
 187          """Test multiple P2SPKH inputs in one transaction."""
 188          wallet, addr = self._create_spk_wallet(node, "spk_multi", use_xpub=True)
 189  
 190          # Send two separate UTXOs to the P2SPKH wallet
 191          addr2 = wallet.getnewaddress(address_type="p2spkh")
 192          addr3 = wallet.getnewaddress(address_type="p2spkh")
 193          default_w.sendtoaddress(addr, 3)
 194          default_w.sendtoaddress(addr2, 3)
 195          default_w.sendtoaddress(addr3, 3)
 196          self.generate(node, 1)
 197          assert_equal(wallet.getbalance(), 9)
 198  
 199          # Spend all three UTXOs in one transaction
 200          unspent = wallet.listunspent()
 201          assert_equal(len(unspent), 3)
 202  
 203          dest = default_w.getnewaddress()
 204          raw = wallet.createrawtransaction(
 205              [{"txid": u["txid"], "vout": u["vout"]} for u in unspent],
 206              {dest: 8.999}
 207          )
 208          funded = wallet.fundrawtransaction(raw)
 209          signed = wallet.signrawtransactionwithwallet(funded['hex'])
 210          assert signed['complete']
 211          txid = wallet.sendrawtransaction(signed['hex'])
 212          self.generate(node, 1)
 213          tx = wallet.gettransaction(txid)
 214          assert_greater_than(tx['confirmations'], 0)
 215  
 216          self.log.info("Multiple spends: PASS")
 217  
 218  
 219  if __name__ == '__main__':
 220      P2SPKHTest(__file__).main()
 221