136
136
"""Create a FakeClient.
138
138
:param responses: A list of response-tuple, body-data pairs to be sent
139
back to callers. A special case is if the response-tuple is
140
'unknown verb', then a UnknownSmartMethod will be raised for that
141
call, using the second element of the tuple as the verb in the
141
144
self.responses = responses
143
146
self.expecting_body = False
144
147
_SmartClient.__init__(self, FakeMedium(self._calls), fake_medium_base)
149
def _get_next_response(self):
150
response_tuple = self.responses.pop(0)
151
if response_tuple[0][0] == 'unknown verb':
152
raise errors.UnknownSmartMethod(response_tuple[0][1])
153
return response_tuple
146
155
def call(self, method, *args):
147
156
self._calls.append(('call', method, args))
148
return self.responses.pop(0)[0]
157
return self._get_next_response()[0]
150
159
def call_expecting_body(self, method, *args):
151
160
self._calls.append(('call_expecting_body', method, args))
152
result = self.responses.pop(0)
161
result = self._get_next_response()
153
162
self.expecting_body = True
154
163
return result[0], FakeProtocol(result[1], self)
156
165
def call_with_body_bytes_expecting_body(self, method, args, body):
157
166
self._calls.append(('call_with_body_bytes_expecting_body', method,
159
result = self.responses.pop(0)
168
result = self._get_next_response()
160
169
self.expecting_body = True
161
170
return result[0], FakeProtocol(result[1], self)
315
324
RemoteBzrDirFormat.probe_transport, OldServerTransport())
327
class TestBzrDirOpenRepository(tests.TestCase):
329
def test_backwards_compat_1_2(self):
330
transport = MemoryTransport()
331
transport.mkdir('quack')
332
transport = transport.clone('quack')
333
client = FakeClient([
334
(('unknown verb', 'RemoteRepository.find_repositoryV2'), ''),
335
(('ok', '', 'no', 'no'), ''),],
337
bzrdir = RemoteBzrDir(transport, _client=client)
338
repo = bzrdir.open_repository()
340
[('call', 'BzrDir.find_repositoryV2', ('quack/',)),
341
('call', 'BzrDir.find_repository', ('quack/',))],
318
345
class OldSmartClient(object):
319
346
"""A fake smart client for test_old_version that just returns a version one
320
347
response to the 'hello' (query version) command.
497
class TestBranchSetLastRevisionInfo(tests.TestCase):
499
def test_set_last_revision_info(self):
500
# set_last_revision_info(num, 'rev-id') is translated to calling
501
# Branch.set_last_revision_info(num, 'rev-id') on the wire.
502
transport = MemoryTransport()
503
transport.mkdir('branch')
504
transport = transport.clone('branch')
505
client = FakeClient([
507
(('ok', 'branch token', 'repo token'), ),
508
# set_last_revision_info
511
(('ok',), )], transport.base)
513
bzrdir = RemoteBzrDir(transport, _client=False)
514
branch = RemoteBranch(bzrdir, None, _client=client)
515
# This is a hack to work around the problem that RemoteBranch currently
516
# unnecessarily invokes _ensure_real upon a call to lock_write.
517
branch._ensure_real = lambda: None
518
# Lock the branch, reset the record of remote calls.
521
result = branch.set_last_revision_info(1234, 'a-revision-id')
523
[('call', 'Branch.set_last_revision_info',
524
('branch/', 'branch token', 'repo token',
525
'1234', 'a-revision-id'))],
527
self.assertEqual(None, result)
529
def test_no_such_revision(self):
530
# A response of 'NoSuchRevision' is translated into an exception.
531
client = FakeClient([
533
(('ok', 'branch token', 'repo token'), ),
534
# set_last_revision_info
535
(('NoSuchRevision', 'revid'), ),
539
transport = MemoryTransport()
540
transport.mkdir('branch')
541
transport = transport.clone('branch')
543
bzrdir = RemoteBzrDir(transport, _client=False)
544
branch = RemoteBranch(bzrdir, None, _client=client)
545
# This is a hack to work around the problem that RemoteBranch currently
546
# unnecessarily invokes _ensure_real upon a call to lock_write.
547
branch._ensure_real = lambda: None
548
# Lock the branch, reset the record of remote calls.
553
errors.NoSuchRevision, branch.set_last_revision_info, 123, 'revid')
556
def lock_remote_branch(self, branch):
557
"""Trick a RemoteBranch into thinking it is locked."""
558
branch._lock_mode = 'w'
559
branch._lock_count = 2
560
branch._lock_token = 'branch token'
561
branch._repo_lock_token = 'repo token'
563
def test_backwards_compatibility(self):
564
"""If the server does not support the Branch.set_last_revision_info
565
verb (which is new in 1.4), then the client falls back to VFS methods.
567
# This test is a little messy. Unlike most tests in this file, it
568
# doesn't purely test what a Remote* object sends over the wire, and
569
# how it reacts to responses from the wire. It instead relies partly
570
# on asserting that the RemoteBranch will call
571
# self._real_branch.set_last_revision_info(...).
573
# First, set up our RemoteBranch with a FakeClient that raises
574
# UnknownSmartMethod, and a StubRealBranch that logs how it is called.
575
transport = MemoryTransport()
576
transport.mkdir('branch')
577
transport = transport.clone('branch')
579
[(('unknown verb', 'Branch.set_last_revision_info',), ),],
581
bzrdir = RemoteBzrDir(transport, _client=False)
582
branch = RemoteBranch(bzrdir, None, _client=client)
583
class StubRealBranch(object):
586
def set_last_revision_info(self, revno, revision_id):
588
('set_last_revision_info', revno, revision_id))
589
real_branch = StubRealBranch()
590
branch._real_branch = real_branch
591
self.lock_remote_branch(branch)
593
# Call set_last_revision_info, and verify it behaved as expected.
594
result = branch.set_last_revision_info(1234, 'a-revision-id')
596
[('call', 'Branch.set_last_revision_info',
597
('branch/', 'branch token', 'repo token',
598
'1234', 'a-revision-id')),],
601
[('set_last_revision_info', 1234, 'a-revision-id')],
470
605
class TestBranchControlGetBranchConf(tests.TestCaseWithMemoryTransport):
471
606
"""Test branch.control_files api munging...
536
671
advisory anyway (a transport could be read-write, but then the
537
672
underlying filesystem could be readonly anyway).
539
client = FakeClient([(
540
('error', "Generic bzr smart protocol error: "
541
"bad request 'Transport.is_readonly'"), '')])
542
transport = RemoteTransport('bzr://example.com/', medium=False,
544
self.assertEqual(False, transport.is_readonly())
546
[('call', 'Transport.is_readonly', ())],
549
def test_error_from_old_0_11_server(self):
550
"""Same as test_error_from_old_server, but with the slightly different
551
error message from bzr 0.11 servers.
553
client = FakeClient([(
554
('error', "Generic bzr smart protocol error: "
555
"bad request u'Transport.is_readonly'"), '')])
674
client = FakeClient([(('unknown verb', 'Transport.is_readonly'), '')])
556
675
transport = RemoteTransport('bzr://example.com/', medium=False,
558
677
self.assertEqual(False, transport.is_readonly())
706
825
def test_get_parent_map_reconnects_if_unknown_method(self):
708
"Generic bzr smart protocol error: "
709
"bad request 'Repository.get_parent_map'")
711
(('error', error_msg), ''),
827
(('unknown verb', 'Repository.get_parent_map'), ''),
713
829
transport_path = 'quack'
714
830
repo, client = self.setup_fake_client_and_repository(
715
831
responses, transport_path)
716
832
rev_id = 'revision-id'
717
parents = repo.get_parent_map([rev_id])
833
expected_deprecations = [
834
'bzrlib.remote.RemoteRepository.get_revision_graph was deprecated '
836
parents = self.callDeprecated(
837
expected_deprecations, repo.get_parent_map, [rev_id])
718
838
self.assertEqual(
719
839
[('call_with_body_bytes_expecting_body',
720
840
'Repository.get_parent_map', ('quack/', rev_id), '\n\n0'),
added
removed
bzrlib/tests/test_remote.py
