/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 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,

    )

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

from bzrlib.transport.memory import MemoryTransport

from bzrlib.transport.remote import RemoteTransport

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 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, responses, 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 = responses

        self._calls = []

        self.expecting_body = False

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

    def _get_next_response(self):

        response_tuple = self.responses.pop(0)

        if response_tuple[0][0] == 'unknown verb':

            raise errors.UnknownSmartMethod(response_tuple[0][1])

        return response_tuple

    def call(self, method, *args):

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

        return self._get_next_response()[0]

    def call_expecting_body(self, method, *args):

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

        result = self._get_next_response()

        self.expecting_body = True

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

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

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

            args, body))

        result = self._get_next_response()

        self.expecting_body = True

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

class FakeMedium(object):

    def __init__(self, client_calls):

        self._remote_is_at_least_1_2 = True

        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([(('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_SmartClient_remote_path_from_transport(tests.TestCase):

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

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

        """Assert that the result of _SmartClient.remote_path_from_transport

        is the expected value for a given client_base and transport_base.

        """

        dummy_medium = 'dummy medium'

        client = _SmartClient(dummy_medium, client_base)

        transport = get_transport(transport_base)

        result = client.remote_path_from_transport(transport)

        self.assertEqual(expected, result)

    def test_remote_path_from_transport(self):

        """_SmartClient.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 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.assertRemotePath(

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

            self.assertRemotePath(

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

class TestBzrDirOpenBranch(tests.TestCase):

    def test_branch_present(self):

        transport = MemoryTransport()

        transport.mkdir('quack')

        transport = transport.clone('quack')

        client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no', 'no'), )],

                            transport.base)

        bzrdir = RemoteBzrDir(transport, _client=client)

        result = bzrdir.open_branch()

        self.assertEqual(

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

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

            client._calls)

        self.assertIsInstance(result, RemoteBranch)

        self.assertEqual(bzrdir, result.bzrdir)

    def test_branch_missing(self):

        transport = MemoryTransport()

        transport.mkdir('quack')

        transport = transport.clone('quack')

        client = FakeClient([(('nobranch',), )], transport.base)

        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 = RemoteTransport('bzr://localhost/~hello/')

        client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no', 'no'), )],

                            transport.base)

        bzrdir = RemoteBzrDir(transport, _client=client)

        result = bzrdir.open_branch()

        self.assertEqual(

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

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

            client._calls)

    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(

            [(('ok', '', rich_response, subtree_response, external_lookup), ),],

            transport.base)

        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([

            (('unknown verb', 'RemoteRepository.find_repositoryV2'), ''),

            (('ok', '', 'no', 'no'), ''),],

            transport.base)

        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 TestBranchLastRevisionInfo(tests.TestCase):

    def test_empty_branch(self):

        # in an empty branch we decode the response properly

        transport = MemoryTransport()

        client = FakeClient([(('ok', '0', 'null:'), )], transport.base)

        transport.mkdir('quack')

        transport = transport.clone('quack')

        # we do not want bzrdir to make any remote calls

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        result = branch.last_revision_info()

        self.assertEqual(

            [('call', 'Branch.last_revision_info', ('quack/',))],

            client._calls)

        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([(('ok', '2', revid), )], transport.base)

        transport.mkdir('kwaak')

        transport = transport.clone('kwaak')

        # we do not want bzrdir to make any remote calls

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        result = branch.last_revision_info()

        self.assertEqual(

            [('call', 'Branch.last_revision_info', ('kwaak/',))],

            client._calls)

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

class TestBranchSetLastRevision(tests.TestCase):

    def test_set_empty(self):

        # set_revision_history([]) is translated to calling

        # Branch.set_last_revision(path, '') on the wire.

        transport = MemoryTransport()

        transport.mkdir('branch')

        transport = transport.clone('branch')

        client = FakeClient([

            # lock_write

            (('ok', 'branch token', 'repo token'), ),

            # set_last_revision

            (('ok',), ),

            # unlock

            (('ok',), )],

            transport.base)

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        # This is a hack to work around the problem that RemoteBranch currently

        # unnecessarily invokes _ensure_real upon a call to lock_write.

        branch._ensure_real = lambda: None

        branch.lock_write()

        client._calls = []

        result = branch.set_revision_history([])

        self.assertEqual(

            [('call', 'Branch.set_last_revision',

                ('branch/', 'branch token', 'repo token', 'null:'))],

            client._calls)

        branch.unlock()

        self.assertEqual(None, result)

    def test_set_nonempty(self):

        # set_revision_history([rev-id1, ..., rev-idN]) is translated to calling

        # Branch.set_last_revision(path, rev-idN) on the wire.

        transport = MemoryTransport()

        transport.mkdir('branch')

        transport = transport.clone('branch')

        client = FakeClient([

            # lock_write

            (('ok', 'branch token', 'repo token'), ),

            # set_last_revision

            (('ok',), ),

            # unlock

            (('ok',), )],

            transport.base)

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        # This is a hack to work around the problem that RemoteBranch currently

        # unnecessarily invokes _ensure_real upon a call to lock_write.

        branch._ensure_real = lambda: None

        # Lock the branch, reset the record of remote calls.

        branch.lock_write()

        client._calls = []

        result = branch.set_revision_history(['rev-id1', 'rev-id2'])

        self.assertEqual(

            [('call', 'Branch.set_last_revision',

                ('branch/', 'branch token', 'repo token', 'rev-id2'))],

            client._calls)

        branch.unlock()

        self.assertEqual(None, result)

    def test_no_such_revision(self):

        # A response of 'NoSuchRevision' is translated into an exception.

        client = FakeClient([

            # lock_write

            (('ok', 'branch token', 'repo token'), ),

            # set_last_revision

            (('NoSuchRevision', 'rev-id'), ),

            # unlock

            (('ok',), )])

        transport = MemoryTransport()

        transport.mkdir('branch')

        transport = transport.clone('branch')

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        branch._ensure_real = lambda: None

        branch.lock_write()

        client._calls = []

        self.assertRaises(

            errors.NoSuchRevision, branch.set_revision_history, ['rev-id'])

        branch.unlock()

class TestBranchSetLastRevisionInfo(tests.TestCase):

    def test_set_last_revision_info(self):

        # set_last_revision_info(num, 'rev-id') is translated to calling

        # Branch.set_last_revision_info(num, 'rev-id') on the wire.

        transport = MemoryTransport()

        transport.mkdir('branch')

        transport = transport.clone('branch')

        client = FakeClient([

            # lock_write

            (('ok', 'branch token', 'repo token'), ),

            # set_last_revision_info

            (('ok',), ),

            # unlock

            (('ok',), )], transport.base)

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        # This is a hack to work around the problem that RemoteBranch currently

        # unnecessarily invokes _ensure_real upon a call to lock_write.

        branch._ensure_real = lambda: None

        # Lock the branch, reset the record of remote calls.

        branch.lock_write()

        client._calls = []

        result = branch.set_last_revision_info(1234, 'a-revision-id')

        self.assertEqual(

            [('call', 'Branch.set_last_revision_info',

                ('branch/', 'branch token', 'repo token',

                 '1234', 'a-revision-id'))],

            client._calls)

        self.assertEqual(None, result)

    def test_no_such_revision(self):

        # A response of 'NoSuchRevision' is translated into an exception.

        client = FakeClient([

            # lock_write

            (('ok', 'branch token', 'repo token'), ),

            # set_last_revision_info

            (('NoSuchRevision', 'revid'), ),

            # unlock

            (('ok',), ),

            ])

        transport = MemoryTransport()

        transport.mkdir('branch')

        transport = transport.clone('branch')

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        # This is a hack to work around the problem that RemoteBranch currently

        # unnecessarily invokes _ensure_real upon a call to lock_write.

        branch._ensure_real = lambda: None

        # Lock the branch, reset the record of remote calls.

        branch.lock_write()

        client._calls = []

        self.assertRaises(

            errors.NoSuchRevision, branch.set_last_revision_info, 123, 'revid')

        branch.unlock()

    def lock_remote_branch(self, branch):

        """Trick a RemoteBranch into thinking it is locked."""

        branch._lock_mode = 'w'

        branch._lock_count = 2

        branch._lock_token = 'branch token'

        branch._repo_lock_token = 'repo token'

    def test_backwards_compatibility(self):

        """If the server does not support the Branch.set_last_revision_info

        verb (which is new in 1.4), then the client falls back to VFS methods.

        """

        # This test is a little messy.  Unlike most tests in this file, it

        # doesn't purely test what a Remote* object sends over the wire, and

        # how it reacts to responses from the wire.  It instead relies partly

        # on asserting that the RemoteBranch will call

        # self._real_branch.set_last_revision_info(...).

        # First, set up our RemoteBranch with a FakeClient that raises

        # UnknownSmartMethod, and a StubRealBranch that logs how it is called.

        transport = MemoryTransport()

        transport.mkdir('branch')

        transport = transport.clone('branch')

        client = FakeClient(

            [(('unknown verb', 'Branch.set_last_revision_info',), ),],

            transport.base)

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        class StubRealBranch(object):

            def __init__(self):

                self.calls = []

            def set_last_revision_info(self, revno, revision_id):

                self.calls.append(

                    ('set_last_revision_info', revno, revision_id))

        real_branch = StubRealBranch()

        branch._real_branch = real_branch

        self.lock_remote_branch(branch)

        # Call set_last_revision_info, and verify it behaved as expected.

        result = branch.set_last_revision_info(1234, 'a-revision-id')

        self.assertEqual(

            [('call', 'Branch.set_last_revision_info',

                ('branch/', 'branch token', 'repo token',

                 '1234', 'a-revision-id')),],

            client._calls)

        self.assertEqual(

            [('set_last_revision_info', 1234, 'a-revision-id')],

            real_branch.calls)

class TestBranchControlGetBranchConf(tests.TestCaseWithMemoryTransport):

    """Test branch.control_files api munging...

    We special case RemoteBranch.control_files.get('branch.conf') to

    call a specific API so that RemoteBranch's can intercept configuration

    file reading, allowing them to signal to the client about things like

    'email is configured for commits'.

    """

    def test_get_branch_conf(self):

        # in an empty branch we decode the response properly

        client = FakeClient([(('ok', ), 'config file body')], self.get_url())

        # we need to make a real branch because the remote_branch.control_files

        # will trigger _ensure_real.

        branch = self.make_branch('quack')

        transport = branch.bzrdir.root_transport

        # we do not want bzrdir to make any remote calls

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        result = branch.control_files.get('branch.conf')

        self.assertEqual(

            [('call_expecting_body', 'Branch.get_config_file', ('quack/',))],

            client._calls)

        self.assertEqual('config file body', result.read())

class TestBranchLockWrite(tests.TestCase):

    def test_lock_write_unlockable(self):

        transport = MemoryTransport()

        client = FakeClient([(('UnlockableTransport', ), '')], transport.base)

        transport.mkdir('quack')

        transport = transport.clone('quack')

        # we do not want bzrdir to make any remote calls

        bzrdir = RemoteBzrDir(transport, _client=False)

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

        self.assertRaises(errors.UnlockableTransport, branch.lock_write)

        self.assertEqual(

            [('call', 'Branch.lock_write', ('quack/', '', ''))],

            client._calls)

class TestTransportIsReadonly(tests.TestCase):

    def test_true(self):

        client = FakeClient([(('yes',), '')])

        transport = RemoteTransport('bzr://example.com/', medium=False,