mercurial/util.py

48945:55d132525155

for k in DIGESTS_BY_STRENGTH:
    assert k in DIGESTS


class digester(object):
    """helper to compute digests.

    This helper can be used to compute one or more digests given their name.

    >>> d = digester([b'md5', b'sha1'])

            if k in supported:
                return k
        return None


class digestchecker(object):
    """file handle wrapper that additionally checks content against a given
    size and digests.

        d = digestchecker(fh, size, {'md5': '...'})

_chunksize = 4096


class bufferedinputpipe(object):
    """a manually buffered input pipe

    Python will not let us use buffered IO and lazy reading with 'polling' at
    the same time. We cannot probe the buffer state and select will not detect
    that data are ready to read if they are already buffered.

        if os.fstat(fd).st_size == 0:
            return b''
        raise


class fileobjectproxy(object):
    """A proxy around file objects that tells a watcher when events occur.

    This type is intended to only be used for testing purposes. Think hard
    before using it in important code.
    """

    'gettimeout',
    'setsockopt',
}


class socketproxy(object):
    """A proxy around a socket that tells a watcher when events occur.

    This is like ``fileobjectproxy`` except for sockets.

    This type is intended to only be used for testing purposes. Think hard

        return object.__getattribute__(self, '_observedcall')(
            'setsockopt', *args, **kwargs
        )


class baseproxyobserver(object):
    def __init__(self, fh, name, logdata, logdataapis):
        self.fh = fh
        self.name = name
        self.logdata = logdata
        self.logdataapis = logdataapis

            return cache[args]

    return f


class cow(object):
    """helper class to make copy-on-write easier

    Call preparewrite before doing any writes.
    """

    Be sure to call d = d.preparewrite() before writing to d.
    """


class transactional(object):  # pytype: disable=ignored-metaclass
    """Base class for making a transactional type into a context manager."""

    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod

@contextlib.contextmanager
def nullcontextmanager(enter_result=None):
    yield enter_result


class _lrucachenode(object):
    """A node in a doubly linked list.

    Holds a reference to nodes on either side as well as a key-value
    pair for the dictionary entry.
    """

        self.key = _notset
        self.value = None
        self.cost = 0


class lrucachedict(object):
    """Dict that caches most recent accesses and sets.

    The dict consists of an actual backing dict - indexed by original
    key - and a doubly linked circular list defining the order of entries in
    the cache.

            return cache[args]

    return f


class propertycache(object):
    def __init__(self, func):
        self.func = func
        self.name = func.__name__

    def __get__(self, obj, type=None):

    _re2 = None
except ImportError:
    _re2 = False


class _re(object):
    def _checkre2(self):
        global _re2
        global _re2_input

        check_pattern = br'\[([^\[]+)\]'

            pass
        raise
    return temp


class filestat(object):
    """help to exactly detect change of a file

    'stat' attribute is result of 'os.stat()' if specified 'path'
    exists. Otherwise, it is None. This can avoid preparative
    'exists()' examination on client side of this class.

    def __ne__(self, other):
        return not self == other


class atomictempfile(object):
    """writable file object that atomically updates a file

    All writes will go to a temporary copy of the original file. Call
    close() when you are done writing, and atomictempfile will rename
    the temporary copy to the original name, making the changes

    # type: (bytes, bytes) -> None
    with open(path, b'ab') as fp:
        fp.write(text)


class chunkbuffer(object):
    """Allow arbitrary sized chunks of data to be efficiently read from an
    iterator over chunks of arbitrary size."""

    def __init__(self, in_iter):
        """in_iter is the iterator that's iterating over the input chunks."""

        if limit:
            limit -= len(s)
        yield s


class cappedreader(object):
    """A file object proxy that allows reading up to N bytes.

    Given a source file object, instances of this type allow reading up to
    N bytes from that source file object. Attempts to read past the allowed
    limit are treated as EOF.

    (1, 1 << 10, _(b'%.2f KB')),
    (1, 1, _(b'%.0f bytes')),
)


class transformingwriter(object):
    """Writable file wrapper to transform data by function"""

    def __init__(self, fp, encode):
        self._fp = fp
        self._encode = encode

    (1, 0.000000001, _(b'%.3f ns')),
)


@attr.s
class timedcmstats(object):
    """Stats information produced by the timedcm context manager on entering."""

    # the starting value of the timer as a float (meaning and resulution is
    # platform dependent, see util.timer)
    start = attr.ib(default=attr.Factory(lambda: timer()))

        return int(t)
    except ValueError:
        raise error.ParseError(_(b"couldn't parse size: %s") % s)


class hooks(object):
    """A collection of hook functions that can be used to extend a
    function's behavior. Hooks are called in lexicographic order,
    based on the names of their sources."""

    def __init__(self):

48946:642e31cb55f0

for k in DIGESTS_BY_STRENGTH:
    assert k in DIGESTS


class digester:
    """helper to compute digests.

    This helper can be used to compute one or more digests given their name.

    >>> d = digester([b'md5', b'sha1'])

            if k in supported:
                return k
        return None


class digestchecker:
    """file handle wrapper that additionally checks content against a given
    size and digests.

        d = digestchecker(fh, size, {'md5': '...'})

_chunksize = 4096


class bufferedinputpipe:
    """a manually buffered input pipe

    Python will not let us use buffered IO and lazy reading with 'polling' at
    the same time. We cannot probe the buffer state and select will not detect
    that data are ready to read if they are already buffered.

        if os.fstat(fd).st_size == 0:
            return b''
        raise


class fileobjectproxy:
    """A proxy around file objects that tells a watcher when events occur.

    This type is intended to only be used for testing purposes. Think hard
    before using it in important code.
    """

    'gettimeout',
    'setsockopt',
}


class socketproxy:
    """A proxy around a socket that tells a watcher when events occur.

    This is like ``fileobjectproxy`` except for sockets.

    This type is intended to only be used for testing purposes. Think hard

        return object.__getattribute__(self, '_observedcall')(
            'setsockopt', *args, **kwargs
        )


class baseproxyobserver:
    def __init__(self, fh, name, logdata, logdataapis):
        self.fh = fh
        self.name = name
        self.logdata = logdata
        self.logdataapis = logdataapis

            return cache[args]

    return f


class cow:
    """helper class to make copy-on-write easier

    Call preparewrite before doing any writes.
    """

    Be sure to call d = d.preparewrite() before writing to d.
    """


class transactional:  # pytype: disable=ignored-metaclass
    """Base class for making a transactional type into a context manager."""

    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod

@contextlib.contextmanager
def nullcontextmanager(enter_result=None):
    yield enter_result


class _lrucachenode:
    """A node in a doubly linked list.

    Holds a reference to nodes on either side as well as a key-value
    pair for the dictionary entry.
    """

        self.key = _notset
        self.value = None
        self.cost = 0


class lrucachedict:
    """Dict that caches most recent accesses and sets.

    The dict consists of an actual backing dict - indexed by original
    key - and a doubly linked circular list defining the order of entries in
    the cache.

            return cache[args]

    return f


class propertycache:
    def __init__(self, func):
        self.func = func
        self.name = func.__name__

    def __get__(self, obj, type=None):

    _re2 = None
except ImportError:
    _re2 = False


class _re:
    def _checkre2(self):
        global _re2
        global _re2_input

        check_pattern = br'\[([^\[]+)\]'

            pass
        raise
    return temp


class filestat:
    """help to exactly detect change of a file

    'stat' attribute is result of 'os.stat()' if specified 'path'
    exists. Otherwise, it is None. This can avoid preparative
    'exists()' examination on client side of this class.

    def __ne__(self, other):
        return not self == other


class atomictempfile:
    """writable file object that atomically updates a file

    All writes will go to a temporary copy of the original file. Call
    close() when you are done writing, and atomictempfile will rename
    the temporary copy to the original name, making the changes

    # type: (bytes, bytes) -> None
    with open(path, b'ab') as fp:
        fp.write(text)


class chunkbuffer:
    """Allow arbitrary sized chunks of data to be efficiently read from an
    iterator over chunks of arbitrary size."""

    def __init__(self, in_iter):
        """in_iter is the iterator that's iterating over the input chunks."""

        if limit:
            limit -= len(s)
        yield s


class cappedreader:
    """A file object proxy that allows reading up to N bytes.

    Given a source file object, instances of this type allow reading up to
    N bytes from that source file object. Attempts to read past the allowed
    limit are treated as EOF.

    (1, 1 << 10, _(b'%.2f KB')),
    (1, 1, _(b'%.0f bytes')),
)


class transformingwriter:
    """Writable file wrapper to transform data by function"""

    def __init__(self, fp, encode):
        self._fp = fp
        self._encode = encode

    (1, 0.000000001, _(b'%.3f ns')),
)


@attr.s
class timedcmstats:
    """Stats information produced by the timedcm context manager on entering."""

    # the starting value of the timer as a float (meaning and resulution is
    # platform dependent, see util.timer)
    start = attr.ib(default=attr.Factory(lambda: timer()))

        return int(t)
    except ValueError:
        raise error.ParseError(_(b"couldn't parse size: %s") % s)


class hooks:
    """A collection of hook functions that can be used to extend a
    function's behavior. Hooks are called in lexicographic order,
    based on the names of their sources."""

    def __init__(self):