# subrepo.py - sub-repository classes and factory

#

# Copyright 2009-2010 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 copy

import errno

import os

import re

import stat

import subprocess

import sys

import tarfile

import xml.dom.minidom

from .i18n import _

from .interfaces.types import (

    MatcherT,

)

from .node import (

    bin,

    hex,

    short,

)

from . import (

    cmdutil,

    encoding,

    error,

    exchange,

    logcmdutil,

    match as matchmod,

    merge as merge,

    pathutil,

    phases,

    pycompat,

    scmutil,

    subrepoutil,

    util,

    vfs as vfsmod,

)

from .interfaces import (

    status as istatus,

)

from .utils import (

    dateutil,

    hashutil,

    procutil,

    urlutil,

)

hg = None

reporelpath = subrepoutil.reporelpath

subrelpath = subrepoutil.subrelpath

_abssource = subrepoutil._abssource

propertycache = util.propertycache

def _expandedabspath(path):

    """

    get a path or url and if it is a path expand it and return an absolute path

    """

    expandedpath = urlutil.urllocalpath(util.expandpath(path))

    u = urlutil.url(expandedpath)

    if not u.scheme:

        path = util.normpath(util.abspath(u.path))

    return path

def _getstorehashcachename(remotepath):

    '''get a unique filename for the store hash cache of a remote repository'''

    return hex(hashutil.sha1(_expandedabspath(remotepath)).digest())[0:12]

class SubrepoAbort(error.Abort):

    """Exception class used to avoid handling a subrepo error more than once"""

    def __init__(self, *args, **kw):

        self.subrepo = kw.pop('subrepo', None)

        self.cause = kw.pop('cause', None)

        error.Abort.__init__(self, *args, **kw)

def annotatesubrepoerror(func):

    def decoratedmethod(self, *args, **kargs):

        try:

            res = func(self, *args, **kargs)

        except SubrepoAbort as ex:

            # This exception has already been handled

            raise ex

        except error.Abort as ex:

            subrepo = subrelpath(self)

            errormsg = (

                ex.message + b' ' + _(b'(in subrepository "%s")') % subrepo

            )

            # avoid handling this exception by raising a SubrepoAbort exception

            raise SubrepoAbort(

                errormsg, hint=ex.hint, subrepo=subrepo, cause=sys.exc_info()

            )

        return res

    return decoratedmethod

def _updateprompt(ui, sub, dirty, local, remote):

    if dirty:

        msg = _(

            b' subrepository sources for %s differ\n'

            b'you can use (l)ocal source (%s) or (r)emote source (%s).\n'

            b'what do you want to do?'

            b'$$ &Local $$ &Remote'

        ) % (subrelpath(sub), local, remote)

    else:

        msg = _(

            b' subrepository sources for %s differ (in checked out '

            b'version)\n'

            b'you can use (l)ocal source (%s) or (r)emote source (%s).\n'

            b'what do you want to do?'

            b'$$ &Local $$ &Remote'

        ) % (subrelpath(sub), local, remote)

    return ui.promptchoice(msg, 0)

def _sanitize(ui, vfs, ignore):

    for dirname, dirs, names in vfs.walk():

        for i, d in enumerate(dirs):

            if d.lower() == ignore:

                del dirs[i]

                break

        if vfs.basename(dirname).lower() != b'.hg':

            continue

        for f in names:

            if f.lower() == b'hgrc':

                ui.warn(

                    _(

                        b"warning: removing potentially hostile 'hgrc' "

                        b"in '%s'\n"

                    )

                    % vfs.join(dirname)

                )

                vfs.unlink(vfs.reljoin(dirname, f))

def _auditsubrepopath(repo, path):

    # sanity check for potentially unsafe paths such as '~' and '$FOO'

    if path.startswith(b'~') or b'$' in path or util.expandpath(path) != path:

        raise error.Abort(

            _(b'subrepo path contains illegal component: %s') % path

        )

    # auditor doesn't check if the path itself is a symlink

    pathutil.pathauditor(repo.root)(path)

    if repo.wvfs.islink(path):

        raise error.Abort(_(b"subrepo '%s' traverses symbolic link") % path)

SUBREPO_ALLOWED_DEFAULTS = {

    b'hg': True,

    b'git': False,

    b'svn': False,

}

