mercurial-scm/hg: comparison mercurial/posix.py

equal deleted inserted replaced

-:0a91aba258e0
+:58dff81ffba1
 import sys
 import tempfile
 import unicodedata
 from typing import (
+Iterable,
+Iterator,
 List,
 NoReturn,
 Optional,
+Sequence,
+Union,
 )
 from .i18n import _
 from .pycompat import (
 getattr,
 if nh:
 return nh, ht[1]
 return ht[0] + b'/', ht[1]
-def openhardlinks():
+def openhardlinks() -> bool:
 '''return true if it is safe to hold open file handles to hardlinks'''
 return True
 def nlinks(name: bytes) -> int:
 '''return number of hardlinks for the given file'''
 return os.lstat(name).st_nlink
-def parsepatchoutput(output_line):
+def parsepatchoutput(output_line: bytes) -> bytes:
 """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
 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):
+def sshargs(
+sshcmd: bytes, host: bytes, user: Optional[bytes], port: Optional[bytes]
+) -> bytes:
 '''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
 if port:
 args = b'-p %s %s' % (shellquote(port), args)
 return args
-def isexec(f):
+def isexec(f: bytes) -> bool:
 """check whether a file is executable"""
 return os.lstat(f).st_mode & 0o100 != 0
-def setflags(f, l, x):
+def setflags(f: bytes, l: bool, x: bool) -> None:
 st = os.lstat(f)
 s = st.st_mode
 if l:
 if not stat.S_ISLNK(s):
 # switch file to link
 elif not x and sx:
 # Turn off all +x bits
 os.chmod(f, s & 0o666)
-def copymode(src, dst, mode=None, enforcewritable=False):
+def copymode(
+src: bytes,
+dst: bytes,
+mode: Optional[bytes] = None,
+enforcewritable: bool = False,
+) -> None:
 """Copy the file mode from the file at path src to dst.
 If src doesn't exist, we're using mode instead. If mode is None, we're
 using umask."""
 try:
 st_mode = os.lstat(src).st_mode & 0o777
 new_mode |= stat.S_IWUSR
 os.chmod(dst, new_mode)
-def checkexec(path):
+def checkexec(path: bytes) -> bool:
 """
 Check whether the given path is on a filesystem with UNIX-like exec flags
 Requires a directory (like /foo/.hg)
 """
 except (IOError, OSError):
 # we don't care, the user probably won't be able to commit anyway
 return False
-def checklink(path):
+def checklink(path: bytes) -> bool:
 """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')
 Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
 """
 return getattr(osutil, 'getfstype', lambda x: None)(dirpath)
-def get_password():
+def get_password() -> bytes:
 return encoding.strtolocal(getpass.getpass(''))
-def setbinary(fd):
+def setbinary(fd) -> None:
 pass
-def pconvert(path):
+def pconvert(path: bytes) -> bytes:
 return path
-def localpath(path):
+def localpath(path: bytes) -> bytes:
 return path
 def samefile(fpath1: bytes, fpath2: bytes) -> bool:
 """Returns whether path1 and path2 refer to the same file. This is only
 st2 = os.lstat(fpath2)
 return st1.st_dev == st2.st_dev
 # os.path.normcase is a no-op, which doesn't help us on non-native filesystems
-def normcase(path):
+def normcase(path: bytes) -> bytes:
 return path.lower()
 # what normcase does to ASCII strings
 normcasespec = encoding.normcasespecs.lower
 # fallback normcase function for non-ASCII strings
 normcasefallback = normcase
 if pycompat.isdarwin:
-def normcase(path):
+def normcase(path: bytes) -> bytes:
 """
 Normalize a filename for OS X-compatible comparison:
 - escape-encode invalid characters
 - decompose to NFD
 - lowercase
 except UnicodeDecodeError:
 return normcasefallback(path)
 normcasespec = encoding.normcasespecs.lower
-def normcasefallback(path):
+def normcasefallback(path: bytes) -> bytes:
 try:
 u = path.decode('utf-8')
 except UnicodeDecodeError:
 # OS X percent-encodes any bytes that aren't valid utf-8
 s = b''
 ],
 reverse=True,
 )
 # use upper-ing as normcase as same as NTFS workaround
-def normcase(path):
+def normcase(path: bytes) -> bytes:
 pathlen = len(path)
 if (pathlen == 0) or (path[0] != pycompat.ossep):
 # treat as relative
 return encoding.upper(path)
 # Cygwin translates native ACLs to POSIX permissions,
 # but these translations are not supported by native
 # tools, so the exec bit tends to be set erroneously.
 # Therefore, disable executable bit access on Cygwin.
-def checkexec(path):
+def checkexec(path: bytes) -> bool:
 return False
 # Similarly, Cygwin's symlink emulation is likely to create
 # problems when Mercurial is used from both Cygwin and native
 # Windows, with other native tools, or on shared volumes
-def checklink(path):
+def checklink(path: bytes) -> bool:
 return False
 _needsshellquote = None
-def shellquote(s):
+def shellquote(s: bytes) -> bytes:
 if pycompat.sysplatform == b'OpenVMS':
 return b'"%s"' % s
 global _needsshellquote
 if _needsshellquote is None:
 _needsshellquote = re.compile(br'[^a-zA-Z0-9._/+-]').search
 return s
 else:
 return b"'%s'" % s.replace(b"'", b"'\\''")
-def shellsplit(s):
+def shellsplit(s: bytes) -> List[bytes]:
 """Parse a command string in POSIX shell way (best-effort)"""
 return pycompat.shlexsplit(s, posix=True)
 def testpid(pid: int) -> bool:
 return True
 except OSError as inst:
 return inst.errno != errno.ESRCH
-def isowner(st):
+def isowner(st: os.stat_result) -> bool:
 """Return True if the stat object st is from the current user."""
 return st.st_uid == os.getuid()
-def findexe(command):
+def findexe(command: bytes) -> Optional[bytes]:
 """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):
