/brz/remove-bazaar

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

import zlib

from breezy import (

    bencode,

    branch as _mod_branch,

    controldir,

    errors,

    gpg,

    tests,

    transport,

    urlutils,

    )

from breezy.bzr import (

    branch as _mod_bzrbranch,

    inventory_delta,

    versionedfile,

    )

from breezy.bzr.smart import (

    branch as smart_branch,

    bzrdir as smart_dir,

    repository as smart_repo,

    packrepository as smart_packrepo,

    request as smart_req,

    server,

    vfs,

    )

from breezy.testament import Testament

from breezy.tests import test_server

from breezy.transport import (

    chroot,

    memory,

    )

def load_tests(loader, standard_tests, pattern):

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

    # FindRepository tests.

    scenarios = [

        ("find_repository", {

            "_request_class": smart_dir.SmartServerRequestFindRepositoryV1}),

        ("find_repositoryV2", {

            "_request_class": smart_dir.SmartServerRequestFindRepositoryV2}),

        ("find_repositoryV3", {

            "_request_class": smart_dir.SmartServerRequestFindRepositoryV3}),

        ]

    to_adapt, result = tests.split_suite_by_re(standard_tests,

        "TestSmartServerRequestFindRepository")

    v2_only, v1_and_2 = tests.split_suite_by_re(to_adapt,

        "_v2")

    tests.multiply_tests(v1_and_2, scenarios, result)

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

    # since.

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

    return result

class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):

    def setUp(self):

        self.vfs_transport_factory = memory.MemoryServer

        super(TestCaseWithChrootedTransport, self).setUp()

        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.start_server(self._chroot_server)

        t = transport.get_transport_from_url(self._chroot_server.get_url())

        if relpath is not None:

            t = t.clone(relpath)

        return t

class TestCaseWithSmartMedium(tests.TestCaseWithMemoryTransport):

    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.overrideAttr(self, "transport_server", self.make_transport_server)

    def make_transport_server(self):

        return test_server.SmartTCPServer_for_testing('-' + self.id())

    def get_smart_medium(self):

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

        return self.get_transport().get_smart_medium()

class TestByteStreamToStream(tests.TestCase):

    def test_repeated_substreams_same_kind_are_one_stream(self):

        # Make a stream - an iterable of bytestrings.

        stream = [('text', [versionedfile.FulltextContentFactory(('k1',), None,

            None, 'foo')]), ('text', [

            versionedfile.FulltextContentFactory(('k2',), None, None, 'bar')])]

        fmt = controldir.format_registry.get('pack-0.92')().repository_format

        bytes = smart_repo._stream_to_byte_stream(stream, fmt)

        streams = []

        # Iterate the resulting iterable; checking that we get only one stream

        # out.

        fmt, stream = smart_repo._byte_stream_to_stream(bytes)

        for kind, substream in stream:

            streams.append((kind, list(substream)))

        self.assertLength(1, streams)

        self.assertLength(2, streams[0][1])

