view mercurial/testing/__init__.py @ 52164:e01e84e5e426

rust-revlog: add a Rust-only `InnerRevlog`

This mirrors the Python `InnerRevlog` and will be used in a future patch
to replace said Python implementation. This allows us to start doing more
things in pure Rust, in particular reading and writing operations.

A lot of changes have to be introduced all at once, it wouldn't be very
useful to separate this patch IMO since all of them are either interlocked
or only useful with the rest.

from __future__ import annotations

import os
import time


# work around check-code complains
#
# This is a simple log level module doing simple test related work, we can't
# import more things, and we do not need it.
environ = getattr(os, 'environ')


def wait_on_cfg(ui, cfg, timeout=10):
    """synchronize on the `cfg` config path

    Use this to synchronize commands during race tests.
    """
    full_config = b'sync.' + cfg
    wait_config = full_config + b'-timeout'
    sync_path = ui.config(b'devel', full_config)
    if sync_path is not None:
        timeout = ui.config(b'devel', wait_config)
        ready_path = sync_path + b'.waiting'
        write_file(ready_path)
        wait_file(sync_path, timeout=timeout)


def _timeout_factor():
    """return the current modification to timeout"""
    default = int(environ.get('HGTEST_TIMEOUT_DEFAULT', 360))
    current = int(environ.get('HGTEST_TIMEOUT', default))
    if current == 0:
        return 1
    return current / float(default)


def wait_file(path, timeout=10):
    timeout *= _timeout_factor()
    start = time.time()
    while not os.path.exists(path):
        if timeout and time.time() - start > timeout:
            raise RuntimeError(b"timed out waiting for file: %s" % path)
        time.sleep(0.01)


def write_file(path, content=b''):
    if content:
        write_path = b'%s.tmp' % path
    else:
        write_path = path
    with open(write_path, 'wb') as f:
        f.write(content)
    if path != write_path:
        os.rename(write_path, path)