/brz/remove-bazaar

# Copyright (C) 2008 Jelmer Vernooij <jelmer@samba.org>

#

# 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

"""Foreign branch utilities."""

from bzrlib.branch import Branch

from bzrlib.commands import Command, Option

from bzrlib.repository import Repository

from bzrlib.revision import Revision

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

from bzrlib import (

    errors,

    log,

    osutils,

    registry,

    )

""")

class ForeignBranch(Branch):

    """Branch that exists in a foreign version control system."""

    def __init__(self, mapping):

        self.mapping = mapping

        super(ForeignBranch, self).__init__()

    def dpull(self, source, stop_revision=None):

        """Pull deltas from another branch.

        :note: This does not, like pull, retain the revision ids from 

        the source branch and will, rather than adding bzr-specific metadata,

        push only those semantics of the revision that can be natively 

        represented in this branch.

        :param source: Source branch

        :param stop_revision: Revision to pull, defaults to last revision.

        """

        raise NotImplementedError(self.pull)

class ForeignRepository(Repository):

    def has_foreign_revision(self, foreign_revid):

        raise NotImplementedError(self.has_foreign_revision)

    def all_revision_ids(self, mapping=None):

        raise NotImplementedError(self.all_revision_ids)

    def get_mapping(self):

        raise NotImplementedError(self.get_mapping)

    def get_inventory_xml(self, revision_id):

        """See Repository.get_inventory_xml()."""

        return self.serialise_inventory(self.get_inventory(revision_id))

    def get_inventory_sha1(self, revision_id):

        """Get the sha1 for the XML representation of an inventory.

        :param revision_id: Revision id of the inventory for which to return 

         the SHA1.

        :return: XML string

        """

        return osutils.sha_string(self.get_inventory_xml(revision_id))

    def get_revision_xml(self, revision_id):

        """Return the XML representation of a revision.

        :param revision_id: Revision for which to return the XML.

        :return: XML string

        """

        return self._serializer.write_revision_to_string(self.get_revision(revision_id))

class FakeControlFiles(object):

    """Dummy implementation of ControlFiles.

    This is required as some code relies on controlfiles being 

    available."""

    def get_utf8(self, name):

        raise errors.NoSuchFile(name)

    def get(self, name):

        raise errors.NoSuchFile(name)

    def break_lock(self):

        pass

class cmd_dpush(Command):

    """Push diffs into a foreign version control system without any 

    Bazaar-specific metadata.

    This will afterwards rebase the local Bazaar branch on the remote

    branch unless the --no-rebase option is used, in which case 

    the two branches will be out of sync. 

    """

    takes_args = ['location?']

    takes_options = ['remember', Option('directory',

            help='Branch to push from, '

                 'rather than the one containing the working directory.',

            short_name='d',

            type=unicode,

            ),

            Option('no-rebase', help="Don't rebase after push")]

    def run(self, location=None, remember=False, directory=None, 

            no_rebase=False):

        from bzrlib import urlutils

        from bzrlib.bzrdir import BzrDir

        from bzrlib.errors import BzrCommandError, NoWorkingTree

        from bzrlib.trace import info

        from bzrlib.workingtree import WorkingTree

        if directory is None:

            directory = "."

        try:

            source_wt = WorkingTree.open_containing(directory)[0]

            source_branch = source_wt.branch

        except NoWorkingTree:

            source_branch = Branch.open_containing(directory)[0]

            source_wt = None

        stored_loc = source_branch.get_push_location()

        if location is None:

            if stored_loc is None:

                raise BzrCommandError("No push location known or specified.")

            else:

                display_url = urlutils.unescape_for_display(stored_loc,

                        self.outf.encoding)

                self.outf.write("Using saved location: %s\n" % display_url)

                location = stored_loc

        bzrdir = BzrDir.open(location)

        target_branch = bzrdir.open_branch()

        target_branch.lock_write()

        if not isinstance(target_branch, ForeignBranch):

            info("target branch is not a foreign branch, using regular push.")

            target_branch.pull(source_branch)

            no_rebase = True

        else:

            revid_map = target_branch.dpull(source_branch)

        # We successfully created the target, remember it

        if source_branch.get_push_location() is None or remember:

            source_branch.set_push_location(target_branch.base)

        if not no_rebase:

            _, old_last_revid = source_branch.last_revision_info()

            new_last_revid = revid_map[old_last_revid]

            if source_wt is not None:

                source_wt.pull(target_branch, overwrite=True, 

                               stop_revision=new_last_revid)

            else:

                source_branch.pull(target_branch, overwrite=True, 

                                   stop_revision=new_last_revid)

def test_suite():

    from unittest import TestSuite

    from bzrlib.tests import TestUtil

    loader = TestUtil.TestLoader()

    suite = TestSuite()

    testmod_names = ['test_versionedfiles', ]

    suite.addTest(loader.loadTestsFromModuleNames(testmod_names))

    return suite

def escape_commit_message(message):

    """Replace xml-incompatible control characters."""

    if message is None:

        return None

    import re

    # FIXME: RBC 20060419 this should be done by the revision

    # serialiser not by commit. Then we can also add an unescaper

    # in the deserializer and start roundtripping revision messages

    # precisely. See repository_implementations/test_repository.py

    # Python strings can include characters that can't be

    # represented in well-formed XML; escape characters that

    # aren't listed in the XML specification

    # (http://www.w3.org/TR/REC-xml/#NT-Char).

    message, _ = re.subn(

        u'[^\x09\x0A\x0D\u0020-\uD7FF\uE000-\uFFFD]+',

        lambda match: match.group(0).encode('unicode_escape'),

        message)

    return message