1
# Copyright (C) 2006, 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Tests for remote bzrdir/branch/repo/etc
19
These are proxy objects which act on remote objects by sending messages
20
through a smart client. The proxies are to be created when attempting to open
21
the object given a transport that supports smartserver rpc operations.
24
from cStringIO import StringIO
32
from bzrlib.branch import Branch
33
from bzrlib.bzrdir import BzrDir, BzrDirFormat
34
from bzrlib.remote import (
40
from bzrlib.revision import NULL_REVISION
41
from bzrlib.smart import server
42
from bzrlib.smart.client import SmartClient
43
from bzrlib.transport import remote as remote_transport
44
from bzrlib.transport.memory import MemoryTransport
47
class BasicRemoteObjectTests(tests.TestCaseWithTransport):
50
super(BasicRemoteObjectTests, self).setUp()
51
self.transport_server = server.SmartTCPServer_for_testing
52
self.transport = self.get_transport()
53
self.client = self.transport.get_smart_client()
54
# make a branch that can be opened over the smart transport
55
self.local_wt = BzrDir.create_standalone_workingtree('.')
57
def test_create_remote_bzrdir(self):
58
b = remote.RemoteBzrDir(self.transport)
59
self.assertIsInstance(b, BzrDir)
61
def test_open_remote_branch(self):
62
# open a standalone branch in the working directory
63
b = remote.RemoteBzrDir(self.transport)
64
branch = b.open_branch()
66
def test_remote_repository(self):
67
b = BzrDir.open_from_transport(self.transport)
68
repo = b.open_repository()
69
revid = u'\xc823123123'
70
self.assertFalse(repo.has_revision(revid))
71
self.local_wt.commit(message='test commit', rev_id=revid)
72
self.assertTrue(repo.has_revision(revid))
74
def test_remote_branch_revision_history(self):
75
b = BzrDir.open_from_transport(self.transport).open_branch()
76
self.assertEqual([], b.revision_history())
77
r1 = self.local_wt.commit('1st commit')
78
r2 = self.local_wt.commit('1st commit', rev_id=u'\xc8')
79
self.assertEqual([r1, r2], b.revision_history())
81
def test_find_correct_format(self):
82
"""Should open a RemoteBzrDir over a RemoteTransport"""
83
fmt = BzrDirFormat.find_format(self.transport)
84
self.assertTrue(RemoteBzrDirFormat in BzrDirFormat._control_formats)
85
self.assertIsInstance(fmt, remote.RemoteBzrDirFormat)
87
def test_open_detected_smart_format(self):
88
fmt = BzrDirFormat.find_format(self.transport)
89
d = fmt.open(self.transport)
90
self.assertIsInstance(d, BzrDir)
93
class FakeProtocol(object):
94
"""Lookalike SmartClientRequestProtocolOne allowing body reading tests."""
96
def __init__(self, body):
97
self._body_buffer = StringIO(body)
99
def read_body_bytes(self, count=-1):
100
return self._body_buffer.read(count)
103
class FakeClient(SmartClient):
104
"""Lookalike for SmartClient allowing testing."""
106
def __init__(self, responses):
107
# We don't call the super init because there is no medium.
108
"""create a FakeClient.
110
:param respones: A list of response-tuple, body-data pairs to be sent
113
self.responses = responses
116
def call(self, method, *args):
117
self._calls.append(('call', method, args))
118
return self.responses.pop(0)[0]
120
def call2(self, method, *args):
121
self._calls.append(('call2', method, args))
122
result = self.responses.pop(0)
123
return result[0], FakeProtocol(result[1])
126
class TestBzrDirOpenBranch(tests.TestCase):
128
def test_branch_present(self):
129
client = FakeClient([(('ok', ''), ), (('ok', ''), )])
130
transport = MemoryTransport()
131
transport.mkdir('quack')
132
transport = transport.clone('quack')
133
bzrdir = RemoteBzrDir(transport, _client=client)
134
result = bzrdir.open_branch()
136
[('call', 'BzrDir.open_branch', ('///quack/',)),
137
('call', 'BzrDir.find_repository', ('///quack/',))],
139
self.assertIsInstance(result, RemoteBranch)
140
self.assertEqual(bzrdir, result.bzrdir)
142
def test_branch_missing(self):
143
client = FakeClient([(('nobranch',), )])
144
transport = MemoryTransport()
145
transport.mkdir('quack')
146
transport = transport.clone('quack')
147
bzrdir = RemoteBzrDir(transport, _client=client)
148
self.assertRaises(errors.NotBranchError, bzrdir.open_branch)
150
[('call', 'BzrDir.open_branch', ('///quack/',))],
154
class TestBranchLastRevisionInfo(tests.TestCase):
156
def test_empty_branch(self):
157
# in an empty branch we decode the response properly
158
client = FakeClient([(('ok', '0', ''), )])
159
transport = MemoryTransport()
160
transport.mkdir('quack')
161
transport = transport.clone('quack')
162
# we do not want bzrdir to make any remote calls
163
bzrdir = RemoteBzrDir(transport, _client=False)
164
branch = RemoteBranch(bzrdir, None, _client=client)
165
result = branch.last_revision_info()
168
[('call', 'Branch.last_revision_info', ('///quack/',))],
170
self.assertEqual((0, NULL_REVISION), result)
172
def test_non_empty_branch(self):
173
# in a non-empty branch we also decode the response properly
175
client = FakeClient([(('ok', '2', u'\xc8'.encode('utf8')), )])
176
transport = MemoryTransport()
177
transport.mkdir('kwaak')
178
transport = transport.clone('kwaak')
179
# we do not want bzrdir to make any remote calls
180
bzrdir = RemoteBzrDir(transport, _client=False)
181
branch = RemoteBranch(bzrdir, None, _client=client)
182
result = branch.last_revision_info()
185
[('call', 'Branch.last_revision_info', ('///kwaak/',))],
187
self.assertEqual((2, u'\xc8'), result)
190
class TestBranchSetLastRevision(tests.TestCase):
192
def test_set_empty(self):
193
# set_revision_history([]) is translated to calling
194
# Branch.set_last_revision(path, '') on the wire.
195
client = FakeClient([
197
(('ok', 'branch token', 'repo token'), ),
202
transport = MemoryTransport()
203
transport.mkdir('branch')
204
transport = transport.clone('branch')
206
bzrdir = RemoteBzrDir(transport, _client=False)
207
branch = RemoteBranch(bzrdir, None, _client=client)
208
# This is a hack to work around the problem that RemoteBranch currently
209
# unnecessarily invokes _ensure_real upon a call to lock_write.
210
branch._ensure_real = lambda: None
213
result = branch.set_revision_history([])
215
[('call', 'Branch.set_last_revision',
216
('///branch/', 'branch token', 'repo token', ''))],
219
self.assertEqual(None, result)
221
def test_set_nonempty(self):
222
# set_revision_history([rev-id1, ..., rev-idN]) is translated to calling
223
# Branch.set_last_revision(path, rev-idN) on the wire.
224
client = FakeClient([
226
(('ok', 'branch token', 'repo token'), ),
231
transport = MemoryTransport()
232
transport.mkdir('branch')
233
transport = transport.clone('branch')
235
bzrdir = RemoteBzrDir(transport, _client=False)
236
branch = RemoteBranch(bzrdir, None, _client=client)
237
# This is a hack to work around the problem that RemoteBranch currently
238
# unnecessarily invokes _ensure_real upon a call to lock_write.
239
branch._ensure_real = lambda: None
240
# Lock the branch, reset the record of remote calls.
244
result = branch.set_revision_history(['rev-id1', 'rev-id2'])
246
[('call', 'Branch.set_last_revision',
247
('///branch/', 'branch token', 'repo token', 'rev-id2'))],
250
self.assertEqual(None, result)
252
def test_no_such_revision(self):
253
# A response of 'NoSuchRevision' is translated into an exception.
254
client = FakeClient([
256
(('ok', 'branch token', 'repo token'), ),
258
(('NoSuchRevision', 'rev-id'), ),
261
transport = MemoryTransport()
262
transport.mkdir('branch')
263
transport = transport.clone('branch')
265
bzrdir = RemoteBzrDir(transport, _client=False)
266
branch = RemoteBranch(bzrdir, None, _client=client)
267
branch._ensure_real = lambda: None
272
errors.NoSuchRevision, branch.set_revision_history, ['rev-id'])
276
class TestBranchControlGetBranchConf(tests.TestCase):
277
"""Test branch.control_files api munging...
279
we special case RemoteBranch.control_files.get('branch.conf') to
280
call a specific API so that RemoteBranch's can intercept configuration
281
file reading, allowing them to signal to the client about things like
282
'email is configured for commits'.
285
def test_get_branch_conf(self):
286
# in an empty branch we decode the response properly
287
client = FakeClient([(('ok', ), 'config file body')])
288
transport = MemoryTransport()
289
transport.mkdir('quack')
290
transport = transport.clone('quack')
291
# we do not want bzrdir to make any remote calls
292
bzrdir = RemoteBzrDir(transport, _client=False)
293
branch = RemoteBranch(bzrdir, None, _client=client)
294
result = branch.control_files.get('branch.conf')
296
[('call2', 'Branch.get_config_file', ('///quack/',))],
298
self.assertEqual('config file body', result.read())
301
class TestRemoteRepository(tests.TestCase):
303
def setup_fake_client_and_repository(self, responses, transport_path):
304
"""Create the fake client and repository for testing with."""
305
client = FakeClient(responses)
306
transport = MemoryTransport()
307
transport.mkdir(transport_path)
308
transport = transport.clone(transport_path)
309
# we do not want bzrdir to make any remote calls
310
bzrdir = RemoteBzrDir(transport, _client=False)
311
repo = RemoteRepository(bzrdir, None, _client=client)
315
class TestRepositoryGatherStats(TestRemoteRepository):
317
def test_revid_none(self):
318
# ('ok',), body with revisions and size
319
responses = [(('ok', ), 'revisions: 2\nsize: 18\n')]
320
transport_path = 'quack'
321
repo, client = self.setup_fake_client_and_repository(
322
responses, transport_path)
323
result = repo.gather_stats(None)
325
[('call2', 'Repository.gather_stats', ('///quack/','','no'))],
327
self.assertEqual({'revisions': 2, 'size': 18}, result)
329
def test_revid_no_committers(self):
330
# ('ok',), body without committers
331
responses = [(('ok', ),
332
'firstrev: 123456.300 3600\n'
333
'latestrev: 654231.400 0\n'
336
transport_path = 'quick'
338
repo, client = self.setup_fake_client_and_repository(
339
responses, transport_path)
340
result = repo.gather_stats(revid)
342
[('call2', 'Repository.gather_stats',
343
('///quick/', revid.encode('utf8'), 'no'))],
345
self.assertEqual({'revisions': 2, 'size': 18,
346
'firstrev': (123456.300, 3600),
347
'latestrev': (654231.400, 0),},
350
def test_revid_with_committers(self):
351
# ('ok',), body with committers
352
responses = [(('ok', ),
354
'firstrev: 123456.300 3600\n'
355
'latestrev: 654231.400 0\n'
358
transport_path = 'buick'
360
repo, client = self.setup_fake_client_and_repository(
361
responses, transport_path)
362
result = repo.gather_stats(revid, True)
364
[('call2', 'Repository.gather_stats',
365
('///buick/', revid.encode('utf8'), 'yes'))],
367
self.assertEqual({'revisions': 2, 'size': 18,
369
'firstrev': (123456.300, 3600),
370
'latestrev': (654231.400, 0),},
374
class TestRepositoryGetRevisionGraph(TestRemoteRepository):
376
def test_null_revision(self):
377
# a null revision has the predictable result {}, we should have no wire
378
# traffic when calling it with this argument
379
responses = [(('notused', ), '')]
380
transport_path = 'empty'
381
repo, client = self.setup_fake_client_and_repository(
382
responses, transport_path)
383
result = repo.get_revision_graph(NULL_REVISION)
384
self.assertEqual([], client._calls)
385
self.assertEqual({}, result)
387
def test_none_revision(self):
388
# with none we want the entire graph
391
lines = [' '.join([r2, r1]), r1]
392
encoded_body = '\n'.join(lines).encode('utf8')
394
responses = [(('ok', ), encoded_body)]
395
transport_path = 'sinhala'
396
repo, client = self.setup_fake_client_and_repository(
397
responses, transport_path)
398
result = repo.get_revision_graph()
400
[('call2', 'Repository.get_revision_graph', ('///sinhala/', ''))],
402
self.assertEqual({r1: [], r2: [r1]}, result)
404
def test_specific_revision(self):
405
# with a specific revision we want the graph for that
406
# with none we want the entire graph
410
lines = [' '.join([r2, r11, r12]), r11, r12]
411
encoded_body = '\n'.join(lines).encode('utf8')
413
responses = [(('ok', ), encoded_body)]
414
transport_path = 'sinhala'
415
repo, client = self.setup_fake_client_and_repository(
416
responses, transport_path)
417
result = repo.get_revision_graph(r2)
419
[('call2', 'Repository.get_revision_graph', ('///sinhala/', r2.encode('utf8')))],
421
self.assertEqual({r11: [], r12: [], r2: [r11, r12], }, result)
423
def test_no_such_revision(self):
425
responses = [(('nosuchrevision', revid), '')]
426
transport_path = 'sinhala'
427
repo, client = self.setup_fake_client_and_repository(
428
responses, transport_path)
429
# also check that the right revision is reported in the error
430
self.assertRaises(errors.NoSuchRevision,
431
repo.get_revision_graph, revid)
433
[('call2', 'Repository.get_revision_graph', ('///sinhala/', revid))],
437
class TestRepositoryIsShared(TestRemoteRepository):
439
def test_is_shared(self):
440
# ('yes', ) for Repository.is_shared -> 'True'.
441
responses = [(('yes', ), )]
442
transport_path = 'quack'
443
repo, client = self.setup_fake_client_and_repository(
444
responses, transport_path)
445
result = repo.is_shared()
447
[('call', 'Repository.is_shared', ('///quack/',))],
449
self.assertEqual(True, result)
451
def test_is_not_shared(self):
452
# ('no', ) for Repository.is_shared -> 'False'.
453
responses = [(('no', ), )]
454
transport_path = 'qwack'
455
repo, client = self.setup_fake_client_and_repository(
456
responses, transport_path)
457
result = repo.is_shared()
459
[('call', 'Repository.is_shared', ('///qwack/',))],
461
self.assertEqual(False, result)
464
class TestRepositoryLockWrite(TestRemoteRepository):
466
def test_lock_write(self):
467
responses = [(('ok', 'a token'), '')]
468
transport_path = 'quack'
469
repo, client = self.setup_fake_client_and_repository(
470
responses, transport_path)
471
result = repo.lock_write()
473
[('call', 'Repository.lock_write', ('///quack/',))],
475
self.assertEqual('a token', result)
477
def test_lock_write_already_locked(self):
478
responses = [(('LockContention', ), '')]
479
transport_path = 'quack'
480
repo, client = self.setup_fake_client_and_repository(
481
responses, transport_path)
482
self.assertRaises(errors.LockContention, repo.lock_write)
484
[('call', 'Repository.lock_write', ('///quack/',))],
488
class TestRepositoryUnlock(TestRemoteRepository):
490
def test_unlock(self):
491
responses = [(('ok', 'a token'), ''),
493
transport_path = 'quack'
494
repo, client = self.setup_fake_client_and_repository(
495
responses, transport_path)
499
[('call', 'Repository.lock_write', ('///quack/',)),
500
('call', 'Repository.unlock', ('///quack/', 'a token'))],
503
def test_unlock_wrong_token(self):
504
# If somehow the token is wrong, unlock will raise TokenMismatch.
505
responses = [(('ok', 'a token'), ''),
506
(('TokenMismatch',), '')]
507
transport_path = 'quack'
508
repo, client = self.setup_fake_client_and_repository(
509
responses, transport_path)
511
self.assertRaises(errors.TokenMismatch, repo.unlock)
514
class TestRepositoryHasRevision(TestRemoteRepository):
517
# repo.has_revision(None) should not cause any traffic.
518
transport_path = 'quack'
520
repo, client = self.setup_fake_client_and_repository(
521
responses, transport_path)
523
# The null revision is always there, so has_revision(None) == True.
524
self.assertEqual(True, repo.has_revision(None))
526
# The remote repo shouldn't be accessed.
527
self.assertEqual([], client._calls)
added
removed
bzrlib/tests/test_remote.py
