/brz/remove-bazaar

# Copyright (C) 2006, 2007 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

"""Tests for the smart wire/domain protocol.

This module contains tests for the domain-level smart requests and responses,

such as the 'Branch.lock_write' request. Many of these use specific disk

formats to exercise calls that only make sense for formats with specific

properties.

Tests for low-level protocol encoding are found in test_smart_transport.

"""

import bz2

from cStringIO import StringIO

import tarfile

from bzrlib import (

    bzrdir,

    errors,

    pack,

    smart,

    tests,

    urlutils,

    )

from bzrlib.branch import Branch, BranchReferenceFormat

import bzrlib.smart.branch

import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir

import bzrlib.smart.packrepository

import bzrlib.smart.repository

from bzrlib.smart.request import (

    FailedSmartServerResponse,

    SmartServerRequest,

    SmartServerResponse,

    SuccessfulSmartServerResponse,

    )

from bzrlib.tests import (

    split_suite_by_re,

    )

from bzrlib.transport import chroot, get_transport

from bzrlib.util import bencode

def load_tests(standard_tests, module, loader):

    """Multiply tests version and protocol consistency."""

    # FindRepository tests.

    bzrdir_mod = bzrlib.smart.bzrdir

    scenarios = [

        ("find_repository", {

            "_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),

        ("find_repositoryV2", {

            "_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),

        ("find_repositoryV3", {

            "_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV3}),

        ]

    to_adapt, result = split_suite_by_re(standard_tests,

        "TestSmartServerRequestFindRepository")

    v2_only, v1_and_2 = split_suite_by_re(to_adapt,

        "_v2")

    tests.multiply_tests(v1_and_2, scenarios, result)

    # The first scenario is only applicable to v1 protocols, it is deleted

    # since.

    tests.multiply_tests(v2_only, scenarios[1:], result)

    return result

class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):

    def setUp(self):

        tests.TestCaseWithTransport.setUp(self)

        self._chroot_server = None

    def get_transport(self, relpath=None):

        if self._chroot_server is None:

            backing_transport = tests.TestCaseWithTransport.get_transport(self)

            self._chroot_server = chroot.ChrootServer(backing_transport)

            self._chroot_server.setUp()

            self.addCleanup(self._chroot_server.tearDown)

        t = get_transport(self._chroot_server.get_url())

        if relpath is not None:

            t = t.clone(relpath)

        return t

class TestCaseWithSmartMedium(tests.TestCaseWithTransport):

    def setUp(self):

        super(TestCaseWithSmartMedium, self).setUp()

        # We're allowed to set  the transport class here, so that we don't use

        # the default or a parameterized class, but rather use the

        # TestCaseWithTransport infrastructure to set up a smart server and

        # transport.

        self.transport_server = self.make_transport_server

    def make_transport_server(self):

        return smart.server.SmartTCPServer_for_testing('-' + self.id())

    def get_smart_medium(self):

        """Get a smart medium to use in tests."""

        return self.get_transport().get_smart_medium()

class TestSmartServerResponse(tests.TestCase):

    def test__eq__(self):

        self.assertEqual(SmartServerResponse(('ok', )),

            SmartServerResponse(('ok', )))

        self.assertEqual(SmartServerResponse(('ok', ), 'body'),

            SmartServerResponse(('ok', ), 'body'))

        self.assertNotEqual(SmartServerResponse(('ok', )),

            SmartServerResponse(('notok', )))

        self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),

            SmartServerResponse(('ok', )))

        self.assertNotEqual(None,

            SmartServerResponse(('ok', )))

    def test__str__(self):

        """SmartServerResponses can be stringified."""

        self.assertEqual(

            "<SuccessfulSmartServerResponse args=('args',) body='body'>",

            str(SuccessfulSmartServerResponse(('args',), 'body')))

        self.assertEqual(

            "<FailedSmartServerResponse args=('args',) body='body'>",

            str(FailedSmartServerResponse(('args',), 'body')))

class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):

    def test_translate_client_path(self):

        transport = self.get_transport()

        request = SmartServerRequest(transport, 'foo/')

        self.assertEqual('./', request.translate_client_path('foo/'))

        self.assertRaises(

            errors.InvalidURLJoin, request.translate_client_path, 'foo/..')

        self.assertRaises(

            errors.PathNotChild, request.translate_client_path, '/')

        self.assertRaises(

            errors.PathNotChild, request.translate_client_path, 'bar/')

        self.assertEqual('./baz', request.translate_client_path('foo/baz'))

    def test_transport_from_client_path(self):

        transport = self.get_transport()

        request = SmartServerRequest(transport, 'foo/')

        self.assertEqual(

            transport.base,

            request.transport_from_client_path('foo/').base)

class TestSmartServerBzrDirRequestCloningMetaDir(

    tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.cloning_metadir."""

    def test_cloning_metadir(self):

        """When there is a bzrdir present, the call succeeds."""

        backing = self.get_transport()

        dir = self.make_bzrdir('.')

        local_result = dir.cloning_metadir()

        request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir

        request = request_class(backing)

        expected = SuccessfulSmartServerResponse(

            (local_result.network_name(),

            local_result.repository_format.network_name(),

            ('branch', local_result.get_branch_format().network_name())))

        self.assertEqual(expected, request.execute('', 'False'))

    def test_cloning_metadir_reference(self):

        """The request fails when bzrdir contains a branch reference."""

        backing = self.get_transport()

        referenced_branch = self.make_branch('referenced')

        dir = self.make_bzrdir('.')

        local_result = dir.cloning_metadir()

        reference = BranchReferenceFormat().initialize(dir, referenced_branch)

        reference_url = BranchReferenceFormat().get_reference(dir)

        # The server shouldn't try to follow the branch reference, so it's fine

        # if the referenced branch isn't reachable.

        backing.rename('referenced', 'moved')

        request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir

        request = request_class(backing)

        expected = FailedSmartServerResponse(('BranchReference',))

        self.assertEqual(expected, request.execute('', 'False'))

class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.create_repository."""

    def test_makes_repository(self):

        """When there is a bzrdir present, the call succeeds."""

        backing = self.get_transport()

        self.make_bzrdir('.')

        request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository

        request = request_class(backing)

        reference_bzrdir_format = bzrdir.format_registry.get('default')()

        reference_format = reference_bzrdir_format.repository_format

        network_name = reference_format.network_name()

        expected = SuccessfulSmartServerResponse(

            ('ok', 'no', 'no', 'no', network_name))

        self.assertEqual(expected, request.execute('', network_name, 'True'))

class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.find_repository."""

    def test_no_repository(self):

        """When there is no repository to be found, ('norepository', ) is returned."""

        backing = self.get_transport()

        request = self._request_class(backing)

        self.make_bzrdir('.')

        self.assertEqual(SmartServerResponse(('norepository', )),

            request.execute(''))

    def test_nonshared_repository(self):

        # nonshared repositorys only allow 'find' to return a handle when the

        # path the repository is being searched on is the same as that that

        # the repository is at.

        backing = self.get_transport()

        request = self._request_class(backing)

        result = self._make_repository_and_result()

        self.assertEqual(result, request.execute(''))

        self.make_bzrdir('subdir')

        self.assertEqual(SmartServerResponse(('norepository', )),

            request.execute('subdir'))

    def _make_repository_and_result(self, shared=False, format=None):

        """Convenience function to setup a repository.

        :result: The SmartServerResponse to expect when opening it.

        """

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

        if repo.supports_rich_root():

            rich_root = 'yes'

        else:

            rich_root = 'no'

        if repo._format.supports_tree_reference:

            subtrees = 'yes'

        else:

            subtrees = 'no'

        if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==

            self._request_class):

            return SuccessfulSmartServerResponse(

                ('ok', '', rich_root, subtrees, 'no',

                 repo._format.network_name()))

        elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==

            self._request_class):

            # All tests so far are on formats, and for non-external

            # repositories.

            return SuccessfulSmartServerResponse(

                ('ok', '', rich_root, subtrees, 'no'))

        else:

            return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))

    def test_shared_repository(self):

        """When there is a shared repository, we get 'ok', 'relpath-to-repo'."""

        backing = self.get_transport()

        request = self._request_class(backing)

        result = self._make_repository_and_result(shared=True)

        self.assertEqual(result, request.execute(''))

        self.make_bzrdir('subdir')

        result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])

        self.assertEqual(result2,

            request.execute('subdir'))

        self.make_bzrdir('subdir/deeper')

        result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])

        self.assertEqual(result3,

            request.execute('subdir/deeper'))

    def test_rich_root_and_subtree_encoding(self):

        """Test for the format attributes for rich root and subtree support."""

        backing = self.get_transport()

        request = self._request_class(backing)

        result = self._make_repository_and_result(format='dirstate-with-subtree')

        # check the test will be valid

        self.assertEqual('yes', result.args[2])

        self.assertEqual('yes', result.args[3])

        self.assertEqual(result, request.execute(''))

    def test_supports_external_lookups_no_v2(self):

        """Test for the supports_external_lookups attribute."""

        backing = self.get_transport()

        request = self._request_class(backing)

        result = self._make_repository_and_result(format='dirstate-with-subtree')

        # check the test will be valid

        self.assertEqual('no', result.args[4])

        self.assertEqual(result, request.execute(''))

class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):

    def test_empty_dir(self):

        """Initializing an empty dir should succeed and do it."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)

        self.assertEqual(SmartServerResponse(('ok', )),

            request.execute(''))

        made_dir = bzrdir.BzrDir.open_from_transport(backing)

        # no branch, tree or repository is expected with the current

        # default formart.

        self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)

        self.assertRaises(errors.NotBranchError, made_dir.open_branch)

        self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)

    def test_missing_dir(self):

        """Initializing a missing directory should fail like the bzrdir api."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)

        self.assertRaises(errors.NoSuchFile,

            request.execute, 'subdir')

    def test_initialized_dir(self):

        """Initializing an extant bzrdir should fail like the bzrdir api."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)

        self.make_bzrdir('subdir')

        self.assertRaises(errors.FileExists,

            request.execute, 'subdir')

class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):

    def test_no_branch(self):

        """When there is no branch, ('nobranch', ) is returned."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestOpenBranch(backing)

        self.make_bzrdir('.')

        self.assertEqual(SmartServerResponse(('nobranch', )),

            request.execute(''))

    def test_branch(self):

        """When there is a branch, 'ok' is returned."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestOpenBranch(backing)

        self.make_branch('.')

        self.assertEqual(SmartServerResponse(('ok', '')),

            request.execute(''))

    def test_branch_reference(self):

        """When there is a branch reference, the reference URL is returned."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestOpenBranch(backing)

        branch = self.make_branch('branch')

        checkout = branch.create_checkout('reference',lightweight=True)

        reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)

        self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')

        self.assertEqual(SmartServerResponse(('ok', reference_url)),

            request.execute('reference'))

class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):

    def test_no_branch(self):

        """When there is no branch, ('nobranch', ) is returned."""

        backing = self.get_transport()

        self.make_bzrdir('.')

        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)

        self.assertEqual(SmartServerResponse(('nobranch', )),

            request.execute(''))

    def test_branch(self):

        """When there is a branch, 'ok' is returned."""

        backing = self.get_transport()

        expected = self.make_branch('.')._format.network_name()

        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)

        self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),

            request.execute(''))

    def test_branch_reference(self):

        """When there is a branch reference, the reference URL is returned."""

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)

        branch = self.make_branch('branch')

        checkout = branch.create_checkout('reference',lightweight=True)

        reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)

        self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')

        self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),

            request.execute('reference'))

    def test_stacked_branch(self):

        """Opening a stacked branch does not open the stacked-on branch."""

        trunk = self.make_branch('trunk')

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

        feature.set_stacked_on_url(trunk.base)

        opened_branches = []

        Branch.hooks.install_named_hook('open', opened_branches.append, None)

        backing = self.get_transport()

        request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)

        request.setup_jail()

        try:

            response = request.execute('feature')

        finally:

            request.teardown_jail()

        expected_format = feature._format.network_name()

        self.assertEqual(

            SuccessfulSmartServerResponse(('branch', expected_format)),

            response)

        self.assertLength(1, opened_branches)

class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):

    def test_empty(self):

        """For an empty branch, the body is empty."""

        backing = self.get_transport()

        request = smart.branch.SmartServerRequestRevisionHistory(backing)

        self.make_branch('.')

        self.assertEqual(SmartServerResponse(('ok', ), ''),

            request.execute(''))

    def test_not_empty(self):

        """For a non-empty branch, the body is empty."""

        backing = self.get_transport()

        request = smart.branch.SmartServerRequestRevisionHistory(backing)

        tree = self.make_branch_and_memory_tree('.')

        tree.lock_write()

        tree.add('')

        r1 = tree.commit('1st commit')

        r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))

        tree.unlock()

        self.assertEqual(

            SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),

            request.execute(''))

class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):

    def test_no_branch(self):

        """When there is a bzrdir and no branch, NotBranchError is raised."""

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequest(backing)

        self.make_bzrdir('.')

        self.assertRaises(errors.NotBranchError,

            request.execute, '')

    def test_branch_reference(self):

        """When there is a branch reference, NotBranchError is raised."""

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequest(backing)

        branch = self.make_branch('branch')

        checkout = branch.create_checkout('reference',lightweight=True)

        self.assertRaises(errors.NotBranchError,

            request.execute, 'checkout')

class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):

    def test_empty(self):

        """For an empty branch, the result is ('ok', '0', 'null:')."""

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)

        self.make_branch('.')

        self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),

            request.execute(''))

    def test_not_empty(self):

        """For a non-empty branch, the result is ('ok', 'revno', 'revid')."""

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)

        tree = self.make_branch_and_memory_tree('.')

        tree.lock_write()

        tree.add('')

        rev_id_utf8 = u'\xc8'.encode('utf-8')

        r1 = tree.commit('1st commit')

        r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)

        tree.unlock()

        self.assertEqual(

            SmartServerResponse(('ok', '2', rev_id_utf8)),

            request.execute(''))

class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):

    def test_default(self):

        """With no file, we get empty content."""

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchGetConfigFile(backing)

        branch = self.make_branch('.')

        # there should be no file by default

        content = ''

        self.assertEqual(SmartServerResponse(('ok', ), content),

            request.execute(''))

    def test_with_content(self):

        # SmartServerBranchGetConfigFile should return the content from

        # branch.control_files.get('branch.conf') for now - in the future it may

        # perform more complex processing.

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchGetConfigFile(backing)

        branch = self.make_branch('.')

        branch._transport.put_bytes('branch.conf', 'foo bar baz')

        self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),

            request.execute(''))

class SetLastRevisionTestBase(tests.TestCaseWithMemoryTransport):

    """Base test case for verbs that implement set_last_revision."""

    def setUp(self):

        tests.TestCaseWithMemoryTransport.setUp(self)

        backing_transport = self.get_transport()

        self.request = self.request_class(backing_transport)

        self.tree = self.make_branch_and_memory_tree('.')

    def lock_branch(self):

        b = self.tree.branch

        branch_token = b.lock_write()

        repo_token = b.repository.lock_write()

        b.repository.unlock()

        return branch_token, repo_token

    def unlock_branch(self):

        self.tree.branch.unlock()

    def set_last_revision(self, revision_id, revno):

        branch_token, repo_token = self.lock_branch()

        response = self._set_last_revision(

            revision_id, revno, branch_token, repo_token)

        self.unlock_branch()

        return response

    def assertRequestSucceeds(self, revision_id, revno):

        response = self.set_last_revision(revision_id, revno)

        self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)

class TestSetLastRevisionVerbMixin(object):

    """Mixin test case for verbs that implement set_last_revision."""

    def test_set_null_to_null(self):

        """An empty branch can have its last revision set to 'null:'."""

        self.assertRequestSucceeds('null:', 0)

    def test_NoSuchRevision(self):

        """If the revision_id is not present, the verb returns NoSuchRevision.

        """

        revision_id = 'non-existent revision'

        self.assertEqual(

            FailedSmartServerResponse(('NoSuchRevision', revision_id)),

            self.set_last_revision(revision_id, 1))

    def make_tree_with_two_commits(self):

        self.tree.lock_write()

        self.tree.add('')

        rev_id_utf8 = u'\xc8'.encode('utf-8')

        r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)

        r2 = self.tree.commit('2nd commit', rev_id='rev-2')

        self.tree.unlock()

    def test_branch_last_revision_info_is_updated(self):

        """A branch's tip can be set to a revision that is present in its

        repository.

        """

        # Make a branch with an empty revision history, but two revisions in

        # its repository.

        self.make_tree_with_two_commits()

        rev_id_utf8 = u'\xc8'.encode('utf-8')

        self.tree.branch.set_revision_history([])

        self.assertEqual(

            (0, 'null:'), self.tree.branch.last_revision_info())

        # We can update the branch to a revision that is present in the

        # repository.

        self.assertRequestSucceeds(rev_id_utf8, 1)

        self.assertEqual(

            (1, rev_id_utf8), self.tree.branch.last_revision_info())

    def test_branch_last_revision_info_rewind(self):

        """A branch's tip can be set to a revision that is an ancestor of the

        current tip.

        """

        self.make_tree_with_two_commits()

        rev_id_utf8 = u'\xc8'.encode('utf-8')

        self.assertEqual(

            (2, 'rev-2'), self.tree.branch.last_revision_info())

        self.assertRequestSucceeds(rev_id_utf8, 1)

        self.assertEqual(

            (1, rev_id_utf8), self.tree.branch.last_revision_info())

    def test_TipChangeRejected(self):

        """If a pre_change_branch_tip hook raises TipChangeRejected, the verb

        returns TipChangeRejected.

        """

        rejection_message = u'rejection message\N{INTERROBANG}'

        def hook_that_rejects(params):

            raise errors.TipChangeRejected(rejection_message)

        Branch.hooks.install_named_hook(

            'pre_change_branch_tip', hook_that_rejects, None)

        self.assertEqual(

            FailedSmartServerResponse(

                ('TipChangeRejected', rejection_message.encode('utf-8'))),

            self.set_last_revision('null:', 0))

class TestSmartServerBranchRequestSetLastRevision(

        SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):

    """Tests for Branch.set_last_revision verb."""

    request_class = smart.branch.SmartServerBranchRequestSetLastRevision

    def _set_last_revision(self, revision_id, revno, branch_token, repo_token):

        return self.request.execute(

            '', branch_token, repo_token, revision_id)

class TestSmartServerBranchRequestSetLastRevisionInfo(

        SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):

    """Tests for Branch.set_last_revision_info verb."""

    request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo

    def _set_last_revision(self, revision_id, revno, branch_token, repo_token):

        return self.request.execute(

            '', branch_token, repo_token, revno, revision_id)

    def test_NoSuchRevision(self):

        """Branch.set_last_revision_info does not have to return

        NoSuchRevision if the revision_id is absent.

        """

        raise tests.TestNotApplicable()

class TestSmartServerBranchRequestSetLastRevisionEx(

        SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):

    """Tests for Branch.set_last_revision_ex verb."""

    request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx

    def _set_last_revision(self, revision_id, revno, branch_token, repo_token):

        return self.request.execute(

            '', branch_token, repo_token, revision_id, 0, 0)

    def assertRequestSucceeds(self, revision_id, revno):

        response = self.set_last_revision(revision_id, revno)

        self.assertEqual(

            SuccessfulSmartServerResponse(('ok', revno, revision_id)),

            response)

    def test_branch_last_revision_info_rewind(self):