+def findexisting(executable: bytes) -> Optional[bytes]:
 b'Will return executable if existing file'
 if os.path.isfile(executable) and os.access(executable, os.X_OK):
 return executable
 return None
 _wantedkinds = {stat.S_IFREG, stat.S_IFLNK}
-def statfiles(files):
+def statfiles(files: Sequence[bytes]) -> Iterator[Optional[os.stat_result]]:
 """Stat each file in files. Yield each stat, or None if a file does not
 exist or has a type we don't care about."""
 lstat = os.lstat
 getkind = stat.S_IFMT
 for nf in files:
 def getuser() -> bytes:
 '''return name of current user'''
 return pycompat.fsencode(getpass.getuser())
-def username(uid=None):
+def username(uid: Optional[int] = None) -> Optional[bytes]:
 """Return the name of the user with the given uid.
 If uid is None, return the name of the current user."""
 if uid is None:
 return pycompat.fsencode(pwd.getpwuid(uid)[0])
 except KeyError:
 return b'%d' % uid
-def groupname(gid=None):
+def groupname(gid: Optional[int] = None) -> Optional[bytes]:
 """Return the name of the group with the given gid.
 If gid is None, return the name of the current group."""
 if gid is None:
 return pycompat.fsencode(grp.getgrgid(gid)[0])
 except KeyError:
 return pycompat.bytestr(gid)
-def groupmembers(name):
+def groupmembers(name: bytes) -> List[bytes]:
 """Return the list of members of the group with the given
 name, KeyError if the group does not exist.
 """
 name = pycompat.fsdecode(name)
 return pycompat.rapply(pycompat.fsencode, list(grp.getgrnam(name).gr_mem))
 def makedir(path: bytes, notindexed: bool) -> None:
 os.mkdir(path)
-def lookupreg(key, name=None, scope=None):
+def lookupreg(
+key: bytes,
+name: Optional[bytes] = None,
+scope: Optional[Union[int, Iterable[int]]] = None,
+) -> Optional[bytes]:
 return None
 def hidewindow() -> None:
 """Hide current shell window.
 def __ne__(self, other):
 return not self == other
-def statislink(st):
+def statislink(st: Optional[os.stat_result]) -> bool:
 '''check whether a stat result is a symlink'''
-return st and stat.S_ISLNK(st.st_mode)
+return stat.S_ISLNK(st.st_mode) if st else False
-def statisexec(st):
+def statisexec(st: Optional[os.stat_result]) -> bool:
 '''check whether a stat result is an executable file'''
-return st and (st.st_mode & 0o100 != 0)
+return (st.st_mode & 0o100 != 0) if st else False
 def poll(fds):
 """block until something happens on any file descriptor
 except ValueError:  # out of range file descriptor
 raise NotImplementedError()
 return sorted(list(set(sum(res, []))))
-def readpipe(pipe):
+def readpipe(pipe) -> bytes:
 """Read all available data from a pipe."""
 # We can't fstat() a pipe because Linux will always report 0.
 # So, we set the pipe to non-blocking mode and read everything
 # that's available.
 flags = fcntl.fcntl(pipe, fcntl.F_GETFL)
 return b''.join(chunks)
 finally:
 fcntl.fcntl(pipe, fcntl.F_SETFL, oldflags)
-def bindunixsocket(sock, path):
+def bindunixsocket(sock, path: bytes) -> None:
 """Bind the UNIX domain socket to the specified path"""
 # use relative path instead of full path at bind() if possible, since
 # AF_UNIX path has very small length limit (107 chars) on common
 # platforms (see sys/un.h)
 dirname, basename = os.path.split(path)

changeset 49812	58dff81ffba1
parent 49811	0a91aba258e0
child 49815	464fe8b8f474