# util.py - Mercurial utility functions and platform specific implementations

#

#  Copyright 2005 K. Thananchayan <thananck@yahoo.com>

#  Copyright 2005-2007 Olivia Mackall <olivia@selenic.com>

#  Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

"""Mercurial utility functions and platform specific implementations.

This contains helper routines that are independent of the SCM core and

hide platform-specific details from the core.

"""

from __future__ import annotations

import abc

import collections

import contextlib

import errno

import gc

import hashlib

import io

import itertools

import locale

import mmap

import os

# provides util.pickle symbol

import pickle  # noqa: F401 (ignore imported but not used)

import re as remod

import shutil

import stat

import sys

import time

import traceback

import typing

import warnings

from typing import (

    Any,

    BinaryIO,

    Callable,

    Iterable,

    Iterator,

    List,

    Optional,

    Tuple,

    Type,

    TypeVar,

)

from .node import hex

from .thirdparty import attr

# Force pytype to use the non-vendored package

if typing.TYPE_CHECKING:

    # noinspection PyPackageRequirements

    import attr

from hgdemandimport import tracing

from . import (

    encoding,

    error,

    i18n,

    policy,

    pycompat,

    typelib,

    urllibcompat,

)

from .interfaces import (

    modules as intmod,

)

from .utils import (

    compression,

    hashutil,

    procutil,

    stringutil,

)

# keeps pyflakes happy

assert [

    Iterable,

    Iterator,

    List,

    Optional,

    Tuple,

]

if typing.TYPE_CHECKING:

    from typing_extensions import (

        Self,

    )

    _Tcow = TypeVar('_Tcow', bound="cow")

_KT = TypeVar("_KT")

"""An unconstrained key type for container classes (see typing.KT)."""

_VT = TypeVar("_VT")

"""An unconstrained value type for container classes (see typing.VT)."""

base85: intmod.Base85 = policy.importmod('base85')

osutil = policy.importmod('osutil')

b85decode = base85.b85decode

b85encode = base85.b85encode

cookielib = pycompat.cookielib

httplib = pycompat.httplib

safehasattr = pycompat.safehasattr

socketserver = pycompat.socketserver

bytesio = io.BytesIO

# TODO deprecate stringio name, as it is a lie on Python 3.

stringio = bytesio

xmlrpclib = pycompat.xmlrpclib

httpserver = urllibcompat.httpserver

urlerr = urllibcompat.urlerr

urlreq = urllibcompat.urlreq

# workaround for win32mbcs

_filenamebytestr = pycompat.bytestr

if pycompat.iswindows:

    from . import windows as platform

else:

    from . import posix as platform

_ = i18n._

abspath = platform.abspath

bindunixsocket = platform.bindunixsocket

cachestat = platform.cachestat

checkexec = platform.checkexec

checklink = platform.checklink

copymode = platform.copymode

expandglobs = platform.expandglobs

getfsmountpoint = platform.getfsmountpoint

getfstype = platform.getfstype

get_password = platform.get_password

groupmembers = platform.groupmembers

groupname = platform.groupname

isexec = platform.isexec

isowner = platform.isowner

listdir = osutil.listdir

localpath = platform.localpath

lookupreg = platform.lookupreg

makedir = platform.makedir

nlinks = platform.nlinks

normpath = platform.normpath

normcase = platform.normcase

normcasespec = platform.normcasespec

normcasefallback = platform.normcasefallback

openhardlinks = platform.openhardlinks

oslink = platform.oslink

parsepatchoutput = platform.parsepatchoutput

pconvert = platform.pconvert

poll = platform.poll

posixfile = platform.posixfile

readlink = platform.readlink

rename = platform.rename

removedirs = platform.removedirs

samedevice = platform.samedevice

samefile = platform.samefile

samestat = platform.samestat

setflags = platform.setflags

split = platform.split

statfiles = getattr(osutil, 'statfiles', platform.statfiles)

statisexec = platform.statisexec

statislink = platform.statislink

umask = platform.umask

unlink = platform.unlink

username = platform.username

if typing.TYPE_CHECKING:

    _Tfilestat = TypeVar('_Tfilestat', bound='filestat')

