1  #!/usr/bin/env python3
  2  # Copyright (c) 2018-present The Bitcoin Core developers
  3  # Distributed under the MIT software license, see the accompanying
  4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  5  """Test RPC help output."""
  6  
  7  from test_framework.test_framework import BitcoinTestFramework
  8  from test_framework.util import assert_equal, assert_raises_rpc_error
  9  
 10  from collections import defaultdict
 11  import os
 12  import re
 13  
 14  
 15  def parse_string(s):
 16      assert s[0] == '"'
 17      assert s[-1] == '"'
 18      return s[1:-1]
 19  
 20  def process_mapping(fname):
 21      """Find and parse conversion table in implementation file `fname`."""
 22      cmds = []
 23      string_params = []
 24      in_rpcs = False
 25      with open(fname, "r") as f:
 26          for line in f:
 27              line = line.rstrip()
 28              if not in_rpcs:
 29                  if re.match(r'static const CRPCConvertParam vRPCConvertParams\[] =', line):
 30                      in_rpcs = True
 31              else:
 32                  if line.startswith('};'):
 33                      in_rpcs = False
 34                  elif '{' in line and '"' in line:
 35                      # Match lines with ParamFormat::STRING
 36                      m_string = re.search(r'{ *("[^"]*") *, *([0-9]+) *, *("[^"]*") *, *ParamFormat::STRING *},?', line)
 37                      if m_string:
 38                          name = parse_string(m_string.group(1))
 39                          idx = int(m_string.group(2))
 40                          argname = parse_string(m_string.group(3))
 41                          string_params.append((name, idx, argname))
 42                          continue
 43  
 44                      # Match lines with ParamFormat::JSON and ParamFormat::JSON_OR_STRING
 45                      m_json = re.search(r'{ *("[^"]*") *, *([0-9]+) *, *("[^"]*") *(?:, *ParamFormat::(JSON_OR_STRING|JSON))? *},?', line)
 46                      if m_json:
 47                          name = parse_string(m_json.group(1))
 48                          idx = int(m_json.group(2))
 49                          argname = parse_string(m_json.group(3))
 50                          cmds.append((name, idx, argname))
 51  
 52      assert not in_rpcs
 53      return cmds, string_params
 54  
 55  class HelpRpcTest(BitcoinTestFramework):
 56      def set_test_params(self):
 57          self.num_nodes = 1
 58          self.uses_wallet = None
 59  
 60      def run_test(self):
 61          self.test_client_conversion_table()
 62          self.test_client_string_conversion_table()
 63          self.test_categories()
 64          self.dump_help()
 65          if self.is_wallet_compiled():
 66              self.wallet_help()
 67  
 68      def test_client_conversion_table(self):
 69          file_conversion_table = os.path.join(self.config["environment"]["SRCDIR"], 'src', 'rpc', 'client.cpp')
 70          mapping_client, _ = process_mapping(file_conversion_table)
 71          # Ignore echojson in client table
 72          mapping_client = [m for m in mapping_client if m[0] != 'echojson']
 73  
 74          mapping_server = self.nodes[0].help("dump_all_command_conversions")
 75          # Filter all RPCs whether they need conversion
 76          mapping_server_conversion = [tuple(m[:3]) for m in mapping_server if not m[3]]
 77  
 78          # Only check if all RPC methods have been compiled (i.e. wallet is enabled)
 79          if self.is_wallet_compiled() and sorted(mapping_client) != sorted(mapping_server_conversion):
 80              raise AssertionError("RPC client conversion table ({}) and RPC server named arguments mismatch!\n{}".format(
 81                  file_conversion_table,
 82                  set(mapping_client).symmetric_difference(mapping_server_conversion),
 83              ))
 84  
 85          # Check for conversion difference by argument name.
 86          # It is preferable for API consistency that arguments with the same name
 87          # have the same conversion, so bin by argument name.
 88          all_methods_by_argname = defaultdict(list)
 89          converts_by_argname = defaultdict(list)
 90          for m in mapping_server:
 91              all_methods_by_argname[m[2]].append(m[0])
 92              converts_by_argname[m[2]].append(m[3])
 93  
 94          for argname, convert in converts_by_argname.items():
 95              if all(convert) != any(convert):
 96                  # Only allow dummy and psbt to fail consistency check
 97                  assert argname in ['dummy', "psbt"], ('WARNING: conversion mismatch for argument named %s (%s)' % (argname, list(zip(all_methods_by_argname[argname], converts_by_argname[argname]))))
 98  
 99      def test_client_string_conversion_table(self):
100          file_conversion_table = os.path.join(self.config["environment"]["SRCDIR"], 'src', 'rpc', 'client.cpp')
101          _, string_params_client = process_mapping(file_conversion_table)
102          mapping_server = self.nodes[0].help("dump_all_command_conversions")
103          server_tuples = {tuple(m[:3]) for m in mapping_server}
104  
105          # Filter string parameters based on wallet compilation status
106          if self.is_wallet_compiled():
107              # Check that every entry in string parameters exists on the server
108              stale_entries = [entry for entry in string_params_client if entry not in server_tuples]
109              if stale_entries:
110                  raise AssertionError(f"String parameters contains entries not present on the server: {stale_entries}")
111              filtered_string_params = string_params_client
112          else:
113              available_string_params = [entry for entry in string_params_client if entry in server_tuples]
114              filtered_string_params = available_string_params
115  
116          # Validate that all entries are legitimate server parameters
117          server_method_param_tuples = {(m[0], m[1], m[2]) for m in mapping_server}
118          invalid_entries = [entry for entry in filtered_string_params if entry not in server_method_param_tuples]
119          if invalid_entries:
120              raise AssertionError(f"String parameters contains invalid entries: {invalid_entries}")
121  
122      def test_categories(self):
123          node = self.nodes[0]
124  
125          # wrong argument count
126          assert_raises_rpc_error(-1, 'help', node.help, 'foo', 'bar')
127  
128          # invalid argument
129          if not self.options.usecli:
130              assert_raises_rpc_error(-3, "JSON value of type number is not of expected type string", node.help, 0)
131  
132          # help of unknown command
133          assert_equal(node.help('foo'), 'help: unknown command: foo')
134  
135          # command titles
136          titles = [line[3:-3] for line in node.help().splitlines() if line.startswith('==')]
137  
138          components = ['Blockchain', 'Control', 'Mining', 'Network', 'Rawtransactions', 'Util']
139  
140          if self.is_wallet_compiled():
141              components.append('Wallet')
142  
143          if self.is_external_signer_compiled():
144              components.append('Signer')
145  
146          if self.is_zmq_compiled():
147              components.append('Zmq')
148  
149          assert_equal(titles, sorted(components))
150  
151      def dump_help(self):
152          dump_dir = os.path.join(self.options.tmpdir, 'rpc_help_dump')
153          os.mkdir(dump_dir)
154          calls = [line.split(' ', 1)[0] for line in self.nodes[0].help().splitlines() if line and not line.startswith('==')]
155          for call in calls:
156              with open(os.path.join(dump_dir, call), 'w') as f:
157                  # Make sure the node can generate the help at runtime without crashing
158                  f.write(self.nodes[0].help(call))
159  
160      def wallet_help(self):
161          assert 'getnewaddress ( "label" "address_type" )' in self.nodes[0].help('getnewaddress')
162          self.restart_node(0, extra_args=['-nowallet=1'])
163          assert 'getnewaddress ( "label" "address_type" )' in self.nodes[0].help('getnewaddress')
164  
165  if __name__ == '__main__':
166      HelpRpcTest(__file__).main()