bitcoin-mirror/test/functional/rpc_help.py

#!/usr/bin/env python3
# Copyright (c) 2018-present 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 = []
    string_params = []
    in_rpcs = False
    with open(fname, "r") as f:
        for line in f:
            line = line.rstrip()
            if not in_rpcs:
                if re.match(r'static const CRPCConvertParam vRPCConvertParams\[] =', line):
                    in_rpcs = True
            else:
                if line.startswith('};'):
                    in_rpcs = False
                elif '{' in line and '"' in line:
                    # Match lines with ParamFormat::STRING
                    m_string = re.search(r'{ *("[^"]*") *, *([0-9]+) *, *("[^"]*") *, *ParamFormat::STRING *},?', line)
                    if m_string:
                        name = parse_string(m_string.group(1))
                        idx = int(m_string.group(2))
                        argname = parse_string(m_string.group(3))
                        string_params.append((name, idx, argname))
                        continue

                    # Match lines with ParamFormat::JSON and ParamFormat::JSON_OR_STRING
                    m_json = re.search(r'{ *("[^"]*") *, *([0-9]+) *, *("[^"]*") *(?:, *ParamFormat::(JSON_OR_STRING|JSON))? *},?', line)
                    if m_json:
                        name = parse_string(m_json.group(1))
                        idx = int(m_json.group(2))
                        argname = parse_string(m_json.group(3))
                        cmds.append((name, idx, argname))

    assert not in_rpcs
    return cmds, string_params

class HelpRpcTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.uses_wallet = None

    def run_test(self):
        self.test_client_conversion_table()
        self.test_client_string_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 and psbt to fail consistency check
                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]))))

    def test_client_string_conversion_table(self):
        file_conversion_table = os.path.join(self.config["environment"]["SRCDIR"], 'src', 'rpc', 'client.cpp')
        _, string_params_client = process_mapping(file_conversion_table)
        mapping_server = self.nodes[0].help("dump_all_command_conversions")
        server_tuples = {tuple(m[:3]) for m in mapping_server}

        # Filter string parameters based on wallet compilation status
        if self.is_wallet_compiled():
            # Check that every entry in string parameters exists on the server
            stale_entries = [entry for entry in string_params_client if entry not in server_tuples]
            if stale_entries:
                raise AssertionError(f"String parameters contains entries not present on the server: {stale_entries}")
            filtered_string_params = string_params_client
        else:
            available_string_params = [entry for entry in string_params_client if entry in server_tuples]
            filtered_string_params = available_string_params

        # Validate that all entries are legitimate server parameters
        server_method_param_tuples = {(m[0], m[1], m[2]) for m in mapping_server}
        invalid_entries = [entry for entry in filtered_string_params if entry not in server_method_param_tuples]
        if invalid_entries:
            raise AssertionError(f"String parameters contains invalid entries: {invalid_entries}")

    def test_categories(self):
        node = self.nodes[0]

        # wrong argument count
        assert_raises_rpc_error(-1, 'help', node.help, 'foo', 'bar')

        # invalid argument
        if not self.options.usecli:
            assert_raises_rpc_error(-3, "JSON value of type number is not of expected type string", 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', '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') 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(__file__).main()