Repository: bitcoin/bitcoin
File: test/functional/interface_rpc.py (Python)

Code before:
        assert_equal(command['method'], 'getrpcinfo')
        assert_greater_than_or_equal(command['duration'], 0)
        assert_equal(info['logpath'], os.path.join(self.nodes[0].chain_path, 'debug.log'))

    def test_batch_request(self):
        self.log.info("Testing basic JSON-RPC batch request...")

        results = self.nodes[0].batch([
            # A basic request that will work fine.
            {"method": "getblockcount", "id": 1},
            # Request that will fail.  The whole batch request should still
            # work fine.
            {"method": "invalidmethod", "id": 2},
            # Another call that should succeed.
            {"method": "getblockhash", "id": 3, "params": [0]},
        ])

        result_by_id = {}
        for res in results:
            result_by_id[res["id"]] = res

        assert_equal(result_by_id[1]['error'], None)
        assert_equal(result_by_id[1]['result'], 0)

        assert_equal(result_by_id[2]['error']['code'], -32601)
        assert_equal(result_by_id[2]['result'], None)

        assert_equal(result_by_id[3]['error'], None)
        assert result_by_id[3]['result'] is not None

    def test_http_status_codes(self):
        self.log.info("Testing HTTP status codes for JSON-RPC requests...")

        expect_http_status(404, -32601, self.nodes[0].invalidmethod)
        expect_http_status(500, -8, self.nodes[0].getblockhash, 42)

    def test_work_queue_exceeded(self):
        self.log.info("Testing work queue exceeded...")
        self.restart_node(0, ['-rpcworkqueue=1', '-rpcthreads=1'])
        got_exceeded_error = []
        threads = []
        for _ in range(3):
            t = Thread(target=test_work_queue_getblock, args=(self.nodes[0], got_exceeded_error))
            t.start()
            threads.append(t)
        for t in threads:
            t.join()

    def run_test(self):
        self.test_getrpcinfo()
        self.test_batch_request()

Review comment (suggestion):
Just thinking out loud here; we use a code comment to indicate what the error code represents but in many of the other tests it appears we don't (admittedly here it is the point of the test, and in the other locations it's not). I wonder whether it might be clearer to have these enumerated in a single location, e.g. _util.py_, and then they could be used by name here and elsewhere? (this might be a change someone else will just make in the future otherwise, too).

I think it's probably better left for a followup though, if at all.

Code after:

def format_response(options, idx, fields):
    if options.version == 2 and options.notification:
        return None
    response = {}
    if not options.notification:
        response.update(id=idx)
    if options.version == 2:
        response.update(jsonrpc="2.0")
    else:
        response.update(result=None, error=None)
    response.update(fields)
    if options.response_fields:
        response.update(options.response_fields)
    return response


def send_raw_rpc(node, raw_body: bytes) -> tuple[object, int]:
    return node._request("POST", "/", raw_body)


def send_json_rpc(node, body: object) -> tuple[object, int]:
    raw = json.dumps(body).encode("utf-8")
    return send_raw_rpc(node, raw)


def expect_http_rpc_status(expected_http_status, expected_rpc_error_code, node, method, params, version=1, notification=False):
    req = format_request(BatchOptions(version, notification), 0, {"method": method, "params": params})
    response, status = send_json_rpc(node, req)

    if expected_rpc_error_code is not None:
        assert_equal(response["error"]["code"], expected_rpc_error_code)

    assert_equal(status, expected_http_status)


def test_work_queue_getblock(node, got_exceeded_error):
    while not got_exceeded_error:
        try:
            node.cli("waitfornewblock", "500").send_cli()
        except subprocess.CalledProcessError as e:
            assert_equal(e.output, 'error: Server response: Work queue depth exceeded\n')
            got_exceeded_error.append(True)


class RPCInterfaceTest(BitcoinTestFramework):
    def set_test_params(self):
        self.num_nodes = 1
        self.setup_clean_chain = True
        self.supports_cli = False