test: Test for wallet txs with alternate wtxids

test/functional/wallet_listtransactions.py

@@ -11,11 +11,13 @@ import shutil
 
 from test_framework.blocktools import MAX_FUTURE_BLOCK_TIME
 from test_framework.descriptors import descsum_create
+from test_framework.extendedkey import ExtendedPrivateKey
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_not_equal,
     assert_array_result,
     assert_equal,
+    assert_greater_than,
     assert_raises_rpc_error,
     find_vout_for_address,
 )
@@ -96,6 +98,7 @@ class ListTransactionsTest(BitcoinTestFramework):
         self.run_coinjoin_test()
         self.run_invalid_parameters_test()
         self.test_op_return()
+        self.test_alternate_witness_tx()
         self.test_from_me_status_change()
 
     def run_externally_generated_address_test(self):
@@ -242,5 +245,117 @@ class ListTransactionsTest(BitcoinTestFramework):
             assert "fee" in tx_info
             assert_equal(any(detail["category"] == "send" for detail in tx_info["details"]), True)
 
+    def check_tx_variants(self, wallet, txid, canonical_tx_hex, canonical_wtxid, alternate_wtxids):
+        """Assert gettransaction and listtransactions report tx variants properly"""
+        tx_info = wallet.gettransaction(txid)
+        assert_equal(tx_info["hex"], canonical_tx_hex)
+        assert_equal(tx_info["wtxid"], canonical_wtxid)
+        # alternate_wtxids lists the other variants, never the canonical one
+        assert canonical_wtxid not in tx_info["alternate_wtxids"]
+        assert_equal(set(tx_info["alternate_wtxids"]), set(alternate_wtxids))
+
+        # listtransactions exposes the same alternate_wtxids field as gettransaction
+        list_entry = next(entry for entry in wallet.listtransactions() if entry["txid"] == txid)
+        assert_equal(list_entry["alternate_wtxids"], tx_info["alternate_wtxids"])
+
+    # Returns the finalized psbt transaction and its wtxid
+    def finalize_tx_variant(self, wallet, psbt, spend_path):
+        # First check the expected spend path is being used
+        sig_field = {"script": "taproot_script_path_sigs", "key": "taproot_key_path_sig"}
+        present, absent = sig_field[spend_path], sig_field["key" if spend_path == "script" else "script"]
+        psbt_input = self.nodes[0].decodepsbt(psbt)["inputs"][0]
+        assert present in psbt_input and absent not in psbt_input
+
+        # Then finalize and decode
+        tx = wallet.finalizepsbt(psbt)["hex"]
+        return tx, wallet.decoderawtransaction(tx)["hash"]
+
+    def test_alternate_witness_tx(self):
+        self.log.info("Test gettransaction and listtransactions report alternate witnesses and canonical variant")
+        self.nodes[0].createwallet("altwit")
+        default_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
+        wallet = self.nodes[0].get_wallet_rpc("altwit")
+
+        xprvs = [ExtendedPrivateKey.generate() for _ in range(0, 2)]
+        xpubs = [xprv.pubkey() for xprv in xprvs]
+
+        # Import a taproot descriptor with script paths
+        desc = descsum_create(f"tr({xpubs[0].to_string()}/*,pk({xprvs[1].to_string()}/*))")
+        assert_equal(wallet.importdescriptors([{"desc": desc, "active": True, "timestamp": "now"}])[0]["success"], True)
+        default_wallet.sendtoaddress(wallet.getnewaddress(address_type="bech32m"), 1)
+        self.generate(self.nodes[0], 1, sync_fun=self.no_op)
+        # Isolate node0 for later reorg coverage
+        self.disconnect_nodes(0, 1)
+        self.disconnect_nodes(0, 2)
+
+        # Create output psbt
+        psbt = wallet.walletcreatefundedpsbt(outputs=[{default_wallet.getnewaddress(): 0.5}])["psbt"]
+
+        # Create a script path spend and relay it. With only one variant known it
+        # is trivially canonical and has no alternates
+        self.log.info("Test the only known variant is canonical with no alternates")
+        script_path_psbt = wallet.walletprocesspsbt(psbt=psbt, finalize=False)["psbt"]
+        script_path_tx, script_path_wtxid = self.finalize_tx_variant(wallet, script_path_psbt, spend_path="script")
+        txid = self.nodes[0].sendrawtransaction(script_path_tx)
+        self.check_tx_variants(wallet, txid, script_path_tx, script_path_wtxid, alternate_wtxids=[])
+
+        # Make a key path spend separate from the wallet
+        key_path_desc = descsum_create(f"tr({xprvs[0].to_string()}/*,pk({xpubs[1].to_string()}/*))")
+        key_path_psbt = self.nodes[0].descriptorprocesspsbt(psbt=psbt, descriptors=[{"desc": key_path_desc}], finalize=False)["psbt"]
+        key_path_tx, key_path_wtxid = self.finalize_tx_variant(wallet, key_path_psbt, spend_path="key")
+
+        # Ensure variants share the same txid but differ in wtxid, and the key path is the lighter of the two
+        assert_equal(txid, self.nodes[0].decoderawtransaction(key_path_tx)["txid"])
+        assert_not_equal(script_path_wtxid, key_path_wtxid)
+        assert_greater_than(
+            self.nodes[0].decoderawtransaction(script_path_tx)["weight"],
+            self.nodes[0].decoderawtransaction(key_path_tx)["weight"],
+        )
+
+        # The wallet only learns the key path witness from a block (the mempool
+        # holds one transaction per txid). Mine the key path: a confirmed variant
+        # is canonical, with the script path now listed as its alternate.
+        block = self.generateblock(self.nodes[0], default_wallet.getnewaddress(), [key_path_tx], sync_fun=self.no_op)["hash"]
+        self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid])
+
+        # Reorg that block out so both variants are known and unconfirmed. With no
+        # confirmation to force the choice, the lighter key path is canonical.
+        self.log.info("Test the lighter variant is canonical when both are known and unconfirmed")
+        self.nodes[0].invalidateblock(block)
+        self.nodes[0].syncwithvalidationinterfacequeue()
+        assert_equal(wallet.gettransaction(txid)["confirmations"], 0)
+        self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid])
+
+        # The canonical choice and alternates survive a wallet reload
+        wallet.unloadwallet()
+        self.nodes[0].loadwallet("altwit")
+        self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid])
+
+        # Now confirm the heavier script path instead, on a longer competing chain
+        # from node1, and reconnect so node0 reorgs onto it. The confirmed variant
+        # is canonical even though it is the heavier one.
+        self.log.info("Test a confirmed variant is canonical even when it is the heavier one")
+        self.generate(self.nodes[1], 3, sync_fun=self.no_op)
+        block = self.generateblock(self.nodes[1], default_wallet.getnewaddress(), [script_path_tx], sync_fun=self.no_op)["hash"]
+        self.connect_nodes(0, 1)
+        self.connect_nodes(0, 2)
+        self.sync_all()
+        self.check_tx_variants(wallet, txid, script_path_tx, script_path_wtxid, alternate_wtxids=[key_path_wtxid])
+
+        # The confirmed-canonical choice survive a reload
+        wallet.unloadwallet()
+        self.nodes[0].loadwallet("altwit")
+        self.check_tx_variants(wallet, txid, script_path_tx, script_path_wtxid, alternate_wtxids=[key_path_wtxid])
+
+        self.log.info("Test canonical reverts to the lighter variant when the confirmed one is reorged out")
+        # Both variants are unconfirmed again, so the lighter key path is canonical once more
+        self.disconnect_nodes(0, 1)
+        self.disconnect_nodes(0, 2)
+        self.nodes[0].invalidateblock(block)
+        self.nodes[0].syncwithvalidationinterfacequeue()
+        assert_equal(wallet.gettransaction(txid)["confirmations"], 0)
+        self.check_tx_variants(wallet, txid, key_path_tx, key_path_wtxid, alternate_wtxids=[script_path_wtxid])
+
+
 if __name__ == '__main__':
     ListTransactionsTest(__file__).main()