# Copyright (C) 2009 Aaron Bentley # # 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 cStringIO import StringIO import os from bzrlib import ( branch, bzrdir, conflicts, directory_service, errors, revision, switch, workingtree, tests, transform, ) from bzrlib.plugins.pipeline.pipeline import ( BranchInTheWay, ChangesAlreadyStored, DuplicatePipe, dwim_pipe, NoSuchPipe, PipeManager, PipeStorage, SwitchWithConflicts, tree_to_pipeline, UnknownRemotePipe, ) from bzrlib.plugins.pipeline.tests import TestCaseWithPipes class TestPipeStorage(TestCaseWithPipes): def test_connect(self): bar = self.make_branch('bar') baz = self.make_branch('baz') PipeStorage.connect(bar, baz) bar_manager = PipeStorage(bar) self.assertEqual('../baz/', bar_manager.get_next()) self.assertIs(None, bar_manager.get_prev()) baz_manager = PipeStorage(baz) self.assertEqual('../bar/', baz_manager.get_prev()) self.assertIs(None, baz_manager.get_next()) def test_connect_relative_url(self): foo = self.make_branch('foo') bar = self.make_branch('bar') PipeStorage.connect(foo, bar) self.assertEqual('../bar/', PipeStorage(foo).get_next()) self.assertEqual('../foo/', PipeStorage(bar).get_prev()) def test_disconnect_middle(self): foo = self.make_branch('foo') foo_manager = PipeManager(foo) bar = foo_manager.insert_pipe('bar') bar_manager = PipeManager(bar) baz = bar_manager.insert_pipe('baz') bar_manager.storage.disconnect() self.assertEqual('../baz/', foo_manager.storage.get_next()) self.assertEqual('../foo/', PipeStorage(baz).get_prev()) self.assertIs(None, bar_manager.get_next_pipe()) def test_disconnect_first(self): foo = self.make_branch('foo') foo_manager = PipeManager(foo) bar = foo_manager.insert_pipe('bar') bar_manager = PipeStorage(bar) foo_manager.storage.disconnect() self.assertIs(None, foo_manager.get_next_pipe()) self.assertIs(None, bar_manager.get_prev()) def test_disconnect_last(self): foo = self.make_branch('foo') foo_manager = PipeManager(foo) bar = foo_manager.insert_pipe('bar') bar_manager = PipeStorage(bar) bar_manager.disconnect() self.assertIs(None, foo_manager.get_next_pipe()) self.assertIs(None, bar_manager.get_prev()) def test_get_put_transform(self): manager = PipeStorage(self.make_branch('branch')) self.assertIs(None, manager._get_transform()) manager._put_transform(StringIO('Hello')) self.assertEqual('Hello', manager._get_transform().read()) def test_put_transform_none(self): manager = PipeStorage(self.make_branch('branch')) manager._put_transform(StringIO('Hello')) manager._put_transform(None) self.assertIs(None, manager._get_transform()) manager._put_transform(None) class TestPipeManager(TestCaseWithPipes): def assertBaseEqual(self, left, right): self.assertEqual([l.base for l in left], [r.base for r in right]) def test_insert_pipe(self): foo = self.make_branch('foo') baz = self.make_branch('baz') PipeStorage.connect(foo, baz) manager = PipeManager(foo) bar = manager.insert_pipe('bar') self.assertBaseEqual([foo, bar, baz], manager.list_pipes()) def test_insert_pipe_before(self): foo = self.make_branch('foo') manager = PipeManager(foo) bar = manager.insert_pipe('bar', before=True) self.assertBaseEqual([bar, foo], manager.list_pipes()) def test_insert_pipe_before_middle(self): foo = self.make_branch('foo') baz = self.make_branch('baz') PipeStorage.connect(foo, baz) manager = PipeManager(baz) bar = manager.insert_pipe('bar', before=True) self.assertBaseEqual([foo, bar, baz], manager.list_pipes()) def test_insert_pipe_revision(self): foo = self.make_branch_and_tree('foo') rev1 = foo.commit('rev1') foo.commit('rev2') manager = PipeManager(foo.branch) bar = manager.insert_pipe('bar', rev1) self.assertEqual(rev1, bar.last_revision()) def test_insert_pipe_checks_dupes(self): foo = self.make_branch_and_tree('foo') manager = PipeManager(foo.branch) e = self.assertRaises(DuplicatePipe, manager.insert_pipe, 'foo') self.assertEqual('There is already a pipe named "foo."', str(e)) def create_long_pipeline(self): foo = self.make_branch('foo') bar = self.make_branch('bar') baz = self.make_branch('baz') qux = self.make_branch('qux') quxx = self.make_branch('quxx') self.connect_many(foo, bar, baz, qux, quxx) return [foo, bar, baz, qux, quxx] def test_list_pipes(self): pipe_list = self.create_long_pipeline() self.assertBaseEqual(pipe_list, PipeManager(pipe_list[2]).list_pipes()) def test_find_pipe(self): pipe_list = self.create_long_pipeline() manager = PipeManager(pipe_list[2]) found = manager.find_pipe('foo') self.assertEqual(pipe_list[0].base, manager.find_pipe('foo').base) self.assertEqual(pipe_list[4].base, manager.find_pipe('quxx').base) def test_find_pipe_not_present(self): pipe_list = self.create_long_pipeline() manager = PipeManager(pipe_list[2]) e = self.assertRaises(NoSuchPipe, manager.find_pipe, 'not-present') self.assertEqual('There is no pipe with nick "not-present".', str(e)) def test_get_first_pipe(self): pipe_list = self.create_long_pipeline() manager = PipeManager(pipe_list[2]) self.assertEqual(pipe_list[0].base, manager.get_first_pipe().base) def test_get_first_pipe_from_first(self): pipe_list = self.create_long_pipeline() manager = PipeManager(pipe_list[0]) self.assertEqual(pipe_list[0].base, manager.get_first_pipe().base) def test_get_last_pipe(self): pipe_list = self.create_long_pipeline() manager = PipeManager(pipe_list[2]) self.assertEqual(pipe_list[-1].base, manager.get_last_pipe().base) def test_get_first_pipe_from_first(self): pipe_list = self.create_long_pipeline() manager = PipeManager(pipe_list[-1]) self.assertEqual(pipe_list[-1].base, manager.get_last_pipe().base) def store_uncommitted(self): tree = self.make_branch_and_tree('tree') tree.commit('get root in there') self.build_tree_contents([('tree/file', 'content')]) tree.add('file', 'file-id') manager = PipeManager(tree.branch) manager.store_uncommitted(tree) return tree, manager def test_store_uncommitted(self): self.store_uncommitted() self.failIfExists('tree/file') def test_store_uncommitted_no_change(self): tree = self.make_branch_and_tree('tree') tree.commit('get root in there') manager = PipeManager(tree.branch) manager.store_uncommitted(tree) self.assertFalse(manager.has_stored_changes()) def test_store_uncommitted_errors_on_overwrite(self): branch = self.make_branch('branch') checkout = branch.create_checkout('checkout', lightweight=True) manager = PipeManager(branch) manager.storage._put_transform(StringIO('Hello')) self.build_tree(['checkout/file']) checkout.add('file') e = self.assertRaises(ChangesAlreadyStored, manager.store_uncommitted, checkout) self.assertEqual('Cannot store uncommitted changes because this pipe' ' already stores uncommitted changes.', str(e)) def test_store_all(self): tree = self.make_branch_and_tree('tree') self.build_tree_contents([('tree/file', 'contents1')]) tree.add('file') tree.commit('message') self.build_tree_contents([('tree/file', 'contents2')]) PipeManager(tree.branch).store_all(tree) self.assertFileEqual('contents1', 'tree/file') def test_restore_uncommitted(self): tree, manager = self.store_uncommitted() manager.restore_uncommitted(tree) self.failUnlessExists('tree/file') self.assertFalse(None, manager.has_stored_changes()) def test_restore_uncommitted_none(self): tree = self.make_branch_and_tree('tree') manager = PipeManager(tree.branch) manager.restore_uncommitted(tree) def test_restore_uncommitted_no_delete(self): tree, manager = self.store_uncommitted() manager.restore_uncommitted(tree, delete=False) self.assertTrue(manager.has_stored_changes()) def make_checkout_first_second(self): tree = self.make_branch_and_checkout('first') tree.commit('get root in there', rev_id='rev1-id') manager = PipeManager(tree.branch) second = manager.insert_pipe('second') tree.commit('second commit.', rev_id='rev2-id') return tree, manager, second def test_switch_to_pipe(self): tree, manager, second = self.make_checkout_first_second() manager.switch_to_pipe(tree, second) tree = workingtree.WorkingTree.open('checkout') self.assertEqual(second.base, tree.branch.base) self.assertEqual('rev1-id', tree.branch.last_revision()) def test_switch_to_pipe_store_restore(self): tree, manager, second = self.make_checkout_first_second() self.build_tree(['checkout/file']) tree.add('file') manager.switch_to_pipe(tree, second) self.failIfExists('checkout/file') tree = workingtree.WorkingTree.open('checkout') first = branch.Branch.open('first') manager = PipeManager(tree.branch).switch_to_pipe(tree, first) self.failUnlessExists('checkout/file') self.assertFalse(manager.storage.has_stored_changes()) tree = workingtree.WorkingTree.open('checkout') self.assertEqual(manager.storage.branch.base, tree.branch.base) self.assertEqual('rev2-id', tree.last_revision()) def test_switch_to_pipe_restore_outdated(self): tree, manager, second = self.make_checkout_first_second() checkout2 = tree.branch.create_checkout('checkout2', lightweight=True) self.build_tree(['checkout/file']) tree.add('file') manager.switch_to_pipe(tree, second) self.build_tree(['checkout2/file2']) checkout2.add('file2') checkout2.commit('rev3', rev_id='rev3-id') tree = workingtree.WorkingTree.open('checkout') first = branch.Branch.open('first') manager = PipeManager(tree.branch).switch_to_pipe(tree, first) self.assertFalse(manager.storage.has_stored_changes()) tree = workingtree.WorkingTree.open('checkout') self.assertEqual(manager.storage.branch.base, tree.branch.base) self.assertEqual('rev3-id', tree.last_revision()) self.failUnlessExists('checkout/file2') def test_switch_to_pipe_local_conflicts(self): tree, manager, second = self.make_checkout_first_second() tree.set_conflicts(conflicts.ConflictList( [conflicts.ContentsConflict('lala')])) self.assertRaises(SwitchWithConflicts, manager.switch_to_pipe, tree, second) def do_conflict_free_pipeline_merge(self): one = self.make_branch('one') checkout = one.create_checkout('checkout', lightweight=True) checkout.commit('rev1') pipe_one = PipeManager(one) two = pipe_one.insert_pipe('two') self.build_tree(['checkout/one']) checkout.add('one') rev2 = checkout.commit('rev2', rev_id='rev2-id') switch.switch(checkout.bzrdir, two) checkout = workingtree.WorkingTree.open('checkout') self.build_tree(['checkout/two']) checkout.add('two') rev3 = checkout.commit('rev3', rev_id='rev3-id') switch.switch(checkout.bzrdir, one) checkout = workingtree.WorkingTree.open('checkout') pipe_one.pipeline_merge(checkout) checkout = workingtree.WorkingTree.open('checkout') self.assertEqual('one', checkout.branch.nick) return two.basis_tree() def test_pipeline_merge(self): """When we do a pipeline merge, we merge into later pipes. When finished, we return to the initial pipe. """ two_basis = self.do_conflict_free_pipeline_merge() self.assertIsNot(None, two_basis.path2id('two')) self.assertIsNot(None, two_basis.path2id('one')) def test_pipeline_merge_updates_parents(self): """When we do a pipeline merge, we merge into later pipes. When finished, we return to the initial pipe. """ two_basis = self.do_conflict_free_pipeline_merge() self.assertEqual(['rev3-id', 'rev2-id'], two_basis.get_parent_ids()) def pipeline_merge(self, trees): self.connect_many(*[t.branch for t in trees]) checkout = trees[0].branch.create_checkout('checkout', lightweight=True) PipeManager(checkout.branch).pipeline_merge(checkout) return workingtree.WorkingTree.open('checkout') def test_pipeline_merge_stops_on_conflict(self): """When a merge is encountered, pipeline merging stops.""" one = self.make_branch_and_tree('one') self.build_tree_contents([('one/conflictable', 'base')]) one.add('conflictable') one.commit('base commit') two = one.bzrdir.sprout('two').open_workingtree() self.build_tree_contents([('one/conflictable', 'other')]) one.commit('other commit') self.build_tree_contents([('two/conflictable', 'this')]) this_revision = two.commit('this commit') checkout = self.pipeline_merge([one, two]) self.assertEqual('two', checkout.branch.nick) self.assertEqual(this_revision, checkout.branch.last_revision()) self.assertFileEqual('this', 'checkout/conflictable.THIS') self.assertFileEqual('other', 'checkout/conflictable.OTHER') def test_pipeline_merge_skips_unchanged(self): """Changes are only committed if there was an actual change.""" one = self.make_branch_and_tree('one') one.commit('base commit', rev_id='rev1-id') two = one.bzrdir.sprout('two').open_workingtree() three = one.bzrdir.sprout('three').open_workingtree() two.commit('new commit', rev_id='rev2-id') self.pipeline_merge([one, two, three]) self.assertEqual('rev2-id', two.branch.last_revision()) self.assertEqual('rev2-id', three.branch.last_revision()) def test_pipeline_merge_stores_and_restores_uncommitted(self): """When a merge is encountered, pipeline merging stops.""" one = self.make_branch('one') checkout = one.create_checkout('checkout', lightweight=True) self.build_tree(['checkout/file', 'checkout/uncommitted']) checkout.add('file') checkout.commit('base commit') manager = PipeManager(one) two = manager.insert_pipe('two') checkout.commit('pointless commit') checkout.add('uncommitted') manager.pipeline_merge(checkout) self.assertIs(None, two.basis_tree().path2id('uncommitted')) self.assertIsNot(None, checkout.path2id('uncommitted')) def test_pipeline_merge_from_submit(self): submit = self.make_branch_and_tree('submit') submit.commit('empty commit') one = submit.bzrdir.sprout('one').open_branch() one.set_submit_branch(submit.branch.base) self.build_tree(['submit/submit-file']) submit.add('submit-file') submit.commit('added submit-file.') checkout = one.create_checkout('checkout', lightweight=True) PipeManager(checkout.branch).pipeline_merge(checkout, from_submit=True) basis = checkout.branch.basis_tree() self.assertIsNot(None, basis.path2id('submit-file')) self.failUnlessExists('checkout/submit-file') def test_pipeline_merge_from_submit_no_change(self): submit = self.make_branch_and_tree('submit') submit.commit('empty commit') one = submit.bzrdir.sprout('one').open_branch() one.set_submit_branch(submit.branch.base) self.build_tree(['submit/submit-file']) checkout = one.create_checkout('checkout', lightweight=True) PipeManager(checkout.branch).pipeline_merge(checkout, from_submit=True) def test_pipeline_merge_pulls_if_possible(self): tree = self.make_branch_and_checkout('first') tree.commit('first commit') first_manager = PipeManager(tree.branch) second = first_manager.insert_pipe('second') third = PipeManager(second).insert_pipe('third') tree.commit('second commit') first_manager.switch_to_pipe(tree, second) tree = workingtree.WorkingTree.open('checkout') tree.commit('second pipe commit') first_manager.pipeline_merge(tree) self.assertEqual(third.last_revision(), second.last_revision()) def test_sync_pipeline_creates_branches(self): self.build_tree(['a/', 'b/']) a_foo = PipeManager(self.make_branch('a/foo')) a_foo.insert_pipe('bar') a_foo.sync_pipeline('b/foo') b_foo = PipeManager(branch.Branch.open('b/foo')) b_bar = PipeManager(branch.Branch.open('b/bar')) self.assertEqual('../bar/', b_foo.storage.get_next()) self.assertEqual('../foo/', b_bar.storage.get_prev()) def test_sync_add_pipe(self): self.build_tree(['local/', 'remote/']) foo = PipeManager(self.make_branch('local/foo')) foo.sync_pipeline('remote/foo') remote_foo = PipeManager(branch.Branch.open('remote/foo')) self.assertIs(None, remote_foo.get_next_pipe()) foo.insert_pipe('bar') foo.sync_pipeline('remote/foo') self.assertEqual('../bar/', remote_foo.storage.get_next()) def test_sync_add_pipe_three(self): self.build_tree(['local/', 'remote/']) foo = PipeManager(self.make_branch('local/foo')) bar = PipeManager(foo.insert_pipe('bar')) foo.sync_pipeline('remote/foo') remote_foo = PipeManager(branch.Branch.open('remote/foo')) self.assertEqual('../bar/', remote_foo.storage.get_next()) bar.insert_pipe('baz') foo.sync_pipeline('remote/foo') self.assertEqual('../bar/', remote_foo.storage.get_next()) def test_sync_pipeline_push_pull(self): self.build_tree(['local/', 'remote/']) foo = PipeManager(self.make_branch('local/foo')) checkout = foo.storage.branch.create_checkout('local/checkout', lightweight=True) checkout.commit('rev1', rev_id='rev1-id') bar = foo.insert_pipe('bar') foo.insert_pipe('baz') # Not part of the test. Just the easiest way to copy a pipeline. foo.sync_pipeline('remote/foo') checkout.commit('rev2', rev_id='rev2-id') remote_bar = branch.Branch.open('remote/bar') remote_checkout = remote_bar.create_checkout('remote/checkout', lightweight=True) remote_checkout.commit('rev3', rev_id='rev3-id') foo.sync_pipeline('remote/foo') remote_foo = branch.Branch.open('remote/foo') self.assertEqual('rev2-id', remote_foo.last_revision()) self.assertEqual('rev3-id', bar.last_revision()) def test_sync_pipeline_diverged(self): self.build_tree(['local/', 'remote/']) foo = PipeManager(self.make_branch('local/foo')) foo.sync_pipeline('remote/foo') checkout = foo.storage.branch.create_checkout('local/checkout', lightweight=True) checkout.commit('rev1', rev_id='rev1-id') remote_foo = branch.Branch.open('remote/foo') remote_checkout = remote_foo.create_checkout('remote/checkout', lightweight=True) remote_checkout.commit('rev3', rev_id='rev3-id') self.assertRaises(errors.DivergedBranches, foo.sync_pipeline, 'remote/foo') def test_sync_pipeline_sets_push_location(self): self.build_tree(['local/', 'remote/']) foo = self.make_branch('local/foo') self.assertIs(None, foo.get_push_location()) foo_manager = PipeManager(foo) foo_manager.sync_pipeline('remote/foo') self.assertContainsRe(foo.get_push_location(), 'remote/foo/$') def test_sync_pipeline_retains_previous_push_location(self): self.build_tree(['local/', 'remote/']) foo = self.make_branch('local/foo') foo.set_push_location('bar') foo_manager = PipeManager(foo) foo_manager.sync_pipeline('remote/foo') self.assertEqual(foo.get_push_location(), 'bar') def test_sync_pipeline_preserves_revision_id_on_create(self): tree = self.make_branch_and_checkout('first') tree.commit('rev1') first_manager = PipeManager(tree.branch) second_pipe = first_manager.insert_pipe('second') first_manager.switch_to_pipe(tree, second_pipe) tree = workingtree.WorkingTree.open('checkout') tree.commit('rev2', rev_id='rev2-id') os.mkdir('remote') first_manager.sync_pipeline('remote/first') remote_second = branch.Branch.open('remote/second') self.assertEqual('rev2-id', remote_second.last_revision()) def test_sync_pipeline_errors_on_bad_target(self): tree = self.make_branch_and_checkout('first') manager = PipeManager(tree.branch) os.mkdir('remote') e = self.assertRaises(UnknownRemotePipe, manager.sync_pipeline, 'remote/random') self.assertEqual('Pipeline has no pipe named "random".', str(e)) def prepare_use_existing(self): os.mkdir('local') tree = self.make_branch_and_checkout('local/first') manager = PipeManager(tree.branch) manager.insert_pipe('second') os.mkdir('remote') remote_second = self.make_branch_and_tree('remote/second') return tree, manager, remote_second def test_sync_pipeline_uses_existing_remote_branches(self): tree, manager, remote_second = self.prepare_use_existing() manager.sync_pipeline('remote/first') def test_sync_pipeline_errors_if_remote_is_not_related(self): tree, manager, remote_second = self.prepare_use_existing() tree.commit('foo') tree.branch.push(manager.get_next_pipe()) remote_second.commit('bar') e = self.assertRaises(BranchInTheWay, manager.sync_pipeline, 'remote/first') def test_sync_pipeline_errors_if_remote_in_pipeline(self): tree, manager, remote_second = self.prepare_use_existing() PipeManager(remote_second.branch).insert_pipe('third') e = self.assertRaises(BranchInTheWay, manager.sync_pipeline, 'remote/first') def test_merge_nick_order(self): """All nicks must appear once in the result. Order is preserved withinin non-matching sections. Order is based on the first item of a non-matching section.""" result = PipeManager._merge_nick_order(['a', 'b', 'd'], ['a', 'c', 'd']) self.assertEqual(['a', 'b', 'c', 'd'], result) result = PipeManager._merge_nick_order(['a', 'c', 'd'], ['a', 'b', 'd']) self.assertEqual(['a', 'b', 'c', 'd'], result) result = PipeManager._merge_nick_order(['a', 'b', 'z'], ['a', 'c', 'y']) self.assertEqual(['a', 'b', 'z', 'c', 'y'], result) def test_connections_to_create(self): connections = PipeManager._connections_to_create(['a', 'b', 'c'], []) self.assertEqual([('a', 'b'), ('b', 'c')], connections) connections = PipeManager._connections_to_create(['a', 'b', 'c'], ['a', 'b']) self.assertEqual([('b', 'c')], connections) connections = PipeManager._connections_to_create(['a', 'c'], ['a', 'b', 'c']) self.assertEqual([('a', 'c')], connections) def test_get_prev_revision_id_second(self): checkout, manager, second = self.make_checkout_first_second() self.assertEqual('rev2-id', PipeManager(second).get_prev_revision_id()) def test_get_prev_revision_id_first(self): branch = self.make_branch('first') self.assertIs(None, PipeManager(branch).get_prev_revision_id()) def prepare_related(self): parent = self.make_branch_and_tree('parent') parent.commit('parent', rev_id='parent-1') child = self.make_branch_and_tree('child') child.pull(parent.branch) child.lock_write() self.addCleanup(child.unlock) parent.commit('parent', rev_id='parent-2') child.commit('child', rev_id='child-1') return parent, child def test_get_prev_revision_id_parent(self): parent, child = self.prepare_related() child.branch.set_parent(parent.branch.base) manager = PipeManager(child.branch) self.assertEqual('parent-1', manager.get_prev_revision_id()) def test_get_prev_revision_id_submit(self): parent, child = self.prepare_related() child.branch.set_submit_branch(parent.branch.base) manager = PipeManager(child.branch) self.assertEqual('parent-1', manager.get_prev_revision_id()) class TestGetDiff(tests.TestCaseWithMemoryTransport): def commit_file(self, branch, path, contents): basis = branch.basis_tree() tt = transform.TransformPreview(basis) if basis.get_root_id() is None: root = tt.new_directory('', transform.ROOT_PARENT, 'root-id') existing_id = basis.path2id(path) if existing_id is None: tt.new_file(path, root, contents, path + '-id') else: trans_id = tt.trans_id_file_id(existing_id) tt.delete_contents(trans_id) tt.create_file(contents, trans_id) tt.commit(branch, 'message') def make_writeable_branch(self, path): branch = self.make_branch(path) branch.lock_write() self.addCleanup(branch.unlock) return branch def make_branch_with_parent(self): parent = self.make_writeable_branch('parent') self.commit_file(parent, 'file', 'A\n') child_url = parent.bzrdir.root_transport.clone('../child').base child = parent.bzrdir.sprout(child_url).open_branch() self.commit_file(parent, 'file', 'B\n') child.lock_write() self.addCleanup(child.unlock) self.commit_file(child, 'file', 'C\n') return child def test_write_patch_no_parent(self): """If no parent is defined for the first pipe, return None.""" first_pipe = self.make_branch('first') patch = PipeManager(first_pipe).get_patch() self.assertIs(None, patch) def test_write_patch_first_parent(self): """If a parent is defined for the first pipe, use that.""" first_pipe = self.make_branch_with_parent() patch = PipeManager(first_pipe).get_patch() self.assertContainsRe(patch, '-A\n\+C') def test_write_patch_first_submit(self): """If a submit branch is defined, it overrides the parent.""" child = self.make_branch_with_parent() submit = self.make_writeable_branch('submit') submit.pull(child) child.set_submit_branch(submit.base) self.commit_file(child, 'file', 'D\n') patch = PipeManager(child).get_patch() self.assertContainsRe(patch, '-C\n\+D') def test_write_patch_second(self): """The previous pipe, if defined, overrides the submit branch.""" submit = self.make_writeable_branch('submit') first = self.make_writeable_branch('first') self.commit_file(first, 'file', 'A\n') second = PipeManager(first).insert_pipe('second') second.lock_write() self.addCleanup(second.unlock) second.set_submit_branch(submit.base) self.commit_file(second, 'file', 'B\n') patch = PipeManager(second).get_patch() self.assertContainsRe(patch, '-A\n\+B') class TestPipeDirectory(TestCaseWithPipes): def test_pipe_directories(self): foo = self.make_branch('foo') bar = self.make_branch('bar') baz = self.make_branch('baz') self.connect_many(foo, bar, baz) os.chdir('bar') self.assertEqual(baz.base, directory_service.directories.dereference(':next')) self.assertEqual(foo.base, directory_service.directories.dereference(':prev')) self.assertEqual(baz.base, directory_service.directories.dereference(':pipe:baz')) self.assertEqual(foo.base, directory_service.directories.dereference(':first')) self.assertEqual(baz.base, directory_service.directories.dereference(':last')) class TestDwimPipe(TestCaseWithPipes): def test_dwim_pipe_no_location(self): checkout = self.make_branch_and_checkout('test') manager = PipeManager(checkout.branch) self.assertRaises(NoSuchPipe, dwim_pipe, manager, None) class TestTreeToPipeline(TestCaseWithPipes): def test_tree_to_pipeline(self): tree = self.make_branch_and_tree('foo') tree.branch.nick = 'bar' tree.commit('rev1', rev_id='rev1-id') tree_to_pipeline(tree) b = branch.Branch.open('foo/.bzr/pipes/bar') self.assertEqual('rev1-id', b.last_revision()) self.assertRaises(errors.NoWorkingTree, workingtree.WorkingTree.open, 'foo/.bzr/pipes/bar') self.assertTrue(b.repository.is_shared()) self.assertEqual(b.base, branch.Branch.open('foo').base) self.assertFalse(tree.is_ignored('pipes')) def test_tree_to_pipeline_lightweight_checkout(self): pipe_branch = self.make_branch('branch') pipe_checkout = pipe_branch.create_checkout('checkout', lightweight=True) self.assertRaises(errors.AlreadyLightweightCheckout, tree_to_pipeline, pipe_checkout) def test_tree_to_pipeline_reuse_shared_repo(self): repo = self.make_repository('repo', shared=True) branch = bzrdir.BzrDir.create_branch_convenience('repo/tree') tree = branch.bzrdir.open_workingtree() checkout = tree_to_pipeline(tree) self.assertEqual(repo.bzrdir.root_transport.base, checkout.branch.repository.bzrdir.root_transport.base)