135
136
"""Create a FakeClient.
137
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
140
144
self.responses = responses
142
146
self.expecting_body = False
143
_SmartClient.__init__(self, FakeMedium(fake_medium_base, self._calls))
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
145
155
def call(self, method, *args):
146
156
self._calls.append(('call', method, args))
147
return self.responses.pop(0)[0]
157
return self._get_next_response()[0]
149
159
def call_expecting_body(self, method, *args):
150
160
self._calls.append(('call_expecting_body', method, args))
151
result = self.responses.pop(0)
161
result = self._get_next_response()
152
162
self.expecting_body = True
153
163
return result[0], FakeProtocol(result[1], self)
155
165
def call_with_body_bytes_expecting_body(self, method, args, body):
156
166
self._calls.append(('call_with_body_bytes_expecting_body', method,
158
result = self.responses.pop(0)
168
result = self._get_next_response()
159
169
self.expecting_body = True
160
170
return result[0], FakeProtocol(result[1], self)
163
173
class FakeMedium(object):
165
def __init__(self, base, client_calls):
167
self.connection = FakeConnection(client_calls)
168
self._client_calls = client_calls
171
class FakeConnection(object):
173
175
def __init__(self, client_calls):
174
176
self._remote_is_at_least_1_2 = True
175
177
self._client_calls = client_calls
191
193
self.assertTrue(result)
196
class Test_SmartClient_remote_path_from_transport(tests.TestCase):
197
"""Tests for the behaviour of _SmartClient.remote_path_from_transport."""
199
def assertRemotePath(self, expected, client_base, transport_base):
200
"""Assert that the result of _SmartClient.remote_path_from_transport
201
is the expected value for a given client_base and transport_base.
203
dummy_medium = 'dummy medium'
204
client = _SmartClient(dummy_medium, client_base)
205
transport = get_transport(transport_base)
206
result = client.remote_path_from_transport(transport)
207
self.assertEqual(expected, result)
209
def test_remote_path_from_transport(self):
210
"""_SmartClient.remote_path_from_transport calculates a URL for the
211
given transport relative to the root of the client base URL.
213
self.assertRemotePath('xyz/', 'bzr://host/path', 'bzr://host/xyz')
214
self.assertRemotePath(
215
'path/xyz/', 'bzr://host/path', 'bzr://host/path/xyz')
217
def test_remote_path_from_transport_http(self):
218
"""Remote paths for HTTP transports are calculated differently to other
219
transports. They are just relative to the client base, not the root
220
directory of the host.
222
for scheme in ['http:', 'https:', 'bzr+http:', 'bzr+https:']:
223
self.assertRemotePath(
224
'../xyz/', scheme + '//host/path', scheme + '//host/xyz')
225
self.assertRemotePath(
226
'xyz/', scheme + '//host/path', scheme + '//host/path/xyz')
194
229
class TestBzrDirOpenBranch(tests.TestCase):
196
231
def test_branch_present(self):
289
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/',))],
292
345
class OldSmartClient(object):
293
346
"""A fake smart client for test_old_version that just returns a version one
294
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')],
444
605
class TestBranchControlGetBranchConf(tests.TestCaseWithMemoryTransport):
445
606
"""Test branch.control_files api munging...
510
671
advisory anyway (a transport could be read-write, but then the
511
672
underlying filesystem could be readonly anyway).
513
client = FakeClient([(
514
('error', "Generic bzr smart protocol error: "
515
"bad request 'Transport.is_readonly'"), '')])
516
transport = RemoteTransport('bzr://example.com/', medium=False,
518
self.assertEqual(False, transport.is_readonly())
520
[('call', 'Transport.is_readonly', ())],
523
def test_error_from_old_0_11_server(self):
524
"""Same as test_error_from_old_server, but with the slightly different
525
error message from bzr 0.11 servers.
527
client = FakeClient([(
528
('error', "Generic bzr smart protocol error: "
529
"bad request u'Transport.is_readonly'"), '')])
674
client = FakeClient([(('unknown verb', 'Transport.is_readonly'), '')])
530
675
transport = RemoteTransport('bzr://example.com/', medium=False,
532
677
self.assertEqual(False, transport.is_readonly())
680
825
def test_get_parent_map_reconnects_if_unknown_method(self):
682
"Generic bzr smart protocol error: "
683
"bad request 'Repository.get_parent_map'")
685
(('error', error_msg), ''),
827
(('unknown verb', 'Repository.get_parent_map'), ''),
687
829
transport_path = 'quack'
688
830
repo, client = self.setup_fake_client_and_repository(
689
831
responses, transport_path)
690
832
rev_id = 'revision-id'
691
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])
692
838
self.assertEqual(
693
839
[('call_with_body_bytes_expecting_body',
694
840
'Repository.get_parent_map', ('quack/', rev_id), '\n\n0'),
added
removed
bzrlib/tests/test_remote.py
