# windows.py - Windows utility function implementations for Mercurial

#

#  Copyright 2005-2009 Olivia Mackall <olivia@selenic.com> 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 getpass

import msvcrt  # pytype: disable=import-error

import os

import re

import stat

import string

import sys

import typing

import winreg  # pytype: disable=import-error

from typing import (

    AnyStr,

    BinaryIO,

    Iterable,

    Iterator,

    List,

    Mapping,

    NoReturn,

    Optional,

    Pattern,

    Sequence,

    Tuple,

    Union,

)

from .i18n import _

from . import (

    encoding,

    error,

    policy,

    pycompat,

    typelib,

    win32,

)

osutil = policy.importmod('osutil')

getfsmountpoint = win32.getvolumename

getfstype = win32.getfstype

getuser = win32.getuser

hidewindow = win32.hidewindow

makedir = win32.makedir

nlinks = win32.nlinks

oslink = win32.oslink

samedevice = win32.samedevice

samefile = win32.samefile

setsignalhandler = win32.setsignalhandler

spawndetached = win32.spawndetached

split = os.path.split

testpid = win32.testpid

unlink = win32.unlink

if typing.TYPE_CHECKING:

    # Replace the various overloads that come along with aliasing stdlib methods

    # with the narrow definition that we care about in the type checking phase

    # only.  This ensures that both Windows and POSIX see only the definition

    # that is actually available.

    #

    # Note that if we check pycompat.TYPE_CHECKING here, it is always False, and

    # the methods aren't replaced.

    def split(p: bytes) -> Tuple[bytes, bytes]:

        raise NotImplementedError

umask: int = 0o022

class mixedfilemodewrapper:

    """Wraps a file handle when it is opened in read/write mode.

    fopen() and fdopen() on Windows have a specific-to-Windows requirement

    that files opened with mode r+, w+, or a+ make a call to a file positioning

    function when switching between reads and writes. Without this extra call,

    Python will raise a not very intuitive "IOError: [Errno 0] Error."

    This class wraps posixfile instances when the file is opened in read/write

    mode and automatically adds checks or inserts appropriate file positioning

    calls when necessary.

    """

    OPNONE = 0

    OPREAD = 1

    OPWRITE = 2

    def __init__(self, fp):

        object.__setattr__(self, '_fp', fp)

        object.__setattr__(self, '_lastop', 0)

    def __enter__(self):

        self._fp.__enter__()

        return self

    def __exit__(self, exc_type, exc_val, exc_tb):

        self._fp.__exit__(exc_type, exc_val, exc_tb)

    def __getattr__(self, name):

        return getattr(self._fp, name)

    def __setattr__(self, name, value):

        return self._fp.__setattr__(name, value)

    def _noopseek(self):

        self._fp.seek(0, os.SEEK_CUR)

    def seek(self, *args, **kwargs):

        object.__setattr__(self, '_lastop', self.OPNONE)

        return self._fp.seek(*args, **kwargs)

    def write(self, d):

        if self._lastop == self.OPREAD:

            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPWRITE)

        return self._fp.write(d)

    def writelines(self, *args, **kwargs):

        if self._lastop == self.OPREAD:

            self._noopeseek()

        object.__setattr__(self, '_lastop', self.OPWRITE)

        return self._fp.writelines(*args, **kwargs)

    def read(self, *args, **kwargs):

        if self._lastop == self.OPWRITE:

            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPREAD)

        return self._fp.read(*args, **kwargs)

    def readline(self, *args, **kwargs):

        if self._lastop == self.OPWRITE:

            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPREAD)

        return self._fp.readline(*args, **kwargs)

    def readlines(self, *args, **kwargs):

        if self._lastop == self.OPWRITE:

            self._noopseek()

        object.__setattr__(self, '_lastop', self.OPREAD)

        return self._fp.readlines(*args, **kwargs)

class fdproxy:

    """Wraps osutil.posixfile() to override the name attribute to reflect the

    underlying file name.

    """

    def __init__(self, name, fp):

        self.name = name

        self._fp = fp

    def __enter__(self):

        self._fp.__enter__()

        # Return this wrapper for the context manager so that the name is

        # still available.

        return self

    def __exit__(self, exc_type, exc_value, traceback):

        self._fp.__exit__(exc_type, exc_value, traceback)

    def __iter__(self):

        return iter(self._fp)

    def __getattr__(self, name):

        return getattr(self._fp, name)

def posixfile(name, mode=b'r', buffering=-1):

    '''Open a file with even more POSIX-like semantics'''

    try:

        fp = osutil.posixfile(name, mode, buffering)  # may raise WindowsError

        # PyFile_FromFd() ignores the name, and seems to report fp.name as the

        # underlying file descriptor.

        fp = fdproxy(name, fp)

        # The position when opening in append mode is implementation defined, so

        # make it consistent with other platforms, which position at EOF.

        if b'a' in mode:

            fp.seek(0, os.SEEK_END)

        if b'+' in mode:

            return mixedfilemodewrapper(fp)

        return fp

    except WindowsError as err:  # pytype: disable=name-error

        # convert to a friendlier exception

        raise IOError(

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

        )

# may be wrapped by win32mbcs extension

listdir = osutil.listdir