def setumask(val: int) -> None:

    '''updates the umask. used by chg server'''

    if pycompat.iswindows:

        return

    os.umask(val)

    global umask

    platform.umask = umask = val & 0o777

# small compat layer

compengines = compression.compengines

SERVERROLE = compression.SERVERROLE

CLIENTROLE = compression.CLIENTROLE

# Python compatibility

_notset = object()

def bitsfrom(container):

    bits = 0

    for bit in container:

        bits |= bit

    return bits

# python 2.6 still have deprecation warning enabled by default. We do not want

# to display anything to standard user so detect if we are running test and

# only use python deprecation warning in this case.

_dowarn = bool(encoding.environ.get(b'HGEMITWARNINGS'))

if _dowarn:

    # explicitly unfilter our warning for python 2.7

    #

    # The option of setting PYTHONWARNINGS in the test runner was investigated.

    # However, module name set through PYTHONWARNINGS was exactly matched, so

    # we cannot set 'mercurial' and have it match eg: 'mercurial.scmutil'. This

    # makes the whole PYTHONWARNINGS thing useless for our usecase.

    warnings.filterwarnings('default', '', DeprecationWarning, 'mercurial')

    warnings.filterwarnings('default', '', DeprecationWarning, 'hgext')

    warnings.filterwarnings('default', '', DeprecationWarning, 'hgext3rd')

if _dowarn:

    # silence warning emitted by passing user string to re.sub()

    warnings.filterwarnings(

        'ignore', 'bad escape', DeprecationWarning, 'mercurial'

    )

    warnings.filterwarnings(

        'ignore', 'invalid escape sequence', DeprecationWarning, 'mercurial'

    )

    # TODO: reinvent imp.is_frozen()

    warnings.filterwarnings(

        'ignore',

        'the imp module is deprecated',

        DeprecationWarning,

        'mercurial',

    )

def nouideprecwarn(msg, version, stacklevel=1):

    """Issue an python native deprecation warning

    This is a noop outside of tests, use 'ui.deprecwarn' when possible.

    """

    if _dowarn:

        msg += (

            b"\n(compatibility will be dropped after Mercurial-%s,"

            b" update your code.)"

        ) % version

        warnings.warn(pycompat.sysstr(msg), DeprecationWarning, stacklevel + 1)

        # on python 3 with chg, we will need to explicitly flush the output

        sys.stderr.flush()

DIGESTS = {

    b'md5': hashlib.md5,

    b'sha1': hashutil.sha1,

    b'sha512': hashlib.sha512,

}

# List of digest types from strongest to weakest

DIGESTS_BY_STRENGTH = [b'sha512', b'sha1', b'md5']

for k in DIGESTS_BY_STRENGTH:

    assert k in DIGESTS

class digester:

    """helper to compute digests.

    This helper can be used to compute one or more digests given their name.

    >>> d = digester([b'md5', b'sha1'])

    >>> d.update(b'foo')

    >>> [k for k in sorted(d)]

    ['md5', 'sha1']

    >>> d[b'md5']

    'acbd18db4cc2f85cedef654fccc4a4d8'

    >>> d[b'sha1']

    '0beec7b5ea3f0fdbc95d0dd47f3c5bc275da8a33'

    >>> digester.preferred([b'md5', b'sha1'])

    'sha1'

    """

    def __init__(self, digests, s=b''):

        self._hashes = {}

        for k in digests:

            if k not in DIGESTS:

                raise error.Abort(_(b'unknown digest type: %s') % k)

            self._hashes[k] = DIGESTS[k]()

        if s:

            self.update(s)

    def update(self, data):

        for h in self._hashes.values():

            h.update(data)

    def __getitem__(self, key):

        if key not in DIGESTS:

            raise error.Abort(_(b'unknown digest type: %s') % k)

        return hex(self._hashes[key].digest())

    def __iter__(self):

        return iter(self._hashes)

    @staticmethod

    def preferred(supported):

        """returns the strongest digest type in both supported and DIGESTS."""

        for k in DIGESTS_BY_STRENGTH:

            if k in supported:

                return k

        return None

