~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_remote.py

• browse files at revision 4706
• compare with another revision
• download diff
• download tarball
• view history from revision 4706

• reverse the comparison (4706 to 4443.2.5)
• stop comparing with revision 4443.2.5

Show diffs side-by-side

added

removed

bzrlib/tests/test_remote.py

280

280

        self.expecting_body = True

281

281

        return result[1], FakeProtocol(result[2], self)

282

282

 

 

283

    def call_with_body_bytes(self, method, args, body):

 

284

        self._check_call(method, args)

 

285

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

 

286

        result = self._get_next_response()

 

287

        return result[1], FakeProtocol(result[2], self)

 

288

 

283

289

    def call_with_body_bytes_expecting_body(self, method, args, body):

284

290

        self._check_call(method, args)

285

291

        self._calls.append(('call_with_body_bytes_expecting_body', method,

683

689

        # fallback all the way to the first version.

684

690

        reference_format = self.get_repo_format()

685

691

        network_name = reference_format.network_name()

686

 

        client = FakeClient('bzr://example.com/')

 

692

        server_url = 'bzr://example.com/'

 

693

        self.permit_url(server_url)

 

694

        client = FakeClient(server_url)

687

695

        client.add_unknown_method_response('BzrDir.find_repositoryV3')

688

696

        client.add_unknown_method_response('BzrDir.find_repositoryV2')

689

697

        client.add_success_response('ok', '', 'no', 'no')

695

703

            reference_format.get_format_string(), 'ok')

696

704

        # PackRepository wants to do a stat

697

705

        client.add_success_response('stat', '0', '65535')

698

 

        remote_transport = RemoteTransport('bzr://example.com/quack/', medium=False,

 

706

        remote_transport = RemoteTransport(server_url + 'quack/', medium=False,

699

707

            _client=client)

700

708

        bzrdir = RemoteBzrDir(remote_transport, remote.RemoteBzrDirFormat(),

701

709

            _client=client)

715

723

        # fallback to find_repositoryV2

716

724

        reference_format = self.get_repo_format()

717

725

        network_name = reference_format.network_name()

718

 

        client = FakeClient('bzr://example.com/')

 

726

        server_url = 'bzr://example.com/'

 

727

        self.permit_url(server_url)

 

728

        client = FakeClient(server_url)

719

729

        client.add_unknown_method_response('BzrDir.find_repositoryV3')

720

730

        client.add_success_response('ok', '', 'no', 'no', 'no')

721

731

        # A real repository instance will be created to determine the network

726

736

            reference_format.get_format_string(), 'ok')

727

737

        # PackRepository wants to do a stat

728

738

        client.add_success_response('stat', '0', '65535')

729

 

        remote_transport = RemoteTransport('bzr://example.com/quack/', medium=False,

 

739

        remote_transport = RemoteTransport(server_url + 'quack/', medium=False,

730

740

            _client=client)

731

741

        bzrdir = RemoteBzrDir(remote_transport, remote.RemoteBzrDirFormat(),

732

742

            _client=client)

851

861

 

852

862

class RemoteBranchTestCase(RemoteBzrDirTestCase):

853

863

 

 

864

    def lock_remote_branch(self, branch):

 

865

        """Trick a RemoteBranch into thinking it is locked."""

 

866

        branch._lock_mode = 'w'

 

867

        branch._lock_count = 2

 

868

        branch._lock_token = 'branch token'

 

869

        branch._repo_lock_token = 'repo token'

 

870

        branch.repository._lock_mode = 'w'

 

871

        branch.repository._lock_count = 2

 

872

        branch.repository._lock_token = 'repo token'

 

873

 

854

874

    def make_remote_branch(self, transport, client):

855

875

        """Make a RemoteBranch using 'client' as its _SmartClient.

856

876

 

995

1015

        self.assertEqual({}, result)

996

1016

 

997

1017

 

 

1018

class TestBranchSetTagsBytes(RemoteBranchTestCase):

 

1019

 

 

1020

    def test_trivial(self):

 

1021

        transport = MemoryTransport()

 

1022

        client = FakeClient(transport.base)

 

1023

        client.add_expected_call(

 

1024

            'Branch.get_stacked_on_url', ('quack/',),

 

1025

            'error', ('NotStacked',))

 

1026

        client.add_expected_call(

 

1027

            'Branch.set_tags_bytes', ('quack/', 'branch token', 'repo token'),

 

1028

            'success', ('',))

 

1029

        transport.mkdir('quack')

 

1030

        transport = transport.clone('quack')

 

1031

        branch = self.make_remote_branch(transport, client)

 

1032

        self.lock_remote_branch(branch)

 

1033

        branch._set_tags_bytes('tags bytes')

 

1034

        self.assertFinished(client)

 

1035

        self.assertEqual('tags bytes', client._calls[-1][-1])

 

1036

 

 

1037

    def test_backwards_compatible(self):

 

1038

        transport = MemoryTransport()

 

1039

        client = FakeClient(transport.base)

 

1040

        client.add_expected_call(

 

1041

            'Branch.get_stacked_on_url', ('quack/',),

 

1042

            'error', ('NotStacked',))

 

1043

        client.add_expected_call(

 

1044

            'Branch.set_tags_bytes', ('quack/', 'branch token', 'repo token'),

 

1045

            'unknown', ('Branch.set_tags_bytes',))

 

1046

        transport.mkdir('quack')

 

1047

        transport = transport.clone('quack')

 

1048

        branch = self.make_remote_branch(transport, client)

 

1049

        self.lock_remote_branch(branch)

 

1050

        class StubRealBranch(object):

 

1051

            def __init__(self):

 

1052

                self.calls = []

 

1053

            def _set_tags_bytes(self, bytes):

 

1054

                self.calls.append(('set_tags_bytes', bytes))

 

1055

        real_branch = StubRealBranch()

 

1056

        branch._real_branch = real_branch

 

1057

        branch._set_tags_bytes('tags bytes')

 

1058

        # Call a second time, to exercise the 'remote version already inferred'

 

1059

        # code path.

 

1060

        branch._set_tags_bytes('tags bytes')

 

1061

        self.assertFinished(client)

 

1062

        self.assertEqual(

 

1063

            [('set_tags_bytes', 'tags bytes')] * 2, real_branch.calls)

 

1064

 

 

1065

 

998

1066

class TestBranchLastRevisionInfo(RemoteBranchTestCase):

999

1067

 

1000

1068

    def test_empty_branch(self):

1342

1410

            errors.NoSuchRevision, branch.set_last_revision_info, 123, 'revid')

1343

1411

        branch.unlock()

1344

1412

 

1345

 

    def lock_remote_branch(self, branch):

1346

 

        """Trick a RemoteBranch into thinking it is locked."""

1347

 

        branch._lock_mode = 'w'

1348

 

        branch._lock_count = 2

1349

 

        branch._lock_token = 'branch token'

1350

 

        branch._repo_lock_token = 'repo token'

1351

 

        branch.repository._lock_mode = 'w'

1352

 

        branch.repository._lock_count = 2

1353

 

        branch.repository._lock_token = 'repo token'

1354

 

 

1355

1413

    def test_backwards_compatibility(self):

1356

1414

        """If the server does not support the Branch.set_last_revision_info

1357

1415

        verb (which is new in 1.4), then the client falls back to VFS methods.

1945

2003

    def test_allows_new_revisions(self):

1946

2004

        """get_parent_map's results can be updated by commit."""

1947

2005

        smart_server = server.SmartTCPServer_for_testing()

1948

 

        smart_server.setUp()

1949

 

        self.addCleanup(smart_server.tearDown)

 

2006

        self.start_server(smart_server)

1950

2007

        self.make_branch('branch')

1951

2008

        branch = Branch.open(smart_server.get_url() + '/branch')

1952

2009

        tree = branch.create_checkout('tree', lightweight=True)

2326

2383

        class FakeRealRepository:

2327

2384

            def _get_sink(self):

2328

2385

                return fake_real_sink

 

2386

            def is_in_write_group(self):

 

2387

                return False

 

2388

            def refresh_data(self):

 

2389

                return True

2329

2390

        repo._real_repository = FakeRealRepository()

2330

2391

        sink = repo._get_sink()

2331

2392

        fmt = repository.RepositoryFormat.get_default_format()

2781

2842

        stacked_branch.set_stacked_on_url('../base')

2782

2843

        # start a server looking at this

2783

2844

        smart_server = server.SmartTCPServer_for_testing()

2784

 

        smart_server.setUp()

2785

 

        self.addCleanup(smart_server.tearDown)

 

2845

        self.start_server(smart_server)

2786

2846

        remote_bzrdir = BzrDir.open(smart_server.get_url() + '/stacked')

2787

2847

        # can get its branch and repository

2788

2848

        remote_branch = remote_bzrdir.open_branch()

2943

3003

        # Create a smart server that publishes whatever the backing VFS server

2944

3004

        # does.

2945

3005

        self.smart_server = server.SmartTCPServer_for_testing()

2946

 

        self.smart_server.setUp(self.get_server())

2947

 

        self.addCleanup(self.smart_server.tearDown)

 

3006

        self.start_server(self.smart_server, self.get_server())

2948

3007

        # Log all HPSS calls into self.hpss_calls.

2949

3008

        _SmartClient.hooks.install_named_hook(

2950

3009

            'call', self.capture_hpss_call, None)

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1