/brz/remove-bazaar

# Copyright (C) 2005-2011 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

from __future__ import absolute_import

import warnings

from .lazy_import import lazy_import

lazy_import(globals(), """

from breezy import (

    branch as _mod_branch,

    cleanup,

    conflicts as _mod_conflicts,

    debug,

    generate_ids,

    graph as _mod_graph,

    merge3,

    osutils,

    patiencediff,

    revision as _mod_revision,

    textfile,

    trace,

    transform,

    tree as _mod_tree,

    tsort,

    ui,

    workingtree,

    )

from breezy.bzr import (

    versionedfile,

    )

from breezy.i18n import gettext

""")

from . import (

    decorators,

    errors,

    hooks,

    registry,

    )

from .sixish import (

    text_type,

    viewitems,

    )

# TODO: Report back as changes are merged in

def transform_tree(from_tree, to_tree, interesting_files=None):

    from_tree.lock_tree_write()

    operation = cleanup.OperationWithCleanups(merge_inner)

    operation.add_cleanup(from_tree.unlock)

    operation.run_simple(from_tree.branch, to_tree, from_tree,

        ignore_zero=True, this_tree=from_tree,

        interesting_files=interesting_files)

class MergeHooks(hooks.Hooks):

    def __init__(self):

        hooks.Hooks.__init__(self, "breezy.merge", "Merger.hooks")

        self.add_hook('merge_file_content',

            "Called with a breezy.merge.Merger object to create a per file "

            "merge object when starting a merge. "

            "Should return either None or a subclass of "

            "``breezy.merge.AbstractPerFileMerger``. "

            "Such objects will then be called per file "

            "that needs to be merged (including when one "

            "side has deleted the file and the other has changed it). "

            "See the AbstractPerFileMerger API docs for details on how it is "

            "used by merge.",

            (2, 1))

        self.add_hook('pre_merge',

            'Called before a merge. '

            'Receives a Merger object as the single argument.',

            (2, 5))

        self.add_hook('post_merge',

            'Called after a merge. '

            'Receives a Merger object as the single argument. '

            'The return value is ignored.',

            (2, 5))

class AbstractPerFileMerger(object):

    """PerFileMerger objects are used by plugins extending merge for breezy.

    See ``breezy.plugins.news_merge.news_merge`` for an example concrete class.

    :ivar merger: The Merge3Merger performing the merge.

    """

    def __init__(self, merger):

        """Create a PerFileMerger for use with merger."""

        self.merger = merger

    def merge_contents(self, merge_params):

        """Attempt to merge the contents of a single file.

        :param merge_params: A breezy.merge.MergeFileHookParams

        :return: A tuple of (status, chunks), where status is one of

            'not_applicable', 'success', 'conflicted', or 'delete'.  If status

            is 'success' or 'conflicted', then chunks should be an iterable of

            strings for the new file contents.

        """

        return ('not applicable', None)

class PerFileMerger(AbstractPerFileMerger):

    """Merge individual files when self.file_matches returns True.

    This class is intended to be subclassed.  The file_matches and

    merge_matching methods should be overridden with concrete implementations.

    """

    def file_matches(self, params):

        """Return True if merge_matching should be called on this file.

        Only called with merges of plain files with no clear winner.

        Subclasses must override this.

        """

        raise NotImplementedError(self.file_matches)

    def merge_contents(self, params):

        """Merge the contents of a single file."""

        # Check whether this custom merge logic should be used.

        if (

            # OTHER is a straight winner, rely on default merge.

            params.winner == 'other' or

            # THIS and OTHER aren't both files.

            not params.is_file_merge() or

            # The filename doesn't match

            not self.file_matches(params)):

            return 'not_applicable', None

        return self.merge_matching(params)

    def merge_matching(self, params):

        """Merge the contents of a single file that has matched the criteria

        in PerFileMerger.merge_contents (is a conflict, is a file,

        self.file_matches is True).

        Subclasses must override this.

        """

        raise NotImplementedError(self.merge_matching)

class ConfigurableFileMerger(PerFileMerger):

    """Merge individual files when configured via a .conf file.

    This is a base class for concrete custom file merging logic. Concrete

    classes should implement ``merge_text``.

    See ``breezy.plugins.news_merge.news_merge`` for an example concrete class.

    :ivar affected_files: The configured file paths to merge.

    :cvar name_prefix: The prefix to use when looking up configuration

        details. <name_prefix>_merge_files describes the files targeted by the

        hook for example.

    :cvar default_files: The default file paths to merge when no configuration

        is present.

    """

    name_prefix = None

    default_files = None

    def __init__(self, merger):

        super(ConfigurableFileMerger, self).__init__(merger)

        self.affected_files = None

        self.default_files = self.__class__.default_files or []

        self.name_prefix = self.__class__.name_prefix

        if self.name_prefix is None:

            raise ValueError("name_prefix must be set.")

    def file_matches(self, params):

        """Check whether the file should call the merge hook.

        <name_prefix>_merge_files configuration variable is a list of files

        that should use the hook.

        """

        affected_files = self.affected_files

        if affected_files is None:

            config = self.merger.this_branch.get_config()

            # Until bzr provides a better policy for caching the config, we

            # just add the part we're interested in to the params to avoid

            # reading the config files repeatedly (breezy.conf, location.conf,

            # branch.conf).

            config_key = self.name_prefix + '_merge_files'

            affected_files = config.get_user_option_as_list(config_key)

            if affected_files is None:

                # If nothing was specified in the config, use the default.

                affected_files = self.default_files

            self.affected_files = affected_files

        if affected_files:

            filepath = params.this_path

            if filepath in affected_files:

                return True

        return False

    def merge_matching(self, params):

        return self.merge_text(params)

    def merge_text(self, params):

        """Merge the byte contents of a single file.

        This is called after checking that the merge should be performed in

        merge_contents, and it should behave as per

        ``breezy.merge.AbstractPerFileMerger.merge_contents``.

        """

        raise NotImplementedError(self.merge_text)

class MergeFileHookParams(object):

    """Object holding parameters passed to merge_file_content hooks.

    There are some fields hooks can access:

    :ivar file_id: the file ID of the file being merged

    :ivar base_path: Path in base tree

    :ivar other_path: Path in other tree

    :ivar this_path: Path in this tree

    :ivar trans_id: the transform ID for the merge of this file

    :ivar this_kind: kind of file_id in 'this' tree

    :ivar other_kind: kind of file_id in 'other' tree

    :ivar winner: one of 'this', 'other', 'conflict'

    """

    def __init__(self, merger, file_id, paths, trans_id, this_kind, other_kind,

            winner):

        self._merger = merger

        self.file_id = file_id

        self.paths = paths

        self.base_path, self.other_path, self.this_path = paths

        self.trans_id = trans_id

        self.this_kind = this_kind

        self.other_kind = other_kind

        self.winner = winner

    def is_file_merge(self):

        """True if this_kind and other_kind are both 'file'."""

        return self.this_kind == 'file' and self.other_kind == 'file'

    @decorators.cachedproperty

    def base_lines(self):

        """The lines of the 'base' version of the file."""

        return self._merger.get_lines(self._merger.base_tree, self.base_path, self.file_id)

    @decorators.cachedproperty

    def this_lines(self):

        """The lines of the 'this' version of the file."""

        return self._merger.get_lines(self._merger.this_tree, self.this_path, self.file_id)

    @decorators.cachedproperty

    def other_lines(self):

        """The lines of the 'other' version of the file."""

        return self._merger.get_lines(self._merger.other_tree, self.other_path, self.file_id)

