comparison mercurial/win32.py @ 49905:a9faacdc5943

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.

49904:7a80a614c9e5

import errno
import msvcrt
import os
import random
import subprocess

from . import (
    encoding,
    pycompat,
)

    ctypes.c_void_p,
]
_kernel32.PeekNamedPipe.restype = _BOOL


def _raiseoserror(name):
    # 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

    raise OSError(
        err.errno, '%s: %s' % (encoding.strfromlocal(name), err.strerror)
    )


def _getfileinfo(name):
    fh = _kernel32.CreateFileA(
        name,
        0,
        _FILE_SHARE_READ | _FILE_SHARE_WRITE | _FILE_SHARE_DELETE,
        None,

        return fi
    finally:
        _kernel32.CloseHandle(fh)


def checkcertificatechain(cert, build=True):
    """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

        if pchainctx:
            _crypt32.CertFreeCertificateChain(pchainctx)
        _crypt32.CertFreeCertificateContext(certctx)


def oslink(src, dst):
    try:
        if not _kernel32.CreateHardLinkA(dst, src, None):
            _raiseoserror(src)
    except AttributeError:  # Wine doesn't support this function
        _raiseoserror(src)


def nlinks(name):
    '''return number of hardlinks for the given file'''
    return _getfileinfo(name).nNumberOfLinks


def samefile(path1, path2):
    '''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):
    '''Returns whether path1 and path2 are on the same device.'''
    res1 = _getfileinfo(path1)
    res2 = _getfileinfo(path2)
    return res1.dwVolumeSerialNumber == res2.dwVolumeSerialNumber


def peekpipe(pipe):
    handle = msvcrt.get_osfhandle(pipe.fileno())  # pytype: disable=module-attr
    avail = _DWORD()

    if not _kernel32.PeekNamedPipe(
        handle, None, 0, None, ctypes.byref(avail), None

        raise ctypes.WinError(err)  # pytype: disable=module-attr

    return avail.value


def lasterrorwaspipeerror(err):
    if err.errno != errno.EINVAL:
        return False
    err = _kernel32.GetLastError()
    return err == _ERROR_BROKEN_PIPE or err == _ERROR_NO_DATA


def testpid(pid):
    """return True if pid is still running or unable to
    determine, False otherwise"""
    h = _kernel32.OpenProcess(_PROCESS_QUERY_INFORMATION, False, pid)
    if h:
        try:

        finally:
            _kernel32.CloseHandle(h)
    return _kernel32.GetLastError() != _ERROR_INVALID_PARAMETER


def executablepath():
    '''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

        raise ctypes.WinError(_ERROR_INSUFFICIENT_BUFFER)
    # pytype: enable=module-attr
    return buf.value


def getvolumename(path):
    """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.
    """

        raise ctypes.WinError()  # pytype: disable=module-attr

    return buf.value


def getfstype(path):
    """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)

        raise ctypes.WinError()  # pytype: disable=module-attr

    return name.value


def getuser():
    '''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 = []


def setsignalhandler():
    """Register a termination handler for console events including
    CTRL+C. python signal handlers do not work well with socket
    operations.
    """

    _signalhandler.append(h)  # needed to prevent garbage collection
    if not _kernel32.SetConsoleCtrlHandler(h, True):
        raise ctypes.WinError()  # pytype: disable=module-attr


def hidewindow():
    def callback(hwnd, pid):
        wpid = _DWORD()
        _user32.GetWindowThreadProcessId(hwnd, ctypes.byref(wpid))
        if pid == wpid.value:
            _user32.ShowWindow(hwnd, _SW_HIDE)

    pid = _kernel32.GetCurrentProcessId()
    _user32.EnumWindows(_WNDENUMPROC(callback), pid)


def termsize():
    # 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

    width = csbi.srWindow.Right - csbi.srWindow.Left  # don't '+ 1'
    height = csbi.srWindow.Bottom - csbi.srWindow.Top + 1
    return width, height


def enablevtmode():
    """Enable virtual terminal mode for the associated console.  Return True if
    enabled, else False."""

    ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x4

            return False

    return True


def spawndetached(args):
    # 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.

    _kernel32.CloseHandle(pi.hThread)

    return pi.dwProcessId


def unlink(f):
    '''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

            # Leaking a tempfile is the lesser evil than aborting here and
            # leaving some potentially serious inconsistencies.
            pass