class digestchecker:

    """file handle wrapper that additionally checks content against a given

    size and digests.

        d = digestchecker(fh, size, {'md5': '...'})

    When multiple digests are given, all of them are validated.

    """

    def __init__(self, fh, size, digests):

        self._fh = fh

        self._size = size

        self._got = 0

        self._digests = dict(digests)

        self._digester = digester(self._digests.keys())

    def read(self, length=-1):

        content = self._fh.read(length)

        self._digester.update(content)

        self._got += len(content)

        return content

    def validate(self):

        if self._size != self._got:

            raise error.Abort(

                _(b'size mismatch: expected %d, got %d')

                % (self._size, self._got)

            )

        for k, v in self._digests.items():

            if v != self._digester[k]:

                # i18n: first parameter is a digest name

                raise error.Abort(

                    _(b'%s mismatch: expected %s, got %s')

                    % (k, v, self._digester[k])

                )

try:

    buffer = buffer  # pytype: disable=name-error

except NameError:

    def buffer(sliceable, offset=0, length=None):

        if length is not None:

            view = memoryview(sliceable)[offset : offset + length]

        else:

            view = memoryview(sliceable)[offset:]

        return view.toreadonly()

_chunksize = 4096

class bufferedinputpipe:

    """a manually buffered input pipe

    Python will not let us use buffered IO and lazy reading with 'polling' at

    the same time. We cannot probe the buffer state and select will not detect

    that data are ready to read if they are already buffered.

    This class let us work around that by implementing its own buffering

    (allowing efficient readline) while offering a way to know if the buffer is

    empty from the output (allowing collaboration of the buffer with polling).

    This class lives in the 'util' module because it makes use of the 'os'

    module from the python stdlib.

    """

    def __new__(cls, fh):

        # If we receive a fileobjectproxy, we need to use a variation of this

        # class that notifies observers about activity.

        if isinstance(fh, fileobjectproxy):

            cls = observedbufferedinputpipe

        return super().__new__(cls)

    def __init__(self, input):

        self._input = input

        self._buffer = []

        self._eof = False

        self._lenbuf = 0

    @property

    def hasbuffer(self):

        """True is any data is currently buffered

        This will be used externally a pre-step for polling IO. If there is

        already data then no polling should be set in place."""

        return bool(self._buffer)

    @property

    def closed(self):

        return self._input.closed

    def fileno(self):

        return self._input.fileno()

    def close(self):

        return self._input.close()

    def read(self, size):

        while (not self._eof) and (self._lenbuf < size):

            self._fillbuffer()

        return self._frombuffer(size)

    def unbufferedread(self, size):

        if not self._eof and self._lenbuf == 0:

            self._fillbuffer(max(size, _chunksize))

        return self._frombuffer(min(self._lenbuf, size))

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

        if len(self._buffer) > 1:

            # this should not happen because both read and readline end with a

            # _frombuffer call that collapse it.

            self._buffer = [b''.join(self._buffer)]

            self._lenbuf = len(self._buffer[0])

        lfi = -1

        if self._buffer:

            lfi = self._buffer[-1].find(b'\n')

        while (not self._eof) and lfi < 0:

            self._fillbuffer()

            if self._buffer:

                lfi = self._buffer[-1].find(b'\n')

        size = lfi + 1

        if lfi < 0:  # end of file

            size = self._lenbuf

        elif len(self._buffer) > 1:

            # we need to take previous chunks into account

            size += self._lenbuf - len(self._buffer[-1])

        return self._frombuffer(size)

    def _frombuffer(self, size):

        """return at most 'size' data from the buffer

        The data are removed from the buffer."""

        if size == 0 or not self._buffer:

            return b''

        buf = self._buffer[0]

        if len(self._buffer) > 1:

            buf = b''.join(self._buffer)

        data = buf[:size]

        buf = buf[len(data) :]

        if buf:

            self._buffer = [buf]

            self._lenbuf = len(buf)

        else:

            self._buffer = []

            self._lenbuf = 0

        return data

    def _fillbuffer(self, size=_chunksize):

        """read data to the buffer"""

        data = os.read(self._input.fileno(), size)

        if not data:

            self._eof = True

        else:

            self._lenbuf += len(data)

            self._buffer.append(data)

        return data

def has_mmap_populate():

    return hasattr(osutil, "background_mmap_populate") or hasattr(

        mmap, 'MAP_POPULATE'

    )