« back to all changes in this revision
Viewing changes to bzrlib/tests/test_remote.py
- browse files at revision 2399.1.16
- compare with another revision
- compare with revision 2052.6.2
- download diff
- download tarball
- view history from revision 2399.1.16
- Committer: John Arbash Meinel
- Date: 2007-04-28 15:04:17 UTC
- mfrom: (2466 +trunk)
- mto: This revision was merged to the branch mainline in revision 2566.
- Revision ID: john@arbash-meinel.com-20070428150417-trp3pi0pzd411pu4
[merge] bzr.dev 2466
- files added:
- bzrlib/breakin.py
- man1
- files removed:
- build-api
- files modified:
- HACKING
- setup.py *
added
removed
bzrlib/tests/test_remote.py
1
# Copyright (C) 2006, 2007 Canonical Ltd
2
#
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.
7
#
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.
12
#
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
16
17
"""Tests for remote bzrdir/branch/repo/etc
18
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.
22
23
These tests correspond to tests.test_smart, which exercises the server side.
24
"""
25
26
from cStringIO import StringIO
27
28
from bzrlib import (
29
bzrdir,
30
errors,
31
remote,
32
repository,
33
tests,
34
)
35
from bzrlib.branch import Branch
36
from bzrlib.bzrdir import BzrDir, BzrDirFormat
37
from bzrlib.remote import (
38
RemoteBranch,
39
RemoteBzrDir,
40
RemoteBzrDirFormat,
41
RemoteRepository,
42
)
43
from bzrlib.revision import NULL_REVISION
44
from bzrlib.smart import server, medium
45
from bzrlib.smart.client import _SmartClient
46
from bzrlib.transport.memory import MemoryTransport
47
48
49
class BasicRemoteObjectTests(tests.TestCaseWithTransport):
50
51
def setUp(self):
52
self.transport_server = server.SmartTCPServer_for_testing
53
super(BasicRemoteObjectTests, self).setUp()
54
self.transport = self.get_transport()
55
self.client = self.transport.get_smart_client()
56
# make a branch that can be opened over the smart transport
57
self.local_wt = BzrDir.create_standalone_workingtree('.')
58
59
def tearDown(self):
60
self.transport.disconnect()
61
tests.TestCaseWithTransport.tearDown(self)
62
63
def test_is_readonly(self):
64
# XXX: this is a poor way to test RemoteTransport, but currently there's
65
# no easy way to substitute in a fake client on a transport like we can
66
# with RemoteBzrDir/Branch/Repository.
67
self.assertEqual(self.transport.is_readonly(), False)
68
69
def test_create_remote_bzrdir(self):
70
b = remote.RemoteBzrDir(self.transport)
71
self.assertIsInstance(b, BzrDir)
72
73
def test_open_remote_branch(self):
74
# open a standalone branch in the working directory
75
b = remote.RemoteBzrDir(self.transport)
76
branch = b.open_branch()
77
self.assertIsInstance(branch, Branch)
78
79
def test_remote_repository(self):
80
b = BzrDir.open_from_transport(self.transport)
81
repo = b.open_repository()
82
revid = u'\xc823123123'.encode('utf8')
83
self.assertFalse(repo.has_revision(revid))
84
self.local_wt.commit(message='test commit', rev_id=revid)
85
self.assertTrue(repo.has_revision(revid))
86
87
def test_remote_branch_revision_history(self):
88
b = BzrDir.open_from_transport(self.transport).open_branch()
89
self.assertEqual([], b.revision_history())
90
r1 = self.local_wt.commit('1st commit')
91
r2 = self.local_wt.commit('1st commit', rev_id=u'\xc8'.encode('utf8'))
92
self.assertEqual([r1, r2], b.revision_history())
93
94
def test_find_correct_format(self):
95
"""Should open a RemoteBzrDir over a RemoteTransport"""
96
fmt = BzrDirFormat.find_format(self.transport)
97
self.assertTrue(RemoteBzrDirFormat
98
in BzrDirFormat._control_server_formats)
99
self.assertIsInstance(fmt, remote.RemoteBzrDirFormat)
100
101
def test_open_detected_smart_format(self):
102
fmt = BzrDirFormat.find_format(self.transport)
103
d = fmt.open(self.transport)
104
self.assertIsInstance(d, BzrDir)
105
106
107
class ReadonlyRemoteTransportTests(tests.TestCaseWithTransport):
108
109
def setUp(self):
110
self.transport_server = server.ReadonlySmartTCPServer_for_testing
111
super(ReadonlyRemoteTransportTests, self).setUp()
112
113
def test_is_readonly_yes(self):
114
# XXX: this is a poor way to test RemoteTransport, but currently there's
115
# no easy way to substitute in a fake client on a transport like we can
116
# with RemoteBzrDir/Branch/Repository.
117
transport = self.get_readonly_transport()
118
self.assertEqual(transport.is_readonly(), True)
119
120
121
class FakeProtocol(object):
122
"""Lookalike SmartClientRequestProtocolOne allowing body reading tests."""
123
124
def __init__(self, body):
125
self._body_buffer = StringIO(body)
126
127
def read_body_bytes(self, count=-1):
128
return self._body_buffer.read(count)
129
130
131
class FakeClient(_SmartClient):
132
"""Lookalike for _SmartClient allowing testing."""
133
134
def __init__(self, responses):
135
# We don't call the super init because there is no medium.
136
"""create a FakeClient.
137
138
:param respones: A list of response-tuple, body-data pairs to be sent
139
back to callers.
140
"""
141
self.responses = responses
142
self._calls = []
143
144
def call(self, method, *args):
145
self._calls.append(('call', method, args))
146
return self.responses.pop(0)[0]
147
148
def call_expecting_body(self, method, *args):
149
self._calls.append(('call_expecting_body', method, args))
150
result = self.responses.pop(0)
151
return result[0], FakeProtocol(result[1])
152
153
154
class TestBzrDirOpenBranch(tests.TestCase):
155
156
def test_branch_present(self):
157
client = FakeClient([(('ok', ''), ), (('ok', '', 'no', 'no'), )])
158
transport = MemoryTransport()
159
transport.mkdir('quack')
160
transport = transport.clone('quack')
161
bzrdir = RemoteBzrDir(transport, _client=client)
162
result = bzrdir.open_branch()
163
self.assertEqual(
164
[('call', 'BzrDir.open_branch', ('///quack/',)),
165
('call', 'BzrDir.find_repository', ('///quack/',))],
166
client._calls)
167
self.assertIsInstance(result, RemoteBranch)
168
self.assertEqual(bzrdir, result.bzrdir)
169
170
def test_branch_missing(self):
171
client = FakeClient([(('nobranch',), )])
172
transport = MemoryTransport()
173
transport.mkdir('quack')
174
transport = transport.clone('quack')
175
bzrdir = RemoteBzrDir(transport, _client=client)
176
self.assertRaises(errors.NotBranchError, bzrdir.open_branch)
177
self.assertEqual(
178
[('call', 'BzrDir.open_branch', ('///quack/',))],
179
client._calls)
180
181
def check_open_repository(self, rich_root, subtrees):
182
if rich_root:
183
rich_response = 'yes'
184
else:
185
rich_response = 'no'
186
if subtrees:
187
subtree_response = 'yes'
188
else:
189
subtree_response = 'no'
190
client = FakeClient([(('ok', '', rich_response, subtree_response), ),])
191
transport = MemoryTransport()
192
transport.mkdir('quack')
193
transport = transport.clone('quack')
194
bzrdir = RemoteBzrDir(transport, _client=client)
195
result = bzrdir.open_repository()
196
self.assertEqual(
197
[('call', 'BzrDir.find_repository', ('///quack/',))],
198
client._calls)
199
self.assertIsInstance(result, RemoteRepository)
200
self.assertEqual(bzrdir, result.bzrdir)
201
self.assertEqual(rich_root, result._format.rich_root_data)
202
self.assertEqual(subtrees, result._format.supports_tree_reference)
203
204
def test_open_repository_sets_format_attributes(self):
205
self.check_open_repository(True, True)
206
self.check_open_repository(False, True)
207
self.check_open_repository(True, False)
208
self.check_open_repository(False, False)
209
210
def test_old_server(self):
211
"""RemoteBzrDirFormat should fail to probe if the server version is too
212
old.
213
"""
214
self.assertRaises(errors.NotBranchError,
215
RemoteBzrDirFormat.probe_transport, OldServerTransport())
216
217
218
class OldSmartClient(object):
219
"""A fake smart client for test_old_version that just returns a version one
220
response to the 'hello' (query version) command.
221
"""
222
223
def get_request(self):
224
input_file = StringIO('ok\x011\n')
225
output_file = StringIO()
226
client_medium = medium.SmartSimplePipesClientMedium(
227
input_file, output_file)
228
return medium.SmartClientStreamMediumRequest(client_medium)
229
230
231
class OldServerTransport(object):
232
"""A fake transport for test_old_server that reports it's smart server
233
protocol version as version one.
234
"""
235
236
def __init__(self):
237
self.base = 'fake:'
238
239
def get_smart_client(self):
240
return OldSmartClient()
241
242
243
class TestBranchLastRevisionInfo(tests.TestCase):
244
245
def test_empty_branch(self):
246
# in an empty branch we decode the response properly
247
client = FakeClient([(('ok', '0', 'null:'), )])
248
transport = MemoryTransport()
249
transport.mkdir('quack')
250
transport = transport.clone('quack')
251
# we do not want bzrdir to make any remote calls
252
bzrdir = RemoteBzrDir(transport, _client=False)
253
branch = RemoteBranch(bzrdir, None, _client=client)
254
result = branch.last_revision_info()
255
256
self.assertEqual(
257
[('call', 'Branch.last_revision_info', ('///quack/',))],
258
client._calls)
259
self.assertEqual((0, NULL_REVISION), result)
260
261
def test_non_empty_branch(self):
262
# in a non-empty branch we also decode the response properly
263
revid = u'\xc8'.encode('utf8')
264
client = FakeClient([(('ok', '2', revid), )])
265
transport = MemoryTransport()
266
transport.mkdir('kwaak')
267
transport = transport.clone('kwaak')
268
# we do not want bzrdir to make any remote calls
269
bzrdir = RemoteBzrDir(transport, _client=False)
270
branch = RemoteBranch(bzrdir, None, _client=client)
271
result = branch.last_revision_info()
272
273
self.assertEqual(
274
[('call', 'Branch.last_revision_info', ('///kwaak/',))],
275
client._calls)
276
self.assertEqual((2, revid), result)
277
278
279
class TestBranchSetLastRevision(tests.TestCase):
280
281
def test_set_empty(self):
282
# set_revision_history([]) is translated to calling
283
# Branch.set_last_revision(path, '') on the wire.
284
client = FakeClient([
285
# lock_write
286
(('ok', 'branch token', 'repo token'), ),
287
# set_last_revision
288
(('ok',), ),
289
# unlock
290
(('ok',), )])
291
transport = MemoryTransport()
292
transport.mkdir('branch')
293
transport = transport.clone('branch')
294
295
bzrdir = RemoteBzrDir(transport, _client=False)
296
branch = RemoteBranch(bzrdir, None, _client=client)
297
# This is a hack to work around the problem that RemoteBranch currently
298
# unnecessarily invokes _ensure_real upon a call to lock_write.
299
branch._ensure_real = lambda: None
300
branch.lock_write()
301
client._calls = []
302
result = branch.set_revision_history([])
303
self.assertEqual(
304
[('call', 'Branch.set_last_revision',
305
('///branch/', 'branch token', 'repo token', 'null:'))],
306
client._calls)
307
branch.unlock()
308
self.assertEqual(None, result)
309
310
def test_set_nonempty(self):
311
# set_revision_history([rev-id1, ..., rev-idN]) is translated to calling
312
# Branch.set_last_revision(path, rev-idN) on the wire.
313
client = FakeClient([
314
# lock_write
315
(('ok', 'branch token', 'repo token'), ),
316
# set_last_revision
317
(('ok',), ),
318
# unlock
319
(('ok',), )])
320
transport = MemoryTransport()
321
transport.mkdir('branch')
322
transport = transport.clone('branch')
323
324
bzrdir = RemoteBzrDir(transport, _client=False)
325
branch = RemoteBranch(bzrdir, None, _client=client)
326
# This is a hack to work around the problem that RemoteBranch currently
327
# unnecessarily invokes _ensure_real upon a call to lock_write.
328
branch._ensure_real = lambda: None
329
# Lock the branch, reset the record of remote calls.
330
branch.lock_write()
331
client._calls = []
332
333
result = branch.set_revision_history(['rev-id1', 'rev-id2'])
334
self.assertEqual(
335
[('call', 'Branch.set_last_revision',
336
('///branch/', 'branch token', 'repo token', 'rev-id2'))],
337
client._calls)
338
branch.unlock()
339
self.assertEqual(None, result)
340
341
def test_no_such_revision(self):
342
# A response of 'NoSuchRevision' is translated into an exception.
343
client = FakeClient([
344
# lock_write
345
(('ok', 'branch token', 'repo token'), ),
346
# set_last_revision
347
(('NoSuchRevision', 'rev-id'), ),
348
# unlock
349
(('ok',), )])
350
transport = MemoryTransport()
351
transport.mkdir('branch')
352
transport = transport.clone('branch')
353
354
bzrdir = RemoteBzrDir(transport, _client=False)
355
branch = RemoteBranch(bzrdir, None, _client=client)
356
branch._ensure_real = lambda: None
357
branch.lock_write()
358
client._calls = []
359
360
self.assertRaises(
361
errors.NoSuchRevision, branch.set_revision_history, ['rev-id'])
362
branch.unlock()
363
364
365
class TestBranchControlGetBranchConf(tests.TestCaseWithMemoryTransport):
366
"""Test branch.control_files api munging...
367
368
We special case RemoteBranch.control_files.get('branch.conf') to
369
call a specific API so that RemoteBranch's can intercept configuration
370
file reading, allowing them to signal to the client about things like
371
'email is configured for commits'.
372
"""
373
374
def test_get_branch_conf(self):
375
# in an empty branch we decode the response properly
376
client = FakeClient([(('ok', ), 'config file body')])
377
# we need to make a real branch because the remote_branch.control_files
378
# will trigger _ensure_real.
379
branch = self.make_branch('quack')
380
transport = branch.bzrdir.root_transport
381
# we do not want bzrdir to make any remote calls
382
bzrdir = RemoteBzrDir(transport, _client=False)
383
branch = RemoteBranch(bzrdir, None, _client=client)
384
result = branch.control_files.get('branch.conf')
385
self.assertEqual(
386
[('call_expecting_body', 'Branch.get_config_file', ('///quack/',))],
387
client._calls)
388
self.assertEqual('config file body', result.read())
389
390
391
class TestBranchLockWrite(tests.TestCase):
392
393
def test_lock_write_unlockable(self):
394
client = FakeClient([(('UnlockableTransport', ), '')])
395
transport = MemoryTransport()
396
transport.mkdir('quack')
397
transport = transport.clone('quack')
398
# we do not want bzrdir to make any remote calls
399
bzrdir = RemoteBzrDir(transport, _client=False)
400
branch = RemoteBranch(bzrdir, None, _client=client)
401
self.assertRaises(errors.UnlockableTransport, branch.lock_write)
402
self.assertEqual(
403
[('call', 'Branch.lock_write', ('///quack/', '', ''))],
404
client._calls)
405
406
407
class TestRemoteRepository(tests.TestCase):
408
"""Base for testing RemoteRepository protocol usage.
409
410
These tests contain frozen requests and responses. We want any changes to
411
what is sent or expected to be require a thoughtful update to these tests
412
because they might break compatibility with different-versioned servers.
413
"""
414
415
def setup_fake_client_and_repository(self, responses, transport_path):
416
"""Create the fake client and repository for testing with.
417
418
There's no real server here; we just have canned responses sent
419
back one by one.
420
421
:param transport_path: Path below the root of the MemoryTransport
422
where the repository will be created.
423
"""
424
client = FakeClient(responses)
425
transport = MemoryTransport()
426
transport.mkdir(transport_path)
427
transport = transport.clone(transport_path)
428
# we do not want bzrdir to make any remote calls
429
bzrdir = RemoteBzrDir(transport, _client=False)
430
repo = RemoteRepository(bzrdir, None, _client=client)
431
return repo, client
432
433
434
class TestRepositoryGatherStats(TestRemoteRepository):
435
436
def test_revid_none(self):
437
# ('ok',), body with revisions and size
438
responses = [(('ok', ), 'revisions: 2\nsize: 18\n')]
439
transport_path = 'quack'
440
repo, client = self.setup_fake_client_and_repository(
441
responses, transport_path)
442
result = repo.gather_stats(None)
443
self.assertEqual(
444
[('call_expecting_body', 'Repository.gather_stats',
445
('///quack/','','no'))],
446
client._calls)
447
self.assertEqual({'revisions': 2, 'size': 18}, result)
448
449
def test_revid_no_committers(self):
450
# ('ok',), body without committers
451
responses = [(('ok', ),
452
'firstrev: 123456.300 3600\n'
453
'latestrev: 654231.400 0\n'
454
'revisions: 2\n'
455
'size: 18\n')]
456
transport_path = 'quick'
457
revid = u'\xc8'.encode('utf8')
458
repo, client = self.setup_fake_client_and_repository(
459
responses, transport_path)
460
result = repo.gather_stats(revid)
461
self.assertEqual(
462
[('call_expecting_body', 'Repository.gather_stats',
463
('///quick/', revid, 'no'))],
464
client._calls)
465
self.assertEqual({'revisions': 2, 'size': 18,
466
'firstrev': (123456.300, 3600),
467
'latestrev': (654231.400, 0),},
468
result)
469
470
def test_revid_with_committers(self):
471
# ('ok',), body with committers
472
responses = [(('ok', ),
473
'committers: 128\n'
474
'firstrev: 123456.300 3600\n'
475
'latestrev: 654231.400 0\n'
476
'revisions: 2\n'
477
'size: 18\n')]
478
transport_path = 'buick'
479
revid = u'\xc8'.encode('utf8')
480
repo, client = self.setup_fake_client_and_repository(
481
responses, transport_path)
482
result = repo.gather_stats(revid, True)
483
self.assertEqual(
484
[('call_expecting_body', 'Repository.gather_stats',
485
('///buick/', revid, 'yes'))],
486
client._calls)
487
self.assertEqual({'revisions': 2, 'size': 18,
488
'committers': 128,
489
'firstrev': (123456.300, 3600),
490
'latestrev': (654231.400, 0),},
491
result)
492
493
494
class TestRepositoryGetRevisionGraph(TestRemoteRepository):
495
496
def test_null_revision(self):
497
# a null revision has the predictable result {}, we should have no wire
498
# traffic when calling it with this argument
499
responses = [(('notused', ), '')]
500
transport_path = 'empty'
501
repo, client = self.setup_fake_client_and_repository(
502
responses, transport_path)
503
result = repo.get_revision_graph(NULL_REVISION)
504
self.assertEqual([], client._calls)
505
self.assertEqual({}, result)
506
507
def test_none_revision(self):
508
# with none we want the entire graph
509
r1 = u'\u0e33'.encode('utf8')
510
r2 = u'\u0dab'.encode('utf8')
511
lines = [' '.join([r2, r1]), r1]
512
encoded_body = '\n'.join(lines)
513
514
responses = [(('ok', ), encoded_body)]
515
transport_path = 'sinhala'
516
repo, client = self.setup_fake_client_and_repository(
517
responses, transport_path)
518
result = repo.get_revision_graph()
519
self.assertEqual(
520
[('call_expecting_body', 'Repository.get_revision_graph',
521
('///sinhala/', ''))],
522
client._calls)
523
self.assertEqual({r1: [], r2: [r1]}, result)
524
525
def test_specific_revision(self):
526
# with a specific revision we want the graph for that
527
# with none we want the entire graph
528
r11 = u'\u0e33'.encode('utf8')
529
r12 = u'\xc9'.encode('utf8')
530
r2 = u'\u0dab'.encode('utf8')
531
lines = [' '.join([r2, r11, r12]), r11, r12]
532
encoded_body = '\n'.join(lines)
533
534
responses = [(('ok', ), encoded_body)]
535
transport_path = 'sinhala'
536
repo, client = self.setup_fake_client_and_repository(
537
responses, transport_path)
538
result = repo.get_revision_graph(r2)
539
self.assertEqual(
540
[('call_expecting_body', 'Repository.get_revision_graph',
541
('///sinhala/', r2))],
542
client._calls)
543
self.assertEqual({r11: [], r12: [], r2: [r11, r12], }, result)
544
545
def test_no_such_revision(self):
546
revid = '123'
547
responses = [(('nosuchrevision', revid), '')]
548
transport_path = 'sinhala'
549
repo, client = self.setup_fake_client_and_repository(
550
responses, transport_path)
551
# also check that the right revision is reported in the error
552
self.assertRaises(errors.NoSuchRevision,
553
repo.get_revision_graph, revid)
554
self.assertEqual(
555
[('call_expecting_body', 'Repository.get_revision_graph',
556
('///sinhala/', revid))],
557
client._calls)
558
559
560
class TestRepositoryIsShared(TestRemoteRepository):
561
562
def test_is_shared(self):
563
# ('yes', ) for Repository.is_shared -> 'True'.
564
responses = [(('yes', ), )]
565
transport_path = 'quack'
566
repo, client = self.setup_fake_client_and_repository(
567
responses, transport_path)
568
result = repo.is_shared()
569
self.assertEqual(
570
[('call', 'Repository.is_shared', ('///quack/',))],
571
client._calls)
572
self.assertEqual(True, result)
573
574
def test_is_not_shared(self):
575
# ('no', ) for Repository.is_shared -> 'False'.
576
responses = [(('no', ), )]
577
transport_path = 'qwack'
578
repo, client = self.setup_fake_client_and_repository(
579
responses, transport_path)
580
result = repo.is_shared()
581
self.assertEqual(
582
[('call', 'Repository.is_shared', ('///qwack/',))],
583
client._calls)
584
self.assertEqual(False, result)
585
586
587
class TestRepositoryLockWrite(TestRemoteRepository):
588
589
def test_lock_write(self):
590
responses = [(('ok', 'a token'), '')]
591
transport_path = 'quack'
592
repo, client = self.setup_fake_client_and_repository(
593
responses, transport_path)
594
result = repo.lock_write()
595
self.assertEqual(
596
[('call', 'Repository.lock_write', ('///quack/', ''))],
597
client._calls)
598
self.assertEqual('a token', result)
599
600
def test_lock_write_already_locked(self):
601
responses = [(('LockContention', ), '')]
602
transport_path = 'quack'
603
repo, client = self.setup_fake_client_and_repository(
604
responses, transport_path)
605
self.assertRaises(errors.LockContention, repo.lock_write)
606
self.assertEqual(
607
[('call', 'Repository.lock_write', ('///quack/', ''))],
608
client._calls)
609
610
def test_lock_write_unlockable(self):
611
responses = [(('UnlockableTransport', ), '')]
612
transport_path = 'quack'
613
repo, client = self.setup_fake_client_and_repository(
614
responses, transport_path)
615
self.assertRaises(errors.UnlockableTransport, repo.lock_write)
616
self.assertEqual(
617
[('call', 'Repository.lock_write', ('///quack/', ''))],
618
client._calls)
619
620
621
class TestRepositoryUnlock(TestRemoteRepository):
622
623
def test_unlock(self):
624
responses = [(('ok', 'a token'), ''),
625
(('ok',), '')]
626
transport_path = 'quack'
627
repo, client = self.setup_fake_client_and_repository(
628
responses, transport_path)
629
repo.lock_write()
630
repo.unlock()
631
self.assertEqual(
632
[('call', 'Repository.lock_write', ('///quack/', '')),
633
('call', 'Repository.unlock', ('///quack/', 'a token'))],
634
client._calls)
635
636
def test_unlock_wrong_token(self):
637
# If somehow the token is wrong, unlock will raise TokenMismatch.
638
responses = [(('ok', 'a token'), ''),
639
(('TokenMismatch',), '')]
640
transport_path = 'quack'
641
repo, client = self.setup_fake_client_and_repository(
642
responses, transport_path)
643
repo.lock_write()
644
self.assertRaises(errors.TokenMismatch, repo.unlock)
645
646
647
class TestRepositoryHasRevision(TestRemoteRepository):
648
649
def test_none(self):
650
# repo.has_revision(None) should not cause any traffic.
651
transport_path = 'quack'
652
responses = None
653
repo, client = self.setup_fake_client_and_repository(
654
responses, transport_path)
655
656
# The null revision is always there, so has_revision(None) == True.
657
self.assertEqual(True, repo.has_revision(None))
658
659
# The remote repo shouldn't be accessed.
660
self.assertEqual([], client._calls)
661
662
663
class TestRepositoryTarball(TestRemoteRepository):
664
665
# This is a canned tarball reponse we can validate against
666
tarball_content = (
667
'QlpoOTFBWSZTWdGkj3wAAWF/k8aQACBIB//A9+8cIX/v33AACEAYABAECEACNz'
668
'JqsgJJFPTSnk1A3qh6mTQAAAANPUHkagkSTEkaA09QaNAAAGgAAAcwCYCZGAEY'
669
'mJhMJghpiaYBUkKammSHqNMZQ0NABkNAeo0AGneAevnlwQoGzEzNVzaYxp/1Uk'
670
'xXzA1CQX0BJMZZLcPBrluJir5SQyijWHYZ6ZUtVqqlYDdB2QoCwa9GyWwGYDMA'
671
'OQYhkpLt/OKFnnlT8E0PmO8+ZNSo2WWqeCzGB5fBXZ3IvV7uNJVE7DYnWj6qwB'
672
'k5DJDIrQ5OQHHIjkS9KqwG3mc3t+F1+iujb89ufyBNIKCgeZBWrl5cXxbMGoMs'
673
'c9JuUkg5YsiVcaZJurc6KLi6yKOkgCUOlIlOpOoXyrTJjK8ZgbklReDdwGmFgt'
674
'dkVsAIslSVCd4AtACSLbyhLHryfb14PKegrVDba+U8OL6KQtzdM5HLjAc8/p6n'
675
'0lgaWU8skgO7xupPTkyuwheSckejFLK5T4ZOo0Gda9viaIhpD1Qn7JqqlKAJqC'
676
'QplPKp2nqBWAfwBGaOwVrz3y1T+UZZNismXHsb2Jq18T+VaD9k4P8DqE3g70qV'
677
'JLurpnDI6VS5oqDDPVbtVjMxMxMg4rzQVipn2Bv1fVNK0iq3Gl0hhnnHKm/egy'
678
'nWQ7QH/F3JFOFCQ0aSPfA='
679
).decode('base64')
680
681
def test_repository_tarball(self):
682
# Test that Repository.tarball generates the right operations
683
transport_path = 'repo'
684
expected_responses = [(('ok',), self.tarball_content),
685
]
686
expected_calls = [('call_expecting_body', 'Repository.tarball',
687
('///repo/', 'bz2',),),
688
]
689
remote_repo, client = self.setup_fake_client_and_repository(
690
expected_responses, transport_path)
691
# Now actually ask for the tarball
692
tarball_file = remote_repo._get_tarball('bz2')
693
try:
694
self.assertEqual(expected_calls, client._calls)
695
self.assertEqual(self.tarball_content, tarball_file.read())
696
finally:
697
tarball_file.close()
698
699
def test_sprout_uses_tarball(self):
700
# RemoteRepository.sprout should try to use the
701
# tarball command rather than accessing all the files
702
transport_path = 'srcrepo'
703
expected_responses = [(('ok',), self.tarball_content),
704
]
705
expected_calls = [('call2', 'Repository.tarball', ('///srcrepo/', 'bz2',),),
706
]
707
remote_repo, client = self.setup_fake_client_and_repository(
708
expected_responses, transport_path)
709
# make a regular local repository to receive the results
710
dest_transport = MemoryTransport()
711
dest_transport.mkdir('destrepo')
712
bzrdir_format = bzrdir.format_registry.make_bzrdir('default')
713
dest_bzrdir = bzrdir_format.initialize_on_transport(dest_transport)
714
# try to copy...
715
remote_repo.sprout(dest_bzrdir)
716
717
718
class TestRemoteRepositoryCopyContent(tests.TestCaseWithTransport):
719
"""RemoteRepository.copy_content_into optimizations"""
720
721
def test_copy_content_remote_to_local(self):
722
self.transport_server = server.SmartTCPServer_for_testing
723
src_repo = self.make_repository('repo1')
724
src_repo = repository.Repository.open(self.get_url('repo1'))
725
# At the moment the tarball-based copy_content_into can't write back
726
# into a smart server. It would be good if it could upload the
727
# tarball; once that works we'd have to create repositories of
728
# different formats. -- mbp 20070410
729
dest_url = self.get_vfs_only_url('repo2')
730
dest_bzrdir = BzrDir.create(dest_url)
731
dest_repo = dest_bzrdir.create_repository()
732
self.assertFalse(isinstance(dest_repo, RemoteRepository))
733
self.assertTrue(isinstance(src_repo, RemoteRepository))
734
src_repo.copy_content_into(dest_repo)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1