/brz/remove-bazaar

# Copyright (C) 2006, 2007, 2008 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 remote bzrdir/branch/repo/etc

These are proxy objects which act on remote objects by sending messages

through a smart client.  The proxies are to be created when attempting to open

the object given a transport that supports smartserver rpc operations. 

These tests correspond to tests.test_smart, which exercises the server side.

"""

import bz2

from cStringIO import StringIO

from bzrlib import (

    errors,

    graph,

    pack,

    remote,

    repository,

    tests,

    urlutils,

    )

from bzrlib.branch import Branch

from bzrlib.bzrdir import BzrDir, BzrDirFormat

from bzrlib.remote import (

    RemoteBranch,

    RemoteBzrDir,

    RemoteBzrDirFormat,

    RemoteRepository,

    )

from bzrlib.revision import NULL_REVISION

from bzrlib.smart import server, medium

from bzrlib.smart.client import _SmartClient

from bzrlib.symbol_versioning import one_four

from bzrlib.transport import get_transport, http

from bzrlib.transport.memory import MemoryTransport

from bzrlib.transport.remote import RemoteTransport, RemoteTCPTransport

class BasicRemoteObjectTests(tests.TestCaseWithTransport):

    def setUp(self):

        self.transport_server = server.SmartTCPServer_for_testing

        super(BasicRemoteObjectTests, self).setUp()

        self.transport = self.get_transport()

        # make a branch that can be opened over the smart transport

        self.local_wt = BzrDir.create_standalone_workingtree('.')

    def tearDown(self):

        self.transport.disconnect()

        tests.TestCaseWithTransport.tearDown(self)

    def test_create_remote_bzrdir(self):

        b = remote.RemoteBzrDir(self.transport)

        self.assertIsInstance(b, BzrDir)

    def test_open_remote_branch(self):

        # open a standalone branch in the working directory

        b = remote.RemoteBzrDir(self.transport)

        branch = b.open_branch()

        self.assertIsInstance(branch, Branch)

    def test_remote_repository(self):

        b = BzrDir.open_from_transport(self.transport)

        repo = b.open_repository()

        revid = u'\xc823123123'.encode('utf8')

        self.assertFalse(repo.has_revision(revid))

        self.local_wt.commit(message='test commit', rev_id=revid)

        self.assertTrue(repo.has_revision(revid))

    def test_remote_branch_revision_history(self):

        b = BzrDir.open_from_transport(self.transport).open_branch()

        self.assertEqual([], b.revision_history())

        r1 = self.local_wt.commit('1st commit')

        r2 = self.local_wt.commit('1st commit', rev_id=u'\xc8'.encode('utf8'))

        self.assertEqual([r1, r2], b.revision_history())

    def test_find_correct_format(self):

        """Should open a RemoteBzrDir over a RemoteTransport"""

        fmt = BzrDirFormat.find_format(self.transport)

        self.assertTrue(RemoteBzrDirFormat

                        in BzrDirFormat._control_server_formats)

        self.assertIsInstance(fmt, remote.RemoteBzrDirFormat)

    def test_open_detected_smart_format(self):

        fmt = BzrDirFormat.find_format(self.transport)

        d = fmt.open(self.transport)

        self.assertIsInstance(d, BzrDir)

    def test_remote_branch_repr(self):

        b = BzrDir.open_from_transport(self.transport).open_branch()

        self.assertStartsWith(str(b), 'RemoteBranch(')

class FakeRemoteTransport(object):

    """This class provides the minimum support for use in place of a RemoteTransport.

    It doesn't actually transmit requests, but rather expects them to be

    handled by a FakeClient which holds canned responses.  It does not allow

    any vfs access, therefore is not suitable for testing any operation that

    will fallback to vfs access.  Backing the test by an instance of this

    class guarantees that it's - done using non-vfs operations.

    """

    _default_url = 'fakeremotetransport://host/path/'

    def __init__(self, url=None):

        if url is None:

            url = self._default_url

        self.base = url

    def __repr__(self):

        return "%r(%r)" % (self.__class__.__name__,

            self.base)

    def clone(self, relpath):

        return FakeRemoteTransport(urlutils.join(self.base, relpath))

    def get(self, relpath):

        # only get is specifically stubbed out, because it's usually the first

        # thing we do.  anything else will fail with an AttributeError.

        raise AssertionError("%r doesn't support file access to %r"

            % (self, relpath))

class FakeProtocol(object):

    """Lookalike SmartClientRequestProtocolOne allowing body reading tests."""

    def __init__(self, body, fake_client):

        self.body = body

        self._body_buffer = None

        self._fake_client = fake_client

    def read_body_bytes(self, count=-1):

        if self._body_buffer is None:

            self._body_buffer = StringIO(self.body)

        bytes = self._body_buffer.read(count)

        if self._body_buffer.tell() == len(self._body_buffer.getvalue()):

            self._fake_client.expecting_body = False

        return bytes

    def cancel_read_body(self):

        self._fake_client.expecting_body = False

    def read_streamed_body(self):

        return self.body

class FakeClient(_SmartClient):

    """Lookalike for _SmartClient allowing testing."""

    def __init__(self, fake_medium_base='fake base'):

        """Create a FakeClient.

        :param responses: A list of response-tuple, body-data pairs to be sent

            back to callers.  A special case is if the response-tuple is

            'unknown verb', then a UnknownSmartMethod will be raised for that

            call, using the second element of the tuple as the verb in the

            exception.

        """

        self.responses = []

        self._calls = []

        self.expecting_body = False

        # if non-None, this is the list of expected calls, with only the

        # method name and arguments included.  the body might be hard to

        # compute so is not included

        self._expected_calls = None

        _SmartClient.__init__(self, FakeMedium(self._calls, fake_medium_base))

    def add_expected_call(self, call_name, call_args, response_type,

        response_args, response_body=None):

        if self._expected_calls is None:

            self._expected_calls = []

        self._expected_calls.append((call_name, call_args))

        self.responses.append((response_type, response_args, response_body))

    def add_success_response(self, *args):

        self.responses.append(('success', args, None))

    def add_success_response_with_body(self, body, *args):

        self.responses.append(('success', args, body))

    def add_error_response(self, *args):

        self.responses.append(('error', args))

    def add_unknown_method_response(self, verb):

        self.responses.append(('unknown', verb))

    def finished_test(self):

        if self._expected_calls:

            raise AssertionError("%r finished but was still expecting %r"

                % (self, self._expected_calls[0]))

    def _get_next_response(self):

        try:

            response_tuple = self.responses.pop(0)

        except IndexError, e:

            raise AssertionError("%r didn't expect any more calls"

                % (self,))

        if response_tuple[0] == 'unknown':

            raise errors.UnknownSmartMethod(response_tuple[1])

        elif response_tuple[0] == 'error':

            raise errors.ErrorFromSmartServer(response_tuple[1])

        return response_tuple

    def _check_call(self, method, args):

        if self._expected_calls is None:

            # the test should be updated to say what it expects

            return

        try:

            next_call = self._expected_calls.pop(0)

        except IndexError:

            raise AssertionError("%r didn't expect any more calls "

                "but got %r%r"

                % (self, method, args,))

        if method != next_call[0] or args != next_call[1]:

            raise AssertionError("%r expected %r%r "

                "but got %r%r"

                % (self, next_call[0], next_call[1], method, args,))

    def call(self, method, *args):

        self._check_call(method, args)

        self._calls.append(('call', method, args))

        return self._get_next_response()[1]

    def call_expecting_body(self, method, *args):

        self._check_call(method, args)

        self._calls.append(('call_expecting_body', method, args))

        result = self._get_next_response()

        self.expecting_body = True

        return result[1], FakeProtocol(result[2], self)

    def call_with_body_bytes_expecting_body(self, method, args, body):

        self._check_call(method, args)

        self._calls.append(('call_with_body_bytes_expecting_body', method,

            args, body))

        result = self._get_next_response()

        self.expecting_body = True

        return result[1], FakeProtocol(result[2], self)

class FakeMedium(medium.SmartClientMedium):

    def __init__(self, client_calls, base):

        medium.SmartClientMedium.__init__(self, base)

        self._client_calls = client_calls

    def disconnect(self):

        self._client_calls.append(('disconnect medium',))

class TestVfsHas(tests.TestCase):

    def test_unicode_path(self):

        client = FakeClient('/')

        client.add_success_response('yes',)

        transport = RemoteTransport('bzr://localhost/', _client=client)

        filename = u'/hell\u00d8'.encode('utf8')

        result = transport.has(filename)

        self.assertEqual(

            [('call', 'has', (filename,))],

            client._calls)

        self.assertTrue(result)

class Test_ClientMedium_remote_path_from_transport(tests.TestCase):

    """Tests for the behaviour of client_medium.remote_path_from_transport."""

    def assertRemotePath(self, expected, client_base, transport_base):

        """Assert that the result of

        SmartClientMedium.remote_path_from_transport is the expected value for

        a given client_base and transport_base.

        """

        client_medium = medium.SmartClientMedium(client_base)

        transport = get_transport(transport_base)

        result = client_medium.remote_path_from_transport(transport)

        self.assertEqual(expected, result)

    def test_remote_path_from_transport(self):

        """SmartClientMedium.remote_path_from_transport calculates a URL for

        the given transport relative to the root of the client base URL.

        """

        self.assertRemotePath('xyz/', 'bzr://host/path', 'bzr://host/xyz')

        self.assertRemotePath(

            'path/xyz/', 'bzr://host/path', 'bzr://host/path/xyz')

    def assertRemotePathHTTP(self, expected, transport_base, relpath):

        """Assert that the result of

        HttpTransportBase.remote_path_from_transport is the expected value for

        a given transport_base and relpath of that transport.  (Note that

        HttpTransportBase is a subclass of SmartClientMedium)

        """

        base_transport = get_transport(transport_base)

        client_medium = base_transport.get_smart_medium()

        cloned_transport = base_transport.clone(relpath)

        result = client_medium.remote_path_from_transport(cloned_transport)

        self.assertEqual(expected, result)

    def test_remote_path_from_transport_http(self):

        """Remote paths for HTTP transports are calculated differently to other

        transports.  They are just relative to the client base, not the root

        directory of the host.

        """

        for scheme in ['http:', 'https:', 'bzr+http:', 'bzr+https:']:

            self.assertRemotePathHTTP(

                '../xyz/', scheme + '//host/path', '../xyz/')

            self.assertRemotePathHTTP(

                'xyz/', scheme + '//host/path', 'xyz/')

class Test_ClientMedium_remote_is_at_least(tests.TestCase):

    """Tests for the behaviour of client_medium.remote_is_at_least."""

    def test_initially_unlimited(self):

        """A fresh medium assumes that the remote side supports all

        versions.

        """

        client_medium = medium.SmartClientMedium('dummy base')

        self.assertFalse(client_medium._is_remote_before((99, 99)))

    def test__remember_remote_is_before(self):

        """Calling _remember_remote_is_before ratchets down the known remote

        version.

        """

        client_medium = medium.SmartClientMedium('dummy base')

        # Mark the remote side as being less than 1.6.  The remote side may

        # still be 1.5.

        client_medium._remember_remote_is_before((1, 6))

        self.assertTrue(client_medium._is_remote_before((1, 6)))

        self.assertFalse(client_medium._is_remote_before((1, 5)))

        # Calling _remember_remote_is_before again with a lower value works.

        client_medium._remember_remote_is_before((1, 5))

        self.assertTrue(client_medium._is_remote_before((1, 5)))

        # You cannot call _remember_remote_is_before with a larger value.

        self.assertRaises(

            AssertionError, client_medium._remember_remote_is_before, (1, 9))

class TestBzrDirOpenBranch(tests.TestCase):

    def test_branch_present(self):

        transport = MemoryTransport()

        transport.mkdir('quack')

        transport = transport.clone('quack')

        client = FakeClient(transport.base)

        client.add_expected_call(

            'BzrDir.open_branch', ('quack/',),

            'success', ('ok', ''))

        client.add_expected_call(

            'BzrDir.find_repositoryV2', ('quack/',),

            'success', ('ok', '', 'no', 'no', 'no'))

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('quack/',),

            'error', ('NotStacked',))

        bzrdir = RemoteBzrDir(transport, _client=client)

        result = bzrdir.open_branch()

        self.assertIsInstance(result, RemoteBranch)

        self.assertEqual(bzrdir, result.bzrdir)

        client.finished_test()

    def test_branch_missing(self):

        transport = MemoryTransport()

        transport.mkdir('quack')

        transport = transport.clone('quack')

        client = FakeClient(transport.base)

        client.add_error_response('nobranch')

        bzrdir = RemoteBzrDir(transport, _client=client)

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

        self.assertEqual(

            [('call', 'BzrDir.open_branch', ('quack/',))],

            client._calls)

    def test__get_tree_branch(self):

        # _get_tree_branch is a form of open_branch, but it should only ask for

        # branch opening, not any other network requests.

        calls = []

        def open_branch():

            calls.append("Called")

            return "a-branch"

        transport = MemoryTransport()

        # no requests on the network - catches other api calls being made.

        client = FakeClient(transport.base)

        bzrdir = RemoteBzrDir(transport, _client=client)

        # patch the open_branch call to record that it was called.

        bzrdir.open_branch = open_branch

        self.assertEqual((None, "a-branch"), bzrdir._get_tree_branch())

        self.assertEqual(["Called"], calls)

        self.assertEqual([], client._calls)

    def test_url_quoting_of_path(self):

        # Relpaths on the wire should not be URL-escaped.  So "~" should be

        # transmitted as "~", not "%7E".

        transport = RemoteTCPTransport('bzr://localhost/~hello/')

        client = FakeClient(transport.base)

        client.add_expected_call(

            'BzrDir.open_branch', ('~hello/',),

            'success', ('ok', ''))

        client.add_expected_call(

            'BzrDir.find_repositoryV2', ('~hello/',),

            'success', ('ok', '', 'no', 'no', 'no'))

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('~hello/',),

            'error', ('NotStacked',))

        bzrdir = RemoteBzrDir(transport, _client=client)

        result = bzrdir.open_branch()

        client.finished_test()

    def check_open_repository(self, rich_root, subtrees, external_lookup='no'):

        transport = MemoryTransport()

        transport.mkdir('quack')

        transport = transport.clone('quack')

        if rich_root:

            rich_response = 'yes'

        else:

            rich_response = 'no'

        if subtrees:

            subtree_response = 'yes'

        else:

            subtree_response = 'no'

        client = FakeClient(transport.base)

        client.add_success_response(

            'ok', '', rich_response, subtree_response, external_lookup)

        bzrdir = RemoteBzrDir(transport, _client=client)

        result = bzrdir.open_repository()

        self.assertEqual(

            [('call', 'BzrDir.find_repositoryV2', ('quack/',))],

            client._calls)

        self.assertIsInstance(result, RemoteRepository)

        self.assertEqual(bzrdir, result.bzrdir)

        self.assertEqual(rich_root, result._format.rich_root_data)

        self.assertEqual(subtrees, result._format.supports_tree_reference)

    def test_open_repository_sets_format_attributes(self):

        self.check_open_repository(True, True)

        self.check_open_repository(False, True)

        self.check_open_repository(True, False)

        self.check_open_repository(False, False)

        self.check_open_repository(False, False, 'yes')

    def test_old_server(self):

        """RemoteBzrDirFormat should fail to probe if the server version is too

        old.

        """

        self.assertRaises(errors.NotBranchError,

            RemoteBzrDirFormat.probe_transport, OldServerTransport())

class TestBzrDirOpenRepository(tests.TestCase):

    def test_backwards_compat_1_2(self):

        transport = MemoryTransport()

        transport.mkdir('quack')

        transport = transport.clone('quack')

        client = FakeClient(transport.base)

        client.add_unknown_method_response('RemoteRepository.find_repositoryV2')

        client.add_success_response('ok', '', 'no', 'no')

        bzrdir = RemoteBzrDir(transport, _client=client)

        repo = bzrdir.open_repository()

        self.assertEqual(

            [('call', 'BzrDir.find_repositoryV2', ('quack/',)),

             ('call', 'BzrDir.find_repository', ('quack/',))],

            client._calls)

class OldSmartClient(object):

    """A fake smart client for test_old_version that just returns a version one

    response to the 'hello' (query version) command.

    """

    def get_request(self):

        input_file = StringIO('ok\x011\n')

        output_file = StringIO()

        client_medium = medium.SmartSimplePipesClientMedium(

            input_file, output_file)

        return medium.SmartClientStreamMediumRequest(client_medium)

    def protocol_version(self):

        return 1

class OldServerTransport(object):

    """A fake transport for test_old_server that reports it's smart server

    protocol version as version one.

    """

    def __init__(self):

        self.base = 'fake:'

    def get_smart_client(self):

        return OldSmartClient()

class RemoteBranchTestCase(tests.TestCase):

    def make_remote_branch(self, transport, client):

        """Make a RemoteBranch using 'client' as its _SmartClient.

        A RemoteBzrDir and RemoteRepository will also be created to fill out

        the RemoteBranch, albeit with stub values for some of their attributes.

        """

        # we do not want bzrdir to make any remote calls, so use False as its

        # _client.  If it tries to make a remote call, this will fail

        # immediately.

        bzrdir = RemoteBzrDir(transport, _client=False)

        repo = RemoteRepository(bzrdir, None, _client=client)

        return RemoteBranch(bzrdir, repo, _client=client)

class TestBranchLastRevisionInfo(RemoteBranchTestCase):

    def test_empty_branch(self):

        # in an empty branch we decode the response properly

        transport = MemoryTransport()

        client = FakeClient(transport.base)

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('quack/',),

            'error', ('NotStacked',))

        client.add_expected_call(

            'Branch.last_revision_info', ('quack/',),

            'success', ('ok', '0', 'null:'))

        transport.mkdir('quack')

        transport = transport.clone('quack')

        branch = self.make_remote_branch(transport, client)

        result = branch.last_revision_info()

        client.finished_test()

        self.assertEqual((0, NULL_REVISION), result)

    def test_non_empty_branch(self):

        # in a non-empty branch we also decode the response properly

        revid = u'\xc8'.encode('utf8')

        transport = MemoryTransport()

        client = FakeClient(transport.base)

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('kwaak/',),

            'error', ('NotStacked',))

        client.add_expected_call(

            'Branch.last_revision_info', ('kwaak/',),

            'success', ('ok', '2', revid))

        transport.mkdir('kwaak')

        transport = transport.clone('kwaak')

        branch = self.make_remote_branch(transport, client)

        result = branch.last_revision_info()

        self.assertEqual((2, revid), result)

class TestBranch_get_stacked_on_url(tests.TestCaseWithMemoryTransport):

    """Test Branch._get_stacked_on_url rpc"""

    def test_get_stacked_on_invalid_url(self):

        raise tests.KnownFailure('opening a branch requires the server to open the fallback repository')

        transport = FakeRemoteTransport('fakeremotetransport:///')

        client = FakeClient(transport.base)

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('.',),

            'success', ('ok', 'file:///stacked/on'))

        bzrdir = RemoteBzrDir(transport, _client=client)

        branch = RemoteBranch(bzrdir, None, _client=client)

        result = branch.get_stacked_on_url()

        self.assertEqual(

            'file:///stacked/on', result)

    def test_backwards_compatible(self):

        # like with bzr1.6 with no Branch.get_stacked_on_url rpc

        base_branch = self.make_branch('base', format='1.6')

        stacked_branch = self.make_branch('stacked', format='1.6')

        stacked_branch.set_stacked_on_url('../base')

        client = FakeClient(self.get_url())

        client.add_expected_call(

            'BzrDir.open_branch', ('stacked/',),

            'success', ('ok', ''))

        client.add_expected_call(

            'BzrDir.find_repositoryV2', ('stacked/',),

            'success', ('ok', '', 'no', 'no', 'no'))

        # called twice, once from constructor and then again by us

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('stacked/',),

            'unknown', ('Branch.get_stacked_on_url',))

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('stacked/',),

            'unknown', ('Branch.get_stacked_on_url',))

        # this will also do vfs access, but that goes direct to the transport

        # and isn't seen by the FakeClient.

        bzrdir = RemoteBzrDir(self.get_transport('stacked'), _client=client)

        branch = bzrdir.open_branch()

        result = branch.get_stacked_on_url()

        self.assertEqual('../base', result)

        client.finished_test()

        # it's in the fallback list both for the RemoteRepository and its vfs

        # repository

        self.assertEqual(1, len(branch.repository._fallback_repositories))

        self.assertEqual(1,

            len(branch.repository._real_repository._fallback_repositories))

    def test_get_stacked_on_real_branch(self):

        base_branch = self.make_branch('base', format='1.6')

        stacked_branch = self.make_branch('stacked', format='1.6')

        stacked_branch.set_stacked_on_url('../base')

        client = FakeClient(self.get_url())

        client.add_expected_call(

            'BzrDir.open_branch', ('stacked/',),

            'success', ('ok', ''))

        client.add_expected_call(

            'BzrDir.find_repositoryV2', ('stacked/',),

            'success', ('ok', '', 'no', 'no', 'no'))

        # called twice, once from constructor and then again by us

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('stacked/',),

            'success', ('ok', '../base'))

        client.add_expected_call(

            'Branch.get_stacked_on_url', ('stacked/',),

            'success', ('ok', '../base'))

        bzrdir = RemoteBzrDir(self.get_transport('stacked'), _client=client)