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

Code before:
    Implementation requires python integers, which have a // operator that does floor division.
    Other types like decimal.Decimal whose // operator truncates towards 0 will not work.
    """
    assert isinstance(a, int)
    assert isinstance(b, int)
    return -(-a // b)


def random_bitflip(data):
    data = list(data)
    data[random.randrange(len(data))] ^= (1 << (random.randrange(8)))
    return bytes(data)


def get_fee(tx_size, feerate_btc_kvb):
    """Calculate the fee in BTC given a feerate is BTC/kvB. Reflects CFeeRate::GetFee"""
    feerate_sat_kvb = int(feerate_btc_kvb * Decimal(1e8)) # Fee in sat/kvb as an int to avoid float precision errors
    target_fee_sat = ceildiv(feerate_sat_kvb * tx_size, 1000) # Round calculated fee up to nearest sat
    return target_fee_sat / Decimal(1e8) # Return result in  BTC


def satoshi_round(amount: Union[int, float, str], *, rounding: str) -> Decimal:
    """Rounds a Decimal amount to the nearest satoshi using the specified rounding mode."""
    return Decimal(amount).quantize(SATOSHI_PRECISION, rounding=rounding)


def wait_until_helper_internal(predicate, *, attempts=float('inf'), timeout=float('inf'), lock=None, timeout_factor=1.0):
    """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.
    """
    if attempts == float('inf') and timeout == float('inf'):
        timeout = 60
    timeout = timeout * timeout_factor
    attempt = 0
    time_end = time.time() + timeout

    while attempt < attempts and time.time() < time_end:
        if lock:
            with lock:
                if predicate():
                    return
        else:
            if predicate():
                return
        attempt += 1
        time.sleep(0.05)

Review comment (refactor):
```suggestion
def ensure_helper_internal(predicate, *, duration):
```

forgot to rename this one?

Code after:
    Implementation requires python integers, which have a // operator that does floor division.
    Other types like decimal.Decimal whose // operator truncates towards 0 will not work.
    """
    assert isinstance(a, int)
    assert isinstance(b, int)
    return -(-a // b)


def random_bitflip(data):
    data = list(data)
    data[random.randrange(len(data))] ^= (1 << (random.randrange(8)))
    return bytes(data)


def get_fee(tx_size, feerate_btc_kvb):
    """Calculate the fee in BTC given a feerate is BTC/kvB. Reflects CFeeRate::GetFee"""
    feerate_sat_kvb = int(feerate_btc_kvb * Decimal(1e8)) # Fee in sat/kvb as an int to avoid float precision errors
    target_fee_sat = ceildiv(feerate_sat_kvb * tx_size, 1000) # Round calculated fee up to nearest sat
    return target_fee_sat / Decimal(1e8) # Return result in  BTC


def satoshi_round(amount: Union[int, float, str], *, rounding: str) -> Decimal:
    """Rounds a Decimal amount to the nearest satoshi using the specified rounding mode."""
    return Decimal(amount).quantize(SATOSHI_PRECISION, rounding=rounding)


def ensure_for(*, duration, f, check_interval=0.2):
    """Check if the predicate keeps returning True for duration.

    check_interval can be used to configure the wait time between checks.
    Setting check_interval to 0 will allow to have two checks: one in the
    beginning and one after duration.
    """
    # If check_interval is 0 or negative or larger than duration, we fall back
    # to checking once in the beginning and once at the end of duration
    if check_interval <= 0 or check_interval > duration:
        check_interval = duration
    time_end = time.time() + duration
    predicate_source = "''''\n" + inspect.getsource(f) + "'''"
    while True:
        if not f():
            raise AssertionError(f"Predicate {predicate_source} became false within {duration} seconds")
        if time.time() > time_end:
            return
        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