Repository: bitcoin/bitcoin
File: test/functional/rpc_psbt.py (Python)

Code before:
            assert hash.hex() in res_input[preimage_key]
            assert_equal(res_input[preimage_key][hash.hex()], preimage.hex())

        self.log.info("Test that combining PSBTs with different transactions fails")
        tx = CTransaction()
        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('aa' * 32, 16), n=0), scriptSig=b"")]
        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
        psbt1 = PSBT(g=PSBTMap({PSBT_GLOBAL_UNSIGNED_TX: tx.serialize()}), i=[PSBTMap()], o=[PSBTMap()]).to_base64()
        tx.vout[0].nValue += 1  # slightly modify tx
        psbt2 = PSBT(g=PSBTMap({PSBT_GLOBAL_UNSIGNED_TX: tx.serialize()}), i=[PSBTMap()], o=[PSBTMap()]).to_base64()
        assert_raises_rpc_error(-8, "PSBTs not compatible (different transactions)", self.nodes[0].combinepsbt, [psbt1, psbt2])
        assert_equal(self.nodes[0].combinepsbt([psbt1, psbt1]), psbt1)

        self.log.info("Test that PSBT inputs are being checked via script execution")
        acs_prevout = CTxOut(nValue=0, scriptPubKey=CScript([OP_TRUE]))
        tx = CTransaction()
        tx.vin = [CTxIn(outpoint=COutPoint(hash=int('dd' * 32, 16), n=0), scriptSig=b"")]
        tx.vout = [CTxOut(nValue=0, scriptPubKey=b"")]
        psbt = PSBT()
        psbt.g = PSBTMap({PSBT_GLOBAL_UNSIGNED_TX: tx.serialize()})
        psbt.i = [PSBTMap({bytes([PSBT_IN_WITNESS_UTXO]) : acs_prevout.serialize()})]
        psbt.o = [PSBTMap()]
        assert_equal(self.nodes[0].finalizepsbt(psbt.to_base64()),
            {'hex': '0200000001dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd0000000000000000000100000000000000000000000000', 'complete': True})

        self.log.info("Test we don't crash when making a 0-value funded transaction at 0 fee without forcing an input selection")
        assert_raises_rpc_error(-4, "Transaction requires one destination of non-zero value, a non-zero feerate, or a pre-selected input", self.nodes[0].walletcreatefundedpsbt, [], [{"data": "deadbeef"}], 0, {"fee_rate": "0"})

        self.log.info("Test descriptorprocesspsbt updates and signs a psbt with descriptors")

        self.generate(self.nodes[2], 1)

        # Disable the wallet for node 2 since `descriptorprocesspsbt` does not use the wallet
        self.restart_node(2, extra_args=["-disablewallet"])
        self.connect_nodes(0, 2)
        self.connect_nodes(1, 2)

        key_info = get_generate_key()
        key = key_info.privkey
        address = key_info.p2wpkh_addr

        descriptor = descsum_create(f"wpkh({key})")

        utxo = self.create_outpoints(self.nodes[0], outputs=[{address: 1}])[0]
        self.sync_all()

        psbt = self.nodes[2].createpsbt([utxo], {self.nodes[0].getnewaddress(): 0.99999})
        decoded = self.nodes[2].decodepsbt(psbt)
        test_psbt_input_keys(decoded['inputs'][0], [])

        # Test that even if the wrong descriptor is given, `witness_utxo` and `non_witness_utxo`

Review comment (suggestion):
Nit: Maybe consider putting this in a separate subtest as the current `run_test` is quite long already.

Code after:

        # Same test but for taproot
        privkey, pubkey = generate_keypair(wif=True)

        desc = descsum_create("tr({},pk({}))".format(H_POINT, pubkey.hex()))
        res = watchonly.importdescriptors([{"desc": desc, "timestamp": "now"}])
        assert res[0]["success"]
        addr = self.nodes[0].deriveaddresses(desc)[0]
        self.nodes[0].sendtoaddress(addr, 10)
        self.generate(self.nodes[0], 1)
        self.nodes[0].importdescriptors([{"desc": descsum_create("tr({})".format(privkey)), "timestamp":"now"}])

        psbt = watchonly.sendall([wallet.getnewaddress(), addr])["psbt"]
        processed_psbt = self.nodes[0].walletprocesspsbt(psbt)
        txid = self.nodes[0].sendrawtransaction(processed_psbt["hex"])
        vout = find_vout_for_address(self.nodes[0], txid, addr)

        # Make sure tap tree is in psbt
        parsed_psbt = PSBT.from_base64(psbt)
        assert_greater_than(len(parsed_psbt.o[vout].map[PSBT_OUT_TAP_TREE]), 0)
        assert "taproot_tree" in self.nodes[0].decodepsbt(psbt)["outputs"][vout]
        parsed_psbt.make_blank()
        comb_psbt = self.nodes[0].combinepsbt([psbt, parsed_psbt.to_base64()])
        assert_equal(comb_psbt, psbt)

        self.log.info("Test that walletprocesspsbt both updates and signs a non-updated psbt containing Taproot inputs")
        addr = self.nodes[0].getnewaddress("", "bech32m")
        utxo = self.create_outpoints(self.nodes[0], outputs=[{addr: 1}])[0]
        psbt = self.nodes[0].createpsbt([utxo], [{self.nodes[0].getnewaddress(): 0.9999}])
        signed = self.nodes[0].walletprocesspsbt(psbt)
        rawtx = signed["hex"]
        self.nodes[0].sendrawtransaction(rawtx)
        self.generate(self.nodes[0], 1)

        # Make sure tap tree is not in psbt
        parsed_psbt = PSBT.from_base64(psbt)
        assert PSBT_OUT_TAP_TREE not in parsed_psbt.o[0].map
        assert "taproot_tree" not in self.nodes[0].decodepsbt(psbt)["outputs"][0]
        parsed_psbt.make_blank()
        comb_psbt = self.nodes[0].combinepsbt([psbt, parsed_psbt.to_base64()])
        assert_equal(comb_psbt, psbt)

        self.log.info("Test walletprocesspsbt raises if an invalid sighashtype is passed")
        assert_raises_rpc_error(-8, "'all' is not a valid sighash parameter.", self.nodes[0].walletprocesspsbt, psbt, sighashtype="all")

        self.log.info("Test decoding PSBT with per-input preimage types")
        # note that the decodepsbt RPC doesn't check whether preimages and hashes match
        hash_ripemd160, preimage_ripemd160 = randbytes(20), randbytes(50)
        hash_sha256, preimage_sha256 = randbytes(32), randbytes(50)
        hash_hash160, preimage_hash160 = randbytes(20), randbytes(50)
        hash_hash256, preimage_hash256 = randbytes(32), randbytes(50)