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

from __future__ import annotations

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 (

    Any,

    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:

    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

        # convert to a friendlier exception

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

    while True:

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

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

            break

        if c == '\003':

            raise KeyboardInterrupt

        if c == '\b':

            pw = pw[:-1]

        else:

            pw = pw + c

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

    msvcrt.putwch('\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()

            pass

    def write(self, s):

        try:

            return self.fp.write(s)

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

                raise

            self.close()

    def flush(self):

        try:

            return self.fp.flush()

            if not win32.lasterrorwaspipeerror(inst):

                raise

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[int] = 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)

    'cmd "$var1 ${var2} %var3%" $var1 ${var2} \\'

    >>> # $$ is not special. %% is not special either, but can be the end and

    >>> # start of consecutive variables

    >>> shelltocmdexe(b"cmd $$ %% %var1%%var2%", e)

    'cmd $$ %% %var1%%var2%'

    >>> # No double substitution

    >>> shelltocmdexe(b"$var1 %var1%", {b'var1': b'%var2%', b'var2': b'boom'})

    '%var1% %var1%'

    >>> # Tilde expansion

    >>> shelltocmdexe(b"~/dir ~\dir2 ~tmpfile \~/", {})

    '%USERPROFILE%/dir %USERPROFILE%\\dir2 ~tmpfile ~/'

    """

    if not any(c in path for c in b"$'~"):

        return path

    varchars = pycompat.sysbytes(string.ascii_letters + string.digits) + b'_-'

    res = b''

    index = 0

    pathlen = len(path)

    while index < pathlen:

        c = path[index : index + 1]

        if c == b'\'':  # no expansion within single quotes

            path = path[index + 1 :]

            pathlen = len(path)

            try:

                index = path.index(b'\'')

                res += b'"' + path[:index] + b'"'

            except ValueError:

                res += c + path

                index = pathlen - 1

        elif c == b'%':  # variable

            path = path[index + 1 :]

            pathlen = len(path)

            try:

                index = path.index(b'%')

            except ValueError:

                res += b'%' + path

                index = pathlen - 1

            else:

                var = path[:index]

                res += b'%' + var + b'%'

        elif c == b'$':  # variable

            if path[index + 1 : index + 2] == b'{':

                path = path[index + 2 :]

                pathlen = len(path)

                try:

                    index = path.index(b'}')

                    var = path[:index]

                    # See below for why empty variables are handled specially

                    if env.get(var, b'') != b'':

                        res += b'%' + var + b'%'

                    else:

                        res += b'${' + var + b'}'

                except ValueError:

                    res += b'${' + path

                    index = pathlen - 1

            else:

                var = b''

                index += 1

                c = path[index : index + 1]

                while c != b'' and c in varchars:

                    var += c

                    index += 1

                    c = path[index : index + 1]

                # Some variables (like HG_OLDNODE) may be defined, but have an

                # empty value.  Those need to be skipped because when spawning

                # cmd.exe to run the hook, it doesn't replace %VAR% for an empty

                # VAR, and that really confuses things like revset expressions.

                # OTOH, if it's left in Unix format and the hook runs sh.exe, it

                # will substitute to an empty string, and everything is happy.

                if env.get(var, b'') != b'':

                    res += b'%' + var + b'%'

                else:

                    res += b'$' + var

                if c != b'':

                    index -= 1

        elif (

            c == b'~'

            and index + 1 < pathlen

            and path[index + 1 : index + 2] in (b'\\', b'/')

        ):

            res += b"%USERPROFILE%"

        elif (

            c == b'\\'

            and index + 1 < pathlen

            and path[index + 1 : index + 2] in (b'$', b'~')

        ):

            # Skip '\', but only if it is escaping $ or ~

            res += path[index + 1 : index + 2]

            index += 1

        else:

            res += c

        index += 1

    return res

# A sequence of backslashes is special iff it precedes a double quote: