# extensions.py - extension handling 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.

import ast

import collections

import functools

import importlib

import inspect

import os

import sys

from .i18n import (

    _,

    gettext,

)

from .pycompat import (

    getattr,

    open,

    setattr,

)

from . import (

    cmdutil,

    configitems,

    error,

    pycompat,

    util,

)

from .utils import stringutil

_extensions = {}

_disabledextensions = {}

_aftercallbacks = {}

_order = []

_builtin = {

    b'hbisect',

    b'bookmarks',

    b'color',

    b'parentrevspec',

    b'progress',

    b'interhg',

    b'inotify',

    b'hgcia',

    b'shelve',

}

def extensions(ui=None):

    if ui:

        def enabled(name):

            for format in [b'%s', b'hgext.%s']:

                conf = ui.config(b'extensions', format % name)

                if conf is not None and not conf.startswith(b'!'):

                    return True

    else:

        enabled = lambda name: True

    for name in _order:

        module = _extensions[name]

        if module and enabled(name):

            yield name, module

def find(name):

    '''return module with given extension name'''

    mod = None

    try:

        mod = _extensions[name]

    except KeyError:

        for k, v in _extensions.items():

            if k.endswith(b'.' + name) or k.endswith(b'/' + name):

                mod = v

                break

    if not mod:

        raise KeyError(name)

    return mod

def loadpath(path, module_name):

    module_name = module_name.replace(b'.', b'_')

    path = util.normpath(util.expandpath(path))

    module_name = pycompat.fsdecode(module_name)

    path = pycompat.fsdecode(path)

    if os.path.isdir(path):

        # module/__init__.py style

        init_py_path = os.path.join(path, '__init__.py')

        if not os.path.exists(init_py_path):

            raise ImportError("No module named '%s'" % os.path.basename(path))

        path = init_py_path

    loader = importlib.machinery.SourceFileLoader(module_name, path)

    spec = importlib.util.spec_from_file_location(module_name, loader=loader)

    assert spec is not None  # help Pytype

    module = importlib.util.module_from_spec(spec)

    sys.modules[module_name] = module

    spec.loader.exec_module(module)

    return module

def _importh(name):

    """import and return the <name> module"""

    mod = __import__(pycompat.sysstr(name))

    components = name.split(b'.')

    for comp in components[1:]:

        mod = getattr(mod, comp)

    return mod

def _importext(name, path=None, reportfunc=None):

    if path:

        # the module will be loaded in sys.modules

        # choose an unique name so that it doesn't

        # conflicts with other modules

        mod = loadpath(path, b'hgext.%s' % name)

    else:

        try:

            mod = _importh(b"hgext.%s" % name)

        except ImportError as err:

            if reportfunc:

                reportfunc(err, b"hgext.%s" % name, b"hgext3rd.%s" % name)

            try:

                mod = _importh(b"hgext3rd.%s" % name)

            except ImportError as err:

                if reportfunc:

                    reportfunc(err, b"hgext3rd.%s" % name, name)

                mod = _importh(name)

    return mod

def _reportimporterror(ui, err, failed, next):

    # note: this ui.log happens before --debug is processed,

    #       Use --config ui.debug=1 to see them.

    ui.log(

        b'extension',

        b'    - could not import %s (%s): trying %s\n',

        failed,

        stringutil.forcebytestr(err),

        next,

    )

    if ui.debugflag and ui.configbool(b'devel', b'debug.extensions'):

        ui.traceback()

def _rejectunicode(name, xs):

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

        for x in xs:

            _rejectunicode(name, x)

    elif isinstance(xs, dict):

        for k, v in xs.items():

            _rejectunicode(name, k)

            _rejectunicode(b'%s.%s' % (name, stringutil.forcebytestr(k)), v)

    elif isinstance(xs, type(u'')):

        raise error.ProgrammingError(

            b"unicode %r found in %s" % (xs, name),

            hint=b"use b'' to make it byte string",

        )

# attributes set by registrar.command

_cmdfuncattrs = (b'norepo', b'optionalrepo', b'inferrepo')

def _validatecmdtable(ui, cmdtable):

    """Check if extension commands have required attributes"""

    for c, e in cmdtable.items():

        f = e[0]

        missing = [a for a in _cmdfuncattrs if not util.safehasattr(f, a)]

        if not missing:

            continue

        raise error.ProgrammingError(

            b'missing attributes: %s' % b', '.join(missing),

            hint=b"use @command decorator to register '%s'" % c,

        )

def _validatetables(ui, mod):

    """Sanity check for loadable tables provided by extension module"""

    for t in [b'cmdtable', b'colortable', b'configtable']:

        _rejectunicode(t, getattr(mod, t, {}))

    for t in [

        b'filesetpredicate',

        b'internalmerge',

        b'revsetpredicate',

        b'templatefilter',

        b'templatefunc',

        b'templatekeyword',

    ]:

        o = getattr(mod, t, None)

        if o:

            _rejectunicode(t, o._table)

    _validatecmdtable(ui, getattr(mod, 'cmdtable', {}))

def load(ui, name, path, loadingtime=None):

    if name.startswith(b'hgext.') or name.startswith(b'hgext/'):

        shortname = name[6:]

    else:

        shortname = name

    if shortname in _builtin:

        return None

    if shortname in _extensions:

        return _extensions[shortname]

    ui.log(b'extension', b'  - loading extension: %s\n', shortname)

    _extensions[shortname] = None

    with util.timedcm('load extension %s', shortname) as stats:

        mod = _importext(name, path, bind(_reportimporterror, ui))

    ui.log(b'extension', b'  > %s extension loaded in %s\n', shortname, stats)

    if loadingtime is not None:

        loadingtime[shortname] += stats.elapsed

    # Before we do anything with the extension, check against minimum stated

    # compatibility. This gives extension authors a mechanism to have their

    # extensions short circuit when loaded with a known incompatible version

    # of Mercurial.

    minver = getattr(mod, 'minimumhgversion', None)

    if minver:

        curver = util.versiontuple(n=2)

        extmin = util.versiontuple(stringutil.forcebytestr(minver), 2)

        if None in extmin:

            extmin = (extmin[0] or 0, extmin[1] or 0)

        if None in curver or extmin > curver:

            msg = _(

                b'(third party extension %s requires version %s or newer '

                b'of Mercurial (current: %s); disabling)\n'

            )

            ui.warn(msg % (shortname, minver, util.version()))

            return

    ui.log(b'extension', b'    - validating extension tables: %s\n', shortname)

    _validatetables(ui, mod)

    _extensions[shortname] = mod

    _order.append(shortname)

    ui.log(

        b'extension', b'    - invoking registered callbacks: %s\n', shortname

    )

    with util.timedcm('callbacks extension %s', shortname) as stats:

        for fn in _aftercallbacks.get(shortname, []):

            fn(loaded=True)

    ui.log(b'extension', b'    > callbacks completed in %s\n', stats)

    return mod

def _runuisetup(name, ui):

    uisetup = getattr(_extensions[name], 'uisetup', None)

    if uisetup:

        try:

            uisetup(ui)

        except Exception as inst:

            ui.traceback(force=True)

            msg = stringutil.forcebytestr(inst)

            ui.warn(_(b"*** failed to set up extension %s: %s\n") % (name, msg))

            return False

    return True

def _runextsetup(name, ui):

    extsetup = getattr(_extensions[name], 'extsetup', None)

    if extsetup:

        try:

            extsetup(ui)

        except Exception as inst:

            ui.traceback(force=True)

            msg = stringutil.forcebytestr(inst)

            ui.warn(_(b"*** failed to set up extension %s: %s\n") % (name, msg))

            return False

    return True

def loadall(ui, whitelist=None):

    loadingtime = collections.defaultdict(int)

    result = ui.configitems(b"extensions")

    if whitelist is not None:

        result = [(k, v) for (k, v) in result if k in whitelist]

    result = [(k, v) for (k, v) in result if b':' not in k]

    newindex = len(_order)

    ui.log(

        b'extension',

        b'loading %sextensions\n',

        b'additional ' if newindex else b'',

    )

    ui.log(b'extension', b'- processing %d entries\n', len(result))

    with util.timedcm('load all extensions') as stats:

        default_sub_options = ui.configsuboptions(b"extensions", b"*")[1]

        for (name, path) in result:

            if path:

                if path[0:1] == b'!':

                    if name not in _disabledextensions:

                        ui.log(

                            b'extension',

                            b'  - skipping disabled extension: %s\n',

                            name,

                        )

                    _disabledextensions[name] = path[1:]

                    continue

            try:

                load(ui, name, path, loadingtime)

            except Exception as inst:

                msg = stringutil.forcebytestr(inst)

                if path:

                    error_msg = _(

                        b'failed to import extension "%s" from %s: %s'

                    )

                    error_msg %= (name, path, msg)

                else:

                    error_msg = _(b'failed to import extension "%s": %s')

                    error_msg %= (name, msg)

                options = default_sub_options.copy()

                ext_options = ui.configsuboptions(b"extensions", name)[1]

                options.update(ext_options)

                if stringutil.parsebool(options.get(b"required", b'no')):

                    hint = None

                    if isinstance(inst, error.Hint) and inst.hint:

                        hint = inst.hint

                    if hint is None:

                        hint = _(

                            b"loading of this extension was required, "

                            b"see `hg help config.extensions` for details"

                        )

                    raise error.Abort(error_msg, hint=hint)

                else:

                    ui.warn((b"*** %s\n") % error_msg)

                    if isinstance(inst, error.Hint) and inst.hint:

                        ui.warn(_(b"*** (%s)\n") % inst.hint)

                    ui.traceback()

    ui.log(

        b'extension',

        b'> loaded %d extensions, total time %s\n',

        len(_order) - newindex,

        stats,

    )

    # list of (objname, loadermod, loadername) tuple:

    # - objname is the name of an object in extension module,

    #   from which extra information is loaded

    # - loadermod is the module where loader is placed

    # - loadername is the name of the function,

    #   which takes (ui, extensionname, extraobj) arguments

    #

    # This one is for the list of item that must be run before running any setup

    earlyextraloaders = [

        (b'configtable', configitems, b'loadconfigtable'),

    ]

    ui.log(b'extension', b'- loading configtable attributes\n')

    _loadextra(ui, newindex, earlyextraloaders)

    broken = set()

    ui.log(b'extension', b'- executing uisetup hooks\n')

    with util.timedcm('all uisetup') as alluisetupstats:

        for name in _order[newindex:]:

            ui.log(b'extension', b'  - running uisetup for %s\n', name)

            with util.timedcm('uisetup %s', name) as stats: