comparison mercurial/util.py @ 45942:89a2afe31e82

formating: upgrade to black 20.8b1

This required a couple of small tweaks to un-confuse black, but now it
works. Big formatting changes come from:

 * Dramatically improved collection-splitting logic upstream
 * Black having a strong (correct IMO) opinion that """ is better than '''

Differential Revision: https://phab.mercurial-scm.org/D9430

45941:346af7687c6f

        self._copied = getattr(self, '_copied', 0) + 1
        return self


class sortdict(collections.OrderedDict):
    '''a simple sorted dictionary

    >>> d1 = sortdict([(b'a', 0), (b'b', 1)])
    >>> d2 = d1.copy()
    >>> d2
    sortdict([('a', 0), ('b', 1)])

    >>> list(d2.keys()) # should still be in last-set order
    ['b', 'a']
    >>> d1.insert(1, b'a.5', 0.5)
    >>> d1
    sortdict([('a', 0), ('a.5', 0.5), ('b', 1)])
    '''

    def __setitem__(self, key, value):
        if key in self:
            del self[key]
        super(sortdict, self).__setitem__(key, value)

    if prop in obj.__dict__:
        del obj.__dict__[prop]


def increasingchunks(source, min=1024, max=65536):
    '''return no less than min bytes per chunk while data remains,
    doubling min after each chunk until it reaches max'''

    def log2(x):
        if not x:
            return 0
        i = 0

    # PyPy runs slower with gc disabled
    nogc = lambda x: x


def pathto(root, n1, n2):
    '''return the relative path from one place to another.
    root should use os.sep to separate directories
    n1 should use os.sep to separate directories
    n2 should use "/" to separate directories
    returns an os.sep-separated path.

    If n1 is a relative path, it's assumed it's
    relative to root.
    n2 should always be relative to root.
    '''
    if not n1:
        return localpath(n2)
    if os.path.isabs(n1):
        if os.path.splitdrive(root)[0] != os.path.splitdrive(n1)[0]:
            return os.path.join(root, localpath(n2))

    b'zfs',
}


def copyfile(src, dest, hardlink=False, copystat=False, checkambig=False):
    '''copy a file, preserving mode and optionally other stat info like
    atime/mtime

    checkambig argument is used with filestat, and is useful only if
    destination file is guarded by any lock (e.g. repo.lock or
    repo.wlock).

    copystat and checkambig should be exclusive.
    '''
    assert not (copystat and checkambig)
    oldstat = None
    if os.path.lexists(dest):
        if checkambig:
            oldstat = checkambig and filestat.frompath(dest)

}
_winreservedchars = b':*?"<>|'


def checkwinfilename(path):
    r'''Check that the base-relative path is a valid filename on Windows.
    Returns None if the path is ok, or a UI string describing the problem.

    >>> checkwinfilename(b"just/a/normal/path")
    >>> checkwinfilename(b"foo/bar/con.xml")
    "filename contains 'con', which is reserved on Windows"

    >>> checkwinfilename(b"../bar")
    >>> checkwinfilename(b"foo\\")
    "filename ends with '\\', which is invalid on Windows"
    >>> checkwinfilename(b"foo\\/bar")
    "directory name ends with '\\', which is invalid on Windows"
    '''
    if path.endswith(b'\\'):
        return _(b"filename ends with '\\', which is invalid on Windows")
    if b'\\/' in path:
        return _(b"directory name ends with '\\', which is invalid on Windows")
    for n in path.replace(b'\\', b'/').split(b'/'):

            _re2 = bool(re2.match(r'\[([^\[]+)\]', b'[ui]'))
        except ImportError:
            _re2 = False

    def compile(self, pat, flags=0):
        '''Compile a regular expression, using re2 if possible

        For best performance, use only re2-compatible regexp features. The
        only flags from the re module that are re2-compatible are
        IGNORECASE and MULTILINE.'''
        if _re2 is None:
            self._checkre2()
        if _re2 and (flags & ~(remod.IGNORECASE | remod.MULTILINE)) == 0:
            if flags & remod.IGNORECASE:
                pat = b'(?i)' + pat

                pass
        return remod.compile(pat, flags)

    @propertycache
    def escape(self):
        '''Return the version of escape corresponding to self.compile.

        This is imperfect because whether re2 or re is used for a particular
        function depends on the flags, etc, but it's the best we can do.
        '''
        global _re2
        if _re2 is None:
            self._checkre2()
        if _re2:
            return re2.escape

_fspathcache = {}


def fspath(name, root):
    '''Get name in the case stored in the filesystem

    The name should be relative to root, and be normcase-ed for efficiency.

    Note that this function is unnecessary, and should not be
    called, for case-sensitive filesystems (simply because it's expensive).

    The root should be normcase-ed, too.
    '''

    def _makefspathcacheentry(dir):
        return {normcase(n): n for n in os.listdir(dir)}

    seps = pycompat.ossep

        and path.endswith(pycompat.osaltsep)
    )


def splitpath(path):
    '''Split path by os.sep.
    Note that this function does not use os.altsep because this is
    an alternative of simple "xxx.split(os.sep)".
    It is recommended to use os.path.normpath() before using this
    function if need.'''
    return path.split(pycompat.ossep)


def mktempcopy(name, emptyok=False, createmode=None, enforcewritable=False):
    """Create a temporary file with the same contents from name

    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
    visible. If the object is destroyed without being closed, all your
    writes are discarded.

    checkambig argument of constructor is used with filestat, and is
    useful only if target file is guarded by any lock (e.g. repo.lock
    or repo.wlock).
    '''

    def __init__(self, name, mode=b'w+b', createmode=None, checkambig=False):
        self.__name = name  # permanent name
        self._tempname = mktempcopy(
            name,

timedcm._nested = 0


def timed(func):
    '''Report the execution time of a function call to stderr.

    During development, use as a decorator when you need to measure
    the cost of a function, e.g. as follows:

    @util.timed
    def foo(a, b, c):
        pass
    '''

    def wrapper(*args, **kwargs):
        with timedcm(pycompat.bytestr(func.__name__)) as time_stats:
            result = func(*args, **kwargs)
        stderr = procutil.stderr

    (b'b', 1),
)


def sizetoint(s):
    '''Convert a space specifier to a byte count.

    >>> sizetoint(b'30')
    30
    >>> sizetoint(b'2.2kb')
    2252
    >>> sizetoint(b'6M')
    6291456
    '''
    t = s.strip().lower()
    try:
        for k, u in _sizeunits:
            if t.endswith(k):
                return int(float(t[: -len(k)]) * u)

    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):
        self._hooks = []

    def add(self, source, hook):

            results.append(hook(*args))
        return results


def getstackframes(skip=0, line=b' %-*s in %s\n', fileline=b'%s:%d', depth=0):
    '''Yields lines for a nicely formatted stacktrace.
    Skips the 'skip' last entries, then return the last 'depth' entries.
    Each file+linenumber is formatted according to fileline.
    Each line is formatted according to line.
    If line is None, it yields:
      length of longest filepath+line number,
      filepath+linenumber,
      function

    Not be used in production code but very convenient while developing.
    '''
    entries = [
        (fileline % (pycompat.sysbytes(fn), ln), pycompat.sysbytes(func))
        for fn, ln, func, _text in traceback.extract_stack()[: -skip - 1]
    ][-depth:]
    if entries:

    f=procutil.stderr,
    otherf=procutil.stdout,
    depth=0,
    prefix=b'',
):
    '''Writes a message to f (stderr) with a nicely formatted stacktrace.
    Skips the 'skip' entries closest to the call, then show 'depth' entries.
    By default it will flush stdout first.
    It can be used everywhere and intentionally does not require an ui object.
    Not be used in production code but very convenient while developing.
    '''
    if otherf:
        otherf.flush()
    f.write(b'%s%s at:\n' % (prefix, msg.rstrip()))
    for line in getstackframes(skip + 1, depth=depth):
        f.write(prefix + line)

45942:89a2afe31e82

        self._copied = getattr(self, '_copied', 0) + 1
        return self


class sortdict(collections.OrderedDict):
    """a simple sorted dictionary

    >>> d1 = sortdict([(b'a', 0), (b'b', 1)])
    >>> d2 = d1.copy()
    >>> d2
    sortdict([('a', 0), ('b', 1)])

    >>> list(d2.keys()) # should still be in last-set order
    ['b', 'a']
    >>> d1.insert(1, b'a.5', 0.5)
    >>> d1
    sortdict([('a', 0), ('a.5', 0.5), ('b', 1)])
    """

    def __setitem__(self, key, value):
        if key in self:
            del self[key]
        super(sortdict, self).__setitem__(key, value)

    if prop in obj.__dict__:
        del obj.__dict__[prop]


def increasingchunks(source, min=1024, max=65536):
    """return no less than min bytes per chunk while data remains,
    doubling min after each chunk until it reaches max"""

    def log2(x):
        if not x:
            return 0
        i = 0

    # PyPy runs slower with gc disabled
    nogc = lambda x: x


def pathto(root, n1, n2):
    """return the relative path from one place to another.
    root should use os.sep to separate directories
    n1 should use os.sep to separate directories
    n2 should use "/" to separate directories
    returns an os.sep-separated path.

    If n1 is a relative path, it's assumed it's
    relative to root.
    n2 should always be relative to root.
    """
    if not n1:
        return localpath(n2)
    if os.path.isabs(n1):
        if os.path.splitdrive(root)[0] != os.path.splitdrive(n1)[0]:
            return os.path.join(root, localpath(n2))

    b'zfs',
}


def copyfile(src, dest, hardlink=False, copystat=False, checkambig=False):
    """copy a file, preserving mode and optionally other stat info like
    atime/mtime

    checkambig argument is used with filestat, and is useful only if
    destination file is guarded by any lock (e.g. repo.lock or
    repo.wlock).

    copystat and checkambig should be exclusive.
    """
    assert not (copystat and checkambig)
    oldstat = None
    if os.path.lexists(dest):
        if checkambig:
            oldstat = checkambig and filestat.frompath(dest)

}
_winreservedchars = b':*?"<>|'


def checkwinfilename(path):
    r"""Check that the base-relative path is a valid filename on Windows.
    Returns None if the path is ok, or a UI string describing the problem.

    >>> checkwinfilename(b"just/a/normal/path")
    >>> checkwinfilename(b"foo/bar/con.xml")
    "filename contains 'con', which is reserved on Windows"

    >>> checkwinfilename(b"../bar")
    >>> checkwinfilename(b"foo\\")
    "filename ends with '\\', which is invalid on Windows"
    >>> checkwinfilename(b"foo\\/bar")
    "directory name ends with '\\', which is invalid on Windows"
    """
    if path.endswith(b'\\'):
        return _(b"filename ends with '\\', which is invalid on Windows")
    if b'\\/' in path:
        return _(b"directory name ends with '\\', which is invalid on Windows")
    for n in path.replace(b'\\', b'/').split(b'/'):

            _re2 = bool(re2.match(r'\[([^\[]+)\]', b'[ui]'))
        except ImportError:
            _re2 = False

    def compile(self, pat, flags=0):
        """Compile a regular expression, using re2 if possible

        For best performance, use only re2-compatible regexp features. The
        only flags from the re module that are re2-compatible are
        IGNORECASE and MULTILINE."""
        if _re2 is None:
            self._checkre2()
        if _re2 and (flags & ~(remod.IGNORECASE | remod.MULTILINE)) == 0:
            if flags & remod.IGNORECASE:
                pat = b'(?i)' + pat

                pass
        return remod.compile(pat, flags)

    @propertycache
    def escape(self):
        """Return the version of escape corresponding to self.compile.

        This is imperfect because whether re2 or re is used for a particular
        function depends on the flags, etc, but it's the best we can do.
        """
        global _re2
        if _re2 is None:
            self._checkre2()
        if _re2:
            return re2.escape

_fspathcache = {}


def fspath(name, root):
    """Get name in the case stored in the filesystem

    The name should be relative to root, and be normcase-ed for efficiency.

    Note that this function is unnecessary, and should not be
    called, for case-sensitive filesystems (simply because it's expensive).

    The root should be normcase-ed, too.
    """

    def _makefspathcacheentry(dir):
        return {normcase(n): n for n in os.listdir(dir)}

    seps = pycompat.ossep

        and path.endswith(pycompat.osaltsep)
    )


def splitpath(path):
    """Split path by os.sep.
    Note that this function does not use os.altsep because this is
    an alternative of simple "xxx.split(os.sep)".
    It is recommended to use os.path.normpath() before using this
    function if need."""
    return path.split(pycompat.ossep)


def mktempcopy(name, emptyok=False, createmode=None, enforcewritable=False):
    """Create a temporary file with the same contents from name

    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
    visible. If the object is destroyed without being closed, all your
    writes are discarded.

    checkambig argument of constructor is used with filestat, and is
    useful only if target file is guarded by any lock (e.g. repo.lock
    or repo.wlock).
    """

    def __init__(self, name, mode=b'w+b', createmode=None, checkambig=False):
        self.__name = name  # permanent name
        self._tempname = mktempcopy(
            name,

timedcm._nested = 0


def timed(func):
    """Report the execution time of a function call to stderr.

    During development, use as a decorator when you need to measure
    the cost of a function, e.g. as follows:

    @util.timed
    def foo(a, b, c):
        pass
    """

    def wrapper(*args, **kwargs):
        with timedcm(pycompat.bytestr(func.__name__)) as time_stats:
            result = func(*args, **kwargs)
        stderr = procutil.stderr

    (b'b', 1),
)


def sizetoint(s):
    """Convert a space specifier to a byte count.

    >>> sizetoint(b'30')
    30
    >>> sizetoint(b'2.2kb')
    2252
    >>> sizetoint(b'6M')
    6291456
    """
    t = s.strip().lower()
    try:
        for k, u in _sizeunits:
            if t.endswith(k):
                return int(float(t[: -len(k)]) * u)

    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):
        self._hooks = []

    def add(self, source, hook):

            results.append(hook(*args))
        return results


def getstackframes(skip=0, line=b' %-*s in %s\n', fileline=b'%s:%d', depth=0):
    """Yields lines for a nicely formatted stacktrace.
    Skips the 'skip' last entries, then return the last 'depth' entries.
    Each file+linenumber is formatted according to fileline.
    Each line is formatted according to line.
    If line is None, it yields:
      length of longest filepath+line number,
      filepath+linenumber,
      function

    Not be used in production code but very convenient while developing.
    """
    entries = [
        (fileline % (pycompat.sysbytes(fn), ln), pycompat.sysbytes(func))
        for fn, ln, func, _text in traceback.extract_stack()[: -skip - 1]
    ][-depth:]
    if entries:

    f=procutil.stderr,
    otherf=procutil.stdout,
    depth=0,
    prefix=b'',
):
    """Writes a message to f (stderr) with a nicely formatted stacktrace.
    Skips the 'skip' entries closest to the call, then show 'depth' entries.
    By default it will flush stdout first.
    It can be used everywhere and intentionally does not require an ui object.
    Not be used in production code but very convenient while developing.
    """
    if otherf:
        otherf.flush()
    f.write(b'%s%s at:\n' % (prefix, msg.rstrip()))
    for line in getstackframes(skip + 1, depth=depth):
        f.write(prefix + line)