/brz/remove-bazaar

# Copyright (C) 2006-2007 by Jelmer Vernooij

#

# 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

"""Tests for the rebase code."""

from ....conflicts import ConflictList

from ....errors import (

    UnknownFormatError,

    NoSuchFile,

    ConflictsInTree,

    )

from ....graph import (

    Graph,

    DictParentsProvider,

    )

from ....revision import NULL_REVISION

from ....tests import TestCase, TestCaseWithTransport

from ....tests.matchers import RevisionHistoryMatches

from ..rebase import (

    marshall_rebase_plan,

    unmarshall_rebase_plan,

    CommitBuilderRevisionRewriter,

    generate_simple_plan,

    generate_transpose_plan,

    rebase_todo,

    REBASE_PLAN_FILENAME,

    REBASE_CURRENT_REVID_FILENAME,

    RebaseState1,

    ReplaySnapshotError,

    WorkingTreeRevisionRewriter,

    )

class RebasePlanReadWriterTests(TestCase):

    def test_simple_marshall_rebase_plan(self):

        self.assertEqualDiff(b"""\

# Bazaar rebase plan 1

1 bla

oldrev newrev newparent1 newparent2

""", marshall_rebase_plan((1, b"bla"),

                          {b"oldrev": (b"newrev", (b"newparent1", b"newparent2"))}))

    def test_simple_unmarshall_rebase_plan(self):

        self.assertEquals(((1, b"bla"),

                          {b"oldrev": (b"newrev", (b"newparent1", b"newparent2"))}),

                          unmarshall_rebase_plan(b"""# Bazaar rebase plan 1

1 bla

oldrev newrev newparent1 newparent2

"""))

    def test_unmarshall_rebase_plan_formatunknown(self):

        self.assertRaises(UnknownFormatError,

                          unmarshall_rebase_plan, b"""# Bazaar rebase plan x

1 bla

oldrev newrev newparent1 newparent2

""")

