/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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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 BranchReferenceFormat

import bzrlib.smart.branch

import bzrlib.smart.bzrdir

import bzrlib.smart.repository

from bzrlib.smart.request import (

    FailedSmartServerResponse,

    SmartServerRequest,

    SmartServerResponse,

    SuccessfulSmartServerResponse,

    )

from bzrlib.tests import (

    iter_suite_tests,

    split_suite_by_re,

    TestScenarioApplier,

    )

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

    applier = TestScenarioApplier()

    applier.scenarios = [

        ("find_repository", {

            "_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),

        ("find_repositoryV2", {

            "_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),

        ]

    to_adapt, result = split_suite_by_re(standard_tests,

        "TestSmartServerRequestFindRepository")

    v2_only, v1_and_2 = split_suite_by_re(to_adapt,

        "_v2")

    for test in iter_suite_tests(v1_and_2):

        result.addTests(applier.adapt(test))

    del applier.scenarios[0]

    for test in iter_suite_tests(v2_only):

        result.addTests(applier.adapt(test))

    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 = smart.server.SmartTCPServer_for_testing

    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(

            "<SmartServerResponse status=OK args=('args',) body='body'>",

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

        self.assertEqual(

            "<SmartServerResponse status=ERR 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 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.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 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.control_files.put_utf8('branch.conf', 'foo bar baz')

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

            request.execute(''))

class TestSmartServerBranchRequestSetLastRevision(tests.TestCaseWithMemoryTransport):

    def test_empty(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)

        b = self.make_branch('.')

        branch_token = b.lock_write()

        repo_token = b.repository.lock_write()

        b.repository.unlock()

        try:

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

                request.execute(

                    '', branch_token, repo_token,

                    'null:'))

        finally:

            b.unlock()

    def test_not_present_revision_id(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)

        b = self.make_branch('.')

        branch_token = b.lock_write()

        repo_token = b.repository.lock_write()

        b.repository.unlock()

        try:

            revision_id = 'non-existent revision'

            self.assertEqual(

                SmartServerResponse(('NoSuchRevision', revision_id)),

                request.execute(

                    '', branch_token, repo_token,

                    revision_id))

        finally:

            b.unlock()

    def test_revision_id_present(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevision(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', rev_id=rev_id_utf8)

        r2 = tree.commit('2nd commit')

        tree.unlock()

        branch_token = tree.branch.lock_write()

        repo_token = tree.branch.repository.lock_write()

        tree.branch.repository.unlock()

        try:

            self.assertEqual(

                SmartServerResponse(('ok',)),

                request.execute(

                    '', branch_token, repo_token,

                    rev_id_utf8))

            self.assertEqual([rev_id_utf8], tree.branch.revision_history())

        finally:

            tree.branch.unlock()

    def test_revision_id_present2(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevision(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', rev_id=rev_id_utf8)

        r2 = tree.commit('2nd commit')

        tree.unlock()

        tree.branch.set_revision_history([])

        branch_token = tree.branch.lock_write()

        repo_token = tree.branch.repository.lock_write()

        tree.branch.repository.unlock()

        try:

            self.assertEqual(

                SmartServerResponse(('ok',)),

                request.execute(

                    '', branch_token, repo_token,

                    rev_id_utf8))

            self.assertEqual([rev_id_utf8], tree.branch.revision_history())

        finally:

            tree.branch.unlock()

class TestSmartServerBranchRequestSetLastRevisionInfo(tests.TestCaseWithTransport):

    def lock_branch(self, branch):

        branch_token = branch.lock_write()

        repo_token = branch.repository.lock_write()

        branch.repository.unlock()

        self.addCleanup(branch.unlock)

        return branch_token, repo_token

    def make_locked_branch(self, format=None):

        branch = self.make_branch('.', format=format)

        branch_token, repo_token = self.lock_branch(branch)

        return branch, branch_token, repo_token

    def test_empty(self):

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

        b, branch_token, repo_token = self.make_locked_branch()

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(

            backing)

        response = request.execute('', branch_token, repo_token, '0', 'null:')

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

    def assertBranchLastRevisionInfo(self, expected_info, branch_relpath):

        branch = bzrdir.BzrDir.open(branch_relpath).open_branch()

        self.assertEqual(expected_info, branch.last_revision_info())

    def test_branch_revision_info_is_updated(self):

        """This method really does update the branch last revision info."""

        tree = self.make_branch_and_memory_tree('.')

        tree.lock_write()

        tree.add('')

        tree.commit('First commit', rev_id='revision-1')

        tree.commit('Second commit', rev_id='revision-2')

        tree.unlock()

        branch = tree.branch

        branch_token, repo_token = self.lock_branch(branch)

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(

            backing)

        self.assertBranchLastRevisionInfo((2, 'revision-2'), '.')

        response = request.execute(

            '', branch_token, repo_token, '1', 'revision-1')

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

        self.assertBranchLastRevisionInfo((1, 'revision-1'), '.')

    def test_not_present_revid(self):

        """Some branch formats will check that the revision is present in the

        repository.  When that check fails, a NoSuchRevision error is returned

        to the client.

        """

        # Make a knit format branch, because that format checks the values

        # given to set_last_revision_info.

        b, branch_token, repo_token = self.make_locked_branch(format='knit')

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(

            backing)

        response = request.execute(

            '', branch_token, repo_token, '1', 'not-present')

        self.assertEqual(

            SmartServerResponse(('NoSuchRevision', 'not-present')), response)

class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):

    def setUp(self):

        tests.TestCaseWithMemoryTransport.setUp(self)

    def test_lock_write_on_unlocked_branch(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLockWrite(backing)

        branch = self.make_branch('.', format='knit')

        repository = branch.repository

        response = request.execute('')

        branch_nonce = branch.control_files._lock.peek().get('nonce')

        repository_nonce = repository.control_files._lock.peek().get('nonce')

        self.assertEqual(

            SmartServerResponse(('ok', branch_nonce, repository_nonce)),

            response)

        # The branch (and associated repository) is now locked.  Verify that

        # with a new branch object.

        new_branch = repository.bzrdir.open_branch()

        self.assertRaises(errors.LockContention, new_branch.lock_write)

    def test_lock_write_on_locked_branch(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLockWrite(backing)

        branch = self.make_branch('.')

        branch.lock_write()

        branch.leave_lock_in_place()

        branch.unlock()

        response = request.execute('')

        self.assertEqual(

            SmartServerResponse(('LockContention',)), response)

    def test_lock_write_with_tokens_on_locked_branch(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLockWrite(backing)

        branch = self.make_branch('.', format='knit')

        branch_token = branch.lock_write()

        repo_token = branch.repository.lock_write()

        branch.repository.unlock()

        branch.leave_lock_in_place()

        branch.repository.leave_lock_in_place()

        branch.unlock()

        response = request.execute('',

                                   branch_token, repo_token)

        self.assertEqual(

            SmartServerResponse(('ok', branch_token, repo_token)), response)

    def test_lock_write_with_mismatched_tokens_on_locked_branch(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLockWrite(backing)

        branch = self.make_branch('.', format='knit')

        branch_token = branch.lock_write()

        repo_token = branch.repository.lock_write()

        branch.repository.unlock()

        branch.leave_lock_in_place()

        branch.repository.leave_lock_in_place()

        branch.unlock()

        response = request.execute('',

                                   branch_token+'xxx', repo_token)

        self.assertEqual(

            SmartServerResponse(('TokenMismatch',)), response)

    def test_lock_write_on_locked_repo(self):

        backing = self.get_transport()

        request = smart.branch.SmartServerBranchRequestLockWrite(backing)

        branch = self.make_branch('.', format='knit')

        branch.repository.lock_write()

        branch.repository.leave_lock_in_place()

        branch.repository.unlock()

        response = request.execute('')

        self.assertEqual(

            SmartServerResponse(('LockContention',)), response)

    def test_lock_write_on_readonly_transport(self):

        backing = self.get_readonly_transport()

        request = smart.branch.SmartServerBranchRequestLockWrite(backing)

        branch = self.make_branch('.')

        root = self.get_transport().clone('/')

        path = urlutils.relative_url(root.base, self.get_transport().base)

        response = request.execute(path)

        error_name, lock_str, why_str = response.args

        self.assertFalse(response.is_successful())

        self.assertEqual('LockFailed', error_name)