files: #!/usr/bin/env python3
  # Copyright (c) 2014-2020 The Bitcoin Core developers
  # Distributed under the MIT software license, see the accompanying
  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  """Run regression test suite.
  
  This module calls down into individual test cases via subprocess. It will
  forward all unrecognized arguments onto the individual test scripts.
  
  For a description of arguments recognized by test scripts, see
  `test/functional/test_framework/test_framework.py:BitcoinTestFramework.main`.
  
  """
  
  import argparse
  from collections import deque
  import configparser
  import datetime
  import os
  import time
  import shutil
  import subprocess
  import sys
  import tempfile
  import re
  import logging
  import unittest
  
  # Formatting. Default colors to empty strings.
  BOLD, GREEN, RED, GREY = ("", ""), ("", ""), ("", ""), ("", "")
  try:
      # Make sure python thinks it can write unicode to its stdout
      "\u2713".encode("utf_8").decode(sys.stdout.encoding)
      TICK = "‚úì "
      CROSS = "‚úñ "
      CIRCLE = "‚óã "
  except UnicodeDecodeError:
      TICK = "P "
      CROSS = "x "
      CIRCLE = "o "
  
  if os.name != 'nt' or sys.getwindowsversion() >= (10, 0, 14393):
      if os.name == 'nt':
          import ctypes
          kernel32 = ctypes.windll.kernel32  # type: ignore
          ENABLE_VIRTUAL_TERMINAL_PROCESSING = 4
          STD_OUTPUT_HANDLE = -11
          STD_ERROR_HANDLE = -12
          # Enable ascii color control to stdout
          stdout = kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
          stdout_mode = ctypes.c_int32()
          kernel32.GetConsoleMode(stdout, ctypes.byref(stdout_mode))
          kernel32.SetConsoleMode(stdout, stdout_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING)
          # Enable ascii color control to stderr
          stderr = kernel32.GetStdHandle(STD_ERROR_HANDLE)
          stderr_mode = ctypes.c_int32()
          kernel32.GetConsoleMode(stderr, ctypes.byref(stderr_mode))
          kernel32.SetConsoleMode(stderr, stderr_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING)
      # primitive formatting on supported
      # terminal via ANSI escape sequences:
      BOLD = ('\033[0m', '\033[1m')
      GREEN = ('\033[0m', '\033[0;32m')
      RED = ('\033[0m', '\033[0;31m')
      GREY = ('\033[0m', '\033[1;30m')
  
  TEST_EXIT_PASSED = 0
  TEST_EXIT_SKIPPED = 77
  
  TEST_FRAMEWORK_MODULES = [
      "address",
      "blocktools",
      "muhash",
      "key",
      "script",
      "segwit_addr",
      "util",
  ]
  
  EXTENDED_SCRIPTS = [
      # These tests are not run by default.
      # Longest test should go first, to favor running tests in parallel
      'feature_pruning.py',
      'feature_dbcrash.py',
  ]
  
  BASE_SCRIPTS = [
      # Scripts that are run by default.
      # Longest test should go first, to favor running tests in parallel
      'wallet_hd.py --legacy-wallet',
      'wallet_hd.py --descriptors',
      'wallet_backup.py --legacy-wallet',
      'wallet_backup.py --descriptors',
      # vv Tests less than 5m vv
      'mining_getblocktemplate_longpoll.py',
      'feature_maxuploadtarget.py',
      'feature_block.py',
      'rpc_fundrawtransaction.py --legacy-wallet',
      'rpc_fundrawtransaction.py --descriptors',
      'p2p_compactblocks.py',
      'feature_segwit.py --legacy-wallet',
      # vv Tests less than 2m vv
      'wallet_basic.py --legacy-wallet',
      'wallet_basic.py --descriptors',
      'wallet_labels.py --legacy-wallet',
      'wallet_labels.py --descriptors',
      'p2p_segwit.py',
      'p2p_timeouts.py',
      'p2p_tx_download.py',
      'mempool_updatefromblock.py',
      'wallet_dump.py --legacy-wallet',
      'wallet_listtransactions.py --legacy-wallet',
      'wallet_listtransactions.py --descriptors',
      'feature_taproot.py',
      # vv Tests less than 60s vv
      'p2p_sendheaders.py',
      'wallet_importmulti.py --legacy-wallet',
      'mempool_limit.py',
      'rpc_txoutproof.py',
      'wallet_listreceivedby.py --legacy-wallet',
      'wallet_listreceivedby.py --descriptors',
      'wallet_abandonconflict.py --legacy-wallet',
      'wallet_abandonconflict.py --descriptors',
      'feature_csv_activation.py',
      'rpc_rawtransaction.py --legacy-wallet',
      'rpc_rawtransaction.py --descriptors',
      'wallet_address_types.py --legacy-wallet',
      'wallet_address_types.py --descriptors',
      'feature_bip68_sequence.py',
      'p2p_feefilter.py',
      'feature_reindex.py',
      'feature_abortnode.py',
      # vv Tests less than 30s vv
      'wallet_keypool_topup.py --legacy-wallet',
      'wallet_keypool_topup.py --descriptors',
      'feature_fee_estimation.py',
      'interface_zmq.py',
      'rpc_invalid_address_message.py',
      'interface_bitcoin_cli.py',
      'mempool_resurrect.py',
      'wallet_txn_doublespend.py --mineblock',
      'tool_wallet.py --legacy-wallet',
      'tool_wallet.py --descriptors',
      'wallet_txn_clone.py',
      'wallet_txn_clone.py --segwit',
      'rpc_getchaintips.py',
      'rpc_misc.py',
      'interface_rest.py',
      'mempool_spend_coinbase.py',
      'wallet_avoidreuse.py --legacy-wallet',
      'wallet_avoidreuse.py --descriptors',
      'mempool_reorg.py',
      'mempool_persist.py',
      'wallet_multiwallet.py --legacy-wallet',
      'wallet_multiwallet.py --descriptors',
      'wallet_multiwallet.py --usecli',
      'wallet_createwallet.py --legacy-wallet',
      'wallet_createwallet.py --usecli',
      'wallet_createwallet.py --descriptors',
      'wallet_watchonly.py --legacy-wallet',
      'wallet_watchonly.py --usecli --legacy-wallet',
      'wallet_reorgsrestore.py',
      'interface_http.py',
      'interface_rpc.py',
      'rpc_psbt.py --legacy-wallet',
      'rpc_psbt.py --descriptors',
      'rpc_users.py',
      'rpc_whitelist.py',
      'feature_proxy.py',
      'rpc_signrawtransaction.py --legacy-wallet',
      'rpc_signrawtransaction.py --descriptors',
      'wallet_groups.py --legacy-wallet',
      'p2p_addrv2_relay.py',
      'wallet_groups.py --descriptors',
      'p2p_disconnect_ban.py',
      'rpc_decodescript.py',
      'rpc_blockchain.py',
      'rpc_deprecated.py',
      'wallet_disable.py --legacy-wallet',
      'wallet_disable.py --descriptors',
      'p2p_addr_relay.py',
      'p2p_getaddr_caching.py',
      'p2p_getdata.py',
      'rpc_net.py',
      'wallet_keypool.py --legacy-wallet',
      'wallet_keypool.py --descriptors',
      'wallet_descriptor.py --descriptors',
      'p2p_nobloomfilter_messages.py',
      'p2p_filter.py',
      'rpc_setban.py',
      'p2p_blocksonly.py',
      'mining_prioritisetransaction.py',
      'p2p_invalid_locator.py',
      'p2p_invalid_block.py',
      'p2p_invalid_messages.py',
      'p2p_invalid_tx.py
  
  [... the rest is omitted ...]
files: test/functional/test_runner.py
files: #!/usr/bin/env python3
  # Copyright (c) 2014-2020 The Bitcoin Core developers
  # Distributed under the MIT software license, see the accompanying
  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  """Run regression test suite.
  
  This module calls down into individual test cases via subprocess. It will
  forward all unrecognized arguments onto the individual test scripts.
  
  For a description of arguments recognized by test scripts, see
  `test/functional/test_framework/test_framework.py:BitcoinTestFramework.main`.
  
  """
  
  import argparse
  from collections import deque
  import configparser
  import datetime
  import os
  import time
  import shutil
  import subprocess
  import sys
  import tempfile
  import re
  import logging
  import unittest
  
  # Formatting. Default colors to empty strings.
  BOLD, GREEN, RED, GREY = ("", ""), ("", ""), ("", ""), ("", "")
  try:
      # Make sure python thinks it can write unicode to its stdout
      "\u2713".encode("utf_8").decode(sys.stdout.encoding)
      TICK = "‚úì "
      CROSS = "‚úñ "
      CIRCLE = "‚óã "
  except UnicodeDecodeError:
      TICK = "P "
      CROSS = "x "
      CIRCLE = "o "
  
  if os.name != 'nt' or sys.getwindowsversion() >= (10, 0, 14393):
      if os.name == 'nt':
          import ctypes
          kernel32 = ctypes.windll.kernel32  # type: ignore
          ENABLE_VIRTUAL_TERMINAL_PROCESSING = 4
          STD_OUTPUT_HANDLE = -11
          STD_ERROR_HANDLE = -12
          # Enable ascii color control to stdout
          stdout = kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
          stdout_mode = ctypes.c_int32()
          kernel32.GetConsoleMode(stdout, ctypes.byref(stdout_mode))
          kernel32.SetConsoleMode(stdout, stdout_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING)
          # Enable ascii color control to stderr
          stderr = kernel32.GetStdHandle(STD_ERROR_HANDLE)
          stderr_mode = ctypes.c_int32()
          kernel32.GetConsoleMode(stderr, ctypes.byref(stderr_mode))
          kernel32.SetConsoleMode(stderr, stderr_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING)
      # primitive formatting on supported
      # terminal via ANSI escape sequences:
      BOLD = ('\033[0m', '\033[1m')
      GREEN = ('\033[0m', '\033[0;32m')
      RED = ('\033[0m', '\033[0;31m')
      GREY = ('\033[0m', '\033[1;30m')
  
  TEST_EXIT_PASSED = 0
  TEST_EXIT_SKIPPED = 77
  
  TEST_FRAMEWORK_MODULES = [
      "address",
      "blocktools",
      "muhash",
      "key",
      "script",
      "segwit_addr",
      "util",
  ]
  
  EXTENDED_SCRIPTS = [
      # These tests are not run by default.
      # Longest test should go first, to favor running tests in parallel
      'feature_pruning.py',
      'feature_dbcrash.py',
  ]
  
  BASE_SCRIPTS = [
      # Scripts that are run by default.
      # Longest test should go first, to favor running tests in parallel
      'wallet_hd.py --legacy-wallet',
      'wallet_hd.py --descriptors',
      'wallet_backup.py --legacy-wallet',
      'wallet_backup.py --descriptors',
      # vv Tests less than 5m vv
      'mining_getblocktemplate_longpoll.py',
      'feature_maxuploadtarget.py',
      'feature_block.py',
      'rpc_fundrawtransaction.py --legacy-wallet',
      'rpc_fundrawtransaction.py --descriptors',
      'p2p_compactblocks.py',
      'feature_segwit.py --legacy-wallet',
      # vv Tests less than 2m vv
      'wallet_basic.py --legacy-wallet',
      'wallet_basic.py --descriptors',
      'wallet_labels.py --legacy-wallet',
      'wallet_labels.py --descriptors',
      'p2p_segwit.py',
      'p2p_timeouts.py',
      'p2p_tx_download.py',
      'mempool_updatefromblock.py',
      'wallet_dump.py --legacy-wallet',
      'wallet_listtransactions.py --legacy-wallet',
      'wallet_listtransactions.py --descriptors',
      'feature_taproot.py',
      # vv Tests less than 60s vv
      'p2p_sendheaders.py',
      'wallet_importmulti.py --legacy-wallet',
      'mempool_limit.py',
      'rpc_txoutproof.py',
      'wallet_listreceivedby.py --legacy-wallet',
      'wallet_listreceivedby.py --descriptors',
      'wallet_abandonconflict.py --legacy-wallet',
      'wallet_abandonconflict.py --descriptors',
      'feature_csv_activation.py',
      'rpc_rawtransaction.py --legacy-wallet',
      'rpc_rawtransaction.py --descriptors',
      'wallet_address_types.py --legacy-wallet',
      'wallet_address_types.py --descriptors',
      'feature_bip68_sequence.py',
      'p2p_feefilter.py',
      'feature_reindex.py',
      'feature_abortnode.py',
      # vv Tests less than 30s vv
      'wallet_keypool_topup.py --legacy-wallet',
      'wallet_keypool_topup.py --descriptors',
      'feature_fee_estimation.py',
      'interface_zmq.py',
      'rpc_invalid_address_message.py',
      'interface_bitcoin_cli.py',
      'mempool_resurrect.py',
      'wallet_txn_doublespend.py --mineblock',
      'tool_wallet.py --legacy-wallet',
      'tool_wallet.py --descriptors',
      'wallet_txn_clone.py',
      'wallet_txn_clone.py --segwit',
      'rpc_getchaintips.py',
      'rpc_misc.py',
      'interface_rest.py',
      'mempool_spend_coinbase.py',
      'wallet_avoidreuse.py --legacy-wallet',
      'wallet_avoidreuse.py --descriptors',
      'mempool_reorg.py',
      'mempool_persist.py',
      'wallet_multiwallet.py --legacy-wallet',
      'wallet_multiwallet.py --descriptors',
      'wallet_multiwallet.py --usecli',
      'wallet_createwallet.py --legacy-wallet',
      'wallet_createwallet.py --usecli',
      'wallet_createwallet.py --descriptors',
      'wallet_watchonly.py --legacy-wallet',
      'wallet_watchonly.py --usecli --legacy-wallet',
      'wallet_reorgsrestore.py',
      'interface_http.py',
      'interface_rpc.py',
      'rpc_psbt.py --legacy-wallet',
      'rpc_psbt.py --descriptors',
      'rpc_users.py',
      'rpc_whitelist.py',
      'feature_proxy.py',
      'rpc_signrawtransaction.py --legacy-wallet',
      'rpc_signrawtransaction.py --descriptors',
      'wallet_groups.py --legacy-wallet',
      'p2p_addrv2_relay.py',
      'wallet_groups.py --descriptors',
      'p2p_disconnect_ban.py',
      'rpc_decodescript.py',
      'rpc_blockchain.py',
      'rpc_deprecated.py',
      'wallet_disable.py --legacy-wallet',
      'wallet_disable.py --descriptors',
      'p2p_addr_relay.py',
      'p2p_getaddr_caching.py',
      'p2p_getdata.py',
      'rpc_net.py',
      'wallet_keypool.py --legacy-wallet',
      'wallet_keypool.py --descriptors',
      'wallet_descriptor.py --descriptors',
      'p2p_nobloomfilter_messages.py',
      'p2p_filter.py',
      'rpc_setban.py',
      'p2p_blocksonly.py',
      'mining_prioritisetransaction.py',
      'p2p_invalid_locator.py',
      'p2p_invalid_block.py',
      'p2p_invalid_messages.py',
      'p2p_invalid_tx.py
  
  [... the rest is omitted ...]
files: test/functional/test_runner.py
files: #!/usr/bin/env python3
  # Copyright (c) 2020 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 send RPC command."""
  
  from decimal import Decimal, getcontext
  from itertools import product
  
  from test_framework.authproxy import JSONRPCException
  from test_framework.descriptors import descsum_create
  from test_framework.test_framework import BitcoinTestFramework
  from test_framework.util import (
      assert_equal,
      assert_fee_amount,
      assert_greater_than,
      assert_raises_rpc_error,
  )
  
  class WalletSendTest(BitcoinTestFramework):
      def set_test_params(self):
          self.num_nodes = 2
          # whitelist all peers to speed up tx relay / mempool sync
          self.extra_args = [
              ["-whitelist=127.0.0.1","-walletrbf=1"],
              ["-whitelist=127.0.0.1","-walletrbf=1"],
          ]
          getcontext().prec = 8 # Satoshi precision for Decimal
  
      def skip_test_if_missing_module(self):
          self.skip_if_no_wallet()
  
      def test_send(self, from_wallet, to_wallet=None, amount=None, data=None,
                    arg_conf_target=None, arg_estimate_mode=None, arg_fee_rate=None,
                    conf_target=None, estimate_mode=None, fee_rate=None, add_to_wallet=None, psbt=None,
                    inputs=None, add_inputs=None, change_address=None, change_position=None, change_type=None,
                    include_watching=None, locktime=None, lock_unspents=None, replaceable=None, subtract_fee_from_outputs=None,
                    expect_error=None):
          assert (amount is None) != (data is None)
  
          from_balance_before = from_wallet.getbalance()
          if to_wallet is None:
              assert amount is None
          else:
              to_untrusted_pending_before = to_wallet.getbalances()["mine"]["untrusted_pending"]
  
          if amount:
              dest = to_wallet.getnewaddress()
              outputs = {dest: amount}
          else:
              outputs = {"data": data}
  
          # Construct options dictionary
          options = {}
          if add_to_wallet is not None:
              options["add_to_wallet"] = add_to_wallet
          else:
              if psbt:
                  add_to_wallet = False
              else:
                  add_to_wallet = from_wallet.getwalletinfo()["private_keys_enabled"] # Default value
          if psbt is not None:
              options["psbt"] = psbt
          if conf_target is not None:
              options["conf_target"] = conf_target
          if estimate_mode is not None:
              options["estimate_mode"] = estimate_mode
          if fee_rate is not None:
              options["fee_rate"] = fee_rate
          if inputs is not None:
              options["inputs"] = inputs
          if add_inputs is not None:
              options["add_inputs"] = add_inputs
          if change_address is not None:
              options["change_address"] = change_address
          if change_position is not None:
              options["change_position"] = change_position
          if change_type is not None:
              options["change_type"] = change_type
          if include_watching is not None:
              options["include_watching"] = include_watching
          if locktime is not None:
              options["locktime"] = locktime
          if lock_unspents is not None:
              options["lock_unspents"] = lock_unspents
          if replaceable is None:
              replaceable = True # default
          else:
              options["replaceable"] = replaceable
          if subtract_fee_from_outputs is not None:
              options["subtract_fee_from_outputs"] = subtract_fee_from_outputs
  
          if len(options.keys()) == 0:
              options = None
  
          if expect_error is None:
              res = from_wallet.send(outputs=outputs, conf_target=arg_conf_target, estimate_mode=arg_estimate_mode, fee_rate=arg_fee_rate, options=options)
          else:
              try:
                  assert_raises_rpc_error(expect_error[0], expect_error[1], from_wallet.send,
                      outputs=outputs, conf_target=arg_conf_target, estimate_mode=arg_estimate_mode, fee_rate=arg_fee_rate, options=options)
              except AssertionError:
                  # Provide debug info if the test fails
                  self.log.error("Unexpected successful result:")
                  self.log.error(arg_conf_target)
                  self.log.error(arg_estimate_mode)
                  self.log.error(arg_fee_rate)
                  self.log.error(options)
                  res = from_wallet.send(outputs=outputs, conf_target=arg_conf_target, estimate_mode=arg_estimate_mode, fee_rate=arg_fee_rate, options=options)
                  self.log.error(res)
                  if "txid" in res and add_to_wallet:
                      self.log.error("Transaction details:")
                      try:
                          tx = from_wallet.gettransaction(res["txid"])
                          self.log.error(tx)
                          self.log.error("testmempoolaccept (transaction may already be in mempool):")
                          self.log.error(from_wallet.testmempoolaccept([tx["hex"]]))
                      except JSONRPCException as exc:
                          self.log.error(exc)
  
                  raise
  
              return
  
          if locktime:
              return res
  
          if from_wallet.getwalletinfo()["private_keys_enabled"] and not include_watching:
              assert_equal(res["complete"], True)
              assert "txid" in res
          else:
              assert_equal(res["complete"], False)
              assert not "txid" in res
              assert "psbt" in res
  
          if add_to_wallet and not include_watching:
              # Ensure transaction exists in the wallet:
              tx = from_wallet.gettransaction(res["txid"])
              assert tx
              assert_equal(tx["bip125-replaceable"], "yes" if replaceable else "no")
              # Ensure transaction exists in the mempool:
              tx = from_wallet.getrawtransaction(res["txid"], True)
              assert tx
              if amount:
                  if subtract_fee_from_outputs:
                      assert_equal(from_balance_before - from_wallet.getbalance(), amount)
                  else:
                      assert_greater_than(from_balance_before - from_wallet.getbalance(), amount)
              else:
                  assert next((out for out in tx["vout"] if out["scriptPubKey"]["asm"] == "OP_RETURN 35"), None)
          else:
              assert_equal(from_balance_before, from_wallet.getbalance())
  
          if to_wallet:
              self.sync_mempools()
              if add_to_wallet:
                  if not subtract_fee_from_outputs:
                      assert_equal(to_wallet.getbalances()["mine"]["untrusted_pending"], to_untrusted_pending_before + Decimal(amount if amount else 0))
              else:
                  assert_equal(to_wallet.getbalances()["mine"]["untrusted_pending"], to_untrusted_pending_before)
  
          return res
  
      def run_test(self):
          self.log.info("Setup wallets...")
          # w0 is a wallet with coinbase rewards
          w0 = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
          # w1 is a regular wallet
          self.nodes[1].createwallet(wallet_name="w1")
          w1 = self.nodes[1].get_wallet_rpc("w1")
          # w2 contains the private keys for w3
          self.nodes[1].createwallet(wallet_name="w2", blank=True)
          w2 = self.nodes[1].get_wallet_rpc("w2")
          xpriv = "tprv8ZgxMBicQKsPfHCsTwkiM1KT56RXbGGTqvc2hgqzycpwbHqqpcajQeMRZoBD35kW4RtyCemu6j34Ku5DEspmgjKdt2qe4SvRch5Kk8B8A2v"
          xpub = "tpubD6NzVbkrYhZ4YkEfMbRJkQyZe7wTkbTNRECozCtJPtdLRn6cT1QKb8yHjwAPcAr26eHBFYs5iLiFFnCbwPRsncCKUKCfubHDMGKzMVcN1Jg"
          if self.options.descriptors:
              w2.importdescriptors([{
                  "desc": descsum_create("wpkh(" + xpriv + "/0/0/*)"),
                  "timestamp": "now",
                  "range": [0, 100],
                  "active": True
              },{
                  "desc": descsum_create("wpkh(" + xpriv + "/0/1/*)"),
                  "timestamp": "now",
                  "range": [0, 100],
                  "active": True,
                  "internal": True
              }])
          else:
              w2
  
  [... the rest is omitted ...]
files: test/functional/wallet_send.py
files: #!/usr/bin/env python3
  # Copyright (c) 2020 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 send RPC command."""
  
  from decimal import Decimal, getcontext
  from itertools import product
  
  from test_framework.authproxy import JSONRPCException
  from test_framework.test_framework import BitcoinTestFramework
  from test_framework.util import (
      assert_equal,
      assert_fee_amount,
      assert_greater_than,
      assert_raises_rpc_error,
  )
  
  class WalletSendTest(BitcoinTestFramework):
      def set_test_params(self):
          self.num_nodes = 2
          # whitelist all peers to speed up tx relay / mempool sync
          self.extra_args = [
              ["-whitelist=127.0.0.1","-walletrbf=1"],
              ["-whitelist=127.0.0.1","-walletrbf=1"],
          ]
          getcontext().prec = 8 # Satoshi precision for Decimal
  
      def skip_test_if_missing_module(self):
          self.skip_if_no_wallet()
  
      def test_send(self, from_wallet, to_wallet=None, amount=None, data=None,
                    arg_conf_target=None, arg_estimate_mode=None, arg_fee_rate=None,
                    conf_target=None, estimate_mode=None, fee_rate=None, add_to_wallet=None, psbt=None,
                    inputs=None, add_inputs=None, change_address=None, change_position=None, change_type=None,
                    include_watching=None, locktime=None, lock_unspents=None, replaceable=None, subtract_fee_from_outputs=None,
                    expect_error=None):
          assert (amount is None) != (data is None)
  
          from_balance_before = from_wallet.getbalance()
          if to_wallet is None:
              assert amount is None
          else:
              to_untrusted_pending_before = to_wallet.getbalances()["mine"]["untrusted_pending"]
  
          if amount:
              dest = to_wallet.getnewaddress()
              outputs = {dest: amount}
          else:
              outputs = {"data": data}
  
          # Construct options dictionary
          options = {}
          if add_to_wallet is not None:
              options["add_to_wallet"] = add_to_wallet
          else:
              if psbt:
                  add_to_wallet = False
              else:
                  add_to_wallet = from_wallet.getwalletinfo()["private_keys_enabled"] # Default value
          if psbt is not None:
              options["psbt"] = psbt
          if conf_target is not None:
              options["conf_target"] = conf_target
          if estimate_mode is not None:
              options["estimate_mode"] = estimate_mode
          if fee_rate is not None:
              options["fee_rate"] = fee_rate
          if inputs is not None:
              options["inputs"] = inputs
          if add_inputs is not None:
              options["add_inputs"] = add_inputs
          if change_address is not None:
              options["change_address"] = change_address
          if change_position is not None:
              options["change_position"] = change_position
          if change_type is not None:
              options["change_type"] = change_type
          if include_watching is not None:
              options["include_watching"] = include_watching
          if locktime is not None:
              options["locktime"] = locktime
          if lock_unspents is not None:
              options["lock_unspents"] = lock_unspents
          if replaceable is None:
              replaceable = True # default
          else:
              options["replaceable"] = replaceable
          if subtract_fee_from_outputs is not None:
              options["subtract_fee_from_outputs"] = subtract_fee_from_outputs
  
          if len(options.keys()) == 0:
              options = None
  
          if expect_error is None:
              res = from_wallet.send(outputs=outputs, conf_target=arg_conf_target, estimate_mode=arg_estimate_mode, fee_rate=arg_fee_rate, options=options)
          else:
              try:
                  assert_raises_rpc_error(expect_error[0], expect_error[1], from_wallet.send,
                      outputs=outputs, conf_target=arg_conf_target, estimate_mode=arg_estimate_mode, fee_rate=arg_fee_rate, options=options)
              except AssertionError:
                  # Provide debug info if the test fails
                  self.log.error("Unexpected successful result:")
                  self.log.error(arg_conf_target)
                  self.log.error(arg_estimate_mode)
                  self.log.error(arg_fee_rate)
                  self.log.error(options)
                  res = from_wallet.send(outputs=outputs, conf_target=arg_conf_target, estimate_mode=arg_estimate_mode, fee_rate=arg_fee_rate, options=options)
                  self.log.error(res)
                  if "txid" in res and add_to_wallet:
                      self.log.error("Transaction details:")
                      try:
                          tx = from_wallet.gettransaction(res["txid"])
                          self.log.error(tx)
                          self.log.error("testmempoolaccept (transaction may already be in mempool):")
                          self.log.error(from_wallet.testmempoolaccept([tx["hex"]]))
                      except JSONRPCException as exc:
                          self.log.error(exc)
  
                  raise
  
              return
  
          if locktime:
              return res
  
          if from_wallet.getwalletinfo()["private_keys_enabled"] and not include_watching:
              assert_equal(res["complete"], True)
              assert "txid" in res
          else:
              assert_equal(res["complete"], False)
              assert not "txid" in res
              assert "psbt" in res
  
          if add_to_wallet and not include_watching:
              # Ensure transaction exists in the wallet:
              tx = from_wallet.gettransaction(res["txid"])
              assert tx
              assert_equal(tx["bip125-replaceable"], "yes" if replaceable else "no")
              # Ensure transaction exists in the mempool:
              tx = from_wallet.getrawtransaction(res["txid"], True)
              assert tx
              if amount:
                  if subtract_fee_from_outputs:
                      assert_equal(from_balance_before - from_wallet.getbalance(), amount)
                  else:
                      assert_greater_than(from_balance_before - from_wallet.getbalance(), amount)
              else:
                  assert next((out for out in tx["vout"] if out["scriptPubKey"]["asm"] == "OP_RETURN 35"), None)
          else:
              assert_equal(from_balance_before, from_wallet.getbalance())
  
          if to_wallet:
              self.sync_mempools()
              if add_to_wallet:
                  if not subtract_fee_from_outputs:
                      assert_equal(to_wallet.getbalances()["mine"]["untrusted_pending"], to_untrusted_pending_before + Decimal(amount if amount else 0))
              else:
                  assert_equal(to_wallet.getbalances()["mine"]["untrusted_pending"], to_untrusted_pending_before)
  
          return res
  
      def run_test(self):
          self.log.info("Setup wallets...")
          # w0 is a wallet with coinbase rewards
          w0 = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
          # w1 is a regular wallet
          self.nodes[1].createwallet(wallet_name="w1")
          w1 = self.nodes[1].get_wallet_rpc("w1")
          # w2 contains the private keys for w3
          self.nodes[1].createwallet(wallet_name="w2")
          w2 = self.nodes[1].get_wallet_rpc("w2")
          # w3 is a watch-only wallet, based on w2
          self.nodes[1].createwallet(wallet_name="w3", disable_private_keys=True)
          w3 = self.nodes[1].get_wallet_rpc("w3")
          for _ in range(3):
              a2_receive = w2.getnewaddress()
              a2_change = w2.getrawchangeaddress() # doesn't actually use change derivation
              res = w3.importmulti([{
                  "desc": w2.getaddressinfo(a2_receive)["desc"],
                  "timestamp": "now",
                  "keypool": True,
                  "watchonly": True
              },{
                  "desc": w2.getaddressinfo(a2_change)["desc"],
                  "timestamp": "now",
                  "keypool": True,
                  "internal": True,
                  "watchonly": True
              }])
              assert_equal(res, [{"success": True}, {"success": True}])
  
          w0.sendtoaddress(a2_receive, 10) # fund w3
          self.nodes[0].generate(1)
          self.sync_blocks()
  
          # w4 has private keys enabled, but only contains watch-only keys (from w2)
          self.nodes[1].createwallet(wallet_name="w4", disable_private_keys=False)
          w4 = self.nodes[1].get_wallet_rpc("w4")
         
  
  [... the rest is omitted ...]
files: test/functional/wallet_send.py