/brz/remove-bazaar

# Copyright (C) 2006-2012, 2016 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 brz push."""

import re

from breezy import (

    branch,

    controldir,

    directory_service,

    errors,

    osutils,

    tests,

    transport,

    uncommit,

    urlutils,

    workingtree

    )

from breezy.bzr import (

    bzrdir,

    )

from breezy.bzr import knitrepo

from breezy.tests import (

    http_server,

    scenarios,

    script,

    test_foreign,

    )

from breezy.tests.matchers import ContainsNoVfsCalls

from breezy.transport import memory

load_tests = scenarios.load_tests_apply_scenarios

class TestPush(tests.TestCaseWithTransport):

    def test_push_error_on_vfs_http(self):

        """ pushing a branch to a HTTP server fails cleanly. """

        # the trunk is published on a web server

        self.transport_readonly_server = http_server.HttpServer

        self.make_branch('source')

        public_url = self.get_readonly_url('target')

        self.run_bzr_error(['http does not support mkdir'],

                           ['push', public_url],

                           working_dir='source')

    def test_push_suggests_parent_alias(self):

        """Push suggests using :parent if there is a known parent branch."""

        tree_a = self.make_branch_and_tree('a')

        tree_a.commit('this is a commit')

        tree_b = self.make_branch_and_tree('b')

        # If there is no parent location set, :parent isn't mentioned.

        out = self.run_bzr('push', working_dir='a', retcode=3)

        self.assertEqual(out,

                         ('', 'brz: ERROR: No push location known or specified.\n'))

        # If there is a parent location set, the error suggests :parent.

        tree_a.branch.set_parent(tree_b.branch.base)

        out = self.run_bzr('push', working_dir='a', retcode=3)

        self.assertEqual(out,

                         ('', 'brz: ERROR: No push location known or specified. '

                          'To push to the parent branch '

                          '(at %s), use \'brz push :parent\'.\n' %

                          urlutils.unescape_for_display(tree_b.branch.base, 'utf-8')))

    def test_push_remember(self):

        """Push changes from one branch to another and test push 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.controldir.sprout('branch_b').open_workingtree()

        branch_b = tree_b.branch

        tree_c = branch_a.controldir.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')

        self.build_tree(['branch_b/c'])

        tree_b.add('c')

        tree_b.commit('commit c')

        # initial push location must be empty

        self.assertEqual(None, branch_b.get_push_location())

        # test push for failure without push location set

        out = self.run_bzr('push', working_dir='branch_a', retcode=3)

        self.assertEqual(out,

                         ('', 'brz: ERROR: No push location known or specified.\n'))

        # test not remembered if cannot actually push

        self.run_bzr('push path/which/doesnt/exist',

                     working_dir='branch_a', retcode=3)

        out = self.run_bzr('push', working_dir='branch_a', retcode=3)

        self.assertEqual(

            ('', 'brz: ERROR: No push location known or specified.\n'),

            out)

        # test implicit --remember when no push location set, push fails

        out = self.run_bzr('push ../branch_b',

                           working_dir='branch_a', retcode=3)

        self.assertEqual(out,

                         ('', 'brz: ERROR: These branches have diverged.  '

                          'See "brz help diverged-branches" for more information.\n'))

        # Refresh the branch as 'push' modified it

        branch_a = branch_a.controldir.open_branch()

        self.assertEqual(osutils.abspath(branch_a.get_push_location()),

                         osutils.abspath(branch_b.controldir.root_transport.base))

        # test implicit --remember after resolving previous failure

        uncommit.uncommit(branch=branch_b, tree=tree_b)

        transport.delete('branch_b/c')

        out, err = self.run_bzr('push', working_dir='branch_a')

        # Refresh the branch as 'push' modified it

        branch_a = branch_a.controldir.open_branch()

        path = branch_a.get_push_location()

        self.assertEqual(err,

                         'Using saved push location: %s\n'

                         'All changes applied successfully.\n'

                         'Pushed up to revision 2.\n'

                         % urlutils.local_path_from_url(path))

        self.assertEqual(path,

                         branch_b.controldir.root_transport.base)

        # test explicit --remember

        self.run_bzr('push ../branch_c --remember', working_dir='branch_a')

        # Refresh the branch as 'push' modified it

        branch_a = branch_a.controldir.open_branch()

        self.assertEqual(branch_a.get_push_location(),

                         branch_c.controldir.root_transport.base)

    def test_push_without_tree(self):

        # brz push from a branch that does not have a checkout should work.

        b = self.make_branch('.')

        out, err = self.run_bzr('push pushed-location')

        self.assertEqual('', out)

        self.assertEqual('Created new branch.\n', err)

        b2 = branch.Branch.open('pushed-location')

        self.assertEndsWith(b2.base, 'pushed-location/')

    def test_push_no_tree(self):

        # brz push --no-tree of a branch with working trees

        b = self.make_branch_and_tree('push-from')

        self.build_tree(['push-from/file'])

        b.add('file')

        b.commit('commit 1')

        out, err = self.run_bzr('push --no-tree -d push-from push-to')

        self.assertEqual('', out)

        self.assertEqual('Created new branch.\n', err)

        self.assertPathDoesNotExist('push-to/file')

    def test_push_new_branch_revision_count(self):

        # brz push of a branch with revisions to a new location

        # should print the number of revisions equal to the length of the

        # local branch.

        t = self.make_branch_and_tree('tree')

        self.build_tree(['tree/file'])

        t.add('file')

        t.commit('commit 1')

        out, err = self.run_bzr('push -d tree pushed-to')

        self.assertEqual('', out)

        self.assertEqual('Created new branch.\n', err)

    def test_push_quiet(self):

        # test that using -q makes output quiet

        t = self.make_branch_and_tree('tree')

        self.build_tree(['tree/file'])

        t.add('file')

        t.commit('commit 1')

        self.run_bzr('push -d tree pushed-to')

        # Refresh the branch as 'push' modified it and get the push location

        push_loc = t.branch.controldir.open_branch().get_push_location()

        out, err = self.run_bzr('push', working_dir="tree")

        self.assertEqual('Using saved push location: %s\n'

                         'No new revisions or tags to push.\n' %

                         urlutils.local_path_from_url(push_loc), err)

        out, err = self.run_bzr('push -q', working_dir="tree")

        self.assertEqual('', out)

        self.assertEqual('', err)

    def test_push_only_pushes_history(self):

        # Knit branches should only push the history for the current revision.

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = knitrepo.RepositoryFormatKnit1()

        shared_repo = self.make_repository('repo', format=format, shared=True)

        shared_repo.set_make_working_trees(True)

        def make_shared_tree(path):

            shared_repo.controldir.root_transport.mkdir(path)

            controldir.ControlDir.create_branch_convenience('repo/' + path)

            return workingtree.WorkingTree.open('repo/' + path)

        tree_a = make_shared_tree('a')

        self.build_tree(['repo/a/file'])

        tree_a.add('file')

        tree_a.commit('commit a-1', rev_id=b'a-1')

        f = open('repo/a/file', 'ab')

        f.write(b'more stuff\n')

        f.close()

        tree_a.commit('commit a-2', rev_id=b'a-2')

        tree_b = make_shared_tree('b')

        self.build_tree(['repo/b/file'])

        tree_b.add('file')

        tree_b.commit('commit b-1', rev_id=b'b-1')

        self.assertTrue(shared_repo.has_revision(b'a-1'))

        self.assertTrue(shared_repo.has_revision(b'a-2'))

        self.assertTrue(shared_repo.has_revision(b'b-1'))

        # Now that we have a repository with shared files, make sure

        # that things aren't copied out by a 'push'

        self.run_bzr('push ../../push-b', working_dir='repo/b')

        pushed_tree = workingtree.WorkingTree.open('push-b')

        pushed_repo = pushed_tree.branch.repository

        self.assertFalse(pushed_repo.has_revision(b'a-1'))

        self.assertFalse(pushed_repo.has_revision(b'a-2'))

        self.assertTrue(pushed_repo.has_revision(b'b-1'))

    def test_push_funky_id(self):

        t = self.make_branch_and_tree('tree')

        self.build_tree(['tree/filename'])

        t.add('filename', b'funky-chars<>%&;"\'')

        t.commit('commit filename')

        self.run_bzr('push -d tree new-tree')

    def test_push_dash_d(self):

        t = self.make_branch_and_tree('from')

        t.commit(allow_pointless=True,

                 message='first commit')

        self.run_bzr('push -d from to-one')

        self.assertPathExists('to-one')

        self.run_bzr('push -d %s %s'

                     % tuple(map(urlutils.local_path_to_url, ['from', 'to-two'])))

        self.assertPathExists('to-two')

    def test_push_repository_no_branch_doesnt_fetch_all_revs(self):

        # See https://bugs.launchpad.net/bzr/+bug/465517

        target_repo = self.make_repository('target')

        source = self.make_branch_builder('source')

        source.start_series()

        source.build_snapshot(None, [

            ('add', ('', b'root-id', 'directory', None))],

            revision_id=b'A')

        source.build_snapshot([b'A'], [], revision_id=b'B')

        source.build_snapshot([b'A'], [], revision_id=b'C')

        source.finish_series()

        self.run_bzr('push target -d source')

        self.addCleanup(target_repo.lock_read().unlock)

        # We should have pushed 'C', but not 'B', since it isn't in the

        # ancestry

        self.assertEqual([(b'A',), (b'C',)], sorted(

            target_repo.revisions.keys()))

    def test_push_smart_non_stacked_streaming_acceptance(self):

        self.setup_smart_server_with_call_log()

        t = self.make_branch_and_tree('from')

        t.commit(allow_pointless=True, message='first commit')

        self.reset_smart_call_log()

        self.run_bzr(['push', self.get_url('to-one')], working_dir='from')

        # 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(9, self.hpss_calls)

        self.assertLength(1, self.hpss_connections)

        self.assertThat(self.hpss_calls, ContainsNoVfsCalls)

    def test_push_smart_stacked_streaming_acceptance(self):

        self.setup_smart_server_with_call_log()

        parent = self.make_branch_and_tree('parent', format='1.9')

        parent.commit(message='first commit')

        local = parent.controldir.sprout('local').open_workingtree()

        local.commit(message='local commit')

        self.reset_smart_call_log()

        self.run_bzr(['push', '--stacked', '--stacked-on', '../parent',

                      self.get_url('public')], working_dir='local')

        # 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(15, self.hpss_calls)

        self.assertLength(1, self.hpss_connections)

        self.assertThat(self.hpss_calls, ContainsNoVfsCalls)

        remote = branch.Branch.open('public')

        self.assertEndsWith(remote.get_stacked_on_url(), '/parent')

    def test_push_smart_tags_streaming_acceptance(self):

        self.setup_smart_server_with_call_log()

        t = self.make_branch_and_tree('from')

        rev_id = t.commit(allow_pointless=True, message='first commit')

        t.branch.tags.set_tag('new-tag', rev_id)

        self.reset_smart_call_log()

        self.run_bzr(['push', self.get_url('to-one')], working_dir='from')

        # 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(11, self.hpss_calls)

        self.assertLength(1, self.hpss_connections)

        self.assertThat(self.hpss_calls, ContainsNoVfsCalls)

    def test_push_smart_incremental_acceptance(self):

        self.setup_smart_server_with_call_log()

        t = self.make_branch_and_tree('from')

        rev_id1 = t.commit(allow_pointless=True, message='first commit')

        rev_id2 = t.commit(allow_pointless=True, message='second commit')

        self.run_bzr(

            ['push', self.get_url('to-one'), '-r1'], working_dir='from')

        self.reset_smart_call_log()

        self.run_bzr(['push', self.get_url('to-one')], working_dir='from')

        # 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(11, self.hpss_calls)

        self.assertLength(1, self.hpss_connections)

        self.assertThat(self.hpss_calls, ContainsNoVfsCalls)

    def test_push_smart_with_default_stacking_url_path_segment(self):

        # If the default stacked-on location is a path element then branches

        # we push there over the smart server are stacked and their

        # stacked_on_url is that exact path segment. Added to nail bug 385132.

        self.setup_smart_server_with_call_log()

        self.make_branch('stack-on', format='1.9')

        self.make_controldir('.').get_config().set_default_stack_on(

            '/stack-on')

        self.make_branch('from', format='1.9')

        out, err = self.run_bzr(['push', '-d', 'from', self.get_url('to')])

        b = branch.Branch.open(self.get_url('to'))

        self.assertEqual('/extra/stack-on', b.get_stacked_on_url())

    def test_push_smart_with_default_stacking_relative_path(self):

        # If the default stacked-on location is a relative path then branches

        # we push there over the smart server are stacked and their

        # stacked_on_url is a relative path. Added to nail bug 385132.

        self.setup_smart_server_with_call_log()

        self.make_branch('stack-on', format='1.9')

        self.make_controldir('.').get_config().set_default_stack_on('stack-on')

        self.make_branch('from', format='1.9')

        out, err = self.run_bzr(['push', '-d', 'from', self.get_url('to')])

        b = branch.Branch.open(self.get_url('to'))

        self.assertEqual('../stack-on', b.get_stacked_on_url())

    def create_simple_tree(self):

        tree = self.make_branch_and_tree('tree')

        self.build_tree(['tree/a'])

        tree.add(['a'], [b'a-id'])

        tree.commit('one', rev_id=b'r1')

        return tree

    def test_push_create_prefix(self):

        """'brz push --create-prefix' will create leading directories."""

        tree = self.create_simple_tree()

        self.run_bzr_error(['Parent directory of ../new/tree does not exist'],

                           'push ../new/tree',

                           working_dir='tree')

        self.run_bzr('push ../new/tree --create-prefix',

                     working_dir='tree')

        new_tree = workingtree.WorkingTree.open('new/tree')

        self.assertEqual(tree.last_revision(), new_tree.last_revision())

        self.assertPathExists('new/tree/a')

    def test_push_use_existing(self):

        """'brz push --use-existing-dir' can push into an existing dir.

        By default, 'brz push' will not use an existing, non-versioned dir.

        """

        tree = self.create_simple_tree()

        self.build_tree(['target/'])

        self.run_bzr_error(['Target directory ../target already exists',

                            'Supply --use-existing-dir',

                            ],

                           'push ../target', working_dir='tree')

        self.run_bzr('push --use-existing-dir ../target',

                     working_dir='tree')

        new_tree = workingtree.WorkingTree.open('target')

        self.assertEqual(tree.last_revision(), new_tree.last_revision())

        # The push should have created target/a

        self.assertPathExists('target/a')

    def test_push_use_existing_into_empty_bzrdir(self):

        """'brz push --use-existing-dir' into a dir with an empty .bzr dir

        fails.

        """

        tree = self.create_simple_tree()

        self.build_tree(['target/', 'target/.bzr/'])

        self.run_bzr_error(

            ['Target directory ../target already contains a .bzr directory, '

             'but it is not valid.'],

            'push ../target --use-existing-dir', working_dir='tree')

    def test_push_onto_repo(self):

        """We should be able to 'brz push' into an existing bzrdir."""

        tree = self.create_simple_tree()

        repo = self.make_repository('repo', shared=True)

        self.run_bzr('push ../repo',

                     working_dir='tree')

        # Pushing onto an existing bzrdir will create a repository and

        # branch as needed, but will only create a working tree if there was

        # no BzrDir before.

        self.assertRaises(errors.NoWorkingTree,

                          workingtree.WorkingTree.open, 'repo')

        new_branch = branch.Branch.open('repo')

        self.assertEqual(tree.last_revision(), new_branch.last_revision())

    def test_push_onto_just_bzrdir(self):

        """We don't handle when the target is just a bzrdir.

        Because you shouldn't be able to create *just* a bzrdir in the wild.

        """

        # TODO: jam 20070109 Maybe it would be better to create the repository

        #       if at this point

        tree = self.create_simple_tree()

        a_controldir = self.make_controldir('dir')

        self.run_bzr_error(['At ../dir you have a valid .bzr control'],

                           'push ../dir',

                           working_dir='tree')

    def test_push_with_revisionspec(self):

        """We should be able to push a revision older than the tip."""

        tree_from = self.make_branch_and_tree('from')

        tree_from.commit("One.", rev_id=b"from-1")

        tree_from.commit("Two.", rev_id=b"from-2")

        self.run_bzr('push -r1 ../to', working_dir='from')

        tree_to = workingtree.WorkingTree.open('to')

        repo_to = tree_to.branch.repository

        self.assertTrue(repo_to.has_revision(b'from-1'))

        self.assertFalse(repo_to.has_revision(b'from-2'))

        self.assertEqual(tree_to.branch.last_revision_info()[1], b'from-1')

        self.assertFalse(

            tree_to.changes_from(tree_to.basis_tree()).has_changed())

        self.run_bzr_error(

            ['brz: ERROR: brz push --revision '

             'takes exactly one revision identifier\n'],

            'push -r0..2 ../to', working_dir='from')

    def create_trunk_and_feature_branch(self):

        # We have a mainline

        trunk_tree = self.make_branch_and_tree('target',

                                               format='1.9')

        trunk_tree.commit('mainline')

        # and a branch from it

        branch_tree = self.make_branch_and_tree('branch',

                                                format='1.9')

        branch_tree.pull(trunk_tree.branch)

        branch_tree.branch.set_parent(trunk_tree.branch.base)

        # with some work on it

        branch_tree.commit('moar work plz')

        return trunk_tree, branch_tree

    def assertPublished(self, branch_revid, stacked_on):

        """Assert that the branch 'published' has been published correctly."""

        published_branch = branch.Branch.open('published')

        # The published branch refers to the mainline

        self.assertEqual(stacked_on, published_branch.get_stacked_on_url())

        # and the branch's work was pushed

        self.assertTrue(published_branch.repository.has_revision(branch_revid))

    def test_push_new_branch_stacked_on(self):

        """Pushing a new branch with --stacked-on creates a stacked branch."""

        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()

        # we publish branch_tree with a reference to the mainline.

        out, err = self.run_bzr(['push', '--stacked-on', trunk_tree.branch.base,

                                 self.get_url('published')], working_dir='branch')

        self.assertEqual('', out)

        self.assertEqual('Created new stacked branch referring to %s.\n' %

                         trunk_tree.branch.base, err)

        self.assertPublished(branch_tree.last_revision(),

                             trunk_tree.branch.base)

    def test_push_new_branch_stacked_on(self):

        """Pushing a new branch with --stacked-on can use directory URLs."""

        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()

        class FooDirectory(object):

            def look_up(self, name, url, purpose=None):

                if url == 'foo:':

                    return trunk_tree.branch.base

                return url

        directory_service.directories.register('foo:', FooDirectory, 'Foo directory')

        self.addCleanup(directory_service.directories.remove, 'foo:')

        # we publish branch_tree with a reference to the mainline.

        out, err = self.run_bzr(['push', '--stacked-on', 'foo:',

                                 self.get_url('published')], working_dir='branch')

        self.assertEqual('', out)

        self.assertEqual('Created new stacked branch referring to %s.\n' %

                         trunk_tree.branch.base, err)

        self.assertPublished(branch_tree.last_revision(),

                             trunk_tree.branch.base)

    def test_push_new_branch_stacked_uses_parent_when_no_public_url(self):

        """When the parent has no public url the parent is used as-is."""

        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()

        # now we do a stacked push, which should determine the public location

        # for us.

        out, err = self.run_bzr(['push', '--stacked',

                                 self.get_url('published')], working_dir='branch')

        self.assertEqual('', out)

        self.assertEqual('Created new stacked branch referring to %s.\n' %

                         trunk_tree.branch.base, err)

        self.assertPublished(branch_tree.last_revision(),

                             trunk_tree.branch.base)

    def test_push_new_branch_stacked_uses_parent_public(self):

        """Pushing a new branch with --stacked creates a stacked branch."""

        trunk_tree, branch_tree = self.create_trunk_and_feature_branch()

        # the trunk is published on a web server

        self.transport_readonly_server = http_server.HttpServer

        trunk_public = self.make_branch('public_trunk', format='1.9')

        trunk_public.pull(trunk_tree.branch)

        trunk_public_url = self.get_readonly_url('public_trunk')

        br = trunk_tree.branch

        br.set_public_branch(trunk_public_url)

        # now we do a stacked push, which should determine the public location

        # for us.

        out, err = self.run_bzr(['push', '--stacked',

                                 self.get_url('published')], working_dir='branch')

        self.assertEqual('', out)

        self.assertEqual('Created new stacked branch referring to %s.\n' %

                         trunk_public_url, err)

        self.assertPublished(branch_tree.last_revision(), trunk_public_url)