/brz/remove-bazaar

# Copyright (C) 2006-2013, 2016 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""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 base64

import bz2

import tarfile

import zlib

from .. import (

    bencode,

    branch,

    config,

    controldir,

    errors,

    repository,

    tests,

    transport,

    treebuilder,

    )

from ..branch import Branch

from ..bzr import (

    bzrdir,

    inventory,

    inventory_delta,

    remote,

    versionedfile,

    vf_search,

    )

from ..bzr.bzrdir import (

    BzrDir,

    BzrDirFormat,

    )

from ..bzr import (

    RemoteBzrProber,

    )

from ..bzr.chk_serializer import chk_bencode_serializer

from ..bzr.remote import (

    RemoteBranch,

    RemoteBranchFormat,

    RemoteBzrDir,

    RemoteBzrDirFormat,

    RemoteRepository,

    RemoteRepositoryFormat,

    )

from ..bzr import groupcompress_repo, knitpack_repo

from ..revision import (

    NULL_REVISION,

    Revision,

    )

from ..sixish import (

    BytesIO,

    PY3,

    text_type,

    )

from ..bzr.smart import medium, request

from ..bzr.smart.client import _SmartClient

from ..bzr.smart.repository import (

    SmartServerRepositoryGetParentMap,

    SmartServerRepositoryGetStream_1_19,

    _stream_to_byte_stream,

    )

from . import (

    test_server,

    )

from .scenarios import load_tests_apply_scenarios

from ..transport.memory import MemoryTransport

from ..transport.remote import (

    RemoteTransport,

    RemoteSSHTransport,

    RemoteTCPTransport,

    )

load_tests = load_tests_apply_scenarios

