~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

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

 

        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)

 

            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)

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

 

 

    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/')])

 

 

        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'))])]

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

            "".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")