Differential D4353 Diff 13870 test/functional/rpc_psbt.py

Changeset View

Standalone View

test/functional/rpc_psbt.py

This file was added.

Property	Old Value	New Value
File Mode	null	100755

				#!/usr/bin/env python3
				# Copyright (c) 2018 The Bitcoin Core developers
				# Distributed under the MIT software license, see the accompanying
				# file COPYING or http://www.opensource.org/licenses/mit-license.php.
				"""Test the Partially Signed Transaction RPCs.
				"""

				from test_framework.test_framework import BitcoinTestFramework
				from test_framework.util import (
				assert_equal,
				assert_raises_rpc_error,
				find_output,
				)

				import os
				import json
				FabienUnsubmitted Not Done Inline Actions Nit: put above and sort. Fabien: Nit: put above and sort.

				# Create one-input, one-output, no-fee transaction:


				class PSBTTest(BitcoinTestFramework):

				def set_test_params(self):
				self.setup_clean_chain = False
				self.num_nodes = 3

				def run_test(self):
				# Create and fund a raw tx for sending 10 BTC
				psbtx1 = self.nodes[0].walletcreatefundedpsbt(
				[], {self.nodes[2].getnewaddress(): 10})['psbt']

				# Node 1 should not be able to add anything to it but still return the psbtx same as before
				psbtx = self.nodes[1].walletprocesspsbt(psbtx1)['psbt']
				assert_equal(psbtx1, psbtx)

				# Sign the transaction and send
				signed_tx = self.nodes[0].walletprocesspsbt(psbtx)['psbt']
				final_tx = self.nodes[0].finalizepsbt(signed_tx)['hex']
				self.nodes[0].sendrawtransaction(final_tx)

				# Create p2sh, p2wpkh, and p2wsh addresses
				FabienUnsubmitted Not Done Inline Actions p2pkh and p2sh Fabien: p2pkh and p2sh
				pubkey0 = self.nodes[0].getaddressinfo(
				self.nodes[0].getnewaddress())['pubkey']
				pubkey1 = self.nodes[1].getaddressinfo(
				self.nodes[1].getnewaddress())['pubkey']
				pubkey2 = self.nodes[2].getaddressinfo(
				self.nodes[2].getnewaddress())['pubkey']
				p2sh = self.nodes[1].addmultisigaddress(
				2, [pubkey0, pubkey1, pubkey2], "")['address']
				p2pkh = self.nodes[1].getnewaddress("")

				# fund those addresses
				rawtx = self.nodes[0].createrawtransaction([], {p2sh: 10, p2pkh: 10})
				rawtx = self.nodes[0].fundrawtransaction(rawtx, {"changePosition": 0})
				signed_tx = self.nodes[0].signrawtransactionwithwallet(rawtx['hex'])[
				'hex']
				txid = self.nodes[0].sendrawtransaction(signed_tx)
				self.nodes[0].generate(6)
				self.sync_all()

				# Find the output pos
				p2sh_pos = -1
				p2pkh_pos = -1
				decoded = self.nodes[0].decoderawtransaction(signed_tx)
				for out in decoded['vout']:
				if out['scriptPubKey']['addresses'][0] == p2sh:
				p2sh_pos = out['n']
				elif out['scriptPubKey']['addresses'][0] == p2pkh:
				p2pkh_pos = out['n']

				# spend single key from node 1
				rawtx = self.nodes[1].walletcreatefundedpsbt([{"txid": txid, "vout": p2pkh_pos}], {
				self.nodes[1].getnewaddress(): 9.99})['psbt']
				walletprocesspsbt_out = self.nodes[1].walletprocesspsbt(rawtx)
				assert_equal(walletprocesspsbt_out['complete'], True)
				self.nodes[1].sendrawtransaction(
				self.nodes[1].finalizepsbt(walletprocesspsbt_out['psbt'])['hex'])

				# partially sign multisig things with node 1
				psbtx = self.nodes[1].walletcreatefundedpsbt([{"txid": txid, "vout": p2sh_pos}], {
				self.nodes[1].getnewaddress(): 9.99})['psbt']
				walletprocesspsbt_out = self.nodes[1].walletprocesspsbt(psbtx)
				psbtx = walletprocesspsbt_out['psbt']
				assert_equal(walletprocesspsbt_out['complete'], False)

				# partially sign with node 2. This should be complete and sendable
				walletprocesspsbt_out = self.nodes[2].walletprocesspsbt(psbtx)
				assert_equal(walletprocesspsbt_out['complete'], True)
				self.nodes[2].sendrawtransaction(
				self.nodes[2].finalizepsbt(walletprocesspsbt_out['psbt'])['hex'])

				# check that walletprocesspsbt fails to decode a non-psbt
				rawtx = self.nodes[1].createrawtransaction([{"txid": txid, "vout": p2pkh_pos}], {
				self.nodes[1].getnewaddress(): 9.99})
				assert_raises_rpc_error(-22, "TX decode failed",
				self.nodes[1].walletprocesspsbt, rawtx)

				# Convert a non-psbt to psbt and make sure we can decode it
				rawtx = self.nodes[0].createrawtransaction(
				[], {self.nodes[1].getnewaddress(): 10})
				rawtx = self.nodes[0].fundrawtransaction(rawtx)
				new_psbt = self.nodes[0].converttopsbt(rawtx['hex'])
				self.nodes[0].decodepsbt(new_psbt)
				'''
				# Make sure that a psbt with signatures cannot be converted
				signedtx = self.nodes[0].signrawtransactionwithwallet(rawtx['hex'])
				assert_raises_rpc_error(-22, "TX decode failed",
				self.nodes[0].converttopsbt, signedtx['hex'])
				'''
				FabienUnsubmitted Not Done Inline Actions Is this test failing ? Fabien: Is this test failing ?
				deadalnixAuthorUnsubmitted Done Inline Actions ha yes, this is failing due to some segwit stuff, but we don't have segwit. Removing. deadalnix: ha yes, this is failing due to some segwit stuff, but we don't have segwit. Removing.
				# Explicilty allow converting non-empty txs
				new_psbt = self.nodes[0].converttopsbt(rawtx['hex'])
				self.nodes[0].decodepsbt(new_psbt)

				# Create outputs to nodes 1 and 2
				node1_addr = self.nodes[1].getnewaddress()
				node2_addr = self.nodes[2].getnewaddress()
				txid1 = self.nodes[0].sendtoaddress(node1_addr, 13)
				txid2 = self.nodes[0].sendtoaddress(node2_addr, 13)
				self.nodes[0].generate(6)
				self.sync_all()
				vout1 = find_output(self.nodes[1], txid1, 13)
				vout2 = find_output(self.nodes[2], txid2, 13)

				# Create a psbt spending outputs from nodes 1 and 2
				psbt_orig = self.nodes[0].createpsbt([{"txid": txid1, "vout": vout1}, {
				"txid": txid2, "vout": vout2}], {self.nodes[0].getnewaddress(): 25.999})

				# Update psbts, should only have data for one input and not the other
				psbt1 = self.nodes[1].walletprocesspsbt(psbt_orig)['psbt']
				psbt1_decoded = self.nodes[0].decodepsbt(psbt1)
				assert psbt1_decoded['inputs'][0] and not psbt1_decoded['inputs'][1]
				psbt2 = self.nodes[2].walletprocesspsbt(psbt_orig)['psbt']
				psbt2_decoded = self.nodes[0].decodepsbt(psbt2)
				assert not psbt2_decoded['inputs'][0] and psbt2_decoded['inputs'][1]

				# Combine, finalize, and send the psbts
				combined = self.nodes[0].combinepsbt([psbt1, psbt2])
				finalized = self.nodes[0].finalizepsbt(combined)['hex']
				self.nodes[0].sendrawtransaction(finalized)
				self.nodes[0].generate(6)
				self.sync_all()

				# BIP 174 Test Vectors

				# Check that unknown values are just passed through
				unknown_psbt = "cHNidP8BAD8CAAAAAf//////////////////////////////////////////AAAAAAD/////AQAAAAAAAAAAA2oBAAAAAAAACg8BAgMEBQYHCAkPAQIDBAUGBwgJCgsMDQ4PAAA="
				unknown_out = self.nodes[0].walletprocesspsbt(unknown_psbt)['psbt']
				assert_equal(unknown_psbt, unknown_out)

				# Open the data file
				with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'data/rpc_psbt.json'), encoding='utf-8') as f:
				d = json.load(f)
				invalids = d['invalid']
				valids = d['valid']
				creators = d['creator']
				signers = d['signer']
				combiners = d['combiner']
				finalizers = d['finalizer']
				extractors = d['extractor']

				# Invalid PSBTs
				for invalid in invalids:
				assert_raises_rpc_error(-22, "TX decode failed",
				self.nodes[0].decodepsbt, invalid)

				# Valid PSBTs
				for valid in valids:
				self.nodes[0].decodepsbt(valid)

				# Creator Tests
				for creator in creators:
				created_tx = self.nodes[0].createpsbt(
				creator['inputs'], creator['outputs'])
				assert_equal(created_tx, creator['result'])

				# Signer tests
				for i, signer in enumerate(signers):
				for key in signer['privkeys']:
				self.nodes[i].importprivkey(key)
				signed_tx = self.nodes[i].walletprocesspsbt(signer['psbt'])['psbt']
				assert_equal(signed_tx, signer['result'])

				# Combiner test
				for combiner in combiners:
				combined = self.nodes[2].combinepsbt(combiner['combine'])
				assert_equal(combined, combiner['result'])

				# Finalizer test
				for finalizer in finalizers:
				finalized = self.nodes[2].finalizepsbt(
				finalizer['finalize'], False)['psbt']
				assert_equal(finalized, finalizer['result'])

				# Extractor test
				for extractor in extractors:
				extracted = self.nodes[2].finalizepsbt(
				extractor['extract'], True)['hex']
				assert_equal(extracted, extractor['result'])


				if __name__ == '__main__':
				PSBTTest().main()