# exchange.py - utility to exchange data between repos.

#

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

import weakref

from .i18n import _

from .node import (

    hex,

    nullrev,

)

from . import (

    bookmarks as bookmod,

    bundle2,

    bundlecaches,

    changegroup,

    discovery,

    error,

    lock as lockmod,

    logexchange,

    narrowspec,

    obsolete,

    obsutil,

    phases,

    pushkey,

    pycompat,

    requirements,

    scmutil,

    streamclone,

    url as urlmod,

    util,

    wireprototypes,

)

from .utils import (

    hashutil,

    stringutil,

    urlutil,

)

from .interfaces import repository

urlerr = util.urlerr

urlreq = util.urlreq

_NARROWACL_SECTION = b'narrowacl'

def readbundle(ui, fh, fname, vfs=None):

    header = changegroup.readexactly(fh, 4)

    alg = None

    if not fname:

        fname = b"stream"

        if not header.startswith(b'HG') and header.startswith(b'\0'):

            fh = changegroup.headerlessfixup(fh, header)

            header = b"HG10"

            alg = b'UN'

    elif vfs:

        fname = vfs.join(fname)

    magic, version = header[0:2], header[2:4]

    if magic != b'HG':

        raise error.Abort(_(b'%s: not a Mercurial bundle') % fname)

    if version == b'10':

        if alg is None:

            alg = changegroup.readexactly(fh, 2)

        return changegroup.cg1unpacker(fh, alg)

    elif version.startswith(b'2'):

        return bundle2.getunbundler(ui, fh, magicstring=magic + version)

    elif version == b'S1':

        return streamclone.streamcloneapplier(fh)

    else:

        raise error.Abort(

            _(b'%s: unknown bundle version %s') % (fname, version)

        )

def _format_params(params):

    parts = []

    for key, value in sorted(params.items()):

        value = urlreq.quote(value)

        parts.append(b"%s=%s" % (key, value))

    return b';'.join(parts)

def getbundlespec(ui, fh):

    """Infer the bundlespec from a bundle file handle.

    The input file handle is seeked and the original seek position is not

    restored.

    """

    def speccompression(alg):

        try:

            return util.compengines.forbundletype(alg).bundletype()[0]

        except KeyError:

            return None

    params = {}

    b = readbundle(ui, fh, None)

    if isinstance(b, changegroup.cg1unpacker):

        alg = b._type

        if alg == b'_truncatedBZ':

            alg = b'BZ'

        comp = speccompression(alg)

        if not comp:

            raise error.Abort(_(b'unknown compression algorithm: %s') % alg)

        return b'%s-v1' % comp

    elif isinstance(b, bundle2.unbundle20):

        if b'Compression' in b.params:

            comp = speccompression(b.params[b'Compression'])

            if not comp:

                raise error.Abort(

                    _(b'unknown compression algorithm: %s') % comp

                )

        else:

            comp = b'none'

        version = None

        for part in b.iterparts():

            if part.type == b'changegroup':

                cgversion = part.params[b'version']

                if cgversion in (b'01', b'02'):

                    version = b'v2'

                elif cgversion in (b'03',):

                    version = b'v2'

                    params[b'cg.version'] = cgversion

                else:

                    raise error.Abort(

                        _(

                            b'changegroup version %s does not have '

                            b'a known bundlespec'

                        )

                        % version,

                        hint=_(b'try upgrading your Mercurial client'),

                    )

            elif part.type == b'stream2' and version is None:

                # A stream2 part requires to be part of a v2 bundle

                requirements = urlreq.unquote(part.params[b'requirements'])

                splitted = requirements.split()

                params = bundle2._formatrequirementsparams(splitted)

                return b'none-v2;stream=v2;%s' % params

            elif part.type == b'stream3-exp' and version is None:

                # A stream3 part requires to be part of a v2 bundle

                requirements = urlreq.unquote(part.params[b'requirements'])

                splitted = requirements.split()

                params = bundle2._formatrequirementsparams(splitted)

                return b'none-v2;stream=v3-exp;%s' % params

            elif part.type == b'obsmarkers':

                params[b'obsolescence'] = b'yes'

                if not part.mandatory:

                    params[b'obsolescence-mandatory'] = b'no'

        if not version:

            raise error.Abort(

                _(b'could not identify changegroup version in bundle')

            )

        spec = b'%s-%s' % (comp, version)

        if params:

            spec += b';'

            spec += _format_params(params)

        return spec

    elif isinstance(b, streamclone.streamcloneapplier):

        requirements = streamclone.readbundle1header(fh)[2]

        formatted = bundle2._formatrequirementsparams(requirements)

        return b'none-packed1;%s' % formatted

    else:

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

def _computeoutgoing(repo, heads, common):

    """Computes which revs are outgoing given a set of common

    and a set of heads.

    This is a separate function so extensions can have access to

    the logic.

    Returns a discovery.outgoing object.

    """

    cl = repo.changelog

    if common:

        hasnode = cl.hasnode

        common = [n for n in common if hasnode(n)]

    else:

        common = [repo.nullid]

    if not heads:

        heads = cl.heads()

    return discovery.outgoing(repo, common, heads)