class Merger(object):

    hooks = MergeHooks()

    def __init__(self, this_branch, other_tree=None, base_tree=None,

                 this_tree=None, change_reporter=None,

                 recurse='down', revision_graph=None):

        object.__init__(self)

        self.this_branch = this_branch

        self.this_basis = _mod_revision.ensure_null(

            this_branch.last_revision())

        self.this_rev_id = None

        self.this_tree = this_tree

        self.this_revision_tree = None

        self.this_basis_tree = None

        self.other_tree = other_tree

        self.other_branch = None

        self.base_tree = base_tree

        self.ignore_zero = False

        self.backup_files = False

        self.interesting_files = None

        self.show_base = False

        self.reprocess = False

        self.pp = None

        self.recurse = recurse

        self.change_reporter = change_reporter

        self._cached_trees = {}

        self._revision_graph = revision_graph

        self._base_is_ancestor = None

        self._base_is_other_ancestor = None

        self._is_criss_cross = None

        self._lca_trees = None

    def cache_trees_with_revision_ids(self, trees):

        """Cache any tree in trees if it has a revision_id."""

        for maybe_tree in trees:

            if maybe_tree is None:

                continue

            try:

                rev_id = maybe_tree.get_revision_id()

            except AttributeError:

                continue

            self._cached_trees[rev_id] = maybe_tree

    @property

    def revision_graph(self):

        if self._revision_graph is None:

            self._revision_graph = self.this_branch.repository.get_graph()

        return self._revision_graph

    def _set_base_is_ancestor(self, value):

        self._base_is_ancestor = value

    def _get_base_is_ancestor(self):

        if self._base_is_ancestor is None:

            self._base_is_ancestor = self.revision_graph.is_ancestor(

                self.base_rev_id, self.this_basis)

        return self._base_is_ancestor

    base_is_ancestor = property(_get_base_is_ancestor, _set_base_is_ancestor)

    def _set_base_is_other_ancestor(self, value):

        self._base_is_other_ancestor = value

    def _get_base_is_other_ancestor(self):

        if self._base_is_other_ancestor is None:

            if self.other_basis is None:

                return True

            self._base_is_other_ancestor = self.revision_graph.is_ancestor(

                self.base_rev_id, self.other_basis)

        return self._base_is_other_ancestor

    base_is_other_ancestor = property(_get_base_is_other_ancestor,

                                      _set_base_is_other_ancestor)

    @staticmethod

    def from_uncommitted(tree, other_tree, base_tree=None):

        """Return a Merger for uncommitted changes in other_tree.

        :param tree: The tree to merge into

        :param other_tree: The tree to get uncommitted changes from

        :param base_tree: The basis to use for the merge.  If unspecified,

            other_tree.basis_tree() will be used.

        """

        if base_tree is None:

            base_tree = other_tree.basis_tree()

        merger = Merger(tree.branch, other_tree, base_tree, tree)

        merger.base_rev_id = merger.base_tree.get_revision_id()

        merger.other_rev_id = None

        merger.other_basis = merger.base_rev_id

        return merger

    @classmethod

    def from_mergeable(klass, tree, mergeable):

        """Return a Merger for a bundle or merge directive.

        :param tree: The tree to merge changes into

        :param mergeable: A merge directive or bundle

        """

        mergeable.install_revisions(tree.branch.repository)

        base_revision_id, other_revision_id, verified =\

            mergeable.get_merge_request(tree.branch.repository)

        revision_graph = tree.branch.repository.get_graph()

        if base_revision_id is not None:

            if (base_revision_id != _mod_revision.NULL_REVISION and

                revision_graph.is_ancestor(

                base_revision_id, tree.branch.last_revision())):

                base_revision_id = None

            else:

                trace.warning('Performing cherrypick')

        merger = klass.from_revision_ids(tree, other_revision_id,

                                         base_revision_id, revision_graph=

                                         revision_graph)

        return merger, verified

    @staticmethod

    def from_revision_ids(tree, other, base=None, other_branch=None,

                          base_branch=None, revision_graph=None,

                          tree_branch=None):

        """Return a Merger for revision-ids.

        :param tree: The tree to merge changes into

        :param other: The revision-id to use as OTHER

        :param base: The revision-id to use as BASE.  If not specified, will

            be auto-selected.

        :param other_branch: A branch containing the other revision-id.  If

            not supplied, tree.branch is used.

        :param base_branch: A branch containing the base revision-id.  If

            not supplied, other_branch or tree.branch will be used.

        :param revision_graph: If you have a revision_graph precomputed, pass

            it in, otherwise it will be created for you.

        :param tree_branch: The branch associated with tree.  If not supplied,

            tree.branch will be used.

        """

        if tree_branch is None:

            tree_branch = tree.branch

        merger = Merger(tree_branch, this_tree=tree,

                        revision_graph=revision_graph)

        if other_branch is None:

            other_branch = tree.branch

        merger.set_other_revision(other, other_branch)

        if base is None:

            merger.find_base()

        else:

            if base_branch is None:

                base_branch = other_branch

            merger.set_base_revision(base, base_branch)

        return merger

    def revision_tree(self, revision_id, branch=None):

        if revision_id not in self._cached_trees:

            if branch is None:

                branch = self.this_branch

            try:

                tree = self.this_tree.revision_tree(revision_id)

            except errors.NoSuchRevisionInTree:

                tree = branch.repository.revision_tree(revision_id)

            self._cached_trees[revision_id] = tree

        return self._cached_trees[revision_id]

    def _get_tree(self, treespec, possible_transports=None):

        location, revno = treespec

        if revno is None:

            tree = workingtree.WorkingTree.open_containing(location)[0]

            return tree.branch, tree

        branch = _mod_branch.Branch.open_containing(

            location, possible_transports)[0]

        if revno == -1:

            revision_id = branch.last_revision()

        else:

            revision_id = branch.get_rev_id(revno)

        revision_id = _mod_revision.ensure_null(revision_id)

        return branch, self.revision_tree(revision_id, branch)

    def set_interesting_files(self, file_list):

        self.interesting_files = file_list

    def set_pending(self):

        if (not self.base_is_ancestor or not self.base_is_other_ancestor

            or self.other_rev_id is None):

            return

        self._add_parent()

    def _add_parent(self):

        new_parents = self.this_tree.get_parent_ids() + [self.other_rev_id]

        new_parent_trees = []

        operation = cleanup.OperationWithCleanups(

            self.this_tree.set_parent_trees)

        for revision_id in new_parents:

            try:

                tree = self.revision_tree(revision_id)

            except errors.NoSuchRevision:

                tree = None

            else:

                tree.lock_read()

                operation.add_cleanup(tree.unlock)

            new_parent_trees.append((revision_id, tree))

        operation.run_simple(new_parent_trees, allow_leftmost_as_ghost=True)

    def set_other(self, other_revision, possible_transports=None):

        """Set the revision and tree to merge from.

        This sets the other_tree, other_rev_id, other_basis attributes.

        :param other_revision: The [path, revision] list to merge from.

        """

        self.other_branch, self.other_tree = self._get_tree(other_revision,

                                                            possible_transports)

        if other_revision[1] == -1:

            self.other_rev_id = _mod_revision.ensure_null(

                self.other_branch.last_revision())

            if _mod_revision.is_null(self.other_rev_id):

                raise errors.NoCommits(self.other_branch)

            self.other_basis = self.other_rev_id

        elif other_revision[1] is not None:

            self.other_rev_id = self.other_branch.get_rev_id(other_revision[1])

            self.other_basis = self.other_rev_id

        else:

            self.other_rev_id = None

            self.other_basis = self.other_branch.last_revision()

            if self.other_basis is None:

                raise errors.NoCommits(self.other_branch)

        if self.other_rev_id is not None:

            self._cached_trees[self.other_rev_id] = self.other_tree

        self._maybe_fetch(self.other_branch, self.this_branch, self.other_basis)

    def set_other_revision(self, revision_id, other_branch):

        """Set 'other' based on a branch and revision id

        :param revision_id: The revision to use for a tree

        :param other_branch: The branch containing this tree

        """

        self.other_rev_id = revision_id

        self.other_branch = other_branch

        self._maybe_fetch(other_branch, self.this_branch, self.other_rev_id)

        self.other_tree = self.revision_tree(revision_id)

        self.other_basis = revision_id

    def set_base_revision(self, revision_id, branch):

        """Set 'base' based on a branch and revision id

        :param revision_id: The revision to use for a tree

        :param branch: The branch containing this tree

        """

        self.base_rev_id = revision_id

        self.base_branch = branch

        self._maybe_fetch(branch, self.this_branch, revision_id)

        self.base_tree = self.revision_tree(revision_id)

    def _maybe_fetch(self, source, target, revision_id):

        if not source.repository.has_same_location(target.repository):

            target.fetch(source, revision_id)

    def find_base(self):

        revisions = [_mod_revision.ensure_null(self.this_basis),

                     _mod_revision.ensure_null(self.other_basis)]

        if _mod_revision.NULL_REVISION in revisions:

            self.base_rev_id = _mod_revision.NULL_REVISION

            self.base_tree = self.revision_tree(self.base_rev_id)

            self._is_criss_cross = False

        else:

            lcas = self.revision_graph.find_lca(revisions[0], revisions[1])

            self._is_criss_cross = False

            if len(lcas) == 0:

                self.base_rev_id = _mod_revision.NULL_REVISION

            elif len(lcas) == 1:

                self.base_rev_id = list(lcas)[0]

            else: # len(lcas) > 1

                self._is_criss_cross = True

                if len(lcas) > 2:

                    # find_unique_lca can only handle 2 nodes, so we have to

                    # start back at the beginning. It is a shame to traverse

                    # the graph again, but better than re-implementing

                    # find_unique_lca.

                    self.base_rev_id = self.revision_graph.find_unique_lca(

                                            revisions[0], revisions[1])

                else:

                    self.base_rev_id = self.revision_graph.find_unique_lca(

                                            *lcas)

                sorted_lca_keys = self.revision_graph.find_merge_order(

                    revisions[0], lcas)

                if self.base_rev_id == _mod_revision.NULL_REVISION:

                    self.base_rev_id = sorted_lca_keys[0]

            if self.base_rev_id == _mod_revision.NULL_REVISION:

                raise errors.UnrelatedBranches()

            if self._is_criss_cross:

                trace.warning('Warning: criss-cross merge encountered.  See bzr'

                              ' help criss-cross.')

                trace.mutter('Criss-cross lcas: %r' % lcas)

                if self.base_rev_id in lcas:

                    trace.mutter('Unable to find unique lca. '

                                 'Fallback %r as best option.'

                                 % self.base_rev_id)

                interesting_revision_ids = set(lcas)

                interesting_revision_ids.add(self.base_rev_id)

                interesting_trees = dict((t.get_revision_id(), t)

                    for t in self.this_branch.repository.revision_trees(

                        interesting_revision_ids))

                self._cached_trees.update(interesting_trees)

                if self.base_rev_id in lcas:

                    self.base_tree = interesting_trees[self.base_rev_id]

                else:

                    self.base_tree = interesting_trees.pop(self.base_rev_id)

                self._lca_trees = [interesting_trees[key]

                                   for key in sorted_lca_keys]

            else:

                self.base_tree = self.revision_tree(self.base_rev_id)

        self.base_is_ancestor = True

        self.base_is_other_ancestor = True

        trace.mutter('Base revid: %r' % self.base_rev_id)

    def set_base(self, base_revision):

        """Set the base revision to use for the merge.

        :param base_revision: A 2-list containing a path and revision number.

        """

        trace.mutter("doing merge() with no base_revision specified")

        if base_revision == [None, None]:

            self.find_base()

        else:

            base_branch, self.base_tree = self._get_tree(base_revision)

            if base_revision[1] == -1:

                self.base_rev_id = base_branch.last_revision()

            elif base_revision[1] is None:

                self.base_rev_id = _mod_revision.NULL_REVISION

            else:

                self.base_rev_id = _mod_revision.ensure_null(

                    base_branch.get_rev_id(base_revision[1]))

            self._maybe_fetch(base_branch, self.this_branch, self.base_rev_id)

    def make_merger(self):

        kwargs = {'working_tree': self.this_tree, 'this_tree': self.this_tree,

                  'other_tree': self.other_tree,

                  'interesting_files': self.interesting_files,

                  'this_branch': self.this_branch,

                  'other_branch': self.other_branch,

                  'do_merge': False}

        if self.merge_type.requires_base:

            kwargs['base_tree'] = self.base_tree

        if self.merge_type.supports_reprocess:

            kwargs['reprocess'] = self.reprocess

        elif self.reprocess:

            raise errors.BzrError(

                "Conflict reduction is not supported for merge"

                " type %s." % self.merge_type)

        if self.merge_type.supports_show_base:

            kwargs['show_base'] = self.show_base

        elif self.show_base:

            raise errors.BzrError("Showing base is not supported for this"

                                  " merge type. %s" % self.merge_type)

        if (not getattr(self.merge_type, 'supports_reverse_cherrypick', True)

            and not self.base_is_other_ancestor):

            raise errors.CannotReverseCherrypick()

        if self.merge_type.supports_cherrypick: