# ui.py - user interface bits for mercurial

#

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

#

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

import contextlib

import datetime

import errno

import inspect

import os

import re

import signal

import socket

import subprocess

import sys

import traceback

import typing

from typing import (

    Any,

    Callable,

    Dict,

    List,

    NoReturn,

    Optional,

    Tuple,

    Type,

    TypeVar,

    Union,

    cast,

    overload,

)

from .i18n import _

from .node import hex

from . import (

    color,

    config,

    configitems,

    encoding,

    error,

    extensions,

    formatter,

    loggingutil,

    progress,

    pycompat,

    scmutil,

    util,

)

from .configuration import rcutil

from .utils import (

    dateutil,

    procutil,

    resourceutil,

    stringutil,

    urlutil,

)

_ConfigItems = Dict[Tuple[bytes, bytes], object]  # {(section, name) : value}

# The **opts args of the various write() methods can be basically anything, but

# there's no way to express it as "anything but str".  So type it to be the

# handful of known types that are used.

_MsgOpts = Union[bytes, bool, List["_PromptChoice"]]

_PromptChoice = Tuple[bytes, bytes]

_Tui = TypeVar('_Tui', bound="ui")

urlreq = util.urlreq

# for use with str.translate(None, _keepalnum), to keep just alphanumerics

_keepalnum: bytes = b''.join(

    c for c in map(pycompat.bytechr, range(256)) if not c.isalnum()

)

# The config knobs that will be altered (if unset) by ui.tweakdefaults.

tweakrc: bytes = b"""

[ui]

# The rollback command is dangerous. As a rule, don't use it.

rollback = False

# Make `hg status` report copy information

statuscopies = yes

# Prefer curses UIs when available. Revert to plain-text with `text`.

interface = curses

# Make compatible commands emit cwd-relative paths by default.

relative-paths = yes

[commands]

# Grep working directory by default.

grep.all-files = True

# Refuse to perform an `hg update` that would cause a file content merge

update.check = noconflict

# Show conflicts information in `hg status`

status.verbose = True

# Make `hg resolve` with no action (like `-m`) fail instead of re-merging.

resolve.explicit-re-merge = True

[diff]

git = 1

showfunc = 1

word-diff = 1

"""

samplehgrcs: Dict[bytes, bytes] = {

    b'user': b"""# example user config (see 'hg help config' for more info)

[ui]

# name and email, e.g.

# username = Jane Doe <jdoe@example.com>

username =

# We recommend enabling tweakdefaults to get slight improvements to

# the UI over time. Make sure to set HGPLAIN in the environment when

# writing scripts!

# tweakdefaults = True

# uncomment to disable color in command output

# (see 'hg help color' for details)

# color = never

# uncomment to disable command output pagination

# (see 'hg help pager' for details)

# paginate = never

[extensions]

# uncomment the lines below to enable some popular extensions

# (see 'hg help extensions' for more info)

#

# histedit =

# rebase =

# uncommit =

""",

    b'cloned': b"""# example repository config (see 'hg help config' for more info)

[paths]

default = %s

# path aliases to other clones of this repo in URLs or filesystem paths

# (see 'hg help config.paths' for more info)

#

# default:pushurl = ssh://jdoe@example.net/hg/jdoes-fork

# my-fork         = ssh://jdoe@example.net/hg/jdoes-fork

# my-clone        = /home/jdoe/jdoes-clone

[ui]

# name and email (local to this repository, optional), e.g.

# username = Jane Doe <jdoe@example.com>

""",

    b'local': b"""# example repository config (see 'hg help config' for more info)

[paths]

# path aliases to other clones of this repo in URLs or filesystem paths

# (see 'hg help config.paths' for more info)

#

# default         = http://example.com/hg/example-repo

# default:pushurl = ssh://jdoe@example.net/hg/jdoes-fork

# my-fork         = ssh://jdoe@example.net/hg/jdoes-fork

# my-clone        = /home/jdoe/jdoes-clone

[ui]

# name and email (local to this repository, optional), e.g.

# username = Jane Doe <jdoe@example.com>

""",

    b'global': b"""# example system-wide hg config (see 'hg help config' for more info)

[ui]

# uncomment to disable color in command output

# (see 'hg help color' for details)

# color = never

# uncomment to disable command output pagination

# (see 'hg help pager' for details)

# paginate = never

[extensions]

# uncomment the lines below to enable some popular extensions

# (see 'hg help extensions' for more info)

#

# blackbox =

# churn =

""",

}

def _maybestrurl(maybebytes):

    return pycompat.rapply(pycompat.strurl, maybebytes)

def _maybebytesurl(maybestr):

    return pycompat.rapply(pycompat.bytesurl, maybestr)

class httppasswordmgrdbproxy:

    """Delays loading urllib2 until it's needed."""

    def __init__(self) -> None:

        self._mgr = None

    def _get_mgr(self):

        if self._mgr is None:

            self._mgr = urlreq.httppasswordmgrwithdefaultrealm()

        return self._mgr

    def add_password(self, realm, uris, user, passwd):

        return self._get_mgr().add_password(

            _maybestrurl(realm),

            _maybestrurl(uris),

            _maybestrurl(user),

            _maybestrurl(passwd),

        )

    def find_user_password(self, realm, uri):

        mgr = self._get_mgr()

        return _maybebytesurl(

            mgr.find_user_password(_maybestrurl(realm), _maybestrurl(uri))

        )

def _catchterm(*args) -> NoReturn:

    raise error.SignalInterrupt

# unique object used to detect no default value has been provided when

# retrieving configuration value.

_unset = object()

# _reqexithandlers: callbacks run at the end of a request

_reqexithandlers: List = []

class ui:

    def __init__(self, src: Optional["ui"] = None) -> None:

        """Create a fresh new ui object if no src given

        Use uimod.ui.load() to create a ui which knows global and user configs.

        In most cases, you should use ui.copy() to create a copy of an existing

        ui object.

        """

        # _buffers: used for temporary capture of output

        self._buffers = []

        # 3-tuple describing how each buffer in the stack behaves.

        # Values are (capture stderr, capture subprocesses, apply labels).

        self._bufferstates = []

        # When a buffer is active, defines whether we are expanding labels.

        # This exists to prevent an extra list lookup.

        self._bufferapplylabels = None

        self.quiet = self.verbose = self.debugflag = self.tracebackflag = False

        self._reportuntrusted = True

        self._knownconfig = configitems.coreitems

        self._ocfg = config.config()  # overlay

        self._tcfg = config.config()  # trusted

        self._ucfg = config.config()  # untrusted

        self._trustusers = set()

        self._trustgroups = set()

        self.callhooks = True

        # hold the root to use for each [paths] entry

        self._path_to_root = {}

        # Insecure server connections requested.

        self.insecureconnections = False

        # Blocked time

        self.logblockedtimes = False

        # color mode: see mercurial/color.py for possible value

        self._colormode = None

        self._terminfoparams = {}

        self._styles = {}

        self._uninterruptible = False

        self.showtimestamp = False

        if src:

            self._fout = src._fout

            self._ferr = src._ferr

            self._fin = src._fin

            self._fmsg = src._fmsg

            self._fmsgout = src._fmsgout

            self._fmsgerr = src._fmsgerr

            self._finoutredirected = src._finoutredirected

            self._loggers = src._loggers.copy()

            self.pageractive = src.pageractive

            self._disablepager = src._disablepager

            self._tweaked = src._tweaked

            self._tcfg = src._tcfg.copy()

            self._ucfg = src._ucfg.copy()

            self._ocfg = src._ocfg.copy()

            self._trustusers = src._trustusers.copy()

            self._trustgroups = src._trustgroups.copy()

            self.environ = src.environ

            self.callhooks = src.callhooks

            self._path_to_root = src._path_to_root

            self.insecureconnections = src.insecureconnections

            self._colormode = src._colormode

            self._terminfoparams = src._terminfoparams.copy()

            self._styles = src._styles.copy()

            self.fixconfig()

            self.httppasswordmgrdb = src.httppasswordmgrdb

            self._blockedtimes = src._blockedtimes

        else:

            self._fout = procutil.stdout

            self._ferr = procutil.stderr

            self._fin = procutil.stdin

            self._fmsg = None

            self._fmsgout = self.fout  # configurable

            self._fmsgerr = self.ferr  # configurable

            self._finoutredirected = False

            self._loggers = {}

            self.pageractive = False

            self._disablepager = False

            self._tweaked = False

            # shared read-only environment

            self.environ = encoding.environ

            self.httppasswordmgrdb = httppasswordmgrdbproxy()

            self._blockedtimes = collections.defaultdict(int)

        allowed = self.configlist(b'experimental', b'exportableenviron')

        if b'*' in allowed:

            self._exportableenviron = self.environ

        else:

            self._exportableenviron = {}

            for k in allowed:

                if k in self.environ:

                    self._exportableenviron[k] = self.environ[k]

    def _new_source(self) -> None:

        self._ocfg.new_source()

        self._tcfg.new_source()

        self._ucfg.new_source()

    @classmethod

    def load(cls: Type[_Tui]) -> _Tui:

        """Create a ui and load global and user configs"""

        u = cls()

        # we always trust global config files and environment variables

        for _lvl, t, f in rcutil.rccomponents():

            if t == b'path':

                u.readconfig(f, trust=True)

            elif t == b'resource':

                u.read_resource_config(f, trust=True)

            elif t == b'items':

                u._new_source()

                sections = set()

                for section, name, value, source in f:

                    # do not set u._ocfg

                    # XXX clean this up once immutable config object is a thing

                    u._tcfg.set(section, name, value, source)

                    u._ucfg.set(section, name, value, source)

                    sections.add(section)

                for section in sections:

                    u.fixconfig(section=section)

            else:

                raise error.ProgrammingError(b'unknown rctype: %s' % t)

        u._maybetweakdefaults()

        u._new_source()  # anything after that is a different level

        return u

    def _maybetweakdefaults(self) -> None:

        if not self.configbool(b'ui', b'tweakdefaults'):

            return

        if self._tweaked or self.plain(b'tweakdefaults'):

            return

        # Note: it is SUPER IMPORTANT that you set self._tweaked to

        # True *before* any calls to setconfig(), otherwise you'll get

        # infinite recursion between setconfig and this method.

        #

        # TODO: We should extract an inner method in setconfig() to

        # avoid this weirdness.

        self._tweaked = True

        tmpcfg = config.config()

        tmpcfg.parse(b'<tweakdefaults>', tweakrc)

        for section in tmpcfg:

            for name, value in tmpcfg.items(section):

                if not self.hasconfig(section, name):

                    self.setconfig(section, name, value, b"<tweakdefaults>")

    def copy(self: _Tui) -> _Tui:

        return self.__class__(self)

    def resetstate(self) -> None:

        """Clear internal state that shouldn't persist across commands"""

        if self._progbar:

            self._progbar.resetstate()  # reset last-print time of progress bar

        self.httppasswordmgrdb = httppasswordmgrdbproxy()

    @contextlib.contextmanager

    def timeblockedsection(self, key: bytes):

        # this is open-coded below - search for timeblockedsection to find them

        starttime = util.timer()

        try:

            yield

        finally:

            self._blockedtimes[key + b'_blocked'] += (

                util.timer() - starttime

            ) * 1000

    @contextlib.contextmanager

    def uninterruptible(self):

        """Mark an operation as unsafe.

        Most operations on a repository are safe to interrupt, but a

        few are risky (for example repair.strip). This context manager

        lets you advise Mercurial that something risky is happening so

        that control-C etc can be blocked if desired.

        """

        enabled = self.configbool(b'experimental', b'nointerrupt')

        if enabled and self.configbool(

            b'experimental', b'nointerrupt-interactiveonly'

        ):

            enabled = self.interactive()

        if self._uninterruptible or not enabled:

            # if nointerrupt support is turned off, the process isn't

            # interactive, or we're already in an uninterruptible

            # block, do nothing.

            yield

            return

        def warn():

            self.warn(_(b"shutting down cleanly\n"))

            self.warn(

                _(b"press ^C again to terminate immediately (dangerous)\n")

            )

            return True

        with procutil.uninterruptible(warn):

            try:

                self._uninterruptible = True

                yield

            finally:

                self._uninterruptible = False

    def formatter(self, topic: bytes, opts):

        return formatter.formatter(self, self, topic, opts)

    def _trusted(self, fp, f: bytes) -> bool:

        st = util.fstat(fp)

        if util.isowner(st):

            return True

        tusers, tgroups = self._trustusers, self._trustgroups

        if b'*' in tusers or b'*' in tgroups:

            return True

        user = util.username(st.st_uid)

        group = util.groupname(st.st_gid)

        if user in tusers or group in tgroups or user == util.username():

            return True

        if self._reportuntrusted:

            self.warn(

                _(

                    b'not trusting file %s from untrusted '

                    b'user %s, group %s\n'

                )

                % (f, user, group)

            )

        return False

    def read_resource_config(

        self, name, root=None, trust=False, sections=None, remap=None

    ) -> None:

        try:

            fp = resourceutil.open_resource(name[0], name[1])

            if not sections:  # ignore unless we were looking for something

                return

            raise

        self._readconfig(

            b'resource:%s.%s' % name, fp, root, trust, sections, remap

        )

    def readconfig(

        self, filename, root=None, trust=False, sections=None, remap=None