~bzr-pqm/bzr/bzr.dev

#

# 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

from bzrlib import (

    bencode,

    branch as _mod_branch,

    errors,

    gpg,

    inventory_delta,

 

    def setUp(self):

        self.vfs_transport_factory = memory.MemoryServer

        self._chroot_server = None

 

    def get_transport(self, relpath=None):

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

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

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

        bytes = smart_repo._stream_to_byte_stream(stream, fmt)

        streams = []

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

        self.make_bzrdir('.')

        request_class = smart_dir.SmartServerRequestCreateRepository

        request = request_class(backing)

        reference_format = reference_bzrdir_format.repository_format

        network_name = reference_format.network_name()

        expected = smart_req.SuccessfulSmartServerResponse(

        backing = self.get_transport()

        request = self._request_class(backing)

        result = self._make_repository_and_result(

        # check the test will be valid

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

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

        backing = self.get_transport()

        request = self._request_class(backing)

        result = self._make_repository_and_result(

        # check the test will be valid

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

 

 

        request = smart_dir.SmartServerRequestInitializeBzrDir(backing)

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

            request.execute(''))

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

        # default formart.

        self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)

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

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

                            'False'))

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

        # default format.

        self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)

        self.assertEqual(

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

            request.do_body('foo bar baz'))

            branch.control_transport.get_bytes('branch.conf'),

            'foo bar baz')

        branch.unlock()

        response = request.do_chunk(tag_bytes)

        self.assertEqual(None, response)

        response = request.do_end()

            smart_req.SuccessfulSmartServerResponse(()), response)

        base_branch.unlock()

 

    """Base test case for verbs that implement set_last_revision."""

 

    def setUp(self):

        backing_transport = self.get_transport()

        self.request = self.request_class(backing_transport)

        self.tree = self.make_branch_and_memory_tree('.')

        base_branch = self.make_branch('base')

        request = smart_branch.SmartServerBranchGetParent(self.get_transport())

        response = request.execute('base')

            smart_req.SuccessfulSmartServerResponse(('',)), response)

 

    def test_get_parent_something(self):

        base_branch.set_parent(self.get_url('foo'))

        request = smart_branch.SmartServerBranchGetParent(self.get_transport())

        response = request.execute('base')

            smart_req.SuccessfulSmartServerResponse(("../foo",)),

            response)

 

        finally:

            branch.unlock()

        self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), response)

        self.assertEqual(None, branch.get_parent())

 

    def test_set_parent_something(self):

        branch_token, repo_token = self.get_lock_tokens(branch)

        try:

            response = request.execute('base', branch_token, repo_token,

        finally:

            branch.unlock()

        self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), response)

 

 

class TestSmartServerBranchRequestGetTagsBytes(

        request = smart_branch.SmartServerBranchGetTagsBytes(

            self.get_transport())

        response = request.execute('base')

            smart_req.SuccessfulSmartServerResponse(('',)), response)

 

 

        request = smart_branch.SmartServerBranchRequestGetStackedOnURL(

            self.get_transport())

        response = request.execute('stacked')

            smart_req.SmartServerResponse(('ok', '../base')),

            response)

 

 

class TestSmartServerBranchRequestLockWrite(TestLockedBranch):

 

    def test_lock_write_on_unlocked_branch(self):

        backing = self.get_transport()

        request = smart_branch.SmartServerBranchRequestLockWrite(backing)

 

class TestSmartServerBranchRequestGetPhysicalLockStatus(TestLockedBranch):

 

    def test_true(self):

        backing = self.get_transport()

        request = smart_branch.SmartServerBranchRequestGetPhysicalLockStatus(

            backing)

        branch = self.make_branch('.')

        branch_token, repo_token = self.get_lock_tokens(branch)

        response = request.execute('')

        self.assertEqual(

            smart_req.SmartServerResponse(('yes',)), response)

        request = smart_branch.SmartServerBranchRequestGetPhysicalLockStatus(

            backing)

        branch = self.make_branch('.')

        response = request.execute('')

        self.assertEqual(

            smart_req.SmartServerResponse(('no',)), response)

 

class TestSmartServerBranchRequestUnlock(TestLockedBranch):

 

    def test_unlock_on_locked_branch_and_repo(self):

        backing = self.get_transport()

        request = smart_branch.SmartServerBranchRequestUnlock(backing)

        backing = self.get_transport()

        request = smart_repo.SmartServerRepositoryAllRevisionIds(backing)

        self.make_repository('.')

            smart_req.SuccessfulSmartServerResponse(("ok", ), ""),

            request.execute(''))

 

        tree.commit(rev_id='origineel', message="message")

        tree.commit(rev_id='nog-een-revisie', message="message")

        tree.unlock()

            smart_req.SuccessfulSmartServerResponse(("ok", ),

                "origineel\nnog-een-revisie"),

            request.execute(''))

        response = request.do_body("rev1\nrev2")

        self.assertTrue(response.is_successful())

        # Format 2a uses serializer format 10

 

        self.addCleanup(tree.branch.lock_read().unlock)

        entries = [zlib.compress(record.get_bytes_as("fulltext")) for record in

        response = request.do_body("rev1\nrev2")

        self.assertTrue(response.is_successful())

        # Format 2a uses serializer format 10

 

        contents = "".join(response.body_stream)

 

 

class GetStreamTestBase(tests.TestCaseWithMemoryTransport):

        self.assertIs(None, request.execute(''))

        response = request.do_body("thefileid\0somerev\n")

        self.assertTrue(response.is_successful())

            "ok\x000\n" + zlib.compress("somecontents"))

 

    def test_missing(self):

        self.assertIs(None, request.execute(''))

        response = request.do_body("thefileid\0revision\n")

        self.assertTrue(response.is_successful())

            "absent\x00thefileid\x00revision\x000\n")

 

 

 

class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):

 

    def test_unlock_on_locked_repo(self):

        backing = self.get_transport()

        request = smart_repo.SmartServerRepositoryUnlock(backing)

    def test_without_write_lock(self):

        backing = self.get_transport()

        repo = self.make_repository('.')

        request_class = smart_repo.SmartServerRepositoryGetPhysicalLockStatus

        request = request_class(backing)

        self.assertEqual(smart_req.SuccessfulSmartServerResponse(('no',)),

            None)

        self.server._sockname = ('example.com', 42)

        self.server.run_server_started_hooks()

            [([self.get_transport().base], 'bzr://example.com:42/')])

            [([self.get_transport().base], self.server)])

 

    def test_run_server_started_hooks_ipv6(self):

            lambda backing_urls, url: started_calls.append((backing_urls, url)),

            None)

        self.server.run_server_started_hooks()

                [([self.get_transport().base], 'bzr://:::42/')])

 

    def test_run_server_stopped_hooks(self):

            lambda backing_urls, url: stopped_calls.append((backing_urls, url)),

            None)

        self.server.run_server_stopped_hooks()

            [([self.get_transport().base], 'bzr://example.com:42/')])

 

 

class TestSmartServerRepositoryGetInventories(tests.TestCaseWithTransport):

 

    def _get_serialized_inventory_delta(self, repository, base_revid, revid):

        inv_delta = inv._make_delta(base_inv)

        serializer = inventory_delta.InventoryDeltaSerializer(True, False)

        return "".join(serializer.delta_to_lines(base_revid, revid, inv_delta))

        self.assertIs(None, request.execute('', 'unordered'))

        response = request.do_body("somerev\n")

        self.assertTrue(response.is_successful())

        stream = [('inventory-deltas', [

            versionedfile.FulltextContentFactory('somerev', None, None,

                self._get_serialized_inventory_delta(

                    t.branch.repository, 'null:', 'somerev'))])]

            "".join(response.body_stream),

            "".join(smart_repo._stream_to_byte_stream(stream, fmt)))

 

        self.assertIs(None, request.execute('', 'unordered'))

        response = request.do_body("")

        self.assertTrue(response.is_successful())

            "Bazaar pack format 1 (introduced in 0.18)\nB54\n\nBazaar repository format 2a (needs bzr 1.16 or later)\nE")