class BasicRemoteObjectTests(tests.TestCaseWithTransport):

    scenarios = [

        ('HPSS-v2',

            {'transport_server':

                test_server.SmartTCPServer_for_testing_v2_only}),

        ('HPSS-v3',

            {'transport_server': test_server.SmartTCPServer_for_testing})]

    def setUp(self):

        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('.')

        self.addCleanup(self.transport.disconnect)

    def test_create_remote_bzrdir(self):

        b = remote.RemoteBzrDir(self.transport, RemoteBzrDirFormat())

        self.assertIsInstance(b, BzrDir)

    def test_open_remote_branch(self):

        # open a standalone branch in the working directory

        b = remote.RemoteBzrDir(self.transport, RemoteBzrDirFormat())

        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_find_correct_format(self):

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

        fmt = BzrDirFormat.find_format(self.transport)

        self.assertTrue(RemoteBzrProber

                        in controldir.ControlDirFormat._server_probers)

        self.assertIsInstance(fmt, 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(')

    def test_remote_bzrdir_repr(self):

        b = BzrDir.open_from_transport(self.transport)

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

    def test_remote_branch_format_supports_stacking(self):

        t = self.transport

        self.make_branch('unstackable', format='pack-0.92')

        b = BzrDir.open_from_transport(t.clone('unstackable')).open_branch()

        self.assertFalse(b._format.supports_stacking())

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

        b = BzrDir.open_from_transport(t.clone('stackable')).open_branch()

        self.assertTrue(b._format.supports_stacking())

    def test_remote_repo_format_supports_external_references(self):

        t = self.transport

        bd = self.make_controldir('unstackable', format='pack-0.92')

        r = bd.create_repository()

        self.assertFalse(r._format.supports_external_lookups)

        r = BzrDir.open_from_transport(

            t.clone('unstackable')).open_repository()

        self.assertFalse(r._format.supports_external_lookups)

        bd = self.make_controldir('stackable', format='1.9')

        r = bd.create_repository()

        self.assertTrue(r._format.supports_external_lookups)

        r = BzrDir.open_from_transport(t.clone('stackable')).open_repository()

        self.assertTrue(r._format.supports_external_lookups)

    def test_remote_branch_set_append_revisions_only(self):

        # Make a format 1.9 branch, which supports append_revisions_only

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

        branch.set_append_revisions_only(True)

        config = branch.get_config_stack()

        self.assertEqual(

            True, config.get('append_revisions_only'))

        branch.set_append_revisions_only(False)

        config = branch.get_config_stack()

        self.assertEqual(

            False, config.get('append_revisions_only'))

    def test_remote_branch_set_append_revisions_only_upgrade_reqd(self):

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

        self.assertRaises(

            errors.UpgradeRequired, branch.set_append_revisions_only, True)

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 = BytesIO(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."""

        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. If a call is None, that call can

        # be anything.

        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((b'success', args, None))

    def add_success_response_with_body(self, body, *args):

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

        if self._expected_calls is not None:

            self._expected_calls.append(None)

    def add_error_response(self, *args):

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

    def add_unknown_method_response(self, verb):

        self.responses.append((b'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:

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

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

            raise errors.UnknownSmartMethod(response_tuple[1])

        elif response_tuple[0] == b'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 next_call is None:

            return

        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(self, method, args, body):

        self._check_call(method, args)

        self._calls.append(('call_with_body_bytes', method, args, body))

        result = self._get_next_response()

        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)

    def call_with_body_stream(self, args, stream):

        # Explicitly consume the stream before checking for an error, because

        # that's what happens a real medium.

        stream = list(stream)

        self._check_call(args[0], args[1:])

        self._calls.append(

            ('call_with_body_stream', args[0], args[1:], stream))

        result = self._get_next_response()

        # The second value returned from call_with_body_stream is supposed to

        # be a response_handler object, but so far no tests depend on that.

        response_handler = None

        return result[1], response_handler

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(b'yes',)

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

        filename = u'/hell\u00d8'

        if PY3:

            result = transport.has(filename)

        else:

            result = transport.has(filename.encode('utf-8'))

        self.assertEqual(

            [('call', b'has', (filename.encode('utf-8'),))],

            client._calls)

        self.assertTrue(result)

class TestRemote(tests.TestCaseWithMemoryTransport):

    def get_branch_format(self):

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

        return reference_bzrdir_format.get_branch_format()

    def get_repo_format(self):

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

        return reference_bzrdir_format.repository_format

    def assertFinished(self, fake_client):

        """Assert that all of a FakeClient's expected calls have occurred."""

        fake_client.finished_test()

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)

        t = transport.get_transport(transport_base)

        result = client_medium.remote_path_from_transport(t)

        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 = 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)))

        # If you call _remember_remote_is_before with a higher value it logs a

        # warning, and continues to remember the lower value.

        self.assertNotContainsRe(self.get_log(), '_remember_remote_is_before')

        client_medium._remember_remote_is_before((1, 9))

        self.assertContainsRe(self.get_log(), '_remember_remote_is_before')

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

