~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2006-2012 Canonical Ltd

    bencode,

    inventory_delta,

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)

 

 

            format='development-subtree')

            format='development-subtree')

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

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(

            ("success", ), local_result)

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

 

    def test_empty(self):

        backing = self.get_transport()

        dir = self.make_bzrdir('.')

        request_class = smart_dir.SmartServerBzrDirRequestGetBranches

        request = request_class(backing)

        local_result = bencode.bencode({})

        expected = smart_req.SuccessfulSmartServerResponse(

            ('success',), local_result)

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

 

 

        self.assertRaises(errors.AlreadyControlDirError,

        # Refresh branch as SetParentLocation modified it

        branch = branch.bzrdir.open_branch()

                                       'http://bar/')

        refreshed = _mod_branch.Branch.open(branch.base)

        self.assertEqual('http://bar/', refreshed.get_parent())

class TestSmartServerRepositoryReconcile(tests.TestCaseWithTransport):

 

    def test_reconcile(self):

        backing = self.get_transport()

        repo = self.make_repository('.')

        token = repo.lock_write().repository_token

        self.addCleanup(repo.unlock)

        request_class = smart_repo.SmartServerRepositoryReconcile

        request = request_class(backing)

        self.assertEqual(smart_req.SuccessfulSmartServerResponse(

            ('ok', ),

             'garbage_inventories: 0\n'

             'inconsistent_parents: 0\n'),

            request.execute('', token))

 

 

        self.assertHandlerEqual('BzrDir.checkout_metadir',

            smart_dir.SmartServerBzrDirRequestCheckoutMetaDir)

        self.assertHandlerEqual('BzrDir.get_branches',

            smart_dir.SmartServerBzrDirRequestGetBranches)

        self.assertHandlerEqual('Repository.reconcile',

            smart_repo.SmartServerRepositoryReconcile)

        self.assertHandlerEqual('VersionedFileRepository.get_inventories',

            smart_repo.SmartServerRepositoryGetInventories)

 

class TestSmartServerRepositoryGetInventories(tests.TestCaseWithTransport):

 

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

        base_inv = repository.revision_tree(base_revid).root_inventory

        inv = repository.revision_tree(revid).root_inventory

        inv_delta = inv._make_delta(base_inv)

        serializer = inventory_delta.InventoryDeltaSerializer(True, False)

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

 

    def test_single(self):

        backing = self.get_transport()

        request = smart_repo.SmartServerRepositoryGetInventories(backing)

        t = self.make_branch_and_tree('.', format='2a')

        self.addCleanup(t.lock_write().unlock)

        self.build_tree_contents([("file", "somecontents")])

        t.add(["file"], ["thefileid"])

        t.commit(rev_id='somerev', message="add file")

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

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

        self.assertTrue(response.is_successful())

        self.assertEquals(response.args, ("ok", ))

        stream = [('inventory-deltas', [

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

                self._get_serialized_inventory_delta(

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

        fmt = bzrdir.format_registry.get('2a')().repository_format

        self.assertEquals(

            "".join(response.body_stream),

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

 

    def test_empty(self):

        backing = self.get_transport()

        request = smart_repo.SmartServerRepositoryGetInventories(backing)

        t = self.make_branch_and_tree('.', format='2a')

        self.addCleanup(t.lock_write().unlock)

        self.build_tree_contents([("file", "somecontents")])

        t.add(["file"], ["thefileid"])

        t.commit(rev_id='somerev', message="add file")

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

        response = request.do_body("")

        self.assertTrue(response.is_successful())

        self.assertEquals(response.args, ("ok", ))

        self.assertEquals("".join(response.body_stream),

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