def makedir(path, notindexed):
    os.mkdir(path)
    if notindexed:
        _kernel32.SetFileAttributesA(path, _FILE_ATTRIBUTE_NOT_CONTENT_INDEXED)

49905:a9faacdc5943

import errno
import msvcrt
import os
import random
import subprocess

from typing import (
    List,
    NoReturn,
    Optional,
    Tuple,
)

from . import (
    encoding,
    pycompat,
)

    ctypes.c_void_p,
]
_kernel32.PeekNamedPipe.restype = _BOOL


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

    raise OSError(
        err.errno, '%s: %s' % (encoding.strfromlocal(name), err.strerror)
    )


def _getfileinfo(name: bytes) -> _BY_HANDLE_FILE_INFORMATION:
    fh = _kernel32.CreateFileA(
        name,
        0,
        _FILE_SHARE_READ | _FILE_SHARE_WRITE | _FILE_SHARE_DELETE,
        None,

        return fi
    finally:
        _kernel32.CloseHandle(fh)


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

        if pchainctx:
            _crypt32.CertFreeCertificateChain(pchainctx)
        _crypt32.CertFreeCertificateContext(certctx)


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: bytes) -> int:
    '''return number of hardlinks for the given file'''
    return _getfileinfo(name).nNumberOfLinks


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: 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) -> 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

        raise ctypes.WinError(err)  # pytype: disable=module-attr

    return avail.value


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: 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:

        finally:
            _kernel32.CloseHandle(h)
    return _kernel32.GetLastError() != _ERROR_INVALID_PARAMETER


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

        raise ctypes.WinError(_ERROR_INSUFFICIENT_BUFFER)
    # pytype: enable=module-attr
    return buf.value


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

        raise ctypes.WinError()  # pytype: disable=module-attr

    return buf.value


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)

        raise ctypes.WinError()  # pytype: disable=module-attr

    return name.value


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: List[_SIGNAL_HANDLER] = []


def setsignalhandler() -> None:
    """Register a termination handler for console events including
    CTRL+C. python signal handlers do not work well with socket
    operations.
    """

    _signalhandler.append(h)  # needed to prevent garbage collection
    if not _kernel32.SetConsoleCtrlHandler(h, True):
        raise ctypes.WinError()  # pytype: disable=module-attr


def hidewindow() -> None:
    def callback(hwnd, pid):
        wpid = _DWORD()
        _user32.GetWindowThreadProcessId(hwnd, ctypes.byref(wpid))
        if pid == wpid.value:
            _user32.ShowWindow(hwnd, _SW_HIDE)

    pid = _kernel32.GetCurrentProcessId()
    _user32.EnumWindows(_WNDENUMPROC(callback), pid)


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

    width = csbi.srWindow.Right - csbi.srWindow.Left  # don't '+ 1'
    height = csbi.srWindow.Bottom - csbi.srWindow.Top + 1
    return width, height


def enablevtmode() -> bool:
    """Enable virtual terminal mode for the associated console.  Return True if
    enabled, else False."""

    ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x4

            return False

    return True


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.

    _kernel32.CloseHandle(pi.hThread)

    return pi.dwProcessId


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

            # Leaking a tempfile is the lesser evil than aborting here and
            # leaving some potentially serious inconsistencies.
            pass


def makedir(path: bytes, notindexed: bool) -> None:
    os.mkdir(path)
    if notindexed:
        _kernel32.SetFileAttributesA(path, _FILE_ATTRIBUTE_NOT_CONTENT_INDEXED)