class TestBzrDirCloningMetaDir(TestRemote):

    def test_backwards_compat(self):

        self.setup_smart_server_with_call_log()

        a_dir = self.make_controldir('.')

        self.reset_smart_call_log()

        verb = b'BzrDir.cloning_metadir'

        self.disable_verb(verb)

        a_dir.cloning_metadir()

        call_count = len([call for call in self.hpss_calls if

                          call.call.method == verb])

        self.assertEqual(1, call_count)

    def test_branch_reference(self):

        transport = self.get_transport('quack')

        referenced = self.make_branch('referenced')

        expected = referenced.controldir.cloning_metadir()

        client = FakeClient(transport.base)

        client.add_expected_call(

            b'BzrDir.cloning_metadir', (b'quack/', b'False'),

            b'error', (b'BranchReference',)),

        client.add_expected_call(

            b'BzrDir.open_branchV3', (b'quack/',),

            b'success', (b'ref', self.get_url('referenced').encode('utf-8'))),

        a_controldir = RemoteBzrDir(transport, RemoteBzrDirFormat(),

                                    _client=client)

        result = a_controldir.cloning_metadir()

        # We should have got a control dir matching the referenced branch.

        self.assertEqual(bzrdir.BzrDirMetaFormat1, type(result))

        self.assertEqual(expected._repository_format,

                         result._repository_format)

        self.assertEqual(expected._branch_format, result._branch_format)

        self.assertFinished(client)

    def test_current_server(self):

        transport = self.get_transport('.')

        transport = transport.clone('quack')

        self.make_controldir('quack')

        client = FakeClient(transport.base)

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

        control_name = reference_bzrdir_format.network_name()

        client.add_expected_call(

            b'BzrDir.cloning_metadir', (b'quack/', b'False'),

            b'success', (control_name, b'', (b'branch', b''))),

        a_controldir = RemoteBzrDir(transport, RemoteBzrDirFormat(),

                                    _client=client)

        result = a_controldir.cloning_metadir()

        # We should have got a reference control dir with default branch and

        # repository formats.

        # This pokes a little, just to be sure.

        self.assertEqual(bzrdir.BzrDirMetaFormat1, type(result))

        self.assertEqual(None, result._repository_format)

        self.assertEqual(None, result._branch_format)

        self.assertFinished(client)

    def test_unknown(self):

        transport = self.get_transport('quack')

        referenced = self.make_branch('referenced')

        referenced.controldir.cloning_metadir()

        client = FakeClient(transport.base)

        client.add_expected_call(

            b'BzrDir.cloning_metadir', (b'quack/', b'False'),

            b'success', (b'unknown', b'unknown', (b'branch', b''))),

        a_controldir = RemoteBzrDir(transport, RemoteBzrDirFormat(),

                                    _client=client)

        self.assertRaises(errors.UnknownFormatError,

                          a_controldir.cloning_metadir)

class TestBzrDirCheckoutMetaDir(TestRemote):

    def test__get_checkout_format(self):

        transport = MemoryTransport()

        client = FakeClient(transport.base)

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

        control_name = reference_bzrdir_format.network_name()

        client.add_expected_call(

            b'BzrDir.checkout_metadir', (b'quack/', ),

            b'success', (control_name, b'', b''))

        transport.mkdir('quack')

        transport = transport.clone('quack')

        a_controldir = RemoteBzrDir(transport, RemoteBzrDirFormat(),

                                    _client=client)

        result = a_controldir.checkout_metadir()

        # We should have got a reference control dir with default branch and

        # repository formats.

        self.assertEqual(bzrdir.BzrDirMetaFormat1, type(result))

        self.assertEqual(None, result._repository_format)

        self.assertEqual(None, result._branch_format)

        self.assertFinished(client)

    def test_unknown_format(self):

        transport = MemoryTransport()

        client = FakeClient(transport.base)

        client.add_expected_call(

            b'BzrDir.checkout_metadir', (b'quack/',),

            b'success', (b'dontknow', b'', b''))

        transport.mkdir('quack')

        transport = transport.clone('quack')

        a_controldir = RemoteBzrDir(transport, RemoteBzrDirFormat(),

                                    _client=client)

        self.assertRaises(errors.UnknownFormatError,

                          a_controldir.checkout_metadir)

        self.assertFinished(client)

class TestBzrDirGetBranches(TestRemote):

    def test_get_branches(self):

        transport = MemoryTransport()

        client = FakeClient(transport.base)

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

        branch_name = reference_bzrdir_format.get_branch_format().network_name()

        client.add_success_response_with_body(

            bencode.bencode({

                b"foo": (b"branch", branch_name),

                b"": (b"branch", branch_name)}), b"success")

        client.add_success_response(

            b'ok', b'', b'no', b'no', b'no',

            reference_bzrdir_format.repository_format.network_name())

        client.add_error_response(b'NotStacked')

        client.add_success_response(

            b'ok', b'', b'no', b'no', b'no',

            reference_bzrdir_format.repository_format.network_name())

        client.add_error_response(b'NotStacked')

        transport.mkdir('quack')