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

Code before:
        time.sleep(check_interval)


def wait_until_helper_internal(predicate, *, timeout=60, lock=None, timeout_factor=1.0, check_interval=0.05):
    """Sleep until the predicate resolves to be True.

    Warning: Note that this method is not recommended to be used in tests as it is
    not aware of the context of the test framework. Using the `wait_until()` members
    from `BitcoinTestFramework` or `P2PInterface` class ensures the timeout is
    properly scaled. Furthermore, `wait_until()` from `P2PInterface` class in
    `p2p.py` has a preset lock.
    """
    timeout = timeout * timeout_factor
    time_end = time.time() + timeout

    while time.time() < time_end:
        if lock:
            with lock:
                if predicate():
                    return
        else:
            if predicate():
                return
        time.sleep(check_interval)

    # Print the cause of the timeout
    predicate_source = "''''\n" + inspect.getsource(predicate) + "'''"
    logger.error("wait_until() failed. Predicate: {}".format(predicate_source))
    raise AssertionError("Predicate {} not true after {} seconds".format(predicate_source, timeout))


def bpf_cflags():
    return [
        "-Wno-error=implicit-function-declaration",
        "-Wno-duplicate-decl-specifier",  # https://github.com/bitcoin/bitcoin/issues/32322
    ]


def sha256sum_file(filename):
    h = hashlib.sha256()
    with open(filename, 'rb') as f:
        d = f.read(4096)
        while len(d) > 0:
            h.update(d)
            d = f.read(4096)
    return h.digest()


def util_xor(data, key, *, offset):
    data = bytearray(data)
    for i in range(len(data)):

Review comment (suggestion):
Just to clarify, i'm not saying that it is coupled to the test_framework, but that it is coupled to the config. 
For a followup I could see something like the code below, for this PR I like that it is more or less a code move.

Rough example; **not** ment to be a nit:
```suggestion
def get_binary_paths(bin_path, exe_ext):
    """Get paths of all binaries from environment variables or their default values"""

    paths = types.SimpleNamespace()
    binaries = {
        "bitcoin": "BITCOIN_BIN",
        "bitcoind": "BITCOIND",
        "bitcoin-cli": "BITCOINCLI",
        "bitcoin-util": "BITCOINUTIL",
        "bitcoin-tx": "BITCOINTX",
        "bitcoin-chainstate": "BITCOINCHAINSTATE",
        "bitcoin-wallet": "BITCOINWALLET",
    }
    # Set paths to bitcoin core binaries allowing overrides with environment
    # variables.
    for binary, env_variable_name in binaries.items():
        default_filename = os.path.join(
            bin_path,
            binary + exe_ext,
        )
        setattr(paths, env_variable_name.lower(), os.getenv(env_variable_name, default=default_filename))
```

Code after:
        else:
            return [bin_path]


def get_binary_paths(config):
    """Get paths of all binaries from environment variables or their default values"""

    paths = types.SimpleNamespace()
    binaries = {
        "bitcoin": "BITCOIN_BIN",
        "bitcoind": "BITCOIND",
        "bitcoin-cli": "BITCOINCLI",
        "bitcoin-util": "BITCOINUTIL",
        "bitcoin-tx": "BITCOINTX",
        "bitcoin-chainstate": "BITCOINCHAINSTATE",
        "bitcoin-wallet": "BITCOINWALLET",
    }
    # Set paths to bitcoin core binaries allowing overrides with environment
    # variables.
    for binary, env_variable_name in binaries.items():
        default_filename = os.path.join(
            config["environment"]["BUILDDIR"],
            "bin",
            binary + config["environment"]["EXEEXT"],
        )
        setattr(paths, env_variable_name.lower(), os.getenv(env_variable_name, default=default_filename))
    # BITCOIN_CMD environment variable can be specified to invoke bitcoin
    # wrapper binary instead of other executables.
    paths.bitcoin_cmd = shlex.split(os.getenv("BITCOIN_CMD", "")) or None
    return paths


def export_env_build_path(config):
    os.environ["PATH"] = os.pathsep.join([
        os.path.join(config["environment"]["BUILDDIR"], "bin"),
        os.environ["PATH"],
    ])


def count_bytes(hex_string):
    return len(bytearray.fromhex(hex_string))


def str_to_b64str(string):
    return b64encode(string.encode('utf-8')).decode('ascii')


def ceildiv(a, b):
    """
    Divide 2 ints and round up to next int rather than round down
    Implementation requires python integers, which have a // operator that does floor division.