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

#

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

#  Copyright 2005-2007 Matt Mackall <mpm@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 absolute_import, print_function

import abc

import bz2

import collections

import contextlib

import errno

import gc

import hashlib

import imp

import io

import itertools

import mmap

import os

import platform as pyplatform

import re as remod

import shutil

import signal

import socket

import stat

import subprocess

import sys

import tempfile

import time

import traceback

import warnings

import zlib

from . import (

    encoding,

    error,

    i18n,

    node as nodemod,

    policy,

    pycompat,

    urllibcompat,

)

base85 = policy.importmod(r'base85')

osutil = policy.importmod(r'osutil')

parsers = policy.importmod(r'parsers')

b85decode = base85.b85decode

b85encode = base85.b85encode

cookielib = pycompat.cookielib

empty = pycompat.empty

httplib = pycompat.httplib

pickle = pycompat.pickle

queue = pycompat.queue

socketserver = pycompat.socketserver

stderr = pycompat.stderr

stdin = pycompat.stdin

stdout = pycompat.stdout

bytesio = pycompat.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

def isatty(fp):

    try:

        return fp.isatty()

    except AttributeError:

        return False

# glibc determines buffering on first write to stdout - if we replace a TTY

# destined stdout with a pipe destined stdout (e.g. pager), we want line

# buffering

if isatty(stdout):

    stdout = os.fdopen(stdout.fileno(), r'wb', 1)

if pycompat.iswindows:

    from . import windows as platform

    stdout = platform.winstdout(stdout)

else:

    from . import posix as platform

_ = i18n._

bindunixsocket = platform.bindunixsocket

cachestat = platform.cachestat

checkexec = platform.checkexec

checklink = platform.checklink

copymode = platform.copymode

executablepath = platform.executablepath

expandglobs = platform.expandglobs

explainexit = platform.explainexit

findexe = platform.findexe

getfsmountpoint = platform.getfsmountpoint

getfstype = platform.getfstype

gethgcmd = platform.gethgcmd

getuser = platform.getuser

getpid = os.getpid

groupmembers = platform.groupmembers

groupname = platform.groupname

hidewindow = platform.hidewindow

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

popen = platform.popen

posixfile = platform.posixfile

quotecommand = platform.quotecommand

readpipe = platform.readpipe

rename = platform.rename

removedirs = platform.removedirs

samedevice = platform.samedevice

samefile = platform.samefile

samestat = platform.samestat

setbinary = platform.setbinary

setflags = platform.setflags

setsignalhandler = platform.setsignalhandler

shellquote = platform.shellquote

shellsplit = platform.shellsplit

spawndetached = platform.spawndetached

split = platform.split

sshargs = platform.sshargs

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

statisexec = platform.statisexec

statislink = platform.statislink

testpid = platform.testpid

umask = platform.umask

unlink = platform.unlink

username = platform.username

try:

    recvfds = osutil.recvfds

except AttributeError:

    pass

try:

    setprocname = osutil.setprocname

except AttributeError:

    pass

try:

    unblocksignal = osutil.unblocksignal

except AttributeError:

    pass

# Python compatibility

_notset = object()

def safehasattr(thing, attr):

    return getattr(thing, attr, _notset) is not _notset

def _rapply(f, xs):

    if xs is None:

        # assume None means non-value of optional data

        return xs

    if isinstance(xs, (list, set, tuple)):

        return type(xs)(_rapply(f, x) for x in xs)

    if isinstance(xs, dict):

        return type(xs)((_rapply(f, k), _rapply(f, v)) for k, v in xs.items())

    return f(xs)

def rapply(f, xs):

    """Apply function recursively to every item preserving the data structure

    >>> def f(x):

    ...     return 'f(%s)' % x

    >>> rapply(f, None) is None

    True

    >>> rapply(f, 'a')

    'f(a)'

    >>> rapply(f, {'a'}) == {'f(a)'}

    True

    >>> rapply(f, ['a', 'b', None, {'c': 'd'}, []])

    ['f(a)', 'f(b)', None, {'f(c)': 'f(d)'}, []]

    >>> xs = [object()]

    >>> rapply(pycompat.identity, xs) is xs

    True

    """

    if f is pycompat.identity:

        # fast path mainly for py2

        return xs

    return _rapply(f, xs)

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('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(r'default', r'', DeprecationWarning, r'mercurial')

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

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

if _dowarn and pycompat.ispy3:

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

    warnings.filterwarnings(r'ignore', r'bad escape', DeprecationWarning,

                            r'mercurial')

    warnings.filterwarnings(r'ignore', r'invalid escape sequence',

                            DeprecationWarning, r'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 += ("\n(compatibility will be dropped after Mercurial-%s,"

                " update your code.)") % version

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

DIGESTS = {

    'md5': hashlib.md5,

    'sha1': hashlib.sha1,

    'sha512': hashlib.sha512,

}

# List of digest types from strongest to weakest

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

for k in DIGESTS_BY_STRENGTH:

    assert k in DIGESTS

class digester(object):

    """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=''):

        self._hashes = {}

        for k in digests:

            if k not in DIGESTS:

                raise Abort(_('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 Abort(_('unknown digest type: %s') % k)

        return nodemod.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(object):

    """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 Abort(_('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 Abort(_('%s mismatch: expected %s, got %s') %

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

try:

    buffer = buffer

except NameError:

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

        if length is not None:

            return memoryview(sliceable)[offset:offset + length]

        return memoryview(sliceable)[offset:]

closefds = pycompat.isposix

_chunksize = 4096

class bufferedinputpipe(object):

    """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(bufferedinputpipe, cls).__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