def _checkpublish(pushop):

    repo = pushop.repo

    ui = repo.ui

    behavior = ui.config(b'experimental', b'auto-publish')

    if pushop.publish or behavior not in (b'warn', b'confirm', b'abort'):

        return

    remotephases = listkeys(pushop.remote, b'phases')

    if not remotephases.get(b'publishing', False):

        return

    if pushop.revs is None:

        published = repo.filtered(b'served').revs(b'not public()')

    else:

        published = repo.revs(b'::%ln - public()', pushop.revs)

        # we want to use pushop.revs in the revset even if they themselves are

        # secret, but we don't want to have anything that the server won't see

        # in the result of this expression

        published &= repo.filtered(b'served')

    if published:

        if behavior == b'warn':

            ui.warn(

                _(b'%i changesets about to be published\n') % len(published)

            )

        elif behavior == b'confirm':

            if ui.promptchoice(

                _(b'push and publish %i changesets (yn)?$$ &Yes $$ &No')

                % len(published)

            ):

                raise error.CanceledError(_(b'user quit'))

        elif behavior == b'abort':

            msg = _(b'push would publish %i changesets') % len(published)

            hint = _(

                b"use --publish or adjust 'experimental.auto-publish'"

                b" config"

            )

            raise error.Abort(msg, hint=hint)

def _forcebundle1(op):

    """return true if a pull/push must use bundle1

    This function is used to allow testing of the older bundle version"""

    ui = op.repo.ui

    # The goal is this config is to allow developer to choose the bundle

    # version used during exchanged. This is especially handy during test.

    # Value is a list of bundle version to be picked from, highest version

    # should be used.

    #

    # developer config: devel.legacy.exchange

    exchange = ui.configlist(b'devel', b'legacy.exchange')

    forcebundle1 = b'bundle2' not in exchange and b'bundle1' in exchange

    return forcebundle1 or not op.remote.capable(b'bundle2')

class pushoperation:

    """A object that represent a single push operation

    Its purpose is to carry push related state and very common operations.

    A new pushoperation should be created at the beginning of each push and

    discarded afterward.

    """

    def __init__(

        self,

        repo,

        remote,

        force=False,

        revs=None,

        newbranch=False,

        bookmarks=(),

        publish=False,

        pushvars=None,

    ):

        # repo we push from

        self.repo = repo

        self.ui = repo.ui

        # repo we push to

        self.remote = remote

        # force option provided

        self.force = force

        # revs to be pushed (None is "all")

        self.revs = revs

        # bookmark explicitly pushed

        self.bookmarks = bookmarks

        # allow push of new branch

        self.newbranch = newbranch

        # step already performed

        # (used to check what steps have been already performed through bundle2)

        self.stepsdone = set()

        # Integer version of the changegroup push result

        # - None means nothing to push

        # - 0 means HTTP error

        # - 1 means we pushed and remote head count is unchanged *or*

        #   we have outgoing changesets but refused to push

        # - other values as described by addchangegroup()

        self.cgresult = None

        # Boolean value for the bookmark push

        self.bkresult = None

        # discover.outgoing object (contains common and outgoing data)

        self.outgoing = None

        # all remote topological heads before the push

        self.remoteheads = None

        # Details of the remote branch pre and post push

        #

        # mapping: {'branch': ([remoteheads],

        #                      [newheads],

        #                      [unsyncedheads],

        #                      [discardedheads])}

        # - branch: the branch name

        # - remoteheads: the list of remote heads known locally

        #                None if the branch is new

        # - newheads: the new remote heads (known locally) with outgoing pushed

        # - unsyncedheads: the list of remote heads unknown locally.

        # - discardedheads: the list of remote heads made obsolete by the push

        self.pushbranchmap = None

        # testable as a boolean indicating if any nodes are missing locally.

        self.incoming = None

        # summary of the remote phase situation

        self.remotephases = None

        # phases changes that must be pushed along side the changesets

        self.outdatedphases = None

        # phases changes that must be pushed if changeset push fails

        self.fallbackoutdatedphases = None

        # outgoing obsmarkers

        self.outobsmarkers = set()

        # outgoing bookmarks, list of (bm, oldnode | '', newnode | '')

        self.outbookmarks = []

        # transaction manager

        self.trmanager = None

        # map { pushkey partid -> callback handling failure}

        # used to handle exception from mandatory pushkey part failure

        self.pkfailcb = {}

        # an iterable of pushvars or None

        self.pushvars = pushvars

        # publish pushed changesets

        self.publish = publish

    @util.propertycache

    def futureheads(self):

        """future remote heads if the changeset push succeeds"""

        return self.outgoing.ancestorsof

    @util.propertycache

    def fallbackheads(self):

        """future remote heads if the changeset push fails"""

        if self.revs is None:

            # not target to push, all common are relevant

            return self.outgoing.commonheads

        unfi = self.repo.unfiltered()

        # I want cheads = heads(::push_heads and ::commonheads)

        #

        # To push, we already computed

        #     common = (::commonheads)

        #     missing = ((commonheads::push_heads) - commonheads)

        #

        # So we basically search

        #

        #     almost_heads = heads((parents(missing) + push_heads) & common)

        #

        # We use "almost" here as this can return revision that are ancestors

        # of other in the set and we need to explicitly turn it into an

        # antichain later. We can do so using:

        #

        #     cheads = heads(almost_heads::almost_heads)

        #

        # In pratice the code is a bit more convulted to avoid some extra

        # computation. It aims at doing the same computation as highlighted

        # above however.

        common = self.outgoing.common

        unfi = self.repo.unfiltered()

        cl = unfi.changelog

        to_rev = cl.index.rev

        to_node = cl.node

        parent_revs = cl.parentrevs

        unselected = []

        cheads = set()

        # XXX-perf: `self.revs` and `outgoing.missing` could hold revs directly

        for n in self.revs:

            r = to_rev(n)

            if r in common:

                cheads.add(r)

            else:

                unselected.append(r)

        known_non_heads = cl.ancestors(cheads, inclusive=True)

        if unselected:

            missing_revs = {to_rev(n) for n in self.outgoing.missing}

            missing_revs.add(nullrev)

            root_points = set()

            for r in missing_revs: