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 the smart wire/domain protocol.
19
This module contains tests for the domain-level smart requests and responses,
20
such as the 'Branch.lock_write' request.
22
Tests for low-level protocol encoding are found in test_smart_transport.
25
from StringIO import StringIO
29
from bzrlib import bzrdir, errors, pack, smart, tests
30
from bzrlib.smart.request import SmartServerResponse
31
import bzrlib.smart.bzrdir
32
import bzrlib.smart.branch
33
import bzrlib.smart.repository
34
from bzrlib.util import bencode
37
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
40
super(TestCaseWithSmartMedium, self).setUp()
41
# We're allowed to set the transport class here, so that we don't use
42
# the default or a parameterized class, but rather use the
43
# TestCaseWithTransport infrastructure to set up a smart server and
45
self.transport_server = smart.server.SmartTCPServer_for_testing
47
def get_smart_medium(self):
48
"""Get a smart medium to use in tests."""
49
return self.get_transport().get_smart_medium()
52
class TestSmartServerResponse(tests.TestCase):
55
self.assertEqual(SmartServerResponse(('ok', )),
56
SmartServerResponse(('ok', )))
57
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
58
SmartServerResponse(('ok', ), 'body'))
59
self.assertNotEqual(SmartServerResponse(('ok', )),
60
SmartServerResponse(('notok', )))
61
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
62
SmartServerResponse(('ok', )))
63
self.assertNotEqual(None,
64
SmartServerResponse(('ok', )))
67
class TestSmartServerRequestFindRepository(tests.TestCaseWithTransport):
68
"""Tests for BzrDir.find_repository."""
70
def test_no_repository(self):
71
"""When there is no repository to be found, ('norepository', ) is returned."""
72
backing = self.get_transport()
73
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
75
self.assertEqual(SmartServerResponse(('norepository', )),
76
request.execute(backing.local_abspath('')))
78
def test_nonshared_repository(self):
79
# nonshared repositorys only allow 'find' to return a handle when the
80
# path the repository is being searched on is the same as that that
81
# the repository is at.
82
backing = self.get_transport()
83
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
84
result = self._make_repository_and_result()
85
self.assertEqual(result, request.execute(backing.local_abspath('')))
86
self.make_bzrdir('subdir')
87
self.assertEqual(SmartServerResponse(('norepository', )),
88
request.execute(backing.local_abspath('subdir')))
90
def _make_repository_and_result(self, shared=False, format=None):
91
"""Convenience function to setup a repository.
93
:result: The SmartServerResponse to expect when opening it.
95
repo = self.make_repository('.', shared=shared, format=format)
96
if repo.supports_rich_root():
100
if repo._format.supports_tree_reference:
104
return SmartServerResponse(('ok', '', rich_root, subtrees))
106
def test_shared_repository(self):
107
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
108
backing = self.get_transport()
109
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
110
result = self._make_repository_and_result(shared=True)
111
self.assertEqual(result, request.execute(backing.local_abspath('')))
112
self.make_bzrdir('subdir')
113
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
114
self.assertEqual(result2,
115
request.execute(backing.local_abspath('subdir')))
116
self.make_bzrdir('subdir/deeper')
117
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
118
self.assertEqual(result3,
119
request.execute(backing.local_abspath('subdir/deeper')))
121
def test_rich_root_and_subtree_encoding(self):
122
"""Test for the format attributes for rich root and subtree support."""
123
backing = self.get_transport()
124
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
125
result = self._make_repository_and_result(format='dirstate-with-subtree')
126
# check the test will be valid
127
self.assertEqual('yes', result.args[2])
128
self.assertEqual('yes', result.args[3])
129
self.assertEqual(result, request.execute(backing.local_abspath('')))
132
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithTransport):
134
def test_empty_dir(self):
135
"""Initializing an empty dir should succeed and do it."""
136
backing = self.get_transport()
137
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
138
self.assertEqual(SmartServerResponse(('ok', )),
139
request.execute(backing.local_abspath('.')))
140
made_dir = bzrdir.BzrDir.open_from_transport(backing)
141
# no branch, tree or repository is expected with the current
143
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
144
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
145
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
147
def test_missing_dir(self):
148
"""Initializing a missing directory should fail like the bzrdir api."""
149
backing = self.get_transport()
150
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
151
self.assertRaises(errors.NoSuchFile,
152
request.execute, backing.local_abspath('subdir'))
154
def test_initialized_dir(self):
155
"""Initializing an extant bzrdir should fail like the bzrdir api."""
156
backing = self.get_transport()
157
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
158
self.make_bzrdir('subdir')
159
self.assertRaises(errors.FileExists,
160
request.execute, backing.local_abspath('subdir'))
163
class TestSmartServerRequestOpenBranch(tests.TestCaseWithTransport):
165
def test_no_branch(self):
166
"""When there is no branch, ('nobranch', ) is returned."""
167
backing = self.get_transport()
168
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
169
self.make_bzrdir('.')
170
self.assertEqual(SmartServerResponse(('nobranch', )),
171
request.execute(backing.local_abspath('')))
173
def test_branch(self):
174
"""When there is a branch, 'ok' is returned."""
175
backing = self.get_transport()
176
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
177
self.make_branch('.')
178
self.assertEqual(SmartServerResponse(('ok', '')),
179
request.execute(backing.local_abspath('')))
181
def test_branch_reference(self):
182
"""When there is a branch reference, the reference URL is returned."""
183
backing = self.get_transport()
184
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
185
branch = self.make_branch('branch')
186
checkout = branch.create_checkout('reference',lightweight=True)
187
# TODO: once we have an API to probe for references of any sort, we
189
reference_url = backing.abspath('branch') + '/'
190
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
191
self.assertEqual(SmartServerResponse(('ok', reference_url)),
192
request.execute(backing.local_abspath('reference')))
195
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithTransport):
197
def test_empty(self):
198
"""For an empty branch, the body is empty."""
199
backing = self.get_transport()
200
request = smart.branch.SmartServerRequestRevisionHistory(backing)
201
self.make_branch('.')
202
self.assertEqual(SmartServerResponse(('ok', ), ''),
203
request.execute(backing.local_abspath('')))
205
def test_not_empty(self):
206
"""For a non-empty branch, the body is empty."""
207
backing = self.get_transport()
208
request = smart.branch.SmartServerRequestRevisionHistory(backing)
209
tree = self.make_branch_and_memory_tree('.')
212
r1 = tree.commit('1st commit')
213
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
216
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
217
request.execute(backing.local_abspath('')))
220
class TestSmartServerBranchRequest(tests.TestCaseWithTransport):
222
def test_no_branch(self):
223
"""When there is a bzrdir and no branch, NotBranchError is raised."""
224
backing = self.get_transport()
225
request = smart.branch.SmartServerBranchRequest(backing)
226
self.make_bzrdir('.')
227
self.assertRaises(errors.NotBranchError,
228
request.execute, backing.local_abspath(''))
230
def test_branch_reference(self):
231
"""When there is a branch reference, NotBranchError is raised."""
232
backing = self.get_transport()
233
request = smart.branch.SmartServerBranchRequest(backing)
234
branch = self.make_branch('branch')
235
checkout = branch.create_checkout('reference',lightweight=True)
236
self.assertRaises(errors.NotBranchError,
237
request.execute, backing.local_abspath('checkout'))
240
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithTransport):
242
def test_empty(self):
243
"""For an empty branch, the result is ('ok', '0', 'null:')."""
244
backing = self.get_transport()
245
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
246
self.make_branch('.')
247
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
248
request.execute(backing.local_abspath('')))
250
def test_not_empty(self):
251
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
252
backing = self.get_transport()
253
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
254
tree = self.make_branch_and_memory_tree('.')
257
rev_id_utf8 = u'\xc8'.encode('utf-8')
258
r1 = tree.commit('1st commit')
259
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
262
SmartServerResponse(('ok', '2', rev_id_utf8)),
263
request.execute(backing.local_abspath('')))
266
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithTransport):
268
def test_default(self):
269
"""With no file, we get empty content."""
270
backing = self.get_transport()
271
request = smart.branch.SmartServerBranchGetConfigFile(backing)
272
branch = self.make_branch('.')
273
# there should be no file by default
275
self.assertEqual(SmartServerResponse(('ok', ), content),
276
request.execute(backing.local_abspath('')))
278
def test_with_content(self):
279
# SmartServerBranchGetConfigFile should return the content from
280
# branch.control_files.get('branch.conf') for now - in the future it may
281
# perform more complex processing.
282
backing = self.get_transport()
283
request = smart.branch.SmartServerBranchGetConfigFile(backing)
284
branch = self.make_branch('.')
285
branch.control_files.put_utf8('branch.conf', 'foo bar baz')
286
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
287
request.execute(backing.local_abspath('')))
290
class TestSmartServerBranchRequestSetLastRevision(tests.TestCaseWithTransport):
292
def test_empty(self):
293
backing = self.get_transport()
294
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
295
b = self.make_branch('.')
296
branch_token = b.lock_write()
297
repo_token = b.repository.lock_write()
298
b.repository.unlock()
300
self.assertEqual(SmartServerResponse(('ok',)),
302
backing.local_abspath(''), branch_token, repo_token,
307
def test_not_present_revision_id(self):
308
backing = self.get_transport()
309
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
310
b = self.make_branch('.')
311
branch_token = b.lock_write()
312
repo_token = b.repository.lock_write()
313
b.repository.unlock()
315
revision_id = 'non-existent revision'
317
SmartServerResponse(('NoSuchRevision', revision_id)),
319
backing.local_abspath(''), branch_token, repo_token,
324
def test_revision_id_present(self):
325
backing = self.get_transport()
326
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
327
tree = self.make_branch_and_memory_tree('.')
330
rev_id_utf8 = u'\xc8'.encode('utf-8')
331
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
332
r2 = tree.commit('2nd commit')
334
branch_token = tree.branch.lock_write()
335
repo_token = tree.branch.repository.lock_write()
336
tree.branch.repository.unlock()
339
SmartServerResponse(('ok',)),
341
backing.local_abspath(''), branch_token, repo_token,
343
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
347
def test_revision_id_present2(self):
348
backing = self.get_transport()
349
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
350
tree = self.make_branch_and_memory_tree('.')
353
rev_id_utf8 = u'\xc8'.encode('utf-8')
354
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
355
r2 = tree.commit('2nd commit')
357
tree.branch.set_revision_history([])
358
branch_token = tree.branch.lock_write()
359
repo_token = tree.branch.repository.lock_write()
360
tree.branch.repository.unlock()
363
SmartServerResponse(('ok',)),
365
backing.local_abspath(''), branch_token, repo_token,
367
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
372
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithTransport):
375
tests.TestCaseWithTransport.setUp(self)
376
self.reduceLockdirTimeout()
378
def test_lock_write_on_unlocked_branch(self):
379
backing = self.get_transport()
380
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
381
branch = self.make_branch('.')
382
repository = branch.repository
383
response = request.execute(backing.local_abspath(''))
384
branch_nonce = branch.control_files._lock.peek().get('nonce')
385
repository_nonce = repository.control_files._lock.peek().get('nonce')
387
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
389
# The branch (and associated repository) is now locked. Verify that
390
# with a new branch object.
391
new_branch = repository.bzrdir.open_branch()
392
self.assertRaises(errors.LockContention, new_branch.lock_write)
394
def test_lock_write_on_locked_branch(self):
395
backing = self.get_transport()
396
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
397
branch = self.make_branch('.')
399
branch.leave_lock_in_place()
401
response = request.execute(backing.local_abspath(''))
403
SmartServerResponse(('LockContention',)), response)
405
def test_lock_write_with_tokens_on_locked_branch(self):
406
backing = self.get_transport()
407
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
408
branch = self.make_branch('.')
409
branch_token = branch.lock_write()
410
repo_token = branch.repository.lock_write()
411
branch.repository.unlock()
412
branch.leave_lock_in_place()
413
branch.repository.leave_lock_in_place()
415
response = request.execute(backing.local_abspath(''),
416
branch_token, repo_token)
418
SmartServerResponse(('ok', branch_token, repo_token)), response)
420
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
421
backing = self.get_transport()
422
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
423
branch = self.make_branch('.')
424
branch_token = branch.lock_write()
425
repo_token = branch.repository.lock_write()
426
branch.repository.unlock()
427
branch.leave_lock_in_place()
428
branch.repository.leave_lock_in_place()
430
response = request.execute(backing.local_abspath(''),
431
branch_token+'xxx', repo_token)
433
SmartServerResponse(('TokenMismatch',)), response)
435
def test_lock_write_on_locked_repo(self):
436
backing = self.get_transport()
437
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
438
branch = self.make_branch('.')
439
branch.repository.lock_write()
440
branch.repository.leave_lock_in_place()
441
branch.repository.unlock()
442
response = request.execute(backing.local_abspath(''))
444
SmartServerResponse(('LockContention',)), response)
446
def test_lock_write_on_readonly_transport(self):
447
backing = self.get_readonly_transport()
448
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
449
branch = self.make_branch('.')
450
response = request.execute('')
451
error_name, lock_str, why_str = response.args
452
self.assertFalse(response.is_successful())
453
self.assertEqual('LockFailed', error_name)
456
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithTransport):
459
tests.TestCaseWithTransport.setUp(self)
460
self.reduceLockdirTimeout()
462
def test_unlock_on_locked_branch_and_repo(self):
463
backing = self.get_transport()
464
request = smart.branch.SmartServerBranchRequestUnlock(backing)
465
branch = self.make_branch('.')
467
branch_token = branch.lock_write()
468
repo_token = branch.repository.lock_write()
469
branch.repository.unlock()
470
# Unlock the branch (and repo) object, leaving the physical locks
472
branch.leave_lock_in_place()
473
branch.repository.leave_lock_in_place()
475
response = request.execute(backing.local_abspath(''),
476
branch_token, repo_token)
478
SmartServerResponse(('ok',)), response)
479
# The branch is now unlocked. Verify that with a new branch
481
new_branch = branch.bzrdir.open_branch()
482
new_branch.lock_write()
485
def test_unlock_on_unlocked_branch_unlocked_repo(self):
486
backing = self.get_transport()
487
request = smart.branch.SmartServerBranchRequestUnlock(backing)
488
branch = self.make_branch('.')
489
response = request.execute(
490
backing.local_abspath(''), 'branch token', 'repo token')
492
SmartServerResponse(('TokenMismatch',)), response)
494
def test_unlock_on_unlocked_branch_locked_repo(self):
495
backing = self.get_transport()
496
request = smart.branch.SmartServerBranchRequestUnlock(backing)
497
branch = self.make_branch('.')
498
# Lock the repository.
499
repo_token = branch.repository.lock_write()
500
branch.repository.leave_lock_in_place()
501
branch.repository.unlock()
502
# Issue branch lock_write request on the unlocked branch (with locked
504
response = request.execute(
505
backing.local_abspath(''), 'branch token', repo_token)
507
SmartServerResponse(('TokenMismatch',)), response)
510
class TestSmartServerRepositoryRequest(tests.TestCaseWithTransport):
512
def test_no_repository(self):
513
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
514
# we test this using a shared repository above the named path,
515
# thus checking the right search logic is used - that is, that
516
# its the exact path being looked at and the server is not
518
backing = self.get_transport()
519
request = smart.repository.SmartServerRepositoryRequest(backing)
520
self.make_repository('.', shared=True)
521
self.make_bzrdir('subdir')
522
self.assertRaises(errors.NoRepositoryPresent,
523
request.execute, backing.local_abspath('subdir'))
526
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithTransport):
528
def test_none_argument(self):
529
backing = self.get_transport()
530
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
531
tree = self.make_branch_and_memory_tree('.')
534
r1 = tree.commit('1st commit')
535
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
538
# the lines of revision_id->revision_parent_list has no guaranteed
539
# order coming out of a dict, so sort both our test and response
540
lines = sorted([' '.join([r2, r1]), r1])
541
response = request.execute(backing.local_abspath(''), '')
542
response.body = '\n'.join(sorted(response.body.split('\n')))
545
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
547
def test_specific_revision_argument(self):
548
backing = self.get_transport()
549
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
550
tree = self.make_branch_and_memory_tree('.')
553
rev_id_utf8 = u'\xc9'.encode('utf-8')
554
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
555
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
558
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
559
request.execute(backing.local_abspath(''), rev_id_utf8))
561
def test_no_such_revision(self):
562
backing = self.get_transport()
563
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
564
tree = self.make_branch_and_memory_tree('.')
567
r1 = tree.commit('1st commit')
570
# Note that it still returns body (of zero bytes).
572
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
573
request.execute(backing.local_abspath(''), 'missingrevision'))
576
class TestSmartServerRequestHasRevision(tests.TestCaseWithTransport):
578
def test_missing_revision(self):
579
"""For a missing revision, ('no', ) is returned."""
580
backing = self.get_transport()
581
request = smart.repository.SmartServerRequestHasRevision(backing)
582
self.make_repository('.')
583
self.assertEqual(SmartServerResponse(('no', )),
584
request.execute(backing.local_abspath(''), 'revid'))
586
def test_present_revision(self):
587
"""For a present revision, ('yes', ) is returned."""
588
backing = self.get_transport()
589
request = smart.repository.SmartServerRequestHasRevision(backing)
590
tree = self.make_branch_and_memory_tree('.')
593
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
594
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
596
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
597
self.assertEqual(SmartServerResponse(('yes', )),
598
request.execute(backing.local_abspath(''), rev_id_utf8))
601
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithTransport):
603
def test_empty_revid(self):
604
"""With an empty revid, we get only size an number and revisions"""
605
backing = self.get_transport()
606
request = smart.repository.SmartServerRepositoryGatherStats(backing)
607
repository = self.make_repository('.')
608
stats = repository.gather_stats()
610
expected_body = 'revisions: 0\nsize: %d\n' % size
611
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
612
request.execute(backing.local_abspath(''), '', 'no'))
614
def test_revid_with_committers(self):
615
"""For a revid we get more infos."""
616
backing = self.get_transport()
617
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
618
request = smart.repository.SmartServerRepositoryGatherStats(backing)
619
tree = self.make_branch_and_memory_tree('.')
622
# Let's build a predictable result
623
tree.commit('a commit', timestamp=123456.2, timezone=3600)
624
tree.commit('a commit', timestamp=654321.4, timezone=0,
628
stats = tree.branch.repository.gather_stats()
630
expected_body = ('firstrev: 123456.200 3600\n'
631
'latestrev: 654321.400 0\n'
634
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
635
request.execute(backing.local_abspath(''),
638
def test_not_empty_repository_with_committers(self):
639
"""For a revid and requesting committers we get the whole thing."""
640
backing = self.get_transport()
641
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
642
request = smart.repository.SmartServerRepositoryGatherStats(backing)
643
tree = self.make_branch_and_memory_tree('.')
646
# Let's build a predictable result
647
tree.commit('a commit', timestamp=123456.2, timezone=3600,
649
tree.commit('a commit', timestamp=654321.4, timezone=0,
650
committer='bar', rev_id=rev_id_utf8)
652
stats = tree.branch.repository.gather_stats()
655
expected_body = ('committers: 2\n'
656
'firstrev: 123456.200 3600\n'
657
'latestrev: 654321.400 0\n'
660
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
661
request.execute(backing.local_abspath(''),
665
class TestSmartServerRepositoryIsShared(tests.TestCaseWithTransport):
667
def test_is_shared(self):
668
"""For a shared repository, ('yes', ) is returned."""
669
backing = self.get_transport()
670
request = smart.repository.SmartServerRepositoryIsShared(backing)
671
self.make_repository('.', shared=True)
672
self.assertEqual(SmartServerResponse(('yes', )),
673
request.execute(backing.local_abspath(''), ))
675
def test_is_not_shared(self):
676
"""For a shared repository, ('no', ) is returned."""
677
backing = self.get_transport()
678
request = smart.repository.SmartServerRepositoryIsShared(backing)
679
self.make_repository('.', shared=False)
680
self.assertEqual(SmartServerResponse(('no', )),
681
request.execute(backing.local_abspath(''), ))
684
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithTransport):
687
tests.TestCaseWithTransport.setUp(self)
688
self.reduceLockdirTimeout()
690
def test_lock_write_on_unlocked_repo(self):
691
backing = self.get_transport()
692
request = smart.repository.SmartServerRepositoryLockWrite(backing)
693
repository = self.make_repository('.')
694
response = request.execute(backing.local_abspath(''))
695
nonce = repository.control_files._lock.peek().get('nonce')
696
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
697
# The repository is now locked. Verify that with a new repository
699
new_repo = repository.bzrdir.open_repository()
700
self.assertRaises(errors.LockContention, new_repo.lock_write)
702
def test_lock_write_on_locked_repo(self):
703
backing = self.get_transport()
704
request = smart.repository.SmartServerRepositoryLockWrite(backing)
705
repository = self.make_repository('.')
706
repository.lock_write()
707
repository.leave_lock_in_place()
709
response = request.execute(backing.local_abspath(''))
711
SmartServerResponse(('LockContention',)), response)
713
def test_lock_write_on_readonly_transport(self):
714
backing = self.get_readonly_transport()
715
request = smart.repository.SmartServerRepositoryLockWrite(backing)
716
repository = self.make_repository('.')
717
response = request.execute('')
718
self.assertFalse(response.is_successful())
719
self.assertEqual('LockFailed', response.args[0])
722
class TestSmartServerRepositoryUnlock(tests.TestCaseWithTransport):
725
tests.TestCaseWithTransport.setUp(self)
726
self.reduceLockdirTimeout()
728
def test_unlock_on_locked_repo(self):
729
backing = self.get_transport()
730
request = smart.repository.SmartServerRepositoryUnlock(backing)
731
repository = self.make_repository('.')
732
token = repository.lock_write()
733
repository.leave_lock_in_place()
735
response = request.execute(backing.local_abspath(''), token)
737
SmartServerResponse(('ok',)), response)
738
# The repository is now unlocked. Verify that with a new repository
740
new_repo = repository.bzrdir.open_repository()
741
new_repo.lock_write()
744
def test_unlock_on_unlocked_repo(self):
745
backing = self.get_transport()
746
request = smart.repository.SmartServerRepositoryUnlock(backing)
747
repository = self.make_repository('.')
748
response = request.execute(backing.local_abspath(''), 'some token')
750
SmartServerResponse(('TokenMismatch',)), response)
753
class TestSmartServerRepositoryTarball(tests.TestCaseWithTransport):
755
def test_repository_tarball(self):
756
backing = self.get_transport()
757
request = smart.repository.SmartServerRepositoryTarball(backing)
758
repository = self.make_repository('.')
759
# make some extraneous junk in the repository directory which should
761
self.build_tree(['.bzr/repository/extra-junk'])
762
response = request.execute(backing.local_abspath(''), 'bz2')
763
self.assertEqual(('ok',), response.args)
764
# body should be a tbz2
765
body_file = StringIO(response.body)
766
body_tar = tarfile.open('body_tar.tbz2', fileobj=body_file,
768
# let's make sure there are some key repository components inside it.
769
# the tarfile returns directories with trailing slashes...
770
names = set([n.rstrip('/') for n in body_tar.getnames()])
771
self.assertTrue('.bzr/repository/lock' in names)
772
self.assertTrue('.bzr/repository/format' in names)
773
self.assertTrue('.bzr/repository/extra-junk' not in names,
774
"extraneous file present in tar file")
777
class TestSmartServerRepositoryStreamKnitData(tests.TestCaseWithTransport):
779
def test_fetch_revisions(self):
780
backing = self.get_transport()
781
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
782
tree = self.make_branch_and_memory_tree('.')
785
rev_id1_utf8 = u'\xc8'.encode('utf-8')
786
rev_id2_utf8 = u'\xc9'.encode('utf-8')
787
r1 = tree.commit('1st commit', rev_id=rev_id1_utf8)
788
r1 = tree.commit('2nd commit', rev_id=rev_id2_utf8)
791
response = request.execute(backing.local_abspath(''), rev_id2_utf8)
792
self.assertEqual(('ok',), response.args)
793
from cStringIO import StringIO
794
unpacker = pack.ContainerReader(StringIO(response.body))
796
for [name], read_bytes in unpacker.iter_records():
798
bytes = read_bytes(None)
799
# The bytes should be a valid bencoded string.
800
bencode.bdecode(bytes)
801
# XXX: assert that the bencoded knit records have the right
804
def test_no_such_revision_error(self):
805
backing = self.get_transport()
806
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
807
repo = self.make_repository('.')
808
rev_id1_utf8 = u'\xc8'.encode('utf-8')
809
response = request.execute(backing.local_abspath(''), rev_id1_utf8)
811
SmartServerResponse(('NoSuchRevision', rev_id1_utf8)),
815
class TestSmartServerIsReadonly(tests.TestCaseWithTransport):
817
def test_is_readonly_no(self):
818
backing = self.get_transport()
819
request = smart.request.SmartServerIsReadonly(backing)
820
response = request.execute()
822
SmartServerResponse(('no',)), response)
824
def test_is_readonly_yes(self):
825
backing = self.get_readonly_transport()
826
request = smart.request.SmartServerIsReadonly(backing)
827
response = request.execute()
829
SmartServerResponse(('yes',)), response)
832
class TestHandlers(tests.TestCase):
833
"""Tests for the request.request_handlers object."""
835
def test_registered_methods(self):
836
"""Test that known methods are registered to the correct object."""
838
smart.request.request_handlers.get('Branch.get_config_file'),
839
smart.branch.SmartServerBranchGetConfigFile)
841
smart.request.request_handlers.get('Branch.lock_write'),
842
smart.branch.SmartServerBranchRequestLockWrite)
844
smart.request.request_handlers.get('Branch.last_revision_info'),
845
smart.branch.SmartServerBranchRequestLastRevisionInfo)
847
smart.request.request_handlers.get('Branch.revision_history'),
848
smart.branch.SmartServerRequestRevisionHistory)
850
smart.request.request_handlers.get('Branch.set_last_revision'),
851
smart.branch.SmartServerBranchRequestSetLastRevision)
853
smart.request.request_handlers.get('Branch.unlock'),
854
smart.branch.SmartServerBranchRequestUnlock)
856
smart.request.request_handlers.get('BzrDir.find_repository'),
857
smart.bzrdir.SmartServerRequestFindRepository)
859
smart.request.request_handlers.get('BzrDirFormat.initialize'),
860
smart.bzrdir.SmartServerRequestInitializeBzrDir)
862
smart.request.request_handlers.get('BzrDir.open_branch'),
863
smart.bzrdir.SmartServerRequestOpenBranch)
865
smart.request.request_handlers.get('Repository.gather_stats'),
866
smart.repository.SmartServerRepositoryGatherStats)
868
smart.request.request_handlers.get(
869
'Repository.get_revision_graph'),
870
smart.repository.SmartServerRepositoryGetRevisionGraph)
872
smart.request.request_handlers.get('Repository.has_revision'),
873
smart.repository.SmartServerRequestHasRevision)
875
smart.request.request_handlers.get('Repository.is_shared'),
876
smart.repository.SmartServerRepositoryIsShared)
878
smart.request.request_handlers.get('Repository.lock_write'),
879
smart.repository.SmartServerRepositoryLockWrite)
881
smart.request.request_handlers.get(
882
'Repository.stream_knit_data_for_revisions'),
883
smart.repository.SmartServerRepositoryStreamKnitDataForRevisions)
885
smart.request.request_handlers.get('Repository.tarball'),
886
smart.repository.SmartServerRepositoryTarball)
888
smart.request.request_handlers.get('Repository.unlock'),
889
smart.repository.SmartServerRepositoryUnlock)
891
smart.request.request_handlers.get('Transport.is_readonly'),
892
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
