# Copyright (C) 2005-2010 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
"""Black-box tests for bzr pull."""
import os
import sys
from bzrlib import (
debug,
remote,
urlutils,
)
from bzrlib.branch import Branch
from bzrlib.directory_service import directories
from bzrlib.osutils import pathjoin
from bzrlib.tests.blackbox import ExternalBase
from bzrlib.uncommit import uncommit
from bzrlib.workingtree import WorkingTree
class TestPull(ExternalBase):
def example_branch(self, path='.'):
tree = self.make_branch_and_tree(path)
self.build_tree_contents([
(pathjoin(path, 'hello'), 'foo'),
(pathjoin(path, 'goodbye'), 'baz')])
tree.add('hello')
tree.commit(message='setup')
tree.add('goodbye')
tree.commit(message='setup')
return tree
def test_pull(self):
"""Pull changes from one branch to another."""
a_tree = self.example_branch('a')
os.chdir('a')
self.run_bzr('pull', retcode=3)
self.run_bzr('missing', retcode=3)
self.run_bzr('missing .')
self.run_bzr('missing')
# this will work on windows because we check for the same branch
# in pull - if it fails, it is a regression
self.run_bzr('pull')
self.run_bzr('pull /', retcode=3)
if sys.platform not in ('win32', 'cygwin'):
self.run_bzr('pull')
os.chdir('..')
b_tree = a_tree.bzrdir.sprout('b').open_workingtree()
os.chdir('b')
self.run_bzr('pull')
os.mkdir('subdir')
b_tree.add('subdir')
b_tree.commit(message='blah', allow_pointless=True)
os.chdir('..')
a = Branch.open('a')
b = Branch.open('b')
self.assertEqual(a.revision_history(), b.revision_history()[:-1])
os.chdir('a')
self.run_bzr('pull ../b')
self.assertEqual(a.revision_history(), b.revision_history())
a_tree.commit(message='blah2', allow_pointless=True)
b_tree.commit(message='blah3', allow_pointless=True)
# no overwrite
os.chdir('../b')
self.run_bzr('pull ../a', retcode=3)
os.chdir('..')
b_tree.bzrdir.sprout('overwriteme')
os.chdir('overwriteme')
self.run_bzr('pull --overwrite ../a')
overwritten = Branch.open('.')
self.assertEqual(overwritten.revision_history(),
a.revision_history())
a_tree.merge_from_branch(b_tree.branch)
a_tree.commit(message="blah4", allow_pointless=True)
os.chdir('../b/subdir')
self.run_bzr('pull ../../a')
self.assertEqual(a.revision_history()[-1], b.revision_history()[-1])
sub_tree = WorkingTree.open_containing('.')[0]
sub_tree.commit(message="blah5", allow_pointless=True)
sub_tree.commit(message="blah6", allow_pointless=True)
os.chdir('..')
self.run_bzr('pull ../a')
os.chdir('../a')
a_tree.commit(message="blah7", allow_pointless=True)
a_tree.merge_from_branch(b_tree.branch)
a_tree.commit(message="blah8", allow_pointless=True)
self.run_bzr('pull ../b')
self.run_bzr('pull ../b')
def test_pull_dash_d(self):
self.example_branch('a')
self.make_branch_and_tree('b')
self.make_branch_and_tree('c')
# pull into that branch
self.run_bzr('pull -d b a')
# pull into a branch specified by a url
c_url = urlutils.local_path_to_url('c')
self.assertStartsWith(c_url, 'file://')
self.run_bzr(['pull', '-d', c_url, 'a'])
def test_pull_revision(self):
"""Pull some changes from one branch to another."""
a_tree = self.example_branch('a')
self.build_tree_contents([
('a/hello2', 'foo'),
('a/goodbye2', 'baz')])
a_tree.add('hello2')
a_tree.commit(message="setup")
a_tree.add('goodbye2')
a_tree.commit(message="setup")
b_tree = a_tree.bzrdir.sprout('b',
revision_id=a_tree.branch.get_rev_id(1)).open_workingtree()
os.chdir('b')
self.run_bzr('pull -r 2')
a = Branch.open('../a')
b = Branch.open('.')
self.assertEqual(a.revno(),4)
self.assertEqual(b.revno(),2)
self.run_bzr('pull -r 3')
self.assertEqual(b.revno(),3)
self.run_bzr('pull -r 4')
self.assertEqual(a.revision_history(), b.revision_history())
def test_overwrite_uptodate(self):
# Make sure pull --overwrite overwrites
# even if the target branch has merged
# everything already.
a_tree = self.make_branch_and_tree('a')
self.build_tree_contents([('a/foo', 'original\n')])
a_tree.add('foo')
a_tree.commit(message='initial commit')
b_tree = a_tree.bzrdir.sprout('b').open_workingtree()
self.build_tree_contents([('a/foo', 'changed\n')])
a_tree.commit(message='later change')
self.build_tree_contents([('a/foo', 'a third change')])
a_tree.commit(message='a third change')
rev_history_a = a_tree.branch.revision_history()
self.assertEqual(len(rev_history_a), 3)
b_tree.merge_from_branch(a_tree.branch)
b_tree.commit(message='merge')
self.assertEqual(len(b_tree.branch.revision_history()), 2)
os.chdir('b')
self.run_bzr('pull --overwrite ../a')
rev_history_b = b_tree.branch.revision_history()
self.assertEqual(len(rev_history_b), 3)
self.assertEqual(rev_history_b, rev_history_a)
def test_overwrite_children(self):
# Make sure pull --overwrite sets the revision-history
# to be identical to the pull source, even if we have convergence
a_tree = self.make_branch_and_tree('a')
self.build_tree_contents([('a/foo', 'original\n')])
a_tree.add('foo')
a_tree.commit(message='initial commit')
b_tree = a_tree.bzrdir.sprout('b').open_workingtree()
self.build_tree_contents([('a/foo', 'changed\n')])
a_tree.commit(message='later change')
self.build_tree_contents([('a/foo', 'a third change')])
a_tree.commit(message='a third change')
self.assertEqual(len(a_tree.branch.revision_history()), 3)
b_tree.merge_from_branch(a_tree.branch)
b_tree.commit(message='merge')
self.assertEqual(len(b_tree.branch.revision_history()), 2)
self.build_tree_contents([('a/foo', 'a fourth change\n')])
a_tree.commit(message='a fourth change')
rev_history_a = a_tree.branch.revision_history()
self.assertEqual(len(rev_history_a), 4)
# With convergence, we could just pull over the
# new change, but with --overwrite, we want to switch our history
os.chdir('b')
self.run_bzr('pull --overwrite ../a')
rev_history_b = b_tree.branch.revision_history()
self.assertEqual(len(rev_history_b), 4)
self.assertEqual(rev_history_b, rev_history_a)
def test_pull_remember(self):
"""Pull changes from one branch to another and test parent location."""
transport = self.get_transport()
tree_a = self.make_branch_and_tree('branch_a')
branch_a = tree_a.branch
self.build_tree(['branch_a/a'])
tree_a.add('a')
tree_a.commit('commit a')
tree_b = branch_a.bzrdir.sprout('branch_b').open_workingtree()
branch_b = tree_b.branch
tree_c = branch_a.bzrdir.sprout('branch_c').open_workingtree()
branch_c = tree_c.branch
self.build_tree(['branch_a/b'])
tree_a.add('b')
tree_a.commit('commit b')
# reset parent
parent = branch_b.get_parent()
branch_b.set_parent(None)
self.assertEqual(None, branch_b.get_parent())
# test pull for failure without parent set
os.chdir('branch_b')
out = self.run_bzr('pull', retcode=3)
self.assertEqual(out,
('','bzr: ERROR: No pull location known or specified.\n'))
# test implicit --remember when no parent set, this pull conflicts
self.build_tree(['d'])
tree_b.add('d')
tree_b.commit('commit d')
out = self.run_bzr('pull ../branch_a', retcode=3)
self.assertEqual(out,
('','bzr: ERROR: These branches have diverged.'
' Use the missing command to see how.\n'
'Use the merge command to reconcile them.\n'))
self.assertEqual(branch_b.get_parent(), parent)
# test implicit --remember after resolving previous failure
uncommit(branch=branch_b, tree=tree_b)
transport.delete('branch_b/d')
self.run_bzr('pull')
self.assertEqual(branch_b.get_parent(), parent)
# test explicit --remember
self.run_bzr('pull ../branch_c --remember')
self.assertEqual(branch_b.get_parent(),
branch_c.bzrdir.root_transport.base)
def test_pull_bundle(self):
from bzrlib.testament import Testament
# Build up 2 trees and prepare for a pull
tree_a = self.make_branch_and_tree('branch_a')
f = open('branch_a/a', 'wb')
f.write('hello')
f.close()
tree_a.add('a')
tree_a.commit('message')
tree_b = tree_a.bzrdir.sprout('branch_b').open_workingtree()
# Make a change to 'a' that 'b' can pull
f = open('branch_a/a', 'wb')
f.write('hey there')
f.close()
tree_a.commit('message')
# Create the bundle for 'b' to pull
os.chdir('branch_a')
self.run_bzr('bundle ../branch_b -o ../bundle')
os.chdir('../branch_b')
out, err = self.run_bzr('pull ../bundle')
self.assertEqual(out,
'Now on revision 2.\n')
self.assertEqual(err,
' M a\nAll changes applied successfully.\n')
self.assertEqualDiff(tree_a.branch.revision_history(),
tree_b.branch.revision_history())
testament_a = Testament.from_revision(tree_a.branch.repository,
tree_a.get_parent_ids()[0])
testament_b = Testament.from_revision(tree_b.branch.repository,
tree_b.get_parent_ids()[0])
self.assertEqualDiff(testament_a.as_text(),
testament_b.as_text())
# it is legal to attempt to pull an already-merged bundle
out, err = self.run_bzr('pull ../bundle')
self.assertEqual(err, '')
self.assertEqual(out, 'No revisions to pull.\n')
def test_pull_verbose_no_files(self):
"""Pull --verbose should not list modified files"""
tree_a = self.make_branch_and_tree('tree_a')
self.build_tree(['tree_a/foo'])
tree_a.add('foo')
tree_a.commit('bar')
tree_b = self.make_branch_and_tree('tree_b')
out = self.run_bzr('pull --verbose -d tree_b tree_a')[0]
self.assertContainsRe(out, 'bar')
self.assertNotContainsRe(out, 'added:')
self.assertNotContainsRe(out, 'foo')
def test_pull_quiet(self):
"""Check that bzr pull --quiet does not print anything"""
tree_a = self.make_branch_and_tree('tree_a')
self.build_tree(['tree_a/foo'])
tree_a.add('foo')
revision_id = tree_a.commit('bar')
tree_b = tree_a.bzrdir.sprout('tree_b').open_workingtree()
out, err = self.run_bzr('pull --quiet -d tree_b')
self.assertEqual(out, '')
self.assertEqual(err, '')
self.assertEqual(tree_b.last_revision(), revision_id)
self.build_tree(['tree_a/moo'])
tree_a.add('moo')
revision_id = tree_a.commit('quack')
out, err = self.run_bzr('pull --quiet -d tree_b')
self.assertEqual(out, '')
self.assertEqual(err, '')
self.assertEqual(tree_b.last_revision(), revision_id)
def test_pull_from_directory_service(self):
source = self.make_branch_and_tree('source')
source.commit('commit 1')
target = source.bzrdir.sprout('target').open_workingtree()
source_last = source.commit('commit 2')
class FooService(object):
"""A directory service that always returns source"""
def look_up(self, name, url):
return 'source'
directories.register('foo:', FooService, 'Testing directory service')
self.addCleanup(directories.remove, 'foo:')
self.run_bzr('pull foo:bar -d target')
self.assertEqual(source_last, target.last_revision())
def test_pull_verbose_defaults_to_long(self):
tree = self.example_branch('source')
target = self.make_branch_and_tree('target')
out = self.run_bzr('pull -v source -d target')[0]
self.assertContainsRe(out,
r'revno: 1\ncommitter: .*\nbranch nick: source')
self.assertNotContainsRe(out, r'\n {4}1 .*\n {6}setup\n')
def test_pull_verbose_uses_default_log(self):
tree = self.example_branch('source')
target = self.make_branch_and_tree('target')
target_config = target.branch.get_config()
target_config.set_user_option('log_format', 'short')
out = self.run_bzr('pull -v source -d target')[0]
self.assertContainsRe(out, r'\n {4}1 .*\n {6}setup\n')
self.assertNotContainsRe(
out, r'revno: 1\ncommitter: .*\nbranch nick: source')
def test_pull_smart_stacked_streaming_acceptance(self):
"""'bzr pull -r 123' works on stacked, smart branches, even when the
revision specified by the revno is only present in the fallback
repository.
See <https://launchpad.net/bugs/380314>
"""
self.setup_smart_server_with_call_log()
# Make a stacked-on branch with two commits so that the
# revision-history can't be determined just by looking at the parent
# field in the revision in the stacked repo.
parent = self.make_branch_and_tree('parent', format='1.9')
parent.commit(message='first commit')
parent.commit(message='second commit')
local = parent.bzrdir.sprout('local').open_workingtree()
local.commit(message='local commit')
local.branch.create_clone_on_transport(
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
empty = self.make_branch_and_tree('empty', format='1.9')
self.reset_smart_call_log()
self.run_bzr(['pull', '-r', '1', self.get_url('stacked')],
working_dir='empty')
# This figure represent the amount of work to perform this use case. It
# is entirely ok to reduce this number if a test fails due to rpc_count
# being too low. If rpc_count increases, more network roundtrips have
# become necessary for this use case. Please do not adjust this number
# upwards without agreement from bzr's network support maintainers.
self.assertLength(18, self.hpss_calls)
remote = Branch.open('stacked')
self.assertEndsWith(remote.get_stacked_on_url(), '/parent')
def test_pull_cross_format_warning(self):
"""You get a warning for probably slow cross-format pulls.
"""
# this is assumed to be going through InterDifferingSerializer
from_tree = self.make_branch_and_tree('from', format='2a')
to_tree = self.make_branch_and_tree('to', format='1.14-rich-root')
from_tree.commit(message='first commit')
out, err = self.run_bzr(['pull', '-d', 'to', 'from'])
self.assertContainsRe(err,
"(?m)Doing on-the-fly conversion")
def test_pull_cross_format_warning_no_IDS(self):
"""You get a warning for probably slow cross-format pulls.
"""
# this simulates what would happen across the network, where
# interdifferingserializer is not active
debug.debug_flags.add('IDS_never')
# TestCase take care of restoring them
from_tree = self.make_branch_and_tree('from', format='2a')
to_tree = self.make_branch_and_tree('to', format='1.14-rich-root')
from_tree.commit(message='first commit')
out, err = self.run_bzr(['pull', '-d', 'to', 'from'])
self.assertContainsRe(err,
"(?m)Doing on-the-fly conversion")
def test_pull_cross_format_from_network(self):
self.setup_smart_server_with_call_log()
from_tree = self.make_branch_and_tree('from', format='2a')
to_tree = self.make_branch_and_tree('to', format='1.14-rich-root')
self.assertIsInstance(from_tree.branch, remote.RemoteBranch)
from_tree.commit(message='first commit')
out, err = self.run_bzr(['pull', '-d', 'to',
from_tree.branch.bzrdir.root_transport.base])
self.assertContainsRe(err,
"(?m)Doing on-the-fly conversion")
def test_pull_to_experimental_format_warning(self):
"""You get a warning for pulling into experimental formats.
"""
from_tree = self.make_branch_and_tree('from', format='development-subtree')
to_tree = self.make_branch_and_tree('to', format='development-subtree')
from_tree.commit(message='first commit')
out, err = self.run_bzr(['pull', '-d', 'to', 'from'])
self.assertContainsRe(err,
"(?m)Fetching into experimental format")
def test_pull_cross_to_experimental_format_warning(self):
"""You get a warning for pulling into experimental formats.
"""
from_tree = self.make_branch_and_tree('from', format='2a')
to_tree = self.make_branch_and_tree('to', format='development-subtree')
from_tree.commit(message='first commit')
out, err = self.run_bzr(['pull', '-d', 'to', 'from'])
self.assertContainsRe(err,
"(?m)Fetching into experimental format")
|