text: #!/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.
  """
  
  import json
  import os
  
  from test_framework.test_framework import BitcoinTestFramework
  from test_framework.util import (
      assert_equal,
      assert_raises_rpc_error,
      find_output,
  )
  
  # 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, p2pkh addresses
          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)
  
          # 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):
              self.nodes[2].createwallet("wallet{}".format(i))
              wrpc = self.nodes[2].get_wallet_rpc("wallet{}".format(i))
              for key in signer['privkeys']:
                  wrpc.importprivkey(key)
              signed_tx = wrpc.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()
  