typing: add type hints to mercurial/win32.py

These are the low level functions that are imported by the mercurial.windows
module, which is in turn imported by mercurial.utils as the platform module.
Pretty straightforward, but pytype inferred very little of it, likely because of
the heavy ctypes usage.  It also seems to trigger a pytype bug in procutil, now
that it has an idea of the underlying function type, so disable that warning to
maintain a working test.

mercurial/posix.py

@@ -1,753 +1,753 b''
 # posix.py - Posix utility function implementations for Mercurial
 #
 #  Copyright 2005-2009 Olivia Mackall <[e-mail]> and others
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import errno
 import fcntl
 import getpass
 import grp
 import os
 import pwd
 import re
 import select
 import stat
 import sys
 import tempfile
 import unicodedata
 
 from .i18n import _
 from .pycompat import (
     getattr,
     open,
 )
 from . import (
     encoding,
     error,
     policy,
     pycompat,
 )
 
 osutil = policy.importmod('osutil')
 
 normpath = os.path.normpath
 samestat = os.path.samestat
 abspath = os.path.abspath  # re-exports
 
 try:
     oslink = os.link
 except AttributeError:
     # Some platforms build Python without os.link on systems that are
     # 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
 rename = os.rename
 removedirs = os.removedirs
 expandglobs = False
 
 umask = os.umask(0)
 os.umask(umask)
 
 posixfile = open
 
 
 def split(p):
     """Same as posixpath.split, but faster
 
     >>> import posixpath
     >>> for f in [b'/absolute/path/to/file',
     ...           b'relative/path/to/file',
     ...           b'file_alone',
     ...           b'path/to/directory/',
     ...           b'/multiple/path//separators',
     ...           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 nlinks(name):
     '''return number of hardlinks for the given file'''
     return os.lstat(name).st_nlink
 
 
 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"""
     return os.lstat(f).st_mode & 0o100 != 0
 
 
 def setflags(f, l, x):
     st = os.lstat(f)
     s = st.st_mode
     if l:
         if not stat.S_ISLNK(s):
             # switch file to link
             with open(f, b'rb') as fp:
                 data = fp.read()
             unlink(f)
             try:
                 os.symlink(data, f)
             except OSError:
                 # failed to make a link, rewrite file
                 with open(f, b"wb") as fp:
                     fp.write(data)
 
         # no chmod needed at this point
         return
     if stat.S_ISLNK(s):
         # switch link to file
         data = os.readlink(f)
         unlink(f)
         with open(f, b"wb") as fp:
             fp.write(data)
         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.
         os.chmod(f, s | (s & 0o444) >> 2 & ~umask)
     elif not x and sx:
         # Turn off all +x bits
         os.chmod(f, s & 0o666)
 
 
 def copymode(src, dst, mode=None, enforcewritable=False):
     """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
     except FileNotFoundError:
         st_mode = mode
         if st_mode is None:
             st_mode = ~umask
         st_mode &= 0o666
 
     new_mode = st_mode
 
     if enforcewritable:
         new_mode |= stat.S_IWUSR
 
     os.chmod(dst, new_mode)
 
 
 def checkexec(path):
     """
     Check whether the given path is on a filesystem with UNIX-like exec flags
 
     Requires a directory (like /foo/.hg)
     """
 
     # VFAT on some Linux versions can flip mode but it doesn't persist
     # 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.
                 os.mkdir(cachedir)
                 if os.path.exists(storedir):
                     copymode(storedir, cachedir)
                 else:
                     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 FileNotFoundError:
                 # checkisexec does not exist - fall through ...
                 pass
             else:
                 # checkisexec exists, check if it actually is exec
                 if m & EXECFLAGS != 0:
                     # ensure checkisexec exists, check it isn't exec
                     try:
                         m = os.stat(checknoexec).st_mode
                     except FileNotFoundError:
                         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
                     unlink(checknoexec)
                 # checkisexec exists but is not exec - delete it
                 unlink(checkisexec)
 
             # check using one file, leave it as checkisexec
             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)
                 if os.stat(fn).st_mode & EXECFLAGS != 0:
                     if checkisexec is not None:
                         os.rename(fn, checkisexec)
                         fn = None
                     return True
         finally:
             if fn is not None:
                 unlink(fn)
     except (IOError, OSError):
         # we don't care, the user probably won't be able to commit anyway
         return False
 
 
 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
         else:
             checkdir = path
             cachedir = None
         name = tempfile.mktemp(
             dir=pycompat.fsdecode(checkdir), prefix=r'checklink-'
         )
         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 PermissionError:
                     # 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
                     # seems like the least dangerous way to avoid
                     # data loss.
                     return False
             try:
                 os.symlink(target, name)
                 if cachedir is None:
                     unlink(name)
                 else:
                     try:
                         os.rename(name, checklink)
                     except OSError:
                         unlink(name)
                 return True
             except FileExistsError:
                 # link creation might race, try again
                 continue
             finally:
                 if fd is not None:
                     fd.close()
         except AttributeError:
             return False
         except OSError as inst:
             # sshfs might report failure while successfully creating the link
             if inst.errno == errno.EIO and os.path.exists(name):
                 unlink(name)
             return False
 
 
 def checkosfilename(path):
     """Check that the base-relative path is a valid filename on this platform.
     Returns None if the path is ok, or a UI string describing the problem."""
     return None  # on posix platforms, every path is ok
 
 
 def getfsmountpoint(dirpath):
     """Get the filesystem mount point from a directory (best-effort)
 
     Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
     """
     return getattr(osutil, 'getfsmountpoint', lambda x: None)(dirpath)
 
 
 def getfstype(dirpath):
     """Get the filesystem type name from a directory (best-effort)
 
     Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
     """
     return getattr(osutil, 'getfstype', lambda x: None)(dirpath)
 
 
 def get_password():
     return encoding.strtolocal(getpass.getpass(''))
 
 
 def setbinary(fd):
     pass
 
 
 def pconvert(path):
     return path
 
 
 def localpath(path):
     return path
 
 
 def samefile(fpath1, fpath2):
     """Returns whether path1 and path2 refer to the same file. This is only
     guaranteed to work for files, not directories."""
     return os.path.samefile(fpath1, fpath2)
 
 
 def samedevice(fpath1, fpath2):
     """Returns whether fpath1 and fpath2 are on the same device. This is only
     guaranteed to work for files, not directories."""
     st1 = os.lstat(fpath1)
     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):
     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):
         """
         Normalize a filename for OS X-compatible comparison:
         - escape-encode invalid characters
         - decompose to NFD
         - lowercase
         - omit ignored characters [200c-200f, 202a-202e, 206a-206f,feff]
 
         >>> normcase(b'UPPER')
         'upper'
         >>> normcase(b'Caf\\xc3\\xa9')
         'cafe\\xcc\\x81'
         >>> normcase(b'\\xc3\\x89')
         'e\\xcc\\x81'
         >>> normcase(b'\\xb8\\xca\\xc3\\xca\\xbe\\xc8.JPG') # issue3918
         '%b8%ca%c3\\xca\\xbe%c8.jpg'
         """
 
         try:
             return encoding.asciilower(path)  # exception for non-ASCII
         except UnicodeDecodeError:
             return normcasefallback(path)
 
     normcasespec = encoding.normcasespecs.lower
 
     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')
 
         # Decompose then lowercase (HFS+ technote specifies lower)
         enc = unicodedata.normalize('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)
         if (pathlen == 0) or (path[0] != pycompat.ossep):
             # treat as relative
             return encoding.upper(path)
 
         # to preserve case of mountpoint part
         for mp in cygwinmountpoints:
             if not path.startswith(mp):
                 continue
 
             mplen = len(mp)
             if mplen == pathlen:  # mount point itself
                 return mp
             if path[mplen] == pycompat.ossep:
                 return mp + encoding.upper(path[mplen:])
 
         return encoding.upper(path)
 
     normcasespec = encoding.normcasespecs.other
     normcasefallback = normcase
 
     # 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):
         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):
         return False
 
 
 _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)
 
 
 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:
         return inst.errno != errno.ESRCH
 
 
 def isowner(st):
     """Return True if the stat object st is from the current user."""
     return st.st_uid == os.getuid()
 
 
 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
 
 
 def setsignalhandler():
     pass
 
 
 _wantedkinds = {stat.S_IFREG, stat.S_IFLNK}
 
 
 def statfiles(files):
     """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:
         try:
             st = lstat(nf)
             if getkind(st.st_mode) not in _wantedkinds:
                 st = None
         except (FileNotFoundError, NotADirectoryError):
             st = None
         yield st
 
 
 def getuser():
     '''return name of current user'''
     return pycompat.fsencode(getpass.getuser())
 
 
 def username(uid=None):
     """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:
         uid = os.getuid()
     try:
         return pycompat.fsencode(pwd.getpwuid(uid)[0])
     except KeyError:
         return b'%d' % uid
 
 
 def groupname(gid=None):
     """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:
         gid = os.getgid()
     try:
         return pycompat.fsencode(grp.getgrgid(gid)[0])
     except KeyError:
         return pycompat.bytestr(gid)
 
 
 def groupmembers(name):
     """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 spawndetached(args):
+def spawndetached(args) -> int:
     return os.spawnvp(os.P_NOWAIT | getattr(os, 'P_DETACH', 0), args[0], args)
 
 
 def gethgcmd():
     return sys.argv[:1]
 
 
 def makedir(path, notindexed):
     os.mkdir(path)
 
 
 def lookupreg(key, name=None, scope=None):
     return None
 
 
 def hidewindow():
     """Hide current shell window.
 
     Used to hide the window opened when starting asynchronous
     child process under Windows, unneeded on other systems.
     """
     pass
 
 
 class cachestat:
     def __init__(self, path):
         self.stat = os.stat(path)
 
     def cacheable(self):
         return bool(self.stat.st_ino)
 
     __hash__ = object.__hash__
 
     def __eq__(self, other):
         try:
             # Only dev, ino, size, mtime and atime are likely to change. Out
             # of these, we shouldn't compare atime but should compare the
             # rest. However, one of the other fields changing indicates
             # something fishy going on, so return False if anything but atime
             # changes.
             return (
                 self.stat.st_mode == other.stat.st_mode
                 and self.stat.st_ino == other.stat.st_ino
                 and self.stat.st_dev == other.stat.st_dev
                 and self.stat.st_nlink == other.stat.st_nlink
                 and self.stat.st_uid == other.stat.st_uid
                 and self.stat.st_gid == other.stat.st_gid
                 and self.stat.st_size == other.stat.st_size
                 and self.stat[stat.ST_MTIME] == other.stat[stat.ST_MTIME]
                 and self.stat[stat.ST_CTIME] == other.stat[stat.ST_CTIME]
             )
         except AttributeError:
             return False
 
     def __ne__(self, other):
         return not self == other
 
 
 def statislink(st):
     '''check whether a stat result is a symlink'''
     return st and stat.S_ISLNK(st.st_mode)
 
 
 def statisexec(st):
     '''check whether a stat result is an executable file'''
     return st and (st.st_mode & 0o100 != 0)
 
 
 def poll(fds):
     """block until something happens on any file descriptor
 
     This is a generic helper that will check for any activity
     (read, write.  exception) and return the list of touched files.
 
     In unsupported cases, it will raise a NotImplementedError"""
     try:
         res = select.select(fds, fds, fds)
     except ValueError:  # out of range file descriptor
         raise NotImplementedError()
     return sorted(list(set(sum(res, []))))
 
 
 def readpipe(pipe):
     """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)
     flags |= os.O_NONBLOCK
     oldflags = fcntl.fcntl(pipe, fcntl.F_SETFL, flags)
 
     try:
         chunks = []
         while True:
             try:
                 s = pipe.read()
                 if not s:
                     break
                 chunks.append(s)
             except IOError:
                 break
 
         return b''.join(chunks)
     finally:
         fcntl.fcntl(pipe, fcntl.F_SETFL, oldflags)
 
 
 def bindunixsocket(sock, path):
     """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)
     bakwdfd = None
 
     try:
         if dirname:
             bakwdfd = os.open(b'.', os.O_DIRECTORY)
             os.chdir(dirname)
         sock.bind(basename)
         if bakwdfd:
             os.fchdir(bakwdfd)
     finally:
         if bakwdfd:
             os.close(bakwdfd)

mercurial/utils/procutil.py

@@ -1,795 +1,801 b''
 # procutil.py - utility for managing processes and executable environment
 #
 #  Copyright 2005 K. Thananchayan <[e-mail]>
 #  Copyright 2005-2007 Olivia Mackall <[e-mail]>
 #  Copyright 2006 Vadim Gelfer <[e-mail]>
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import contextlib
 import errno
 import io
 import os
 import signal
 import subprocess
 import sys
 import threading
 import time
 
 from typing import (
     BinaryIO,
 )
 
 from ..i18n import _
 from ..pycompat import (
     getattr,
     open,
 )
 
 from .. import (
     encoding,
     error,
     policy,
     pycompat,
     typelib,
 )
 
 # Import like this to keep import-checker happy
 from ..utils import resourceutil
 
 osutil = policy.importmod('osutil')
 
 if pycompat.iswindows:
     from .. import windows as platform
 else:
     from .. import posix as platform
 
 
 def isatty(fp):
     try:
         return fp.isatty()
     except AttributeError:
         return False
 
 
 class BadFile(io.RawIOBase):
     """Dummy file object to simulate closed stdio behavior"""
 
     def readinto(self, b):
         raise IOError(errno.EBADF, 'Bad file descriptor')
 
     def write(self, b):
         raise IOError(errno.EBADF, 'Bad file descriptor')
 
 
 class LineBufferedWrapper:
     def __init__(self, orig):
         self.orig = orig
 
     def __getattr__(self, attr):
         return getattr(self.orig, attr)
 
     def write(self, s):
         orig = self.orig
         res = orig.write(s)
         if s.endswith(b'\n'):
             orig.flush()
         return res
 
 
 # pytype: disable=attribute-error
 io.BufferedIOBase.register(LineBufferedWrapper)
 # pytype: enable=attribute-error
 
 
 def make_line_buffered(stream):
     # First, check if we need to wrap the stream.
     check_stream = stream
     while True:
         if isinstance(check_stream, WriteAllWrapper):
             check_stream = check_stream.orig
         elif pycompat.iswindows and isinstance(
             check_stream,
             # pytype: disable=module-attr
             platform.winstdout
             # pytype: enable=module-attr
         ):
             check_stream = check_stream.fp
         else:
             break
     if isinstance(check_stream, io.RawIOBase):
         # The stream is unbuffered, we don't need to emulate line buffering.
         return stream
     elif isinstance(check_stream, io.BufferedIOBase):
         # The stream supports some kind of buffering. We can't assume that
         # lines are flushed. Fall back to wrapping the stream.
         pass
     else:
         raise NotImplementedError(
             "can't determine whether stream is buffered or not"
         )
 
     if isinstance(stream, LineBufferedWrapper):
         return stream
     return LineBufferedWrapper(stream)
 
 
 def unwrap_line_buffered(stream):
     if isinstance(stream, LineBufferedWrapper):
         assert not isinstance(stream.orig, LineBufferedWrapper)
         return stream.orig
     return stream
 
 
 class WriteAllWrapper(typelib.BinaryIO_Proxy):
     def __init__(self, orig: BinaryIO):
         self.orig = orig
 
     def __getattr__(self, attr):
         return getattr(self.orig, attr)
 
     def write(self, s):
         write1 = self.orig.write
         m = memoryview(s)
         total_to_write = len(s)
         total_written = 0
         while total_written < total_to_write:
             c = write1(m[total_written:])
             if c:
                 total_written += c
         return total_written
 
 
 # pytype: disable=attribute-error
 io.IOBase.register(WriteAllWrapper)
 # pytype: enable=attribute-error
 
 
 def _make_write_all(stream):
     if isinstance(stream, WriteAllWrapper):
         return stream
     if isinstance(stream, io.BufferedIOBase):
         # The io.BufferedIOBase.write() contract guarantees that all data is
         # written.
         return stream
     # In general, the write() method of streams is free to write only part of
     # the data.
     return WriteAllWrapper(stream)
 
 
 # Python 3 implements its own I/O streams. Unlike stdio of C library,
 # sys.stdin/stdout/stderr may be None if underlying fd is closed.
 
 # TODO: .buffer might not exist if std streams were replaced; we'll need
 # a silly wrapper to make a bytes stream backed by a unicode one.
 
 if sys.stdin is None:
     stdin = BadFile()
 else:
     stdin = sys.stdin.buffer
 if sys.stdout is None:
     stdout = BadFile()
 else:
     stdout = _make_write_all(sys.stdout.buffer)
 if sys.stderr is None:
     stderr = BadFile()
 else:
     stderr = _make_write_all(sys.stderr.buffer)
 
 if pycompat.iswindows:
     # Work around Windows bugs.
     stdout = platform.winstdout(stdout)  # pytype: disable=module-attr
     stderr = platform.winstdout(stderr)  # pytype: disable=module-attr
 if isatty(stdout):
     # The standard library doesn't offer line-buffered binary streams.
     stdout = make_line_buffered(stdout)
 
 findexe = platform.findexe
 _gethgcmd = platform.gethgcmd
 getuser = platform.getuser
 getpid = os.getpid
 hidewindow = platform.hidewindow
 readpipe = platform.readpipe
 setbinary = platform.setbinary
 setsignalhandler = platform.setsignalhandler
 shellquote = platform.shellquote
 shellsplit = platform.shellsplit
 spawndetached = platform.spawndetached
 sshargs = platform.sshargs
 testpid = platform.testpid
 
 try:
     setprocname = osutil.setprocname
 except AttributeError:
     pass
 try:
     unblocksignal = osutil.unblocksignal
 except AttributeError:
     pass
 
 closefds = pycompat.isposix
 
 
 def explainexit(code):
     """return a message describing a subprocess status
     (codes from kill are negative - not os.system/wait encoding)"""
     if code >= 0:
         return _(b"exited with status %d") % code
     return _(b"killed by signal %d") % -code
 
 
 class _pfile:
     """File-like wrapper for a stream opened by subprocess.Popen()"""
 
     def __init__(self, proc, fp):
         self._proc = proc
         self._fp = fp
 
     def close(self):
         # unlike os.popen(), this returns an integer in subprocess coding
         self._fp.close()
         return self._proc.wait()
 
     def __iter__(self):
         return iter(self._fp)
 
     def __getattr__(self, attr):
         return getattr(self._fp, attr)
 
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_value, exc_tb):
         self.close()
 
 
 def popen(cmd, mode=b'rb', bufsize=-1):
     if mode == b'rb':
         return _popenreader(cmd, bufsize)
     elif mode == b'wb':
         return _popenwriter(cmd, bufsize)
     raise error.ProgrammingError(b'unsupported mode: %r' % mode)
 
 
 def _popenreader(cmd, bufsize):
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=bufsize,
         close_fds=closefds,
         stdout=subprocess.PIPE,
     )
     return _pfile(p, p.stdout)
 
 
 def _popenwriter(cmd, bufsize):
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=bufsize,
         close_fds=closefds,
         stdin=subprocess.PIPE,
     )
     return _pfile(p, p.stdin)
 
 
 def popen2(cmd, env=None):
     # Setting bufsize to -1 lets the system decide the buffer size.
     # The default for bufsize is 0, meaning unbuffered. This leads to
     # poor performance on Mac OS X: http://bugs.python.org/issue4194
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=-1,
         close_fds=closefds,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         env=tonativeenv(env),
     )
     return p.stdin, p.stdout
 
 
 def popen3(cmd, env=None):
     stdin, stdout, stderr, p = popen4(cmd, env)
     return stdin, stdout, stderr
 
 
 def popen4(cmd, env=None, bufsize=-1):
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         bufsize=bufsize,
         close_fds=closefds,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
         env=tonativeenv(env),
     )
     return p.stdin, p.stdout, p.stderr, p
 
 
 def pipefilter(s, cmd):
     '''filter string S through command CMD, returning its output'''
     p = subprocess.Popen(
         tonativestr(cmd),
         shell=True,
         close_fds=closefds,
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
     )
     pout, perr = p.communicate(s)
     return pout
 
 
 def tempfilter(s, cmd):
     """filter string S through a pair of temporary files with CMD.
     CMD is used as a template to create the real command to be run,
     with the strings INFILE and OUTFILE replaced by the real names of
     the temporary files generated."""
     inname, outname = None, None
     try:
         infd, inname = pycompat.mkstemp(prefix=b'hg-filter-in-')
         fp = os.fdopen(infd, 'wb')
         fp.write(s)
         fp.close()
         outfd, outname = pycompat.mkstemp(prefix=b'hg-filter-out-')
         os.close(outfd)
         cmd = cmd.replace(b'INFILE', inname)
         cmd = cmd.replace(b'OUTFILE', outname)
         code = system(cmd)
         if pycompat.sysplatform == b'OpenVMS' and code & 1:
             code = 0
         if code:
             raise error.Abort(
                 _(b"command '%s' failed: %s") % (cmd, explainexit(code))
             )
         with open(outname, b'rb') as fp:
             return fp.read()
     finally:
         try:
             if inname:
                 os.unlink(inname)
         except OSError:
             pass
         try:
             if outname:
                 os.unlink(outname)
         except OSError:
             pass
 
 
 _filtertable = {
     b'tempfile:': tempfilter,
     b'pipe:': pipefilter,
 }
 
 
 def filter(s, cmd):
     """filter a string through a command that transforms its input to its
     output"""
     for name, fn in _filtertable.items():
         if cmd.startswith(name):
             return fn(s, cmd[len(name) :].lstrip())
     return pipefilter(s, cmd)
 
 
 _hgexecutable = None
 
 
 def hgexecutable():
     """return location of the 'hg' executable.
 
     Defaults to $HG or 'hg' in the search path.
     """
     if _hgexecutable is None:
         hg = encoding.environ.get(b'HG')
         mainmod = sys.modules['__main__']
         if hg:
             _sethgexecutable(hg)
         elif resourceutil.mainfrozen():
             if getattr(sys, 'frozen', None) == 'macosx_app':
                 # Env variable set by py2app
                 _sethgexecutable(encoding.environ[b'EXECUTABLEPATH'])
             else:
                 _sethgexecutable(pycompat.sysexecutable)
         elif (
             not pycompat.iswindows
             and os.path.basename(getattr(mainmod, '__file__', '')) == 'hg'
         ):
             _sethgexecutable(pycompat.fsencode(mainmod.__file__))
         else:
             _sethgexecutable(
                 findexe(b'hg') or os.path.basename(pycompat.sysargv[0])
             )
     return _hgexecutable
 
 
 def _sethgexecutable(path):
     """set location of the 'hg' executable"""
     global _hgexecutable
     _hgexecutable = path
 
 
 def _testfileno(f, stdf):
     fileno = getattr(f, 'fileno', None)
     try:
         return fileno and fileno() == stdf.fileno()
     except io.UnsupportedOperation:
         return False  # fileno() raised UnsupportedOperation
 
 
 def isstdin(f):
     return _testfileno(f, sys.__stdin__)
 
 
 def isstdout(f):
     return _testfileno(f, sys.__stdout__)
 
 
 def protectstdio(uin, uout):
     """Duplicate streams and redirect original if (uin, uout) are stdio
 
     If uin is stdin, it's redirected to /dev/null. If uout is stdout, it's
     redirected to stderr so the output is still readable.
 
     Returns (fin, fout) which point to the original (uin, uout) fds, but
     may be copy of (uin, uout). The returned streams can be considered
     "owned" in that print(), exec(), etc. never reach to them.
     """
     uout.flush()
     fin, fout = uin, uout
     if _testfileno(uin, stdin):
         newfd = os.dup(uin.fileno())
         nullfd = os.open(os.devnull, os.O_RDONLY)
         os.dup2(nullfd, uin.fileno())
         os.close(nullfd)
         fin = os.fdopen(newfd, 'rb')
     if _testfileno(uout, stdout):
         newfd = os.dup(uout.fileno())
         os.dup2(stderr.fileno(), uout.fileno())
         fout = os.fdopen(newfd, 'wb')
     return fin, fout
 
 
 def restorestdio(uin, uout, fin, fout):
     """Restore (uin, uout) streams from possibly duplicated (fin, fout)"""
     uout.flush()
     for f, uif in [(fin, uin), (fout, uout)]:
         if f is not uif:
             os.dup2(f.fileno(), uif.fileno())
             f.close()
 
 
 def shellenviron(environ=None):
     """return environ with optional override, useful for shelling out"""
 
     def py2shell(val):
         """convert python object into string that is useful to shell"""
         if val is None or val is False:
             return b'0'
         if val is True:
             return b'1'
         return pycompat.bytestr(val)
 
     env = dict(encoding.environ)
     if environ:
         env.update((k, py2shell(v)) for k, v in environ.items())
     env[b'HG'] = hgexecutable()
     return env
 
 
 if pycompat.iswindows:
 
     def shelltonative(cmd, env):
         return platform.shelltocmdexe(  # pytype: disable=module-attr
             cmd, shellenviron(env)
         )
 
     tonativestr = encoding.strfromlocal
 else:
 
     def shelltonative(cmd, env):
         return cmd
 
     tonativestr = pycompat.identity
 
 
 def tonativeenv(env):
     """convert the environment from bytes to strings suitable for Popen(), etc."""
     return pycompat.rapply(tonativestr, env)
 
 
 def system(cmd, environ=None, cwd=None, out=None):
     """enhanced shell command execution.
     run with environment maybe modified, maybe in different dir.
 
     if out is specified, it is assumed to be a file-like object that has a
     write() method. stdout and stderr will be redirected to out."""
     try:
         stdout.flush()
     except Exception:
         pass
     env = shellenviron(environ)
     if out is None or isstdout(out):
         rc = subprocess.call(
             tonativestr(cmd),
             shell=True,
             close_fds=closefds,
             env=tonativeenv(env),
             cwd=pycompat.rapply(tonativestr, cwd),
         )
     else:
         proc = subprocess.Popen(
             tonativestr(cmd),
             shell=True,
             close_fds=closefds,
             env=tonativeenv(env),
             cwd=pycompat.rapply(tonativestr, cwd),
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
         )
         for line in iter(proc.stdout.readline, b''):
             out.write(line)
         proc.wait()
         rc = proc.returncode
     if pycompat.sysplatform == b'OpenVMS' and rc & 1:
         rc = 0
     return rc
 
 
 _is_gui = None
 
 
 def _gui():
     '''Are we running in a GUI?'''
     if pycompat.isdarwin:
         if b'SSH_CONNECTION' in encoding.environ:
             # handle SSH access to a box where the user is logged in
             return False
         elif getattr(osutil, 'isgui', None):
             # check if a CoreGraphics session is available
             return osutil.isgui()
         else:
             # pure build; use a safe default
             return True
     else:
         return (
             pycompat.iswindows
             or encoding.environ.get(b"DISPLAY")
             or encoding.environ.get(b"WAYLAND_DISPLAY")
         )
 
 
 def gui():
     global _is_gui
     if _is_gui is None:
         _is_gui = _gui()
     return _is_gui
 
 
 def hgcmd():
     """Return the command used to execute current hg
 
     This is different from hgexecutable() because on Windows we want
     to avoid things opening new shell windows like batch files, so we
     get either the python call or current executable.
     """
     if resourceutil.mainfrozen():
         if getattr(sys, 'frozen', None) == 'macosx_app':
             # Env variable set by py2app
             return [encoding.environ[b'EXECUTABLEPATH']]
         else:
             return [pycompat.sysexecutable]
     return _gethgcmd()
 
 
-def rundetached(args, condfn):
+def rundetached(args, condfn) -> int:
     """Execute the argument list in a detached process.
 
     condfn is a callable which is called repeatedly and should return
     True once the child process is known to have started successfully.
     At this point, the child process PID is returned. If the child
     process fails to start or finishes before condfn() evaluates to
     True, return -1.
     """
     # Windows case is easier because the child process is either
     # successfully starting and validating the condition or exiting
     # on failure. We just poll on its PID. On Unix, if the child
     # process fails to start, it will be left in a zombie state until
     # the parent wait on it, which we cannot do since we expect a long
     # running process on success. Instead we listen for SIGCHLD telling
     # us our child process terminated.
     terminated = set()
 
     def handler(signum, frame):
         terminated.add(os.wait())
 
     prevhandler = None
     SIGCHLD = getattr(signal, 'SIGCHLD', None)
     if SIGCHLD is not None:
         prevhandler = signal.signal(SIGCHLD, handler)
     try:
         pid = spawndetached(args)
         while not condfn():
             if (pid in terminated or not testpid(pid)) and not condfn():
                 return -1
             time.sleep(0.1)
         return pid
     finally:
         if prevhandler is not None:
             signal.signal(signal.SIGCHLD, prevhandler)
 
+        # pytype seems to get confused by not having a return in the finally
+        # block, and thinks the return value should be Optional[int] here.  It
+        # appears to be https://github.com/google/pytype/issues/938, without
+        # the `with` clause.
+        pass  # pytype: disable=bad-return-type
+
 
 @contextlib.contextmanager
 def uninterruptible(warn):
     """Inhibit SIGINT handling on a region of code.
 
     Note that if this is called in a non-main thread, it turns into a no-op.
 
     Args:
       warn: A callable which takes no arguments, and returns True if the
             previous signal handling should be restored.
     """
 
     oldsiginthandler = [signal.getsignal(signal.SIGINT)]
     shouldbail = []
 
     def disabledsiginthandler(*args):
         if warn():
             signal.signal(signal.SIGINT, oldsiginthandler[0])
             del oldsiginthandler[0]
         shouldbail.append(True)
 
     try:
         try:
             signal.signal(signal.SIGINT, disabledsiginthandler)
         except ValueError:
             # wrong thread, oh well, we tried
             del oldsiginthandler[0]
         yield
     finally:
         if oldsiginthandler:
             signal.signal(signal.SIGINT, oldsiginthandler[0])
         if shouldbail:
             raise KeyboardInterrupt
 
 
 if pycompat.iswindows:
     # no fork on Windows, but we can create a detached process
     # https://msdn.microsoft.com/en-us/library/windows/desktop/ms684863.aspx
     # No stdlib constant exists for this value
     DETACHED_PROCESS = 0x00000008
     # Following creation flags might create a console GUI window.
     # Using subprocess.CREATE_NEW_CONSOLE might helps.
     # See https://phab.mercurial-scm.org/D1701 for discussion
     _creationflags = (
         DETACHED_PROCESS
         | subprocess.CREATE_NEW_PROCESS_GROUP  # pytype: disable=module-attr
     )
 
     def runbgcommand(
         script,
         env,
         shell=False,
         stdout=None,
         stderr=None,
         ensurestart=True,
         record_wait=None,
         stdin_bytes=None,
     ):
         '''Spawn a command without waiting for it to finish.'''
         # we can't use close_fds *and* redirect stdin. I'm not sure that we
         # need to because the detached process has no console connection.
 
         try:
             stdin = None
             if stdin_bytes is not None:
                 stdin = pycompat.unnamedtempfile()
                 stdin.write(stdin_bytes)
                 stdin.flush()
                 stdin.seek(0)
 
             p = subprocess.Popen(
                 pycompat.rapply(tonativestr, script),
                 shell=shell,
                 env=tonativeenv(env),
                 close_fds=True,
                 creationflags=_creationflags,
                 stdin=stdin,
                 stdout=stdout,
                 stderr=stderr,
             )
             if record_wait is not None:
                 record_wait(p.wait)
         finally:
             if stdin is not None:
                 stdin.close()
 
 
 else:
 
     def runbgcommand(
         cmd,
         env,
         shell=False,
         stdout=None,
         stderr=None,
         ensurestart=True,
         record_wait=None,
         stdin_bytes=None,
     ):
         """Spawn a command without waiting for it to finish.
 
 
         When `record_wait` is not None, the spawned process will not be fully
         detached and the `record_wait` argument will be called with a the
         `Subprocess.wait` function for the spawned process.  This is mostly
         useful for developers that need to make sure the spawned process
         finished before a certain point. (eg: writing test)"""
         if pycompat.isdarwin:
             # avoid crash in CoreFoundation in case another thread
             # calls gui() while we're calling fork().
             gui()
 
         if shell:
             script = cmd
         else:
             if isinstance(cmd, bytes):
                 cmd = [cmd]
             script = b' '.join(shellquote(x) for x in cmd)
         if record_wait is None:
             # double-fork to completely detach from the parent process
             script = b'( %s ) &' % script
             start_new_session = True
         else:
             start_new_session = False
             ensurestart = True
 
         stdin = None
 
         try:
             if stdin_bytes is None:
                 stdin = subprocess.DEVNULL
             else:
                 stdin = pycompat.unnamedtempfile()
                 stdin.write(stdin_bytes)
                 stdin.flush()
                 stdin.seek(0)
             if stdout is None:
                 stdout = subprocess.DEVNULL
             if stderr is None:
                 stderr = subprocess.DEVNULL
 
             p = subprocess.Popen(
                 script,
                 shell=True,
                 env=env,
                 close_fds=True,
                 stdin=stdin,
                 stdout=stdout,
                 stderr=stderr,
                 start_new_session=start_new_session,
             )
         except Exception:
             if record_wait is not None:
                 record_wait(255)
             raise
         finally:
             if stdin_bytes is not None and stdin is not None:
                 assert not isinstance(stdin, int)
                 stdin.close()
         if not ensurestart:
             # Even though we're not waiting on the child process,
             # we still must call waitpid() on it at some point so
             # it's not a zombie/defunct. This is especially relevant for
             # chg since the parent process won't die anytime soon.
             # We use a thread to make the overhead tiny.
             t = threading.Thread(target=lambda: p.wait)
             t.daemon = True
             t.start()
         else:
             returncode = p.wait
             if record_wait is not None:
                 record_wait(returncode)

mercurial/win32.py

@@ -1,764 +1,771 b''
 # win32.py - utility functions that use win32 API
 #
 # Copyright 2005-2009 Olivia Mackall <[e-mail]> and others
 #
 # This software may be used and distributed according to the terms of the
 # GNU General Public License version 2 or any later version.
 
 
 import ctypes
 import ctypes.wintypes as wintypes
 import errno
 import msvcrt
 import os
 import random
 import subprocess
 
+from typing import (
+    List,
+    NoReturn,
+    Optional,
+    Tuple,
+)
+
 from . import (
     encoding,
     pycompat,
 )
 
 # pytype: disable=module-attr
 _kernel32 = ctypes.windll.kernel32
 _advapi32 = ctypes.windll.advapi32
 _user32 = ctypes.windll.user32
 _crypt32 = ctypes.windll.crypt32
 # pytype: enable=module-attr
 
 _BOOL = ctypes.c_long
 _WORD = ctypes.c_ushort
 _DWORD = ctypes.c_ulong
 _UINT = ctypes.c_uint
 _LONG = ctypes.c_long
 _LPCSTR = _LPSTR = ctypes.c_char_p
 _HANDLE = ctypes.c_void_p
 _HWND = _HANDLE
 _PCCERT_CONTEXT = ctypes.c_void_p
 _MAX_PATH = wintypes.MAX_PATH
 
 _INVALID_HANDLE_VALUE = _HANDLE(-1).value
 
 # GetLastError
 _ERROR_SUCCESS = 0
 _ERROR_NO_MORE_FILES = 18
 _ERROR_INVALID_PARAMETER = 87
 _ERROR_BROKEN_PIPE = 109
 _ERROR_INSUFFICIENT_BUFFER = 122
 _ERROR_NO_DATA = 232
 
 # WPARAM is defined as UINT_PTR (unsigned type)
 # LPARAM is defined as LONG_PTR (signed type)
 if ctypes.sizeof(ctypes.c_long) == ctypes.sizeof(ctypes.c_void_p):
     _WPARAM = ctypes.c_ulong
     _LPARAM = ctypes.c_long
 elif ctypes.sizeof(ctypes.c_longlong) == ctypes.sizeof(ctypes.c_void_p):
     _WPARAM = ctypes.c_ulonglong
     _LPARAM = ctypes.c_longlong
 
 
 class _FILETIME(ctypes.Structure):
     _fields_ = [('dwLowDateTime', _DWORD), ('dwHighDateTime', _DWORD)]
 
 
 class _BY_HANDLE_FILE_INFORMATION(ctypes.Structure):
     _fields_ = [
         ('dwFileAttributes', _DWORD),
         ('ftCreationTime', _FILETIME),
         ('ftLastAccessTime', _FILETIME),
         ('ftLastWriteTime', _FILETIME),
         ('dwVolumeSerialNumber', _DWORD),
         ('nFileSizeHigh', _DWORD),
         ('nFileSizeLow', _DWORD),
         ('nNumberOfLinks', _DWORD),
         ('nFileIndexHigh', _DWORD),
         ('nFileIndexLow', _DWORD),
     ]
 
 
 # CreateFile
 _FILE_SHARE_READ = 0x00000001
 _FILE_SHARE_WRITE = 0x00000002
 _FILE_SHARE_DELETE = 0x00000004
 
 _OPEN_EXISTING = 3
 
 _FILE_FLAG_BACKUP_SEMANTICS = 0x02000000
 
 # SetFileAttributes
 _FILE_ATTRIBUTE_NORMAL = 0x80
 _FILE_ATTRIBUTE_NOT_CONTENT_INDEXED = 0x2000
 
 # Process Security and Access Rights
 _PROCESS_QUERY_INFORMATION = 0x0400
 
 # GetExitCodeProcess
 _STILL_ACTIVE = 259
 
 
 class _STARTUPINFO(ctypes.Structure):
     _fields_ = [
         ('cb', _DWORD),
         ('lpReserved', _LPSTR),
         ('lpDesktop', _LPSTR),
         ('lpTitle', _LPSTR),
         ('dwX', _DWORD),
         ('dwY', _DWORD),
         ('dwXSize', _DWORD),
         ('dwYSize', _DWORD),
         ('dwXCountChars', _DWORD),
         ('dwYCountChars', _DWORD),
         ('dwFillAttribute', _DWORD),
         ('dwFlags', _DWORD),
         ('wShowWindow', _WORD),
         ('cbReserved2', _WORD),
         ('lpReserved2', ctypes.c_char_p),
         ('hStdInput', _HANDLE),
         ('hStdOutput', _HANDLE),
         ('hStdError', _HANDLE),
     ]
 
 
 class _PROCESS_INFORMATION(ctypes.Structure):
     _fields_ = [
         ('hProcess', _HANDLE),
         ('hThread', _HANDLE),
         ('dwProcessId', _DWORD),
         ('dwThreadId', _DWORD),
     ]
 
 
 _CREATE_NO_WINDOW = 0x08000000
 _SW_HIDE = 0
 
 
 class _COORD(ctypes.Structure):
     _fields_ = [('X', ctypes.c_short), ('Y', ctypes.c_short)]
 
 
 class _SMALL_RECT(ctypes.Structure):
     _fields_ = [
         ('Left', ctypes.c_short),
         ('Top', ctypes.c_short),
         ('Right', ctypes.c_short),
         ('Bottom', ctypes.c_short),
     ]
 
 
 class _CONSOLE_SCREEN_BUFFER_INFO(ctypes.Structure):
     _fields_ = [
         ('dwSize', _COORD),
         ('dwCursorPosition', _COORD),
         ('wAttributes', _WORD),
         ('srWindow', _SMALL_RECT),
         ('dwMaximumWindowSize', _COORD),
     ]
 
 
 _STD_OUTPUT_HANDLE = _DWORD(-11).value
 _STD_ERROR_HANDLE = _DWORD(-12).value
 
 # CERT_TRUST_STATUS dwErrorStatus
 CERT_TRUST_IS_PARTIAL_CHAIN = 0x10000
 
 # CertCreateCertificateContext encodings
 X509_ASN_ENCODING = 0x00000001
 PKCS_7_ASN_ENCODING = 0x00010000
 
 # These structs are only complete enough to achieve what we need.
 class CERT_CHAIN_CONTEXT(ctypes.Structure):
     _fields_ = (
         ("cbSize", _DWORD),
         # CERT_TRUST_STATUS struct
         ("dwErrorStatus", _DWORD),
         ("dwInfoStatus", _DWORD),
         ("cChain", _DWORD),
         ("rgpChain", ctypes.c_void_p),
         ("cLowerQualityChainContext", _DWORD),
         ("rgpLowerQualityChainContext", ctypes.c_void_p),
         ("fHasRevocationFreshnessTime", _BOOL),
         ("dwRevocationFreshnessTime", _DWORD),
     )
 
 
 class CERT_USAGE_MATCH(ctypes.Structure):
     _fields_ = (
         ("dwType", _DWORD),
         # CERT_ENHKEY_USAGE struct
         ("cUsageIdentifier", _DWORD),
         ("rgpszUsageIdentifier", ctypes.c_void_p),  # LPSTR *
     )
 
 
 class CERT_CHAIN_PARA(ctypes.Structure):
     _fields_ = (
         ("cbSize", _DWORD),
         ("RequestedUsage", CERT_USAGE_MATCH),
         ("RequestedIssuancePolicy", CERT_USAGE_MATCH),
         ("dwUrlRetrievalTimeout", _DWORD),
         ("fCheckRevocationFreshnessTime", _BOOL),
         ("dwRevocationFreshnessTime", _DWORD),
         ("pftCacheResync", ctypes.c_void_p),  # LPFILETIME
         ("pStrongSignPara", ctypes.c_void_p),  # PCCERT_STRONG_SIGN_PARA
         ("dwStrongSignFlags", _DWORD),
     )
 
 
 # types of parameters of C functions used (required by pypy)
 
 _crypt32.CertCreateCertificateContext.argtypes = [
     _DWORD,  # cert encoding
     ctypes.c_char_p,  # cert
     _DWORD,
 ]  # cert size
 _crypt32.CertCreateCertificateContext.restype = _PCCERT_CONTEXT
 
 _crypt32.CertGetCertificateChain.argtypes = [
     ctypes.c_void_p,  # HCERTCHAINENGINE
     _PCCERT_CONTEXT,
     ctypes.c_void_p,  # LPFILETIME
     ctypes.c_void_p,  # HCERTSTORE
     ctypes.c_void_p,  # PCERT_CHAIN_PARA
     _DWORD,
     ctypes.c_void_p,  # LPVOID
     ctypes.c_void_p,  # PCCERT_CHAIN_CONTEXT *
 ]
 _crypt32.CertGetCertificateChain.restype = _BOOL
 
 _crypt32.CertFreeCertificateContext.argtypes = [_PCCERT_CONTEXT]
 _crypt32.CertFreeCertificateContext.restype = _BOOL
 
 _kernel32.CreateFileA.argtypes = [
     _LPCSTR,
     _DWORD,
     _DWORD,
     ctypes.c_void_p,
     _DWORD,
     _DWORD,
     _HANDLE,
 ]
 _kernel32.CreateFileA.restype = _HANDLE
 
 _kernel32.GetFileInformationByHandle.argtypes = [_HANDLE, ctypes.c_void_p]
 _kernel32.GetFileInformationByHandle.restype = _BOOL
 
 _kernel32.CloseHandle.argtypes = [_HANDLE]
 _kernel32.CloseHandle.restype = _BOOL
 
 try:
     _kernel32.CreateHardLinkA.argtypes = [_LPCSTR, _LPCSTR, ctypes.c_void_p]
     _kernel32.CreateHardLinkA.restype = _BOOL
 except AttributeError:
     pass
 
 _kernel32.SetFileAttributesA.argtypes = [_LPCSTR, _DWORD]
 _kernel32.SetFileAttributesA.restype = _BOOL
 
 _DRIVE_UNKNOWN = 0
 _DRIVE_NO_ROOT_DIR = 1
 _DRIVE_REMOVABLE = 2
 _DRIVE_FIXED = 3
 _DRIVE_REMOTE = 4
 _DRIVE_CDROM = 5
 _DRIVE_RAMDISK = 6
 
 _kernel32.GetDriveTypeA.argtypes = [_LPCSTR]
 _kernel32.GetDriveTypeA.restype = _UINT
 
 _kernel32.GetVolumeInformationA.argtypes = [
     _LPCSTR,
     ctypes.c_void_p,
     _DWORD,
     ctypes.c_void_p,
     ctypes.c_void_p,
     ctypes.c_void_p,
     ctypes.c_void_p,
     _DWORD,
 ]
 _kernel32.GetVolumeInformationA.restype = _BOOL
 
 _kernel32.GetVolumePathNameA.argtypes = [_LPCSTR, ctypes.c_void_p, _DWORD]
 _kernel32.GetVolumePathNameA.restype = _BOOL
 
 _kernel32.OpenProcess.argtypes = [_DWORD, _BOOL, _DWORD]
 _kernel32.OpenProcess.restype = _HANDLE
 
 _kernel32.GetExitCodeProcess.argtypes = [_HANDLE, ctypes.c_void_p]
 _kernel32.GetExitCodeProcess.restype = _BOOL
 
 _kernel32.GetLastError.argtypes = []
 _kernel32.GetLastError.restype = _DWORD
 
 _kernel32.GetModuleFileNameA.argtypes = [_HANDLE, ctypes.c_void_p, _DWORD]
 _kernel32.GetModuleFileNameA.restype = _DWORD
 
 _kernel32.CreateProcessA.argtypes = [
     _LPCSTR,
     _LPCSTR,
     ctypes.c_void_p,
     ctypes.c_void_p,
     _BOOL,
     _DWORD,
     ctypes.c_void_p,
     _LPCSTR,
     ctypes.c_void_p,
     ctypes.c_void_p,
 ]
 _kernel32.CreateProcessA.restype = _BOOL
 
 _kernel32.ExitProcess.argtypes = [_UINT]
 _kernel32.ExitProcess.restype = None
 
 _kernel32.GetCurrentProcessId.argtypes = []
 _kernel32.GetCurrentProcessId.restype = _DWORD
 
 # pytype: disable=module-attr
 _SIGNAL_HANDLER = ctypes.WINFUNCTYPE(_BOOL, _DWORD)
 # pytype: enable=module-attr
 _kernel32.SetConsoleCtrlHandler.argtypes = [_SIGNAL_HANDLER, _BOOL]
 _kernel32.SetConsoleCtrlHandler.restype = _BOOL
 
 _kernel32.SetConsoleMode.argtypes = [_HANDLE, _DWORD]
 _kernel32.SetConsoleMode.restype = _BOOL
 
 _kernel32.GetConsoleMode.argtypes = [_HANDLE, ctypes.c_void_p]
 _kernel32.GetConsoleMode.restype = _BOOL
 
 _kernel32.GetStdHandle.argtypes = [_DWORD]
 _kernel32.GetStdHandle.restype = _HANDLE
 
 _kernel32.GetConsoleScreenBufferInfo.argtypes = [_HANDLE, ctypes.c_void_p]
 _kernel32.GetConsoleScreenBufferInfo.restype = _BOOL
 
 _advapi32.GetUserNameA.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
 _advapi32.GetUserNameA.restype = _BOOL
 
 _user32.GetWindowThreadProcessId.argtypes = [_HANDLE, ctypes.c_void_p]
 _user32.GetWindowThreadProcessId.restype = _DWORD
 
 _user32.ShowWindow.argtypes = [_HANDLE, ctypes.c_int]
 _user32.ShowWindow.restype = _BOOL
 
 # pytype: disable=module-attr
 _WNDENUMPROC = ctypes.WINFUNCTYPE(_BOOL, _HWND, _LPARAM)
 # pytype: enable=module-attr
 _user32.EnumWindows.argtypes = [_WNDENUMPROC, _LPARAM]
 _user32.EnumWindows.restype = _BOOL
 
 _kernel32.PeekNamedPipe.argtypes = [
     _HANDLE,
     ctypes.c_void_p,
     _DWORD,
     ctypes.c_void_p,
     ctypes.c_void_p,
     ctypes.c_void_p,
 ]
 _kernel32.PeekNamedPipe.restype = _BOOL
 
 
-def _raiseoserror(name):
+def _raiseoserror(name: bytes) -> NoReturn:
     # Force the code to a signed int to avoid an 'int too large' error.
     # See https://bugs.python.org/issue28474
     code = _kernel32.GetLastError()
     if code > 0x7FFFFFFF:
         code -= 2 ** 32
     err = ctypes.WinError(code=code)  # pytype: disable=module-attr
     raise OSError(
         err.errno, '%s: %s' % (encoding.strfromlocal(name), err.strerror)
     )
 
 
-def _getfileinfo(name):
+def _getfileinfo(name: bytes) -> _BY_HANDLE_FILE_INFORMATION:
     fh = _kernel32.CreateFileA(
         name,
         0,
         _FILE_SHARE_READ | _FILE_SHARE_WRITE | _FILE_SHARE_DELETE,
         None,
         _OPEN_EXISTING,
         _FILE_FLAG_BACKUP_SEMANTICS,
         None,
     )
     if fh == _INVALID_HANDLE_VALUE:
         _raiseoserror(name)
     try:
         fi = _BY_HANDLE_FILE_INFORMATION()
         if not _kernel32.GetFileInformationByHandle(fh, ctypes.byref(fi)):
             _raiseoserror(name)
         return fi
     finally:
         _kernel32.CloseHandle(fh)
 
 
-def checkcertificatechain(cert, build=True):
+def checkcertificatechain(cert: bytes, build: bool = True) -> bool:
     """Tests the given certificate to see if there is a complete chain to a
     trusted root certificate.  As a side effect, missing certificates are
     downloaded and installed unless ``build=False``.  True is returned if a
     chain to a trusted root exists (even if built on the fly), otherwise
     False.  NB: A chain to a trusted root does NOT imply that the certificate
     is valid.
     """
 
     chainctxptr = ctypes.POINTER(CERT_CHAIN_CONTEXT)
 
     pchainctx = chainctxptr()
     chainpara = CERT_CHAIN_PARA(
         cbSize=ctypes.sizeof(CERT_CHAIN_PARA), RequestedUsage=CERT_USAGE_MATCH()
     )
 
     certctx = _crypt32.CertCreateCertificateContext(
         X509_ASN_ENCODING, cert, len(cert)
     )
     if certctx is None:
         _raiseoserror(b'CertCreateCertificateContext')
 
     flags = 0
 
     if not build:
         flags |= 0x100  # CERT_CHAIN_DISABLE_AUTH_ROOT_AUTO_UPDATE
 
     try:
         # Building the certificate chain will update root certs as necessary.
         if not _crypt32.CertGetCertificateChain(
             None,  # hChainEngine
             certctx,  # pCertContext
             None,  # pTime
             None,  # hAdditionalStore
             ctypes.byref(chainpara),
             flags,
             None,  # pvReserved
             ctypes.byref(pchainctx),
         ):
             _raiseoserror(b'CertGetCertificateChain')
 
         chainctx = pchainctx.contents
 
         return chainctx.dwErrorStatus & CERT_TRUST_IS_PARTIAL_CHAIN == 0
     finally:
         if pchainctx:
             _crypt32.CertFreeCertificateChain(pchainctx)
         _crypt32.CertFreeCertificateContext(certctx)
 
 
-def oslink(src, dst):
+def oslink(src: bytes, dst: bytes) -> None:
     try:
         if not _kernel32.CreateHardLinkA(dst, src, None):
             _raiseoserror(src)
     except AttributeError:  # Wine doesn't support this function
         _raiseoserror(src)
 
 
-def nlinks(name):
+def nlinks(name: bytes) -> int:
     '''return number of hardlinks for the given file'''
     return _getfileinfo(name).nNumberOfLinks
 
 
-def samefile(path1, path2):
+def samefile(path1: bytes, path2: bytes) -> bool:
     '''Returns whether path1 and path2 refer to the same file or directory.'''
     res1 = _getfileinfo(path1)
     res2 = _getfileinfo(path2)
     return (
         res1.dwVolumeSerialNumber == res2.dwVolumeSerialNumber
         and res1.nFileIndexHigh == res2.nFileIndexHigh
         and res1.nFileIndexLow == res2.nFileIndexLow
     )
 
 
-def samedevice(path1, path2):
+def samedevice(path1: bytes, path2: bytes) -> bool:
     '''Returns whether path1 and path2 are on the same device.'''
     res1 = _getfileinfo(path1)
     res2 = _getfileinfo(path2)
     return res1.dwVolumeSerialNumber == res2.dwVolumeSerialNumber
 
 
-def peekpipe(pipe):
+def peekpipe(pipe) -> int:
     handle = msvcrt.get_osfhandle(pipe.fileno())  # pytype: disable=module-attr
     avail = _DWORD()
 
     if not _kernel32.PeekNamedPipe(
         handle, None, 0, None, ctypes.byref(avail), None
     ):
         err = _kernel32.GetLastError()
         if err == _ERROR_BROKEN_PIPE:
             return 0
         raise ctypes.WinError(err)  # pytype: disable=module-attr
 
     return avail.value
 
 
-def lasterrorwaspipeerror(err):
+def lasterrorwaspipeerror(err) -> bool:
     if err.errno != errno.EINVAL:
         return False
     err = _kernel32.GetLastError()
     return err == _ERROR_BROKEN_PIPE or err == _ERROR_NO_DATA
 
 
-def testpid(pid):
+def testpid(pid: int) -> bool:
     """return True if pid is still running or unable to
     determine, False otherwise"""
     h = _kernel32.OpenProcess(_PROCESS_QUERY_INFORMATION, False, pid)
     if h:
         try:
             status = _DWORD()
             if _kernel32.GetExitCodeProcess(h, ctypes.byref(status)):
                 return status.value == _STILL_ACTIVE
         finally:
             _kernel32.CloseHandle(h)
     return _kernel32.GetLastError() != _ERROR_INVALID_PARAMETER
 
 
-def executablepath():
+def executablepath() -> bytes:
     '''return full path of hg.exe'''
     size = 600
     buf = ctypes.create_string_buffer(size + 1)
     len = _kernel32.GetModuleFileNameA(None, ctypes.byref(buf), size)
     # pytype: disable=module-attr
     if len == 0:
         raise ctypes.WinError()  # Note: WinError is a function
     elif len == size:
         raise ctypes.WinError(_ERROR_INSUFFICIENT_BUFFER)
     # pytype: enable=module-attr
     return buf.value
 
 
-def getvolumename(path):
+def getvolumename(path: bytes) -> Optional[bytes]:
     """Get the mount point of the filesystem from a directory or file
     (best-effort)
 
     Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
     """
     # realpath() calls GetFullPathName()
     realpath = os.path.realpath(path)
 
     # allocate at least MAX_PATH long since GetVolumePathName('c:\\', buf, 4)
     # somehow fails on Windows XP
     size = max(len(realpath), _MAX_PATH) + 1
     buf = ctypes.create_string_buffer(size)
 
     if not _kernel32.GetVolumePathNameA(realpath, ctypes.byref(buf), size):
         # Note: WinError is a function
         raise ctypes.WinError()  # pytype: disable=module-attr
 
     return buf.value
 
 
-def getfstype(path):
+def getfstype(path: bytes) -> Optional[bytes]:
     """Get the filesystem type name from a directory or file (best-effort)
 
     Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.
     """
     volume = getvolumename(path)
 
     t = _kernel32.GetDriveTypeA(volume)
 
     if t == _DRIVE_REMOTE:
         return b'cifs'
     elif t not in (
         _DRIVE_REMOVABLE,
         _DRIVE_FIXED,
         _DRIVE_CDROM,
         _DRIVE_RAMDISK,
     ):
         return None
 
     size = _MAX_PATH + 1
     name = ctypes.create_string_buffer(size)
 
     if not _kernel32.GetVolumeInformationA(
         volume, None, 0, None, None, None, ctypes.byref(name), size
     ):
         # Note: WinError is a function
         raise ctypes.WinError()  # pytype: disable=module-attr
 
     return name.value
 
 
-def getuser():
+def getuser() -> bytes:
     '''return name of current user'''
     size = _DWORD(300)
     buf = ctypes.create_string_buffer(size.value + 1)
     if not _advapi32.GetUserNameA(ctypes.byref(buf), ctypes.byref(size)):
         raise ctypes.WinError()  # pytype: disable=module-attr
     return buf.value
 
 
-_signalhandler = []
+_signalhandler: List[_SIGNAL_HANDLER] = []
 
 
-def setsignalhandler():
+def setsignalhandler() -> None:
     """Register a termination handler for console events including
     CTRL+C. python signal handlers do not work well with socket
     operations.
     """
 
     def handler(event):
         _kernel32.ExitProcess(1)
 
     if _signalhandler:
         return  # already registered
     h = _SIGNAL_HANDLER(handler)
     _signalhandler.append(h)  # needed to prevent garbage collection
     if not _kernel32.SetConsoleCtrlHandler(h, True):
         raise ctypes.WinError()  # pytype: disable=module-attr
 
 
-def hidewindow():
+def hidewindow() -> None:
     def callback(hwnd, pid):
         wpid = _DWORD()
         _user32.GetWindowThreadProcessId(hwnd, ctypes.byref(wpid))
         if pid == wpid.value:
             _user32.ShowWindow(hwnd, _SW_HIDE)
             return False  # stop enumerating windows
         return True
 
     pid = _kernel32.GetCurrentProcessId()
     _user32.EnumWindows(_WNDENUMPROC(callback), pid)
 
 
-def termsize():
+def termsize() -> Tuple[int, int]:
     # cmd.exe does not handle CR like a unix console, the CR is
     # counted in the line length. On 80 columns consoles, if 80
     # characters are written, the following CR won't apply on the
     # current line but on the new one. Keep room for it.
     width = 80 - 1
     height = 25
     # Query stderr to avoid problems with redirections
     screenbuf = _kernel32.GetStdHandle(
         _STD_ERROR_HANDLE
     )  # don't close the handle returned
     if screenbuf is None or screenbuf == _INVALID_HANDLE_VALUE:
         return width, height
     csbi = _CONSOLE_SCREEN_BUFFER_INFO()
     if not _kernel32.GetConsoleScreenBufferInfo(screenbuf, ctypes.byref(csbi)):
         return width, height
     width = csbi.srWindow.Right - csbi.srWindow.Left  # don't '+ 1'
     height = csbi.srWindow.Bottom - csbi.srWindow.Top + 1
     return width, height
 
 
-def enablevtmode():
+def enablevtmode() -> bool:
     """Enable virtual terminal mode for the associated console.  Return True if
     enabled, else False."""
 
     ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x4
 
     handle = _kernel32.GetStdHandle(
         _STD_OUTPUT_HANDLE
     )  # don't close the handle
     if handle == _INVALID_HANDLE_VALUE:
         return False
 
     mode = _DWORD(0)
 
     if not _kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
         return False
 
     if (mode.value & ENABLE_VIRTUAL_TERMINAL_PROCESSING) == 0:
         mode.value |= ENABLE_VIRTUAL_TERMINAL_PROCESSING
 
         if not _kernel32.SetConsoleMode(handle, mode):
             return False
 
     return True
 
 
-def spawndetached(args):
+def spawndetached(args: List[bytes]) -> int:
     # No standard library function really spawns a fully detached
     # process under win32 because they allocate pipes or other objects
     # to handle standard streams communications. Passing these objects
     # to the child process requires handle inheritance to be enabled
     # which makes really detached processes impossible.
     si = _STARTUPINFO()
     si.cb = ctypes.sizeof(_STARTUPINFO)
 
     pi = _PROCESS_INFORMATION()
 
     env = b''
     for k in encoding.environ:
         env += b"%s=%s\0" % (k, encoding.environ[k])
     if not env:
         env = b'\0'
     env += b'\0'
 
     args = subprocess.list2cmdline(pycompat.rapply(encoding.strfromlocal, args))
 
     # TODO: CreateProcessW on py3?
     res = _kernel32.CreateProcessA(
         None,
         encoding.strtolocal(args),
         None,
         None,
         False,
         _CREATE_NO_WINDOW,
         env,
         encoding.getcwd(),
         ctypes.byref(si),
         ctypes.byref(pi),
     )
     if not res:
         raise ctypes.WinError()  # pytype: disable=module-attr
 
     _kernel32.CloseHandle(pi.hProcess)
     _kernel32.CloseHandle(pi.hThread)
 
     return pi.dwProcessId
 
 
-def unlink(f):
+def unlink(f: bytes) -> None:
     '''try to implement POSIX' unlink semantics on Windows'''
 
     if os.path.isdir(f):
         # use EPERM because it is POSIX prescribed value, even though
         # unlink(2) on directories returns EISDIR on Linux
         raise IOError(
             errno.EPERM,
             r"Unlinking directory not permitted: '%s'"
             % encoding.strfromlocal(f),
         )
 
     # POSIX allows to unlink and rename open files. Windows has serious
     # problems with doing that:
     # - Calling os.unlink (or os.rename) on a file f fails if f or any
     #   hardlinked copy of f has been opened with Python's open(). There is no
     #   way such a file can be deleted or renamed on Windows (other than
     #   scheduling the delete or rename for the next reboot).
     # - Calling os.unlink on a file that has been opened with Mercurial's
     #   posixfile (or comparable methods) will delay the actual deletion of
     #   the file for as long as the file is held open. The filename is blocked
     #   during that time and cannot be used for recreating a new file under
     #   that same name ("zombie file"). Directories containing such zombie files
     #   cannot be removed or moved.
     # A file that has been opened with posixfile can be renamed, so we rename
     # f to a random temporary name before calling os.unlink on it. This allows
     # callers to recreate f immediately while having other readers do their
     # implicit zombie filename blocking on a temporary name.
 
     for tries in range(10):
         temp = b'%s-%08x' % (f, random.randint(0, 0xFFFFFFFF))
         try:
             os.rename(f, temp)
             break
         except FileExistsError:
             pass
     else:
         raise IOError(errno.EEXIST, "No usable temporary filename found")
 
     try:
         os.unlink(temp)
     except OSError:
         # The unlink might have failed because the READONLY attribute may heave
         # been set on the original file. Rename works fine with READONLY set,
         # but not os.unlink. Reset all attributes and try again.
         _kernel32.SetFileAttributesA(temp, _FILE_ATTRIBUTE_NORMAL)
         try:
             os.unlink(temp)
         except OSError:
             # The unlink might have failed due to some very rude AV-Scanners.
             # Leaking a tempfile is the lesser evil than aborting here and
             # leaving some potentially serious inconsistencies.
             pass
 
 
-def makedir(path, notindexed):
+def makedir(path: bytes, notindexed: bool) -> None:
     os.mkdir(path)
     if notindexed:
         _kernel32.SetFileAttributesA(path, _FILE_ATTRIBUTE_NOT_CONTENT_INDEXED)