# 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'])
        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")
        wt = wt.controldir.sprout("new").open_workingtree()
        self.build_tree(['new/bfile'])
        wt.add(["bfile"], ids=[b"newid"])
        wt.commit("bla", rev_id=b"newparent")
        replayer = WorkingTreeRevisionRewriter(wt, RebaseState1(wt))
        replayer(b"oldcommit", b"newcommit", [b"newparent"])
        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.timestamp, newrev.timestamp)
        self.assertEquals(oldrev.timezone, newrev.timezone)

    def test_multiple(self):
        # rebase from
        # base: []
        # oldparent: [base]
        # newparent: [base]
        # oldcommit: [oldparent, ghost]
        # create newcommit by rebasing oldcommit from oldparent to newparent,
        # keeping the merge of ghost.
        # Common base:
        wt = self.make_branch_and_tree("old")
        wt.commit("base", rev_id=b"base")
        # oldparent:
        self.build_tree_contents([('old/afile', 'base content')])
        wt.add(["afile"], ids=[b"originalid"])
        wt.commit("bla", rev_id=b"oldparent")
        # oldcommit (the delta getting rebased)
        #  - change the content of afile to be 'bloe'
        with open("old/afile", "w") as f:
            f.write("bloe")
        wt.add_pending_merge(b"ghost")
        wt.commit("bla", rev_id=b"oldcommit")
        # newparent (the new base for the rebased commit)
        new_tree = wt.controldir.sprout(
            "new", revision_id=b'base').open_workingtree()
        new_tree.branch.repository.fetch(wt.branch.repository)
        wt = new_tree
        self.build_tree_contents([('new/afile', 'base content')])
        wt.add(["afile"], ids=[b"originalid"])
        wt.commit("bla", rev_id=b"newparent")
        # And do it!
        wt.lock_write()
        replayer = WorkingTreeRevisionRewriter(wt, RebaseState1(wt))
        replayer(b"oldcommit", b"newcommit", (b"newparent", b"ghost"))
        wt.unlock()
        oldrev = wt.branch.repository.get_revision(b"oldcommit")
        newrev = wt.branch.repository.get_revision(b"newcommit")
        self.assertEquals([b"oldparent", b"ghost"], oldrev.parent_ids)
        self.assertEquals([b"newparent", b"ghost"], newrev.parent_ids)
        self.assertEquals(b"newcommit", newrev.revision_id)
        self.assertEquals(oldrev.timestamp, newrev.timestamp)
        self.assertEquals(oldrev.timezone, newrev.timezone)

    def test_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'

        Ancestry:
        A:
        B: A
        C: A, B
        D: A
        E: A, B, D
        D': A, B, C
        E': A, B, C, D'

        """
        oldwt = self.make_branch_and_tree("old")
        self.build_tree(['old/afile'])
        with open("old/afile", "w") as f:
            f.write("A\n" * 10)
        oldwt.add(["afile"])
        oldwt.commit("base", rev_id=b"A")
        newwt = oldwt.controldir.sprout("new").open_workingtree()
        with open("old/afile", "w") as f:
            f.write("A\n" * 10 + "B\n")
        oldwt.commit("bla", rev_id=b"B")
        with open("old/afile", "w") as f:
            f.write("A\n" * 10 + "C\n")
        oldwt.commit("bla", rev_id=b"C")
        self.build_tree(['new/bfile'])
        newwt.add(["bfile"])
        with open("new/bfile", "w") as f:
            f.write("D\n")
        newwt.commit("bla", rev_id=b"D")
        with open("new/afile", "w") as f:
            f.write("E\n" + "A\n" * 10 + "B\n")
        with open("new/bfile", "w") as f:
            f.write("D\nE\n")
        newwt.add_pending_merge(b"B")
        newwt.commit("bla", rev_id=b"E")
        newwt.branch.repository.fetch(oldwt.branch.repository)
        newwt.lock_write()
        replayer = WorkingTreeRevisionRewriter(newwt, RebaseState1(newwt))
        replayer(b"D", b"D'", [b"C"])
        newwt.unlock()
        oldrev = newwt.branch.repository.get_revision(b"D")
        newrev = newwt.branch.repository.get_revision(b"D'")
        self.assertEquals([b"C"], newrev.parent_ids)
        newwt.lock_write()
        replayer = WorkingTreeRevisionRewriter(newwt, RebaseState1(newwt))
        self.assertRaises(ConflictsInTree, replayer, b"E", b"E'", [b"D'"])
        newwt.unlock()
        with open("new/afile", 'r') as f:
            self.assertEquals("E\n" + "A\n" * 10 + "C\n", f.read())
        newwt.set_conflicts(ConflictList())
        oldrev = newwt.branch.repository.get_revision(b"E")
        replayer.commit_rebase(oldrev, b"E'")
        newrev = newwt.branch.repository.get_revision(b"E'")
        self.assertEquals([b"D'"], newrev.parent_ids)
        self.assertThat(
            newwt.branch,
            RevisionHistoryMatches([b"A", b"B", b"C", b"D'", b"E'"]))


class TestReplaySnapshotError(TestCase):

    def test_create(self):
        ReplaySnapshotError("message")