1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
#!/usr/bin/env python3
# Copyright (c) 2018-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 RPC help output."""
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import assert_equal, assert_raises_rpc_error
from collections import defaultdict
import os
import re
def parse_string(s):
assert s[0] == '"'
assert s[-1] == '"'
return s[1:-1]
def process_mapping(fname):
"""Find and parse conversion table in implementation file `fname`."""
cmds = []
in_rpcs = False
with open(fname, "r", encoding="utf8") as f:
for line in f:
line = line.rstrip()
if not in_rpcs:
if line == 'static const CRPCConvertParam vRPCConvertParams[] =':
in_rpcs = True
else:
if line.startswith('};'):
in_rpcs = False
elif '{' in line and '"' in line:
m = re.search(r'{ *("[^"]*"), *([0-9]+) *, *("[^"]*") *},', line)
assert m, 'No match to table expression: %s' % line
name = parse_string(m.group(1))
idx = int(m.group(2))
argname = parse_string(m.group(3))
cmds.append((name, idx, argname))
assert not in_rpcs and cmds
return cmds
class HelpRpcTest(BitcoinTestFramework):
def set_test_params(self):
self.num_nodes = 1
self.supports_cli = False
def run_test(self):
self.test_client_conversion_table()
self.test_categories()
self.dump_help()
if self.is_wallet_compiled():
self.wallet_help()
def test_client_conversion_table(self):
file_conversion_table = os.path.join(self.config["environment"]["SRCDIR"], 'src', 'rpc', 'client.cpp')
mapping_client = process_mapping(file_conversion_table)
# Ignore echojson in client table
mapping_client = [m for m in mapping_client if m[0] != 'echojson']
mapping_server = self.nodes[0].help("dump_all_command_conversions")
# Filter all RPCs whether they need conversion
mapping_server_conversion = [tuple(m[:3]) for m in mapping_server if not m[3]]
# Only check if all RPC methods have been compiled (i.e. wallet is enabled)
if self.is_wallet_compiled() and sorted(mapping_client) != sorted(mapping_server_conversion):
raise AssertionError("RPC client conversion table ({}) and RPC server named arguments mismatch!\n{}".format(
file_conversion_table,
set(mapping_client).symmetric_difference(mapping_server_conversion),
))
# Check for conversion difference by argument name.
# It is preferable for API consistency that arguments with the same name
# have the same conversion, so bin by argument name.
all_methods_by_argname = defaultdict(list)
converts_by_argname = defaultdict(list)
for m in mapping_server:
all_methods_by_argname[m[2]].append(m[0])
converts_by_argname[m[2]].append(m[3])
for argname, convert in converts_by_argname.items():
if all(convert) != any(convert):
# Only allow dummy to fail consistency check
assert argname == 'dummy', ('WARNING: conversion mismatch for argument named %s (%s)' % (argname, list(zip(all_methods_by_argname[argname], converts_by_argname[argname]))))
def test_categories(self):
node = self.nodes[0]
# wrong argument count
assert_raises_rpc_error(-1, 'help', node.help, 'foo', 'bar')
# invalid argument
assert_raises_rpc_error(-1, 'JSON value is not a string as expected', node.help, 0)
# help of unknown command
assert_equal(node.help('foo'), 'help: unknown command: foo')
# command titles
titles = [line[3:-3] for line in node.help().splitlines() if line.startswith('==')]
components = ['Blockchain', 'Control', 'Generating', 'Mining', 'Network', 'Rawtransactions', 'Util']
if self.is_wallet_compiled():
components.append('Wallet')
if self.is_external_signer_compiled():
components.append('Signer')
if self.is_zmq_compiled():
components.append('Zmq')
assert_equal(titles, sorted(components))
def dump_help(self):
dump_dir = os.path.join(self.options.tmpdir, 'rpc_help_dump')
os.mkdir(dump_dir)
calls = [line.split(' ', 1)[0] for line in self.nodes[0].help().splitlines() if line and not line.startswith('==')]
for call in calls:
with open(os.path.join(dump_dir, call), 'w', encoding='utf-8') as f:
# Make sure the node can generate the help at runtime without crashing
f.write(self.nodes[0].help(call))
def wallet_help(self):
assert 'getnewaddress ( "label" "address_type" )' in self.nodes[0].help('getnewaddress')
self.restart_node(0, extra_args=['-nowallet=1'])
assert 'getnewaddress ( "label" "address_type" )' in self.nodes[0].help('getnewaddress')
if __name__ == '__main__':
HelpRpcTest().main()
|