class ConversionTests(TestCaseWithTransport):

    def test_simple(self):

        wt = self.make_branch_and_tree('.')

        b = wt.branch

        with open('hello', 'w') as f:

            f.write('hello world')

        wt.add('hello')

        wt.commit(message='add hello', rev_id=b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bloe")

        wt.set_last_revision(b"bla")

        b.generate_revision_history(b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bla2")

        wt.branch.repository.lock_write()

        newrev = CommitBuilderRevisionRewriter(wt.branch.repository)(

            b"bla2", b"bla4", (b"bloe",))

        self.assertEqual(b"bla4", newrev)

        self.assertTrue(wt.branch.repository.has_revision(newrev))

        self.assertEqual(

            (b"bloe",), wt.branch.repository.get_parent_map([newrev])[newrev])

        self.assertEqual(

            "bla2", wt.branch.repository.get_revision(newrev).properties["rebase-of"])

        wt.branch.repository.unlock()

class PlanCreatorTests(TestCaseWithTransport):

    def test_simple_plan_creator(self):

        wt = self.make_branch_and_tree('.')

        b = wt.branch

        with open('hello', 'w') as f:

            f.write('hello world')

        wt.add('hello')

        wt.commit(message='add hello', rev_id=b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bloe")

        wt.set_last_revision(b"bla")

        b.generate_revision_history(b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bla2")

        b.repository.lock_read()

        graph = b.repository.get_graph()

        self.assertEquals(

            {b'bla2': (b'newbla2', (b"bloe",))},

            generate_simple_plan(

                graph.find_difference(b.last_revision(), b"bla")[0],

                b"bla2", None, b"bloe", graph, lambda y, _: b"new" + y))

        b.repository.unlock()

    def test_simple_plan_creator_extra_history(self):

        wt = self.make_branch_and_tree('.')

        b = wt.branch

        with open('hello', 'w') as f:

            f.write('hello world')

        wt.add('hello')

        wt.commit(message='add hello', rev_id=b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bloe")

        wt.set_last_revision(b"bla")

        b.generate_revision_history(b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bla2")

        with open('hello', 'w') as f:

            f.write('universe')

        wt.commit(message='change hello again', rev_id=b"bla3")

        with b.repository.lock_read():

            graph = b.repository.get_graph()

            self.assertEquals(

                {b'bla2': (b'newbla2', (b"bloe",)), b'bla3': (b'newbla3', (b'newbla2',))},

                generate_simple_plan(

                    graph.find_difference(b.last_revision(), b"bloe")[0],

                    b"bla2", None, b"bloe",

                    graph, lambda y, _: b"new" + y))

    def test_generate_transpose_plan(self):

        wt = self.make_branch_and_tree('.')

        b = wt.branch

        with open('hello', 'w') as f:

            f.write('hello world')

        wt.add('hello')

        wt.commit(message='add hello', rev_id=b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bloe")

        wt.set_last_revision(b"bla")

        b.generate_revision_history(b"bla")

        with open('hello', 'w') as f:

            f.write('world')

        wt.commit(message='change hello', rev_id=b"bla2")

        with open('hello', 'w') as f:

            f.write('universe')

        wt.commit(message='change hello again', rev_id=b"bla3")

        wt.set_last_revision(b"bla")

        b.generate_revision_history(b"bla")

        with open('hello', 'w') as f:

            f.write('somebar')

        wt.commit(message='change hello yet again', rev_id=b"blie")

        wt.set_last_revision(NULL_REVISION)

        b.generate_revision_history(NULL_REVISION)

        wt.add('hello')

        wt.commit(message='add hello', rev_id=b"lala")

        b.repository.lock_read()

        graph = b.repository.get_graph()

        self.assertEquals({

            b'blie': (b'newblie', (b'lala',))},

            generate_transpose_plan(graph.iter_ancestry(

                [b"blie"]),

                {b"bla": b"lala"}, graph, lambda y, _: b"new" + y))

        self.assertEquals({

            b'bla2': (b'newbla2', (b'lala',)),

            b'bla3': (b'newbla3', (b'newbla2',)),

            b'blie': (b'newblie', (b'lala',)),

            b'bloe': (b'newbloe', (b'lala',))},

            generate_transpose_plan(graph.iter_ancestry(

                b.repository._all_revision_ids()),

                {b"bla": b"lala"}, graph, lambda y, _: b"new" + y))

        b.repository.unlock()

    def test_generate_transpose_plan_one(self):

        graph = Graph(DictParentsProvider({"bla": ("bloe",), "bloe": (), "lala": ()}))

        self.assertEquals(

            {"bla": ("newbla", ("lala",))},

            generate_transpose_plan(graph.iter_ancestry(

                ["bla", "bloe"]), {"bloe": "lala"}, graph, lambda y, _: "new" + y))

    def test_plan_with_already_merged(self):

        r"""We need to use a merge base that makes sense.

        A

        | \

        B  D

        | \|

        C  E

        Rebasing E on C should result in:

        A -> B -> C -> D -> E

        with a plan of:

        D -> (D', [C])

        E -> (E', [D', C])

        """

        parents_map = {

            "A": (),

            "B": ("A",),

            "C": ("B",),

            "D": ("A",),

            "E": ("D", "B")

        }

        graph = Graph(DictParentsProvider(parents_map))

        self.assertEquals(

            {"D": ("D'", ("C",)), "E": ("E'", ("D'",))},

            generate_simple_plan(

                ["D", "E"], "D", None, "C", graph, lambda y, _: y + "'"))

    def test_plan_with_already_merged_skip_merges(self):

        r"""We need to use a merge base that makes sense.

        A

        | \

        B  D

        | \|

        C  E

        Rebasing E on C should result in:

        A -> B -> C -> D'

        with a plan of:

        D -> (D', [C])

        """

        parents_map = {

            "A": (),

            "B": ("A",),

            "C": ("B",),

            "D": ("A",),

            "E": ("D", "B")

        }

        graph = Graph(DictParentsProvider(parents_map))

        self.assertEquals(

            {"D": ("D'", ("C",))},

            generate_simple_plan(

                ["D", "E"], "D", None, "C", graph, lambda y, _: y + "'", True))

class RebaseStateTests(TestCaseWithTransport):

    def setUp(self):

        super(RebaseStateTests, self).setUp()

        self.wt = self.make_branch_and_tree('.')

        self.state = RebaseState1(self.wt)

    def test_rebase_plan_exists_false(self):

        self.assertFalse(self.state.has_plan())

    def test_rebase_plan_exists_empty(self):

        self.wt._transport.put_bytes(REBASE_PLAN_FILENAME, b"")

        self.assertFalse(self.state.has_plan())

    def test_rebase_plan_exists(self):

        self.wt._transport.put_bytes(REBASE_PLAN_FILENAME, b"foo")

        self.assertTrue(self.state.has_plan())

    def test_remove_rebase_plan(self):

        self.wt._transport.put_bytes(REBASE_PLAN_FILENAME, b"foo")

        self.state.remove_plan()

        self.assertFalse(self.state.has_plan())

    def test_remove_rebase_plan_twice(self):

        self.state.remove_plan()

        self.assertFalse(self.state.has_plan())

    def test_write_rebase_plan(self):

        with open('hello', 'w') as f:

            f.write('hello world')

        self.wt.add('hello')

        self.wt.commit(message='add hello', rev_id=b"bla")

        self.state.write_plan(

            {b"oldrev": (b"newrev", [b"newparent1", b"newparent2"])})

        self.assertEqualDiff(b"""# Bazaar rebase plan 1

1 bla

oldrev newrev newparent1 newparent2

""", self.wt._transport.get_bytes(REBASE_PLAN_FILENAME))

    def test_read_rebase_plan_nonexistant(self):

        self.assertRaises(NoSuchFile, self.state.read_plan)

    def test_read_rebase_plan_empty(self):

        self.wt._transport.put_bytes(REBASE_PLAN_FILENAME, b"")

        self.assertRaises(NoSuchFile, self.state.read_plan)

    def test_read_rebase_plan(self):

        self.wt._transport.put_bytes(

            REBASE_PLAN_FILENAME,

            b"""# Bazaar rebase plan 1

1 bla

oldrev newrev newparent1 newparent2

""")

        self.assertEquals(

            ((1, b"bla"), {b"oldrev": (b"newrev", (b"newparent1", b"newparent2"))}),

            self.state.read_plan())

    def test_read_nonexistant(self):

        self.assertIs(None, self.state.read_active_revid())

    def test_read_null(self):

        self.wt._transport.put_bytes(REBASE_CURRENT_REVID_FILENAME, NULL_REVISION)

        self.assertIs(None, self.state.read_active_revid())

    def test_read(self):

        self.wt._transport.put_bytes(REBASE_CURRENT_REVID_FILENAME, b"bla")

        self.assertEquals(b"bla", self.state.read_active_revid())

    def test_write(self):

        self.state.write_active_revid(b"bloe")

        self.assertEquals(b"bloe", self.state.read_active_revid())

    def test_write_null(self):

        self.state.write_active_revid(None)

        self.assertIs(None, self.state.read_active_revid())

class RebaseTodoTests(TestCase):

    def test_done(self):

        class Repository:

            def has_revision(self, revid):

                return revid == "bloe"

        self.assertEquals(

            [], list(rebase_todo(Repository(), {"bla": ("bloe", [])})))

    def test_notstarted(self):

        class Repository:

            def has_revision(self, revid):

                return False

        self.assertEquals(["bla"], list(rebase_todo(Repository(), {"bla": ("bloe", [])})))

    def test_halfway(self):

        class Repository:

            def has_revision(self, revid):

                return revid == "bloe"

        self.assertEquals(

            ["ha"],

            list(rebase_todo(Repository(), {"bla": ("bloe", []), "ha": ("hee", [])})))

class ReplaySnapshotTests(TestCaseWithTransport):

    def test_single_revision(self):

        wt = self.make_branch_and_tree(".")

        self.build_tree(['afile'])

        wt.add(["afile"])

        wt.commit("bla", rev_id=b"oldcommit")

        wt.branch.repository.lock_write()

        CommitBuilderRevisionRewriter(wt.branch.repository)(b"oldcommit", b"newcommit", ())

        wt.branch.repository.unlock()

        oldrev = wt.branch.repository.get_revision(b"oldcommit")

        newrev = wt.branch.repository.get_revision(b"newcommit")

        self.assertEquals([], newrev.parent_ids)

        self.assertEquals(b"newcommit", newrev.revision_id)

        self.assertEquals(oldrev.committer, newrev.committer)

        self.assertEquals(oldrev.timestamp, newrev.timestamp)

        self.assertEquals(oldrev.timezone, newrev.timezone)

        tree = wt.branch.repository.revision_tree(b"newcommit")

        self.assertEquals(b"newcommit", tree.get_file_revision("afile"))

    def test_two_revisions(self):

        wt = self.make_branch_and_tree("old")

        self.build_tree_contents([('old/afile', 'afilecontents'), ('old/notherfile', 'notherfilecontents')])

        wt.add(["afile"], [b"somefileid"])

        wt.commit("bla", rev_id=b"oldparent")

        wt.add(["notherfile"])

        wt.commit("bla", rev_id=b"oldcommit")

        oldrepos = wt.branch.repository

        wt = self.make_branch_and_tree("new")

        self.build_tree_contents([('new/afile', 'afilecontents'), ('new/notherfile', 'notherfilecontents')])

        wt.add(["afile"], [b"afileid"])

        wt.commit("bla", rev_id=b"newparent")

        wt.branch.repository.fetch(oldrepos)

        wt.branch.repository.lock_write()

        CommitBuilderRevisionRewriter(wt.branch.repository)(

            b"oldcommit", b"newcommit", (b"newparent",))

        wt.branch.repository.unlock()

        oldrev = wt.branch.repository.get_revision(b"oldcommit")

        newrev = wt.branch.repository.get_revision(b"newcommit")

        self.assertEquals([b"newparent"], newrev.parent_ids)

        self.assertEquals(b"newcommit", newrev.revision_id)

        self.assertEquals(oldrev.committer, newrev.committer)

        self.assertEquals(oldrev.timestamp, newrev.timestamp)

        self.assertEquals(oldrev.timezone, newrev.timezone)

        tree = wt.branch.repository.revision_tree(b"newcommit")

        self.assertEquals(b"afileid", tree.path2id("afile"))

        self.assertEquals(b"newcommit", tree.get_file_revision("notherfile"))

    def test_two_revisions_no_renames(self):

        wt = self.make_branch_and_tree("old")

        self.build_tree(['old/afile', 'old/notherfile'])

        wt.add(["afile"], [b"somefileid"])

        wt.commit("bla", rev_id=b"oldparent")

        wt.add(["notherfile"])

        wt.commit("bla", rev_id=b"oldcommit")

        oldrepos = wt.branch.repository

        wt = self.make_branch_and_tree("new")

        self.build_tree(['new/afile', 'new/notherfile'])

        wt.add(["afile"], [b"afileid"])

        wt.commit("bla", rev_id=b"newparent")

        wt.branch.repository.fetch(oldrepos)

        wt.branch.repository.lock_write()

        CommitBuilderRevisionRewriter(wt.branch.repository)(

            b"oldcommit", b"newcommit", (b"newparent",))

        wt.branch.repository.unlock()

    def test_multi_revisions(self):

        wt = self.make_branch_and_tree("old")

        self.build_tree_contents(

            [('old/afile', 'afilecontent'), ('old/sfile', 'sfilecontent'), ('old/notherfile', 'notherfilecontent')])

        wt.add(['sfile'])

        wt.add(["afile"], [b"somefileid"])

        wt.commit("bla", rev_id=b"oldgrandparent")

        open("old/afile", "w").write("data")

        wt.commit("bla", rev_id=b"oldparent")

        wt.add(["notherfile"])

        wt.commit("bla", rev_id=b"oldcommit")

        oldrepos = wt.branch.repository

        wt = self.make_branch_and_tree("new")

        self.build_tree_contents(

            [('new/afile', 'afilecontent'), ('new/sfile', 'sfilecontent'), ('new/notherfile', 'notherfilecontent')])

        wt.add(['sfile'])

        wt.add(["afile"], [b"afileid"])

        wt.commit("bla", rev_id=b"newgrandparent")

        open("new/afile", "w").write("data")

        wt.commit("bla", rev_id=b"newparent")

        wt.branch.repository.fetch(oldrepos)

        wt.branch.repository.lock_write()

        CommitBuilderRevisionRewriter(wt.branch.repository)(

            b"oldcommit", b"newcommit", (b"newparent",))

        wt.branch.repository.unlock()

        oldrev = wt.branch.repository.get_revision(b"oldcommit")

        newrev = wt.branch.repository.get_revision(b"newcommit")

        self.assertEquals([b"newparent"], newrev.parent_ids)

        self.assertEquals(b"newcommit", newrev.revision_id)

        self.assertEquals(oldrev.committer, newrev.committer)

        self.assertEquals(oldrev.timestamp, newrev.timestamp)

        self.assertEquals(oldrev.timezone, newrev.timezone)

        tree = wt.branch.repository.revision_tree(b"newcommit")

        self.assertEquals(b"afileid", tree.path2id("afile"))

        self.assertEquals(b"newcommit", tree.get_file_revision("notherfile"))

        self.assertEquals(b"newgrandparent", tree.get_file_revision("sfile"))

    def test_maps_ids(self):

        wt = self.make_branch_and_tree("old")

        wt.commit("base", rev_id=b"base")

        self.build_tree(['old/afile'])

        wt.add(["afile"], ids=[b"originalid"])

        wt.commit("bla", rev_id=b"oldparent")

        with open("old/afile", "w") as f:

            f.write("bloe")

        wt.commit("bla", rev_id=b"oldcommit")

        oldrepos = wt.branch.repository

        wt = self.make_branch_and_tree("new")

        self.build_tree(['new/afile'])

        wt.add(["afile"], ids=[b"newid"])

        wt.commit("bla", rev_id=b"newparent")

        wt.branch.repository.fetch(oldrepos)

        wt.branch.repository.lock_write()

        CommitBuilderRevisionRewriter(wt.branch.repository)(

            b"oldcommit", b"newcommit", (b"newparent",))

        wt.branch.repository.unlock()

        oldrev = wt.branch.repository.get_revision(b"oldcommit")

        newrev = wt.branch.repository.get_revision(b"newcommit")

        self.assertEquals([b"newparent"], newrev.parent_ids)

        self.assertEquals(b"newcommit", newrev.revision_id)

        self.assertEquals(oldrev.committer, newrev.committer)

        self.assertEquals(oldrev.timestamp, newrev.timestamp)

        self.assertEquals(oldrev.timezone, newrev.timezone)

        tree = wt.branch.repository.revision_tree(b"newcommit")

        self.assertEquals(b"newid", tree.path2id("afile"))

        self.assertEquals(b"newcommit", tree.get_file_revision("afile"))

class TestReplayWorkingtree(TestCaseWithTransport):

    def test_conflicts(self):

        wt = self.make_branch_and_tree("old")

        wt.commit("base", rev_id=b"base")

        self.build_tree(['old/afile'])

        wt.add(["afile"], ids=[b"originalid"])

        wt.commit("bla", rev_id=b"oldparent")

        with open("old/afile", "w") as f:

            f.write("bloe")

        wt.commit("bla", rev_id=b"oldcommit")

        oldrepos = wt.branch.repository

        wt = self.make_branch_and_tree("new")

        self.build_tree(['new/afile'])

        wt.add(["afile"], ids=[b"newid"])

        wt.commit("bla", rev_id=b"newparent")

        wt.branch.repository.fetch(oldrepos)

        with wt.lock_write():

            replayer = WorkingTreeRevisionRewriter(wt, RebaseState1(wt))

            self.assertRaises(

                ConflictsInTree,

                replayer, b"oldcommit", b"newcommit",

                [b"newparent"], )

    def test_simple(self):

        wt = self.make_branch_and_tree("old")

        wt.commit("base", rev_id=b"base")

        self.build_tree(['old/afile'])