comparison mercurial/posix.py @ 43077:687b865b95ad

formatting: byteify all mercurial/ and hgext/ string literals

Done with

    python3.7 contrib/byteify-strings.py -i $(hg files 'set:mercurial/**.py - mercurial/thirdparty/** + hgext/**.py - hgext/fsmonitor/pywatchman/** - mercurial/__init__.py')
    black -l 80 -t py33 -S $(hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**" - hgext/fsmonitor/pywatchman/**')

# skip-blame mass-reformatting only

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

43076:2372284d9457

    # vaguely unix-like but don't have hardlink support. For those
    # poor souls, just say we tried and that it failed so we fall back
    # to copies.
    def oslink(src, dst):
        raise OSError(
            errno.EINVAL, 'hardlinks not supported: %s to %s' % (src, dst)
        )


readlink = os.readlink
unlink = os.unlink

    ...           b'/file_at_root',
    ...           b'///multiple_leading_separators_at_root',
    ...           b'']:
    ...     assert split(f) == posixpath.split(f), f
    '''
    ht = p.rsplit('/', 1)
    if len(ht) == 1:
        return '', p
    nh = ht[0].rstrip('/')
    if nh:
        return nh, ht[1]
    return ht[0] + '/', ht[1]


def openhardlinks():
    '''return true if it is safe to hold open file handles to hardlinks'''
    return True

def parsepatchoutput(output_line):
    """parses the output produced by patch and returns the filename"""
    pf = output_line[14:]
    if pycompat.sysplatform == 'OpenVMS':
        if pf[0] == '`':
            pf = pf[1:-1]  # Remove the quotes
    else:
        if pf.startswith("'") and pf.endswith("'") and " " in pf:
            pf = pf[1:-1]  # Remove the quotes
    return pf


def sshargs(sshcmd, host, user, port):
    '''Build argument list for ssh'''
    args = user and ("%s@%s" % (user, host)) or host
    if '-' in args[:1]:
        raise error.Abort(
            _('illegal ssh hostname or username starting with -: %s') % args
        )
    args = shellquote(args)
    if port:
        args = '-p %s %s' % (shellquote(port), args)
    return args


def isexec(f):
    """check whether a file is executable"""

    st = os.lstat(f)
    s = st.st_mode
    if l:
        if not stat.S_ISLNK(s):
            # switch file to link
            fp = open(f, 'rb')
            data = fp.read()
            fp.close()
            unlink(f)
            try:
                os.symlink(data, f)
            except OSError:
                # failed to make a link, rewrite file
                fp = open(f, "wb")
                fp.write(data)
                fp.close()
        # no chmod needed at this point
        return
    if stat.S_ISLNK(s):
        # switch link to file
        data = os.readlink(f)
        unlink(f)
        fp = open(f, "wb")
        fp.write(data)
        fp.close()
        s = 0o666 & ~umask  # avoid restatting for chmod

    sx = s & 0o100
    if st.st_nlink > 1 and bool(x) != bool(sx):
        # the file is a hardlink, break it
        with open(f, "rb") as fp:
            data = fp.read()
        unlink(f)
        with open(f, "wb") as fp:
            fp.write(data)

    if x and not sx:
        # Turn on +x for every +r bit when making a file executable
        # and obey umask.

    # a FS remount. Frequently we can detect it if files are created
    # with exec bit on.

    try:
        EXECFLAGS = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
        basedir = os.path.join(path, '.hg')
        cachedir = os.path.join(basedir, 'wcache')
        storedir = os.path.join(basedir, 'store')
        if not os.path.exists(cachedir):
            try:
                # we want to create the 'cache' directory, not the '.hg' one.
                # Automatically creating '.hg' directory could silently spawn
                # invalid Mercurial repositories. That seems like a bad idea.

                    copymode(basedir, cachedir)
            except (IOError, OSError):
                # we other fallback logic triggers
                pass
        if os.path.isdir(cachedir):
            checkisexec = os.path.join(cachedir, 'checkisexec')
            checknoexec = os.path.join(cachedir, 'checknoexec')

            try:
                m = os.stat(checkisexec).st_mode
            except OSError as e:
                if e.errno != errno.ENOENT:

                    try:
                        m = os.stat(checknoexec).st_mode
                    except OSError as e:
                        if e.errno != errno.ENOENT:
                            raise
                        open(checknoexec, 'w').close()  # might fail
                        m = os.stat(checknoexec).st_mode
                    if m & EXECFLAGS == 0:
                        # check-exec is exec and check-no-exec is not exec
                        return True
                    # checknoexec exists but is exec - delete it

            checkdir = cachedir
        else:
            # check directly in path and don't leave checkisexec behind
            checkdir = path
            checkisexec = None
        fh, fn = pycompat.mkstemp(dir=checkdir, prefix='hg-checkexec-')
        try:
            os.close(fh)
            m = os.stat(fn).st_mode
            if m & EXECFLAGS == 0:
                os.chmod(fn, m & 0o777 | EXECFLAGS)

def checklink(path):
    """check whether the given path is on a symlink-capable filesystem"""
    # mktemp is not racy because symlink creation will fail if the
    # file already exists
    while True:
        cachedir = os.path.join(path, '.hg', 'wcache')
        checklink = os.path.join(cachedir, 'checklink')
        # try fast path, read only
        if os.path.islink(checklink):
            return True
        if os.path.isdir(cachedir):
            checkdir = cachedir

        name = pycompat.fsencode(name)
        try:
            fd = None
            if cachedir is None:
                fd = pycompat.namedtempfile(
                    dir=checkdir, prefix='hg-checklink-'
                )
                target = os.path.basename(fd.name)
            else:
                # create a fixed file to link to; doesn't matter if it
                # already exists.
                target = 'checklink-target'
                try:
                    fullpath = os.path.join(cachedir, target)
                    open(fullpath, 'w').close()
                except IOError as inst:
                    if inst[0] == errno.EACCES:
                        # If we can't write to cachedir, just pretend
                        # that the fs is readonly and by association
                        # that the fs won't support symlinks. This

    def normcasefallback(path):
        try:
            u = path.decode('utf-8')
        except UnicodeDecodeError:
            # OS X percent-encodes any bytes that aren't valid utf-8
            s = ''
            pos = 0
            l = len(path)
            while pos < l:
                try:
                    c = encoding.getutf8char(path, pos)
                    pos += len(c)
                except ValueError:
                    c = '%%%02X' % ord(path[pos : pos + 1])
                    pos += 1
                s += c

            u = s.decode('utf-8')

        enc = unicodedata.normalize(r'NFD', u).lower().encode('utf-8')
        # drop HFS+ ignored characters
        return encoding.hfsignoreclean(enc)


if pycompat.sysplatform == 'cygwin':
    # workaround for cygwin, in which mount point part of path is
    # treated as case sensitive, even though underlying NTFS is case
    # insensitive.

    # default mount points
    cygwinmountpoints = sorted(
        ["/usr/bin", "/usr/lib", "/cygdrive",], reverse=True
    )

    # use upper-ing as normcase as same as NTFS workaround
    def normcase(path):
        pathlen = len(path)

_needsshellquote = None


def shellquote(s):
    if pycompat.sysplatform == 'OpenVMS':
        return '"%s"' % s
    global _needsshellquote
    if _needsshellquote is None:
        _needsshellquote = re.compile(br'[^a-zA-Z0-9._/+-]').search
    if s and not _needsshellquote(s):
        # "s" shouldn't have to be quoted
        return s
    else:
        return "'%s'" % s.replace("'", "'\\''")


def shellsplit(s):
    """Parse a command string in POSIX shell way (best-effort)"""
    return pycompat.shlexsplit(s, posix=True)

    return cmd


def testpid(pid):
    '''return False if pid dead, True if running or not sure'''
    if pycompat.sysplatform == 'OpenVMS':
        return True
    try:
        os.kill(pid, 0)
        return True
    except OSError as inst:

def findexe(command):
    '''Find executable for command searching like which does.
    If command is a basename then PATH is searched for command.
    PATH isn't searched if command is an absolute or relative path.
    If command isn't found None is returned.'''
    if pycompat.sysplatform == 'OpenVMS':
        return command

    def findexisting(executable):
        'Will return executable if existing file'
        if os.path.isfile(executable) and os.access(executable, os.X_OK):
            return executable
        return None

    if pycompat.ossep in command:
        return findexisting(command)

    if pycompat.sysplatform == 'plan9':
        return findexisting(os.path.join('/bin', command))

    for path in encoding.environ.get('PATH', '').split(pycompat.ospathsep):
        executable = findexisting(os.path.join(path, command))
        if executable is not None:
            return executable
    return None

                    break
                chunks.append(s)
            except IOError:
                break

        return ''.join(chunks)
    finally:
        fcntl.fcntl(pipe, fcntl.F_SETFL, oldflags)


def bindunixsocket(sock, path):

    # AF_UNIX path has very small length limit (107 chars) on common
    # platforms (see sys/un.h)
    dirname, basename = os.path.split(path)
    bakwdfd = None
    if dirname:
        bakwdfd = os.open('.', os.O_DIRECTORY)
        os.chdir(dirname)
    sock.bind(basename)
    if bakwdfd:
        os.fchdir(bakwdfd)
        os.close(bakwdfd)

43077:687b865b95ad

    # vaguely unix-like but don't have hardlink support. For those
    # poor souls, just say we tried and that it failed so we fall back
    # to copies.
    def oslink(src, dst):
        raise OSError(
            errno.EINVAL, b'hardlinks not supported: %s to %s' % (src, dst)
        )


readlink = os.readlink
unlink = os.unlink

    ...           b'/file_at_root',
    ...           b'///multiple_leading_separators_at_root',
    ...           b'']:
    ...     assert split(f) == posixpath.split(f), f
    '''
    ht = p.rsplit(b'/', 1)
    if len(ht) == 1:
        return b'', p
    nh = ht[0].rstrip(b'/')
    if nh:
        return nh, ht[1]
    return ht[0] + b'/', ht[1]


def openhardlinks():
    '''return true if it is safe to hold open file handles to hardlinks'''
    return True

def parsepatchoutput(output_line):
    """parses the output produced by patch and returns the filename"""
    pf = output_line[14:]
    if pycompat.sysplatform == b'OpenVMS':
        if pf[0] == b'`':
            pf = pf[1:-1]  # Remove the quotes
    else:
        if pf.startswith(b"'") and pf.endswith(b"'") and b" " in pf:
            pf = pf[1:-1]  # Remove the quotes
    return pf


def sshargs(sshcmd, host, user, port):
    '''Build argument list for ssh'''
    args = user and (b"%s@%s" % (user, host)) or host
    if b'-' in args[:1]:
        raise error.Abort(
            _(b'illegal ssh hostname or username starting with -: %s') % args
        )
    args = shellquote(args)
    if port:
        args = b'-p %s %s' % (shellquote(port), args)
    return args


def isexec(f):
    """check whether a file is executable"""

    st = os.lstat(f)
    s = st.st_mode
    if l:
        if not stat.S_ISLNK(s):
            # switch file to link
            fp = open(f, b'rb')
            data = fp.read()
            fp.close()
            unlink(f)
            try:
                os.symlink(data, f)
            except OSError:
                # failed to make a link, rewrite file
                fp = open(f, b"wb")
                fp.write(data)
                fp.close()
        # no chmod needed at this point
        return
    if stat.S_ISLNK(s):
        # switch link to file
        data = os.readlink(f)
        unlink(f)
        fp = open(f, b"wb")
        fp.write(data)
        fp.close()
        s = 0o666 & ~umask  # avoid restatting for chmod

    sx = s & 0o100
    if st.st_nlink > 1 and bool(x) != bool(sx):
        # the file is a hardlink, break it
        with open(f, b"rb") as fp:
            data = fp.read()
        unlink(f)
        with open(f, b"wb") as fp:
            fp.write(data)

    if x and not sx:
        # Turn on +x for every +r bit when making a file executable
        # and obey umask.

    # a FS remount. Frequently we can detect it if files are created
    # with exec bit on.

    try:
        EXECFLAGS = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
        basedir = os.path.join(path, b'.hg')
        cachedir = os.path.join(basedir, b'wcache')
        storedir = os.path.join(basedir, b'store')
        if not os.path.exists(cachedir):
            try:
                # we want to create the 'cache' directory, not the '.hg' one.
                # Automatically creating '.hg' directory could silently spawn
                # invalid Mercurial repositories. That seems like a bad idea.

                    copymode(basedir, cachedir)
            except (IOError, OSError):
                # we other fallback logic triggers
                pass
        if os.path.isdir(cachedir):
            checkisexec = os.path.join(cachedir, b'checkisexec')
            checknoexec = os.path.join(cachedir, b'checknoexec')

            try:
                m = os.stat(checkisexec).st_mode
            except OSError as e:
                if e.errno != errno.ENOENT:

                    try:
                        m = os.stat(checknoexec).st_mode
                    except OSError as e:
                        if e.errno != errno.ENOENT:
                            raise
                        open(checknoexec, b'w').close()  # might fail
                        m = os.stat(checknoexec).st_mode
                    if m & EXECFLAGS == 0:
                        # check-exec is exec and check-no-exec is not exec
                        return True
                    # checknoexec exists but is exec - delete it

            checkdir = cachedir
        else:
            # check directly in path and don't leave checkisexec behind
            checkdir = path
            checkisexec = None
        fh, fn = pycompat.mkstemp(dir=checkdir, prefix=b'hg-checkexec-')
        try:
            os.close(fh)
            m = os.stat(fn).st_mode
            if m & EXECFLAGS == 0:
                os.chmod(fn, m & 0o777 | EXECFLAGS)

def checklink(path):
    """check whether the given path is on a symlink-capable filesystem"""
    # mktemp is not racy because symlink creation will fail if the
    # file already exists
    while True:
        cachedir = os.path.join(path, b'.hg', b'wcache')
        checklink = os.path.join(cachedir, b'checklink')
        # try fast path, read only
        if os.path.islink(checklink):
            return True
        if os.path.isdir(cachedir):
            checkdir = cachedir

        name = pycompat.fsencode(name)
        try:
            fd = None
            if cachedir is None:
                fd = pycompat.namedtempfile(
                    dir=checkdir, prefix=b'hg-checklink-'
                )
                target = os.path.basename(fd.name)
            else:
                # create a fixed file to link to; doesn't matter if it
                # already exists.
                target = b'checklink-target'
                try:
                    fullpath = os.path.join(cachedir, target)
                    open(fullpath, b'w').close()
                except IOError as inst:
                    if inst[0] == errno.EACCES:
                        # If we can't write to cachedir, just pretend
                        # that the fs is readonly and by association
                        # that the fs won't support symlinks. This

    def normcasefallback(path):
        try:
            u = path.decode('utf-8')
        except UnicodeDecodeError:
            # OS X percent-encodes any bytes that aren't valid utf-8
            s = b''
            pos = 0
            l = len(path)
            while pos < l:
                try:
                    c = encoding.getutf8char(path, pos)
                    pos += len(c)
                except ValueError:
                    c = b'%%%02X' % ord(path[pos : pos + 1])
                    pos += 1
                s += c

            u = s.decode('utf-8')

        enc = unicodedata.normalize(r'NFD', u).lower().encode('utf-8')
        # drop HFS+ ignored characters
        return encoding.hfsignoreclean(enc)


if pycompat.sysplatform == b'cygwin':
    # workaround for cygwin, in which mount point part of path is
    # treated as case sensitive, even though underlying NTFS is case
    # insensitive.

    # default mount points
    cygwinmountpoints = sorted(
        [b"/usr/bin", b"/usr/lib", b"/cygdrive",], reverse=True
    )

    # use upper-ing as normcase as same as NTFS workaround
    def normcase(path):
        pathlen = len(path)

_needsshellquote = None


def shellquote(s):
    if pycompat.sysplatform == b'OpenVMS':
        return b'"%s"' % s
    global _needsshellquote
    if _needsshellquote is None:
        _needsshellquote = re.compile(br'[^a-zA-Z0-9._/+-]').search
    if s and not _needsshellquote(s):
        # "s" shouldn't have to be quoted
        return s
    else:
        return b"'%s'" % s.replace(b"'", b"'\\''")


def shellsplit(s):
    """Parse a command string in POSIX shell way (best-effort)"""
    return pycompat.shlexsplit(s, posix=True)

    return cmd


def testpid(pid):
    '''return False if pid dead, True if running or not sure'''
    if pycompat.sysplatform == b'OpenVMS':
        return True
    try:
        os.kill(pid, 0)
        return True
    except OSError as inst:

def findexe(command):
    '''Find executable for command searching like which does.
    If command is a basename then PATH is searched for command.
    PATH isn't searched if command is an absolute or relative path.
    If command isn't found None is returned.'''
    if pycompat.sysplatform == b'OpenVMS':
        return command

    def findexisting(executable):
        b'Will return executable if existing file'
        if os.path.isfile(executable) and os.access(executable, os.X_OK):
            return executable
        return None

    if pycompat.ossep in command:
        return findexisting(command)

    if pycompat.sysplatform == b'plan9':
        return findexisting(os.path.join(b'/bin', command))

    for path in encoding.environ.get(b'PATH', b'').split(pycompat.ospathsep):
        executable = findexisting(os.path.join(path, command))
        if executable is not None:
            return executable
    return None

                    break
                chunks.append(s)
            except IOError:
                break

        return b''.join(chunks)
    finally:
        fcntl.fcntl(pipe, fcntl.F_SETFL, oldflags)


def bindunixsocket(sock, path):

    # AF_UNIX path has very small length limit (107 chars) on common
    # platforms (see sys/un.h)
    dirname, basename = os.path.split(path)
    bakwdfd = None
    if dirname:
        bakwdfd = os.open(b'.', os.O_DIRECTORY)
        os.chdir(dirname)
    sock.bind(basename)
    if bakwdfd:
        os.fchdir(bakwdfd)
        os.close(bakwdfd)