def _checktype(ui, kind):

    # subrepos.allowed is a master kill switch. If disabled, subrepos are

    # disabled period.

    if not ui.configbool(b'subrepos', b'allowed', True):

        raise error.Abort(

            _(b'subrepos not enabled'),

            hint=_(b"see 'hg help config.subrepos' for details"),

        )

    default = SUBREPO_ALLOWED_DEFAULTS.get(kind, False)

    if not ui.configbool(b'subrepos', b'%s:allowed' % kind, default):

        raise error.Abort(

            _(b'%s subrepos not allowed') % kind,

            hint=_(b"see 'hg help config.subrepos' for details"),

        )

    if kind not in types:

        raise error.Abort(_(b'unknown subrepo type %s') % kind)

def subrepo(ctx, path, allowwdir=False, allowcreate=True):

    """return instance of the right subrepo class for subrepo in path"""

    # subrepo inherently violates our import layering rules

    # because it wants to make repo objects from deep inside the stack

    # so we manually delay the circular imports to not break

    # scripts that don't use our demand-loading

    global hg

    from . import hg as h

    hg = h

    repo = ctx.repo()

    _auditsubrepopath(repo, path)

    state = ctx.substate[path]

    _checktype(repo.ui, state[2])

    if allowwdir:

        state = (state[0], ctx.subrev(path), state[2])

    return types[state[2]](ctx, path, state[:2], allowcreate)

def nullsubrepo(ctx, path, pctx):

    """return an empty subrepo in pctx for the extant subrepo in ctx"""

    # subrepo inherently violates our import layering rules

    # because it wants to make repo objects from deep inside the stack

    # so we manually delay the circular imports to not break

    # scripts that don't use our demand-loading

    global hg

    from . import hg as h

    hg = h

    repo = ctx.repo()

    _auditsubrepopath(repo, path)

    state = ctx.substate[path]

    _checktype(repo.ui, state[2])

    subrev = b''

    if state[2] == b'hg':

        subrev = b"0" * 40

    return types[state[2]](pctx, path, (state[0], subrev), True)

# subrepo classes need to implement the following abstract class:

class abstractsubrepo:

    def __init__(self, ctx, path):

        """Initialize abstractsubrepo part

        ``ctx`` is the context referring this subrepository in the

        parent repository.

        ``path`` is the path to this subrepository as seen from

        innermost repository.

        """

        self.ui = ctx.repo().ui

        self._ctx = ctx

        self._path = path

    def addwebdirpath(self, serverpath, webconf):

        """Add the hgwebdir entries for this subrepo, and any of its subrepos.

        ``serverpath`` is the path component of the URL for this repo.

        ``webconf`` is the dictionary of hgwebdir entries.

        """

        pass

    def storeclean(self, path):

        """

        returns true if the repository has not changed since it was last

        cloned from or pushed to a given repository.

        """

        return False

    def dirty(self, ignoreupdate=False, missing=False):

        """returns true if the dirstate of the subrepo is dirty or does not

        match current stored state. If ignoreupdate is true, only check

        whether the subrepo has uncommitted changes in its dirstate.  If missing

        is true, check for deleted files.

        """

        raise NotImplementedError

    def dirtyreason(self, ignoreupdate=False, missing=False):

        """return reason string if it is ``dirty()``

        Returned string should have enough information for the message

        of exception.

        This returns None, otherwise.

        """

        if self.dirty(ignoreupdate=ignoreupdate, missing=missing):

            return _(b'uncommitted changes in subrepository "%s"') % subrelpath(

                self

            )

    def bailifchanged(self, ignoreupdate=False, hint=None):

        """raise Abort if subrepository is ``dirty()``"""

        dirtyreason = self.dirtyreason(ignoreupdate=ignoreupdate, missing=True)

        if dirtyreason:

            raise error.Abort(dirtyreason, hint=hint)

    def basestate(self):

        """current working directory base state, disregarding .hgsubstate

        state and working directory modifications"""

        raise NotImplementedError

    def checknested(self, path):

        """check if path is a subrepository within this repository"""

        return False

    def commit(self, text, user, date):

        """commit the current changes to the subrepo with the given

        log message. Use given user and date if possible. Return the

        new state of the subrepo.

        """

        raise NotImplementedError

    def phase(self, state):

        """returns phase of specified state in the subrepository."""

        return phases.public

    def remove(self):

        """remove the subrepo

        (should verify the dirstate is not dirty first)

        """

        raise NotImplementedError

    def get(self, state, overwrite=False):

        """run whatever commands are needed to put the subrepo into

        this state

        """

        raise NotImplementedError

    def merge(self, state):

        """merge currently-saved state with the new state."""

        raise NotImplementedError

    def push(self, opts):

        """perform whatever action is analogous to 'hg push'

        This may be a no-op on some systems.

        """

        raise NotImplementedError

    def add(self, ui, match, prefix, uipathfn, explicitonly, **opts):

        return []

    def addremove(self, matcher, prefix, uipathfn, opts):

        self.ui.warn(b"%s: %s" % (prefix, _(b"addremove is not supported")))

        return 1

    def cat(self, match, fm, fntemplate, prefix, **opts):

        return 1

    def status(self, rev2, **opts) -> istatus.Status:

        return scmutil.status([], [], [], [], [], [], [])

    def diff(self, ui, diffopts, node2, match, prefix, **opts):

        pass

    def outgoing(self, ui, dest, opts):

        return 1

    def incoming(self, ui, source, opts):

        return 1

    def files(self):

        """return filename iterator"""

        raise NotImplementedError

    def filedata(self, name, decode):

        """return file data, optionally passed through repo decoders"""

        raise NotImplementedError

    def fileflags(self, name):

        """return file flags"""

        return b''

    def matchfileset(self, cwd, expr, badfn=None):

        """Resolve the fileset expression for this repo"""

        return matchmod.never(badfn=badfn)

    def printfiles(self, ui, m, uipathfn, fm, fmt, subrepos):

        """handle the files command for this subrepo"""

        return 1

    def archive(self, opener, prefix, match: MatcherT, decode=True):

        files = [f for f in self.files() if match(f)]

        total = len(files)

        relpath = subrelpath(self)

        progress = self.ui.makeprogress(

            _(b'archiving (%s)') % relpath, unit=_(b'files'), total=total

        )

        progress.update(0)

        archiver = None

        for name in files:

            flags = self.fileflags(name)

            mode = b'x' in flags and 0o755 or 0o644

            symlink = b'l' in flags

            if archiver is None:

                archiver = opener()

            archiver.addfile(

                prefix + name, mode, symlink, self.filedata(name, decode)

            )

            progress.increment()

        progress.complete()

        return total

    def walk(self, match):

        """

        walk recursively through the directory tree, finding all files

        matched by the match function

        """

    def forget(self, match, prefix, uipathfn, dryrun, interactive):

        return ([], [])

    def removefiles(

        self,

        matcher,

        prefix,

        uipathfn,

        after,

        force,

        subrepos,

        dryrun,

        warnings,

    ):

        """remove the matched files from the subrepository and the filesystem,

        possibly by force and/or after the file has been removed from the

        filesystem.  Return 0 on success, 1 on any warning.

        """

        warnings.append(

            _(b"warning: removefiles not implemented (%s)") % self._path

        )

        return 1

    def revert(self, substate, *pats, **opts):

        self.ui.warn(

            _(b'%s: reverting %s subrepos is unsupported\n')

            % (substate[0], substate[2])

        )

        return []

    def shortid(self, revid):

        return revid

    def unshare(self):

        """

        convert this repository from shared to normal storage.

        """

    def verify(self, onpush=False):

        """verify the revision of this repository that is held in `_state` is

        present and not hidden.  Return 0 on success or warning, 1 on any

        error.  In the case of ``onpush``, warnings or errors will raise an

        exception if the result of pushing would be a broken remote repository.

        """

        return 0

    @propertycache

    def wvfs(self):

        """return vfs to access the working directory of this subrepository"""

        return vfsmod.vfs(self._ctx.repo().wvfs.join(self._path))

    @propertycache

    def _relpath(self):

        """return path to this subrepository as seen from outermost repository"""

        return self.wvfs.reljoin(reporelpath(self._ctx.repo()), self._path)

class hgsubrepo(abstractsubrepo):

    def __init__(self, ctx, path, state, allowcreate):

        super().__init__(ctx, path)

        self._state = state

        r = ctx.repo()

        root = r.wjoin(util.localpath(path))

        create = allowcreate and not r.wvfs.exists(b'%s/.hg' % path)

        # repository constructor does expand variables in path, which is

        # unsafe since subrepo path might come from untrusted source.

        norm_root = os.path.normcase(root)

        real_root = os.path.normcase(os.path.realpath(util.expandpath(root)))

        if real_root != norm_root:

            raise error.Abort(

                _(b'subrepo path contains illegal component: %s') % path

            )

        self._repo = hg.repository(r.baseui, root, create=create)

        if os.path.normcase(self._repo.root) != os.path.normcase(root):

            raise error.ProgrammingError(

                b'failed to reject unsafe subrepo '

                b'path: %s (expanded to %s)' % (root, self._repo.root)