text: #!/usr/bin/env python3
  # Copyright (c) 2022 The Bitcoin and Qogecoin Core Authors
  # Distributed under the MIT software license, see the accompanying
  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  
  """  Tests the coin_selection:* tracepoint API interface.
       See https://github.com/qogecoin/qogecoin/blob/master/doc/tracing.md#context-coin_selection
  """
  
  # Test will be skipped if we don't have bcc installed
  try:
      from bcc import BPF, USDT # type: ignore[import]
  except ImportError:
      pass
  from test_framework.test_framework import QogecoinTestFramework
  from test_framework.util import (
      assert_equal,
      assert_greater_than,
      assert_raises_rpc_error,
  )
  
  coinselection_tracepoints_program = """
  #include <uapi/linux/ptrace.h>
  
  #define WALLET_NAME_LENGTH 16
  #define ALGO_NAME_LENGTH 16
  
  struct event_data
  {
      u8 type;
      char wallet_name[WALLET_NAME_LENGTH];
  
      // selected coins event
      char algo[ALGO_NAME_LENGTH];
      s64 target;
      s64 waste;
      s64 selected_value;
  
      // create tx event
      bool success;
      s64 fee;
      s32 change_pos;
  
      // aps create tx event
      bool use_aps;
  };
  
  BPF_QUEUE(coin_selection_events, struct event_data, 1024);
  
  int trace_selected_coins(struct pt_regs *ctx) {
      struct event_data data;
      __builtin_memset(&data, 0, sizeof(data));
      data.type = 1;
      bpf_usdt_readarg_p(1, ctx, &data.wallet_name, WALLET_NAME_LENGTH);
      bpf_usdt_readarg_p(2, ctx, &data.algo, ALGO_NAME_LENGTH);
      bpf_usdt_readarg(3, ctx, &data.target);
      bpf_usdt_readarg(4, ctx, &data.waste);
      bpf_usdt_readarg(5, ctx, &data.selected_value);
      coin_selection_events.push(&data, 0);
      return 0;
  }
  
  int trace_normal_create_tx(struct pt_regs *ctx) {
      struct event_data data;
      __builtin_memset(&data, 0, sizeof(data));
      data.type = 2;
      bpf_usdt_readarg_p(1, ctx, &data.wallet_name, WALLET_NAME_LENGTH);
      bpf_usdt_readarg(2, ctx, &data.success);
      bpf_usdt_readarg(3, ctx, &data.fee);
      bpf_usdt_readarg(4, ctx, &data.change_pos);
      coin_selection_events.push(&data, 0);
      return 0;
  }
  
  int trace_attempt_aps(struct pt_regs *ctx) {
      struct event_data data;
      __builtin_memset(&data, 0, sizeof(data));
      data.type = 3;
      bpf_usdt_readarg_p(1, ctx, &data.wallet_name, WALLET_NAME_LENGTH);
      coin_selection_events.push(&data, 0);
      return 0;
  }
  
  int trace_aps_create_tx(struct pt_regs *ctx) {
      struct event_data data;
      __builtin_memset(&data, 0, sizeof(data));
      data.type = 4;
      bpf_usdt_readarg_p(1, ctx, &data.wallet_name, WALLET_NAME_LENGTH);
      bpf_usdt_readarg(2, ctx, &data.use_aps);
      bpf_usdt_readarg(3, ctx, &data.success);
      bpf_usdt_readarg(4, ctx, &data.fee);
      bpf_usdt_readarg(5, ctx, &data.change_pos);
      coin_selection_events.push(&data, 0);
      return 0;
  }
  """
  
  
  class CoinSelectionTracepointTest(QogecoinTestFramework):
      def set_test_params(self):
          self.num_nodes = 1
          self.setup_clean_chain = True
  
      def skip_test_if_missing_module(self):
          self.skip_if_platform_not_linux()
          self.skip_if_no_qogecoind_tracepoints()
          self.skip_if_no_python_bcc()
          self.skip_if_no_bpf_permissions()
          self.skip_if_no_wallet()
  
      def get_tracepoints(self, expected_types):
          events = []
          try:
              for i in range(0, len(expected_types) + 1):
                  event = self.bpf["coin_selection_events"].pop()
                  assert_equal(event.wallet_name.decode(), self.default_wallet_name)
                  assert_equal(event.type, expected_types[i])
                  events.append(event)
              else:
                  # If the loop exits successfully instead of throwing a KeyError, then we have had
                  # more events than expected. There should be no more than len(expected_types) events.
                  assert False
          except KeyError:
              assert_equal(len(events), len(expected_types))
              return events
  
  
      def determine_selection_from_usdt(self, events):
          success = None
          use_aps = None
          algo = None
          waste = None
          change_pos = None
  
          is_aps = False
          sc_events = []
          for event in events:
              if event.type == 1:
                  if not is_aps:
                      algo = event.algo.decode()
                      waste = event.waste
                  sc_events.append(event)
              elif event.type == 2:
                  success = event.success
                  if not is_aps:
                      change_pos = event.change_pos
              elif event.type == 3:
                  is_aps = True
              elif event.type == 4:
                  assert is_aps
                  if event.use_aps:
                      use_aps = True
                      assert_equal(len(sc_events), 2)
                      algo = sc_events[1].algo.decode()
                      waste = sc_events[1].waste
                      change_pos = event.change_pos
          return success, use_aps, algo, waste, change_pos
  
      def run_test(self):
          self.log.info("hook into the coin_selection tracepoints")
          ctx = USDT(pid=self.nodes[0].process.pid)
          ctx.enable_probe(probe="coin_selection:selected_coins", fn_name="trace_selected_coins")
          ctx.enable_probe(probe="coin_selection:normal_create_tx_internal", fn_name="trace_normal_create_tx")
          ctx.enable_probe(probe="coin_selection:attempting_aps_create_tx", fn_name="trace_attempt_aps")
          ctx.enable_probe(probe="coin_selection:aps_create_tx_internal", fn_name="trace_aps_create_tx")
          self.bpf = BPF(text=coinselection_tracepoints_program, usdt_contexts=[ctx], debug=0)
  
          self.log.info("Prepare wallets")
          self.generate(self.nodes[0], 101)
          wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
  
          self.log.info("Sending a transaction should result in all tracepoints")
          # We should have 5 tracepoints in the order:
          # 1. selected_coins (type 1)
          # 2. normal_create_tx_internal (type 2)
          # 3. attempting_aps_create_tx (type 3)
          # 4. selected_coins (type 1)
          # 5. aps_create_tx_internal (type 4)
          wallet.sendtoaddress(wallet.getnewaddress(), 10)
          events = self.get_tracepoints([1, 2, 3, 1, 4])
          success, use_aps, algo, waste, change_pos = self.determine_selection_from_usdt(events)
          assert_equal(success, True)
          assert_greater_than(change_pos, -1)
  
          self.log.info("Failing to fund results in 1 tracepoint")
          # We should have 1 tracepoints in the order
          # 1. normal_create_tx_internal (type 2)
          assert_raises_rpc_error(-6, "Insufficient funds", wallet.sendtoaddress, wallet.getnewaddress(), 102 * 50)
          events = self.get_tracepoints([2])
          success, use_aps, algo, waste, change_pos = self.determine_selection_from_usdt(events)
          assert_equal(success, False)
  
          self.log.info("Explicitly enabling APS results in 2 tracepoints")
          # We should have 2 tracepoints in the order
          # 1. selected_coins (type 1)
          # 2. normal_create_tx_internal (type 2)
          wallet.setwalletflag("avoid_reuse")
          wallet.sendtoaddress(address=wallet.getnewaddress(), amount=10, avoid_reuse=True)
          events = self.get_tracepoints([1, 2])
          success, use_aps, algo, waste, change_pos = self.determine_selection_from_usdt(events)
          assert_equal(success, True)
          assert_equal(use_aps, None)
  
          self.bpf.cleanup()
  
  
  if __name__ == '__main__':
      CoinSelectionTracepointTest().main()
  