class TestSmartServerResponse(tests.TestCase):

    def test__eq__(self):

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

            smart_req.SmartServerResponse((b'ok', )))

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

            smart_req.SmartServerResponse((b'ok', ), 'body'))

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

            smart_req.SmartServerResponse((b'notok', )))

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

            smart_req.SmartServerResponse((b'ok', )))

        self.assertNotEqual(None,

            smart_req.SmartServerResponse((b'ok', )))

    def test__str__(self):

        """SmartServerResponses can be stringified."""

        self.assertIn(

            str(smart_req.SuccessfulSmartServerResponse((b'args',), b'body')),

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

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

        self.assertIn(

            str(smart_req.FailedSmartServerResponse((b'args',), b'body')),

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

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

class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):

    def test_translate_client_path(self):

        transport = self.get_transport()

        request = smart_req.SmartServerRequest(transport, 'foo/')

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

        self.assertRaises(

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

        e_acute = u'\N{LATIN SMALL LETTER E WITH ACUTE}'.encode('utf-8')

        self.assertEqual('./' + urlutils.escape(e_acute),

                         request.translate_client_path('foo/' + e_acute))

    def test_translate_client_path_vfs(self):

        """VfsRequests receive escaped paths rather than raw UTF-8."""

        transport = self.get_transport()

        request = vfs.VfsRequest(transport, 'foo/')

        e_acute = u'\N{LATIN SMALL LETTER E WITH ACUTE}'.encode('utf-8')

        escaped = urlutils.escape('foo/' + e_acute)

        self.assertEqual('./' + urlutils.escape(e_acute),

                         request.translate_client_path(escaped))

    def test_transport_from_client_path(self):

        transport = self.get_transport()

        request = smart_req.SmartServerRequest(transport, 'foo/')

        self.assertEqual(

            transport.base,

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

class TestSmartServerBzrDirRequestCloningMetaDir(

    tests.TestCaseWithMemoryTransport):

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

    def test_cloning_metadir(self):

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

        backing = self.get_transport()

        dir = self.make_controldir('.')

        local_result = dir.cloning_metadir()

        request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse(

            (local_result.network_name(),

            local_result.repository_format.network_name(),

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

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

    def test_cloning_metadir_reference(self):

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

        backing = self.get_transport()

        referenced_branch = self.make_branch('referenced')

        dir = self.make_controldir('.')

        local_result = dir.cloning_metadir()

        reference = _mod_bzrbranch.BranchReferenceFormat().initialize(

            dir, target_branch=referenced_branch)

        reference_url = _mod_bzrbranch.BranchReferenceFormat().get_reference(dir)

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

        # if the referenced branch isn't reachable.

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

        request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir

        request = request_class(backing)

        expected = smart_req.FailedSmartServerResponse((b'BranchReference',))

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

class TestSmartServerBzrDirRequestCloningMetaDir(

    tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.checkout_metadir."""

    def test_checkout_metadir(self):

        backing = self.get_transport()

        request = smart_dir.SmartServerBzrDirRequestCheckoutMetaDir(

            backing)

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

        response = request.execute('')

        self.assertEqual(

            smart_req.SmartServerResponse(

                ('Bazaar-NG meta directory, format 1\n',

                 'Bazaar repository format 2a (needs bzr 1.16 or later)\n',

                 'Bazaar Branch Format 7 (needs bzr 1.6)\n')),

            response)

class TestSmartServerBzrDirRequestDestroyBranch(

    tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.destroy_branch."""

    def test_destroy_branch_default(self):

        """The default branch can be removed."""

        backing = self.get_transport()

        dir = self.make_branch('.').controldir

        request_class = smart_dir.SmartServerBzrDirRequestDestroyBranch

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((b'ok',))

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

    def test_destroy_branch_named(self):

        """A named branch can be removed."""

        backing = self.get_transport()

        dir = self.make_repository('.', format="development-colo").controldir

        dir.create_branch(name="branchname")

        request_class = smart_dir.SmartServerBzrDirRequestDestroyBranch

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((b'ok',))

        self.assertEqual(expected, request.execute('', "branchname"))

    def test_destroy_branch_missing(self):

        """An error is raised if the branch didn't exist."""

        backing = self.get_transport()

        dir = self.make_controldir('.', format="development-colo")

        request_class = smart_dir.SmartServerBzrDirRequestDestroyBranch

        request = request_class(backing)

        expected = smart_req.FailedSmartServerResponse((b'nobranch',), None)

        self.assertEqual(expected, request.execute('', "branchname"))

class TestSmartServerBzrDirRequestHasWorkingTree(

    tests.TestCaseWithTransport):

    """Tests for BzrDir.has_workingtree."""

    def test_has_workingtree_yes(self):

        """A working tree is present."""

        backing = self.get_transport()

        dir = self.make_branch_and_tree('.').controldir

        request_class = smart_dir.SmartServerBzrDirRequestHasWorkingTree

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((b'yes',))

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

    def test_has_workingtree_no(self):

        """A working tree is missing."""

        backing = self.get_transport()

        dir = self.make_controldir('.')

        request_class = smart_dir.SmartServerBzrDirRequestHasWorkingTree

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((b'no',))

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

class TestSmartServerBzrDirRequestDestroyRepository(

    tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.destroy_repository."""

    def test_destroy_repository_default(self):

        """The repository can be removed."""

        backing = self.get_transport()

        dir = self.make_repository('.').controldir

        request_class = smart_dir.SmartServerBzrDirRequestDestroyRepository

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((b'ok',))

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

    def test_destroy_repository_missing(self):

        """An error is raised if the repository didn't exist."""

        backing = self.get_transport()

        dir = self.make_controldir('.')

        request_class = smart_dir.SmartServerBzrDirRequestDestroyRepository

        request = request_class(backing)

        expected = smart_req.FailedSmartServerResponse(

            ('norepository',), None)

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

class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):

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

    def test_makes_repository(self):

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

        backing = self.get_transport()

        self.make_controldir('.')

        request_class = smart_dir.SmartServerRequestCreateRepository

        request = request_class(backing)

        reference_bzrdir_format = controldir.format_registry.get('pack-0.92')()

        reference_format = reference_bzrdir_format.repository_format

        network_name = reference_format.network_name()

        expected = smart_req.SuccessfulSmartServerResponse(

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

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

class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):

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

    def test_no_repository(self):

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

        backing = self.get_transport()

        request = self._request_class(backing)

        self.make_controldir('.')

        self.assertEqual(smart_req.SmartServerResponse((b'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(b''))

        self.make_controldir('subdir')

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

            request.execute(b'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 = b'yes'

        else:

            rich_root = b'no'

        if repo._format.supports_tree_reference:

            subtrees = b'yes'

        else:

            subtrees = b'no'

        if repo._format.supports_external_lookups:

            external = b'yes'

        else:

            external = b'no'

        if (smart_dir.SmartServerRequestFindRepositoryV3 ==

            self._request_class):

            return smart_req.SuccessfulSmartServerResponse(

                (b'ok', b'', rich_root, subtrees, external,

                 repo._format.network_name()))

        elif (smart_dir.SmartServerRequestFindRepositoryV2 ==

            self._request_class):

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

            # repositories.

            return smart_req.SuccessfulSmartServerResponse(

                (b'ok', b'', rich_root, subtrees, external))

        else:

            return smart_req.SuccessfulSmartServerResponse(

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

        self.make_controldir('subdir')

        result2 = smart_req.SmartServerResponse(

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

        self.assertEqual(result2,

            request.execute(b'subdir'))

        self.make_controldir('subdir/deeper')

        result3 = smart_req.SmartServerResponse(

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

        self.assertEqual(result3,

            request.execute(b'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='development-subtree')

        # check the test will be valid

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

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

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

    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='development-subtree')

        # check the test will be valid

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

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

class TestSmartServerBzrDirRequestGetConfigFile(

    tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.get_config_file."""

    def test_present(self):

        backing = self.get_transport()

        dir = self.make_controldir('.')

        dir.get_config().set_default_stack_on("/")

        local_result = dir._get_config()._get_config_file().read()

        request_class = smart_dir.SmartServerBzrDirRequestConfigFile

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((), local_result)

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

    def test_missing(self):

        backing = self.get_transport()

        dir = self.make_controldir('.')

        request_class = smart_dir.SmartServerBzrDirRequestConfigFile

        request = request_class(backing)

        expected = smart_req.SuccessfulSmartServerResponse((), '')

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

class TestSmartServerBzrDirRequestGetBranches(

    tests.TestCaseWithMemoryTransport):

    """Tests for BzrDir.get_branches."""

    def test_simple(self):

        backing = self.get_transport()

        branch = self.make_branch('.')

        request_class = smart_dir.SmartServerBzrDirRequestGetBranches

        request = request_class(backing)

        local_result = bencode.bencode(

            {"": ("branch", branch._format.network_name())})

        expected = smart_req.SuccessfulSmartServerResponse(

            (b"success", ), local_result)

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

    def test_empty(self):

        backing = self.get_transport()

        dir = self.make_controldir('.')

        request_class = smart_dir.SmartServerBzrDirRequestGetBranches

        request = request_class(backing)

        local_result = bencode.bencode({})

        expected = smart_req.SuccessfulSmartServerResponse(

            (b'success',), local_result)

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

class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):

    def test_empty_dir(self):

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

        backing = self.get_transport()

        request = smart_dir.SmartServerRequestInitializeBzrDir(backing)

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

            request.execute(b''))

        made_dir = controldir.ControlDir.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_dir.SmartServerRequestInitializeBzrDir(backing)

        self.assertRaises(errors.NoSuchFile,

            request.execute, b'subdir')

    def test_initialized_dir(self):

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

        backing = self.get_transport()

        request = smart_dir.SmartServerRequestInitializeBzrDir(backing)

        self.make_controldir('subdir')

        self.assertRaises(errors.AlreadyControlDirError,

            request.execute, b'subdir')

class TestSmartServerRequestBzrDirInitializeEx(

    tests.TestCaseWithMemoryTransport):

    """Basic tests for BzrDir.initialize_ex_1.16 in the smart server.

    The main unit tests in test_bzrdir exercise the API comprehensively.

    """

    def test_empty_dir(self):

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

        backing = self.get_transport()

        name = self.make_controldir('reference')._format.network_name()

        request = smart_dir.SmartServerRequestBzrDirInitializeEx(backing)

        self.assertEqual(

            smart_req.SmartServerResponse((b'', '', '', '', '', '', name,

                                           'False', '', '', '')),

            request.execute(name, b'', b'True', b'False', b'False', b'', b'', b'', b'',

                            b'False'))

        made_dir = controldir.ControlDir.open_from_transport(backing)

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

        # default format.

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

        name = self.make_controldir('reference')._format.network_name()

        request = smart_dir.SmartServerRequestBzrDirInitializeEx(backing)

        self.assertRaises(errors.NoSuchFile, request.execute, name,

            b'subdir/dir', b'False', b'False', b'False', b'', b'', b'', b'', b'False')

    def test_initialized_dir(self):

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

        backing = self.get_transport()

        name = self.make_controldir('reference')._format.network_name()

        request = smart_dir.SmartServerRequestBzrDirInitializeEx(backing)

        self.make_controldir('subdir')

        self.assertRaises(errors.FileExists, request.execute, name, b'subdir',

            b'False', b'False', b'False', b'', b'', b'', b'', b'False')

class TestSmartServerRequestOpenBzrDir(tests.TestCaseWithMemoryTransport):

    def test_no_directory(self):

        backing = self.get_transport()

        request = smart_dir.SmartServerRequestOpenBzrDir(backing)

        self.assertEqual(smart_req.SmartServerResponse((b'no', )),

            request.execute(b'does-not-exist'))

    def test_empty_directory(self):

        backing = self.get_transport()

        backing.mkdir('empty')

        request = smart_dir.SmartServerRequestOpenBzrDir(backing)

        self.assertEqual(smart_req.SmartServerResponse((b'no', )),