def get_password() -> bytes:

    """Prompt for password with echo off, using Windows getch().

    This shouldn't be called directly- use ``ui.getpass()`` instead, which

    checks if the session is interactive first.

    """

    pw = u""

    while True:

        c = msvcrt.getwch()  # pytype: disable=module-attr

        if c == u'\r' or c == u'\n':

            break

        if c == u'\003':

            raise KeyboardInterrupt

        if c == u'\b':

            pw = pw[:-1]

        else:

            pw = pw + c

    msvcrt.putwch(u'\r')  # pytype: disable=module-attr

    msvcrt.putwch(u'\n')  # pytype: disable=module-attr

    return encoding.unitolocal(pw)

class winstdout(typelib.BinaryIO_Proxy):

    """Some files on Windows misbehave.

    When writing to a broken pipe, EINVAL instead of EPIPE may be raised.

    When writing too many bytes to a console at the same, a "Not enough space"

    error may happen. Python 3 already works around that.

    """

    def __init__(self, fp: BinaryIO):

        self.fp = fp

    def __getattr__(self, key):

        return getattr(self.fp, key)

    def close(self):

        try:

            self.fp.close()

        except IOError:

            pass

    def write(self, s):

        try:

            return self.fp.write(s)

        except IOError as inst:

            if inst.errno != 0 and not win32.lasterrorwaspipeerror(inst):

                raise

            self.close()

            raise IOError(errno.EPIPE, 'Broken pipe')

    def flush(self):

        try:

            return self.fp.flush()

        except IOError as inst:

            if not win32.lasterrorwaspipeerror(inst):

                raise

            raise IOError(errno.EPIPE, 'Broken pipe')

def openhardlinks() -> bool:

    return True

def parsepatchoutput(output_line: bytes) -> bytes:

    """parses the output produced by patch and returns the filename"""

    pf = output_line[14:]

    if pf[0] == b'`':

        pf = pf[1:-1]  # Remove the quotes

    return pf

def sshargs(

    sshcmd: bytes, host: bytes, user: Optional[bytes], port: Optional[bytes]

) -> bytes:

    '''Build argument list for ssh or Plink'''

    pflag = b'plink' in sshcmd.lower() and b'-P' or b'-p'

    args = user and (b"%s@%s" % (user, host)) or host

    if args.startswith(b'-') or args.startswith(b'/'):

        raise error.Abort(

            _(b'illegal ssh hostname or username starting with - or /: %s')

            % args

        )

    args = shellquote(args)

    if port:

        args = b'%s %s %s' % (pflag, shellquote(port), args)

    return args

def setflags(f: bytes, l: bool, x: bool) -> None:

    pass

def copymode(

    src: bytes,

    dst: bytes,

    mode: Optional[bytes] = None,

    enforcewritable: bool = False,

) -> None:

    pass

def checkexec(path: bytes) -> bool:

    return False

def checklink(path: bytes) -> bool:

    return False

def setbinary(fd) -> None:

    # When run without console, pipes may expose invalid

    # fileno(), usually set to -1.

    fno = getattr(fd, 'fileno', None)

    if fno is not None and fno() >= 0:

        msvcrt.setmode(fno(), os.O_BINARY)  # pytype: disable=module-attr

def pconvert(path: bytes) -> bytes:

    return path.replace(pycompat.ossep, b'/')

def localpath(path: bytes) -> bytes:

    return path.replace(b'/', b'\\')

def normpath(path: bytes) -> bytes:

    return pconvert(os.path.normpath(path))

def normcase(path: bytes) -> bytes:

    return encoding.upper(path)  # NTFS compares via upper()

DRIVE_RE_B: Pattern[bytes] = re.compile(b'^[a-z]:')

DRIVE_RE_S: Pattern[str] = re.compile('^[a-z]:')

# TODO: why is this accepting str?

def abspath(path: AnyStr) -> AnyStr:

    abs_path = os.path.abspath(path)  # re-exports

    # Python on Windows is inconsistent regarding the capitalization of drive

    # letter and this cause issue with various path comparison along the way.

    # So we normalize the drive later to upper case here.

    #

    # See https://bugs.python.org/issue40368 for and example of this hell.

    if isinstance(abs_path, bytes):

        if DRIVE_RE_B.match(abs_path):

            abs_path = abs_path[0:1].upper() + abs_path[1:]

    elif DRIVE_RE_S.match(abs_path):

        abs_path = abs_path[0:1].upper() + abs_path[1:]

    return abs_path

# see posix.py for definitions

normcasespec: int = encoding.normcasespecs.upper

normcasefallback = encoding.upperfallback

def samestat(s1: os.stat_result, s2: os.stat_result) -> bool:

    return False

def shelltocmdexe(path: bytes, env: Mapping[bytes, bytes]) -> bytes:

    r"""Convert shell variables in the form $var and ${var} inside ``path``

    to %var% form.  Existing Windows style variables are left unchanged.

    The variables are limited to the given environment.  Unknown variables are

    left unchanged.

    >>> e = {b'var1': b'v1', b'var2': b'v2', b'var3': b'v3'}

    >>> # Only valid values are expanded

    >>> shelltocmdexe(b'cmd $var1 ${var2} %var3% $missing ${missing} %missing%',

    ...               e)

    'cmd %var1% %var2% %var3% $missing ${missing} %missing%'

    >>> # Single quote prevents expansion, as does \$ escaping

    >>> shelltocmdexe(b"cmd '$var1 ${var2} %var3%' \$var1 \${var2} \\", e)