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

Code before:
        """Return number of test framework p2p connections to the node."""
        return len([peer for peer in self.getpeerinfo() if peer['subver'] == P2P_SUBVERSION])

    def disconnect_p2ps(self):
        """Close all p2p connections to the node.
        Use only after each p2p has sent a version message to ensure the wait works."""
        for p in self.p2ps:
            p.peer_disconnect()
        del self.p2ps[:]

        self.wait_until(lambda: self.num_test_p2p_connections() == 0)

    def bumpmocktime(self, seconds):
        """Fast forward using setmocktime to self.mocktime + seconds. Requires setmocktime to have
        been called at some point in the past."""
        assert self.mocktime
        self.mocktime += seconds
        self.setmocktime(self.mocktime)

    def wait_until(self, test_function, timeout=60):
        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor)

    def tx_in_orphanage(self, tx: CTransaction) -> bool:
        """Returns true if the transaction is in the orphanage."""
        found = [o for o in self.getorphantxs(verbosity=1) if o["txid"] == tx.rehash() and o["wtxid"] == tx.getwtxid()]
        return len(found) == 1

class TestNodeCLIAttr:
    def __init__(self, cli, command):
        self.cli = cli
        self.command = command

    def __call__(self, *args, **kwargs):
        return self.cli.send_cli(self.command, *args, **kwargs)

    def get_request(self, *args, **kwargs):
        return lambda: self(*args, **kwargs)


def arg_to_cli(arg):
    if isinstance(arg, bool):
        return str(arg).lower()
    elif arg is None:
        return 'null'
    elif isinstance(arg, dict) or isinstance(arg, list):
        return json.dumps(arg, default=serialization_fallback)
    else:
        return str(arg)


class TestNodeCLI():

Review comment (suggestion):
Feels like `TestNode` itself shouldn't have tx-level code. Maybe `/test/functional/test_framework/blocktools.py` would be slightly better? It has `create_witness_tx(node, ...)` and `send_to_witness(use_p2wsh, node, utxo, ...)`.

Code after:
    def disconnect_p2ps(self):
        """Close all p2p connections to the node.
        Use only after each p2p has sent a version message to ensure the wait works."""
        for p in self.p2ps:
            p.peer_disconnect()
        del self.p2ps[:]

        self.wait_until(lambda: self.num_test_p2p_connections() == 0)

    def bumpmocktime(self, seconds):
        """Fast forward using setmocktime to self.mocktime + seconds. Requires setmocktime to have
        been called at some point in the past."""
        assert self.mocktime
        self.mocktime += seconds
        self.setmocktime(self.mocktime)

    def wait_until(self, test_function, timeout=60):
        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor)


class TestNodeCLIAttr:
    def __init__(self, cli, command):
        self.cli = cli
        self.command = command

    def __call__(self, *args, **kwargs):
        return self.cli.send_cli(self.command, *args, **kwargs)

    def get_request(self, *args, **kwargs):
        return lambda: self(*args, **kwargs)


def arg_to_cli(arg):
    if isinstance(arg, bool):
        return str(arg).lower()
    elif arg is None:
        return 'null'
    elif isinstance(arg, dict) or isinstance(arg, list):
        return json.dumps(arg, default=serialization_fallback)
    else:
        return str(arg)


class TestNodeCLI():
    """Interface to bitcoin-cli for an individual node"""
    def __init__(self, binary, datadir):
        self.options = []
        self.binary = binary
        self.datadir = datadir
        self.input = None
        self.log = logging.getLogger('TestFramework.bitcoincli')