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. Many of these use specific disk
21
formats to exercise calls that only make sense for formats with specific
24
Tests for low-level protocol encoding are found in test_smart_transport.
28
from cStringIO import StringIO
39
from bzrlib.branch import Branch, BranchReferenceFormat
40
import bzrlib.smart.branch
41
import bzrlib.smart.bzrdir
42
import bzrlib.smart.repository
43
from bzrlib.smart.request import (
44
FailedSmartServerResponse,
47
SuccessfulSmartServerResponse,
49
from bzrlib.tests import (
54
from bzrlib.transport import chroot, get_transport
55
from bzrlib.util import bencode
58
def load_tests(standard_tests, module, loader):
59
"""Multiply tests version and protocol consistency."""
60
# FindRepository tests.
61
bzrdir_mod = bzrlib.smart.bzrdir
62
applier = TestScenarioApplier()
65
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),
66
("find_repositoryV2", {
67
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),
69
to_adapt, result = split_suite_by_re(standard_tests,
70
"TestSmartServerRequestFindRepository")
71
v2_only, v1_and_2 = split_suite_by_re(to_adapt,
73
for test in iter_suite_tests(v1_and_2):
74
result.addTests(applier.adapt(test))
75
del applier.scenarios[0]
76
for test in iter_suite_tests(v2_only):
77
result.addTests(applier.adapt(test))
81
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
84
tests.TestCaseWithTransport.setUp(self)
85
self._chroot_server = None
87
def get_transport(self, relpath=None):
88
if self._chroot_server is None:
89
backing_transport = tests.TestCaseWithTransport.get_transport(self)
90
self._chroot_server = chroot.ChrootServer(backing_transport)
91
self._chroot_server.setUp()
92
self.addCleanup(self._chroot_server.tearDown)
93
t = get_transport(self._chroot_server.get_url())
94
if relpath is not None:
99
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
102
super(TestCaseWithSmartMedium, self).setUp()
103
# We're allowed to set the transport class here, so that we don't use
104
# the default or a parameterized class, but rather use the
105
# TestCaseWithTransport infrastructure to set up a smart server and
107
self.transport_server = self.make_transport_server
109
def make_transport_server(self):
110
return smart.server.SmartTCPServer_for_testing('-' + self.id())
112
def get_smart_medium(self):
113
"""Get a smart medium to use in tests."""
114
return self.get_transport().get_smart_medium()
117
class TestSmartServerResponse(tests.TestCase):
119
def test__eq__(self):
120
self.assertEqual(SmartServerResponse(('ok', )),
121
SmartServerResponse(('ok', )))
122
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
123
SmartServerResponse(('ok', ), 'body'))
124
self.assertNotEqual(SmartServerResponse(('ok', )),
125
SmartServerResponse(('notok', )))
126
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
127
SmartServerResponse(('ok', )))
128
self.assertNotEqual(None,
129
SmartServerResponse(('ok', )))
131
def test__str__(self):
132
"""SmartServerResponses can be stringified."""
134
"<SmartServerResponse status=OK args=('args',) body='body'>",
135
str(SuccessfulSmartServerResponse(('args',), 'body')))
137
"<SmartServerResponse status=ERR args=('args',) body='body'>",
138
str(FailedSmartServerResponse(('args',), 'body')))
141
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
143
def test_translate_client_path(self):
144
transport = self.get_transport()
145
request = SmartServerRequest(transport, 'foo/')
146
self.assertEqual('./', request.translate_client_path('foo/'))
148
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
150
errors.PathNotChild, request.translate_client_path, '/')
152
errors.PathNotChild, request.translate_client_path, 'bar/')
153
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
155
def test_transport_from_client_path(self):
156
transport = self.get_transport()
157
request = SmartServerRequest(transport, 'foo/')
160
request.transport_from_client_path('foo/').base)
163
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
164
"""Tests for BzrDir.find_repository."""
166
def test_no_repository(self):
167
"""When there is no repository to be found, ('norepository', ) is returned."""
168
backing = self.get_transport()
169
request = self._request_class(backing)
170
self.make_bzrdir('.')
171
self.assertEqual(SmartServerResponse(('norepository', )),
174
def test_nonshared_repository(self):
175
# nonshared repositorys only allow 'find' to return a handle when the
176
# path the repository is being searched on is the same as that that
177
# the repository is at.
178
backing = self.get_transport()
179
request = self._request_class(backing)
180
result = self._make_repository_and_result()
181
self.assertEqual(result, request.execute(''))
182
self.make_bzrdir('subdir')
183
self.assertEqual(SmartServerResponse(('norepository', )),
184
request.execute('subdir'))
186
def _make_repository_and_result(self, shared=False, format=None):
187
"""Convenience function to setup a repository.
189
:result: The SmartServerResponse to expect when opening it.
191
repo = self.make_repository('.', shared=shared, format=format)
192
if repo.supports_rich_root():
196
if repo._format.supports_tree_reference:
200
if (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
201
self._request_class):
202
# All tests so far are on formats, and for non-external
204
return SuccessfulSmartServerResponse(
205
('ok', '', rich_root, subtrees, 'no'))
207
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
209
def test_shared_repository(self):
210
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
211
backing = self.get_transport()
212
request = self._request_class(backing)
213
result = self._make_repository_and_result(shared=True)
214
self.assertEqual(result, request.execute(''))
215
self.make_bzrdir('subdir')
216
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
217
self.assertEqual(result2,
218
request.execute('subdir'))
219
self.make_bzrdir('subdir/deeper')
220
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
221
self.assertEqual(result3,
222
request.execute('subdir/deeper'))
224
def test_rich_root_and_subtree_encoding(self):
225
"""Test for the format attributes for rich root and subtree support."""
226
backing = self.get_transport()
227
request = self._request_class(backing)
228
result = self._make_repository_and_result(format='dirstate-with-subtree')
229
# check the test will be valid
230
self.assertEqual('yes', result.args[2])
231
self.assertEqual('yes', result.args[3])
232
self.assertEqual(result, request.execute(''))
234
def test_supports_external_lookups_no_v2(self):
235
"""Test for the supports_external_lookups attribute."""
236
backing = self.get_transport()
237
request = self._request_class(backing)
238
result = self._make_repository_and_result(format='dirstate-with-subtree')
239
# check the test will be valid
240
self.assertEqual('no', result.args[4])
241
self.assertEqual(result, request.execute(''))
244
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
246
def test_empty_dir(self):
247
"""Initializing an empty dir should succeed and do it."""
248
backing = self.get_transport()
249
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
250
self.assertEqual(SmartServerResponse(('ok', )),
252
made_dir = bzrdir.BzrDir.open_from_transport(backing)
253
# no branch, tree or repository is expected with the current
255
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
256
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
257
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
259
def test_missing_dir(self):
260
"""Initializing a missing directory should fail like the bzrdir api."""
261
backing = self.get_transport()
262
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
263
self.assertRaises(errors.NoSuchFile,
264
request.execute, 'subdir')
266
def test_initialized_dir(self):
267
"""Initializing an extant bzrdir should fail like the bzrdir api."""
268
backing = self.get_transport()
269
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
270
self.make_bzrdir('subdir')
271
self.assertRaises(errors.FileExists,
272
request.execute, 'subdir')
275
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
277
def test_no_branch(self):
278
"""When there is no branch, ('nobranch', ) is returned."""
279
backing = self.get_transport()
280
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
281
self.make_bzrdir('.')
282
self.assertEqual(SmartServerResponse(('nobranch', )),
285
def test_branch(self):
286
"""When there is a branch, 'ok' is returned."""
287
backing = self.get_transport()
288
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
289
self.make_branch('.')
290
self.assertEqual(SmartServerResponse(('ok', '')),
293
def test_branch_reference(self):
294
"""When there is a branch reference, the reference URL is returned."""
295
backing = self.get_transport()
296
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
297
branch = self.make_branch('branch')
298
checkout = branch.create_checkout('reference',lightweight=True)
299
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
300
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
301
self.assertEqual(SmartServerResponse(('ok', reference_url)),
302
request.execute('reference'))
305
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
307
def test_empty(self):
308
"""For an empty branch, the body is empty."""
309
backing = self.get_transport()
310
request = smart.branch.SmartServerRequestRevisionHistory(backing)
311
self.make_branch('.')
312
self.assertEqual(SmartServerResponse(('ok', ), ''),
315
def test_not_empty(self):
316
"""For a non-empty branch, the body is empty."""
317
backing = self.get_transport()
318
request = smart.branch.SmartServerRequestRevisionHistory(backing)
319
tree = self.make_branch_and_memory_tree('.')
322
r1 = tree.commit('1st commit')
323
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
326
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
330
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
332
def test_no_branch(self):
333
"""When there is a bzrdir and no branch, NotBranchError is raised."""
334
backing = self.get_transport()
335
request = smart.branch.SmartServerBranchRequest(backing)
336
self.make_bzrdir('.')
337
self.assertRaises(errors.NotBranchError,
340
def test_branch_reference(self):
341
"""When there is a branch reference, NotBranchError is raised."""
342
backing = self.get_transport()
343
request = smart.branch.SmartServerBranchRequest(backing)
344
branch = self.make_branch('branch')
345
checkout = branch.create_checkout('reference',lightweight=True)
346
self.assertRaises(errors.NotBranchError,
347
request.execute, 'checkout')
350
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
352
def test_empty(self):
353
"""For an empty branch, the result is ('ok', '0', 'null:')."""
354
backing = self.get_transport()
355
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
356
self.make_branch('.')
357
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
360
def test_not_empty(self):
361
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
362
backing = self.get_transport()
363
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
364
tree = self.make_branch_and_memory_tree('.')
367
rev_id_utf8 = u'\xc8'.encode('utf-8')
368
r1 = tree.commit('1st commit')
369
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
372
SmartServerResponse(('ok', '2', rev_id_utf8)),
376
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
378
def test_default(self):
379
"""With no file, we get empty content."""
380
backing = self.get_transport()
381
request = smart.branch.SmartServerBranchGetConfigFile(backing)
382
branch = self.make_branch('.')
383
# there should be no file by default
385
self.assertEqual(SmartServerResponse(('ok', ), content),
388
def test_with_content(self):
389
# SmartServerBranchGetConfigFile should return the content from
390
# branch.control_files.get('branch.conf') for now - in the future it may
391
# perform more complex processing.
392
backing = self.get_transport()
393
request = smart.branch.SmartServerBranchGetConfigFile(backing)
394
branch = self.make_branch('.')
395
branch._transport.put_bytes('branch.conf', 'foo bar baz')
396
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
400
class SetLastRevisionTestBase(tests.TestCaseWithMemoryTransport):
401
"""Base test case for verbs that implement set_last_revision."""
404
tests.TestCaseWithMemoryTransport.setUp(self)
405
backing_transport = self.get_transport()
406
self.request = self.request_class(backing_transport)
407
self.tree = self.make_branch_and_memory_tree('.')
409
def lock_branch(self):
411
branch_token = b.lock_write()
412
repo_token = b.repository.lock_write()
413
b.repository.unlock()
414
return branch_token, repo_token
416
def unlock_branch(self):
417
self.tree.branch.unlock()
419
def set_last_revision(self, revision_id, revno):
420
branch_token, repo_token = self.lock_branch()
421
response = self._set_last_revision(
422
revision_id, revno, branch_token, repo_token)
426
def assertRequestSucceeds(self, revision_id, revno):
427
response = self.set_last_revision(revision_id, revno)
428
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
431
class TestSetLastRevisionVerbMixin(object):
432
"""Mixin test case for verbs that implement set_last_revision."""
434
def test_set_null_to_null(self):
435
"""An empty branch can have its last revision set to 'null:'."""
436
self.assertRequestSucceeds('null:', 0)
438
def test_NoSuchRevision(self):
439
"""If the revision_id is not present, the verb returns NoSuchRevision.
441
revision_id = 'non-existent revision'
443
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
444
self.set_last_revision(revision_id, 1))
446
def make_tree_with_two_commits(self):
447
self.tree.lock_write()
449
rev_id_utf8 = u'\xc8'.encode('utf-8')
450
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
451
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
454
def test_branch_last_revision_info_is_updated(self):
455
"""A branch's tip can be set to a revision that is present in its
458
# Make a branch with an empty revision history, but two revisions in
460
self.make_tree_with_two_commits()
461
rev_id_utf8 = u'\xc8'.encode('utf-8')
462
self.tree.branch.set_revision_history([])
464
(0, 'null:'), self.tree.branch.last_revision_info())
465
# We can update the branch to a revision that is present in the
467
self.assertRequestSucceeds(rev_id_utf8, 1)
469
(1, rev_id_utf8), self.tree.branch.last_revision_info())
471
def test_branch_last_revision_info_rewind(self):
472
"""A branch's tip can be set to a revision that is an ancestor of the
475
self.make_tree_with_two_commits()
476
rev_id_utf8 = u'\xc8'.encode('utf-8')
478
(2, 'rev-2'), self.tree.branch.last_revision_info())
479
self.assertRequestSucceeds(rev_id_utf8, 1)
481
(1, rev_id_utf8), self.tree.branch.last_revision_info())
483
def test_TipChangeRejected(self):
484
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
485
returns TipChangeRejected.
487
rejection_message = u'rejection message\N{INTERROBANG}'
488
def hook_that_rejects(params):
489
raise errors.TipChangeRejected(rejection_message)
490
Branch.hooks.install_named_hook(
491
'pre_change_branch_tip', hook_that_rejects, None)
493
FailedSmartServerResponse(
494
('TipChangeRejected', rejection_message.encode('utf-8'))),
495
self.set_last_revision('null:', 0))
498
class TestSmartServerBranchRequestSetLastRevision(
499
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
500
"""Tests for Branch.set_last_revision verb."""
502
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
504
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
505
return self.request.execute(
506
'', branch_token, repo_token, revision_id)
509
class TestSmartServerBranchRequestSetLastRevisionInfo(
510
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
511
"""Tests for Branch.set_last_revision_info verb."""
513
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
515
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
516
return self.request.execute(
517
'', branch_token, repo_token, revno, revision_id)
519
def test_NoSuchRevision(self):
520
"""Branch.set_last_revision_info does not have to return
521
NoSuchRevision if the revision_id is absent.
523
raise tests.TestNotApplicable()
526
class TestSmartServerBranchRequestSetLastRevisionEx(
527
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
528
"""Tests for Branch.set_last_revision_ex verb."""
530
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
532
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
533
return self.request.execute(
534
'', branch_token, repo_token, revision_id, 0, 0)
536
def assertRequestSucceeds(self, revision_id, revno):
537
response = self.set_last_revision(revision_id, revno)
539
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
542
def test_branch_last_revision_info_rewind(self):
543
"""A branch's tip can be set to a revision that is an ancestor of the
544
current tip, but only if allow_overwrite_descendant is passed.
546
self.make_tree_with_two_commits()
547
rev_id_utf8 = u'\xc8'.encode('utf-8')
549
(2, 'rev-2'), self.tree.branch.last_revision_info())
550
# If allow_overwrite_descendant flag is 0, then trying to set the tip
551
# to an older revision ID has no effect.
552
branch_token, repo_token = self.lock_branch()
553
response = self.request.execute(
554
'', branch_token, repo_token, rev_id_utf8, 0, 0)
556
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
559
(2, 'rev-2'), self.tree.branch.last_revision_info())
561
# If allow_overwrite_descendant flag is 1, then setting the tip to an
563
response = self.request.execute(
564
'', branch_token, repo_token, rev_id_utf8, 0, 1)
566
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
570
(1, rev_id_utf8), self.tree.branch.last_revision_info())
572
def make_branch_with_divergent_history(self):
573
"""Make a branch with divergent history in its repo.
575
The branch's tip will be 'child-2', and the repo will also contain
576
'child-1', which diverges from a common base revision.
578
self.tree.lock_write()
580
r1 = self.tree.commit('1st commit')
581
revno_1, revid_1 = self.tree.branch.last_revision_info()
582
r2 = self.tree.commit('2nd commit', rev_id='child-1')
583
# Undo the second commit
584
self.tree.branch.set_last_revision_info(revno_1, revid_1)
585
self.tree.set_parent_ids([revid_1])
586
# Make a new second commit, child-2. child-2 has diverged from
588
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
591
def test_not_allow_diverged(self):
592
"""If allow_diverged is not passed, then setting a divergent history
593
returns a Diverged error.
595
self.make_branch_with_divergent_history()
597
FailedSmartServerResponse(('Diverged',)),
598
self.set_last_revision('child-1', 2))
599
# The branch tip was not changed.
600
self.assertEqual('child-2', self.tree.branch.last_revision())
602
def test_allow_diverged(self):
603
"""If allow_diverged is passed, then setting a divergent history
606
self.make_branch_with_divergent_history()
607
branch_token, repo_token = self.lock_branch()
608
response = self.request.execute(
609
'', branch_token, repo_token, 'child-1', 1, 0)
611
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
614
# The branch tip was changed.
615
self.assertEqual('child-1', self.tree.branch.last_revision())
618
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
621
tests.TestCaseWithMemoryTransport.setUp(self)
623
def test_lock_write_on_unlocked_branch(self):
624
backing = self.get_transport()
625
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
626
branch = self.make_branch('.', format='knit')
627
repository = branch.repository
628
response = request.execute('')
629
branch_nonce = branch.control_files._lock.peek().get('nonce')
630
repository_nonce = repository.control_files._lock.peek().get('nonce')
632
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
634
# The branch (and associated repository) is now locked. Verify that
635
# with a new branch object.
636
new_branch = repository.bzrdir.open_branch()
637
self.assertRaises(errors.LockContention, new_branch.lock_write)
639
def test_lock_write_on_locked_branch(self):
640
backing = self.get_transport()
641
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
642
branch = self.make_branch('.')
644
branch.leave_lock_in_place()
646
response = request.execute('')
648
SmartServerResponse(('LockContention',)), response)
650
def test_lock_write_with_tokens_on_locked_branch(self):
651
backing = self.get_transport()
652
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
653
branch = self.make_branch('.', format='knit')
654
branch_token = branch.lock_write()
655
repo_token = branch.repository.lock_write()
656
branch.repository.unlock()
657
branch.leave_lock_in_place()
658
branch.repository.leave_lock_in_place()
660
response = request.execute('',
661
branch_token, repo_token)
663
SmartServerResponse(('ok', branch_token, repo_token)), response)
665
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
666
backing = self.get_transport()
667
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
668
branch = self.make_branch('.', format='knit')
669
branch_token = branch.lock_write()
670
repo_token = branch.repository.lock_write()
671
branch.repository.unlock()
672
branch.leave_lock_in_place()
673
branch.repository.leave_lock_in_place()
675
response = request.execute('',
676
branch_token+'xxx', repo_token)
678
SmartServerResponse(('TokenMismatch',)), response)
680
def test_lock_write_on_locked_repo(self):
681
backing = self.get_transport()
682
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
683
branch = self.make_branch('.', format='knit')
684
branch.repository.lock_write()
685
branch.repository.leave_lock_in_place()
686
branch.repository.unlock()
687
response = request.execute('')
689
SmartServerResponse(('LockContention',)), response)
691
def test_lock_write_on_readonly_transport(self):
692
backing = self.get_readonly_transport()
693
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
694
branch = self.make_branch('.')
695
root = self.get_transport().clone('/')
696
path = urlutils.relative_url(root.base, self.get_transport().base)
697
response = request.execute(path)
698
error_name, lock_str, why_str = response.args
699
self.assertFalse(response.is_successful())
700
self.assertEqual('LockFailed', error_name)
703
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
706
tests.TestCaseWithMemoryTransport.setUp(self)
708
def test_unlock_on_locked_branch_and_repo(self):
709
backing = self.get_transport()
710
request = smart.branch.SmartServerBranchRequestUnlock(backing)
711
branch = self.make_branch('.', format='knit')
713
branch_token = branch.lock_write()
714
repo_token = branch.repository.lock_write()
715
branch.repository.unlock()
716
# Unlock the branch (and repo) object, leaving the physical locks
718
branch.leave_lock_in_place()
719
branch.repository.leave_lock_in_place()
721
response = request.execute('',
722
branch_token, repo_token)
724
SmartServerResponse(('ok',)), response)
725
# The branch is now unlocked. Verify that with a new branch
727
new_branch = branch.bzrdir.open_branch()
728
new_branch.lock_write()
731
def test_unlock_on_unlocked_branch_unlocked_repo(self):
732
backing = self.get_transport()
733
request = smart.branch.SmartServerBranchRequestUnlock(backing)
734
branch = self.make_branch('.', format='knit')
735
response = request.execute(
736
'', 'branch token', 'repo token')
738
SmartServerResponse(('TokenMismatch',)), response)
740
def test_unlock_on_unlocked_branch_locked_repo(self):
741
backing = self.get_transport()
742
request = smart.branch.SmartServerBranchRequestUnlock(backing)
743
branch = self.make_branch('.', format='knit')
744
# Lock the repository.
745
repo_token = branch.repository.lock_write()
746
branch.repository.leave_lock_in_place()
747
branch.repository.unlock()
748
# Issue branch lock_write request on the unlocked branch (with locked
750
response = request.execute(
751
'', 'branch token', repo_token)
753
SmartServerResponse(('TokenMismatch',)), response)
756
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
758
def test_no_repository(self):
759
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
760
# we test this using a shared repository above the named path,
761
# thus checking the right search logic is used - that is, that
762
# its the exact path being looked at and the server is not
764
backing = self.get_transport()
765
request = smart.repository.SmartServerRepositoryRequest(backing)
766
self.make_repository('.', shared=True)
767
self.make_bzrdir('subdir')
768
self.assertRaises(errors.NoRepositoryPresent,
769
request.execute, 'subdir')
772
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
774
def test_trivial_bzipped(self):
775
# This tests that the wire encoding is actually bzipped
776
backing = self.get_transport()
777
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
778
tree = self.make_branch_and_memory_tree('.')
780
self.assertEqual(None,
781
request.execute('', 'missing-id'))
782
# Note that it returns a body (of '' bzipped).
784
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
785
request.do_body('\n\n0\n'))
788
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
790
def test_none_argument(self):
791
backing = self.get_transport()
792
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
793
tree = self.make_branch_and_memory_tree('.')
796
r1 = tree.commit('1st commit')
797
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
800
# the lines of revision_id->revision_parent_list has no guaranteed
801
# order coming out of a dict, so sort both our test and response
802
lines = sorted([' '.join([r2, r1]), r1])
803
response = request.execute('', '')
804
response.body = '\n'.join(sorted(response.body.split('\n')))
807
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
809
def test_specific_revision_argument(self):
810
backing = self.get_transport()
811
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
812
tree = self.make_branch_and_memory_tree('.')
815
rev_id_utf8 = u'\xc9'.encode('utf-8')
816
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
817
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
820
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
821
request.execute('', rev_id_utf8))
823
def test_no_such_revision(self):
824
backing = self.get_transport()
825
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
826
tree = self.make_branch_and_memory_tree('.')
829
r1 = tree.commit('1st commit')
832
# Note that it still returns body (of zero bytes).
834
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
835
request.execute('', 'missingrevision'))
838
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
840
def test_missing_revision(self):
841
"""For a missing revision, ('no', ) is returned."""
842
backing = self.get_transport()
843
request = smart.repository.SmartServerRequestHasRevision(backing)
844
self.make_repository('.')
845
self.assertEqual(SmartServerResponse(('no', )),
846
request.execute('', 'revid'))
848
def test_present_revision(self):
849
"""For a present revision, ('yes', ) is returned."""
850
backing = self.get_transport()
851
request = smart.repository.SmartServerRequestHasRevision(backing)
852
tree = self.make_branch_and_memory_tree('.')
855
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
856
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
858
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
859
self.assertEqual(SmartServerResponse(('yes', )),
860
request.execute('', rev_id_utf8))
863
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
865
def test_empty_revid(self):
866
"""With an empty revid, we get only size an number and revisions"""
867
backing = self.get_transport()
868
request = smart.repository.SmartServerRepositoryGatherStats(backing)
869
repository = self.make_repository('.')
870
stats = repository.gather_stats()
871
expected_body = 'revisions: 0\n'
872
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
873
request.execute('', '', 'no'))
875
def test_revid_with_committers(self):
876
"""For a revid we get more infos."""
877
backing = self.get_transport()
878
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
879
request = smart.repository.SmartServerRepositoryGatherStats(backing)
880
tree = self.make_branch_and_memory_tree('.')
883
# Let's build a predictable result
884
tree.commit('a commit', timestamp=123456.2, timezone=3600)
885
tree.commit('a commit', timestamp=654321.4, timezone=0,
889
stats = tree.branch.repository.gather_stats()
890
expected_body = ('firstrev: 123456.200 3600\n'
891
'latestrev: 654321.400 0\n'
893
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
897
def test_not_empty_repository_with_committers(self):
898
"""For a revid and requesting committers we get the whole thing."""
899
backing = self.get_transport()
900
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
901
request = smart.repository.SmartServerRepositoryGatherStats(backing)
902
tree = self.make_branch_and_memory_tree('.')
905
# Let's build a predictable result
906
tree.commit('a commit', timestamp=123456.2, timezone=3600,
908
tree.commit('a commit', timestamp=654321.4, timezone=0,
909
committer='bar', rev_id=rev_id_utf8)
911
stats = tree.branch.repository.gather_stats()
913
expected_body = ('committers: 2\n'
914
'firstrev: 123456.200 3600\n'
915
'latestrev: 654321.400 0\n'
917
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
922
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
924
def test_is_shared(self):
925
"""For a shared repository, ('yes', ) is returned."""
926
backing = self.get_transport()
927
request = smart.repository.SmartServerRepositoryIsShared(backing)
928
self.make_repository('.', shared=True)
929
self.assertEqual(SmartServerResponse(('yes', )),
930
request.execute('', ))
932
def test_is_not_shared(self):
933
"""For a shared repository, ('no', ) is returned."""
934
backing = self.get_transport()
935
request = smart.repository.SmartServerRepositoryIsShared(backing)
936
self.make_repository('.', shared=False)
937
self.assertEqual(SmartServerResponse(('no', )),
938
request.execute('', ))
941
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
944
tests.TestCaseWithMemoryTransport.setUp(self)
946
def test_lock_write_on_unlocked_repo(self):
947
backing = self.get_transport()
948
request = smart.repository.SmartServerRepositoryLockWrite(backing)
949
repository = self.make_repository('.', format='knit')
950
response = request.execute('')
951
nonce = repository.control_files._lock.peek().get('nonce')
952
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
953
# The repository is now locked. Verify that with a new repository
955
new_repo = repository.bzrdir.open_repository()
956
self.assertRaises(errors.LockContention, new_repo.lock_write)
958
def test_lock_write_on_locked_repo(self):
959
backing = self.get_transport()
960
request = smart.repository.SmartServerRepositoryLockWrite(backing)
961
repository = self.make_repository('.', format='knit')
962
repository.lock_write()
963
repository.leave_lock_in_place()
965
response = request.execute('')
967
SmartServerResponse(('LockContention',)), response)
969
def test_lock_write_on_readonly_transport(self):
970
backing = self.get_readonly_transport()
971
request = smart.repository.SmartServerRepositoryLockWrite(backing)
972
repository = self.make_repository('.', format='knit')
973
response = request.execute('')
974
self.assertFalse(response.is_successful())
975
self.assertEqual('LockFailed', response.args[0])
978
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
981
tests.TestCaseWithMemoryTransport.setUp(self)
983
def test_unlock_on_locked_repo(self):
984
backing = self.get_transport()
985
request = smart.repository.SmartServerRepositoryUnlock(backing)
986
repository = self.make_repository('.', format='knit')
987
token = repository.lock_write()
988
repository.leave_lock_in_place()
990
response = request.execute('', token)
992
SmartServerResponse(('ok',)), response)
993
# The repository is now unlocked. Verify that with a new repository
995
new_repo = repository.bzrdir.open_repository()
996
new_repo.lock_write()
999
def test_unlock_on_unlocked_repo(self):
1000
backing = self.get_transport()
1001
request = smart.repository.SmartServerRepositoryUnlock(backing)
1002
repository = self.make_repository('.', format='knit')
1003
response = request.execute('', 'some token')
1005
SmartServerResponse(('TokenMismatch',)), response)
1008
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1010
def test_is_readonly_no(self):
1011
backing = self.get_transport()
1012
request = smart.request.SmartServerIsReadonly(backing)
1013
response = request.execute()
1015
SmartServerResponse(('no',)), response)
1017
def test_is_readonly_yes(self):
1018
backing = self.get_readonly_transport()
1019
request = smart.request.SmartServerIsReadonly(backing)
1020
response = request.execute()
1022
SmartServerResponse(('yes',)), response)
1025
class TestHandlers(tests.TestCase):
1026
"""Tests for the request.request_handlers object."""
1028
def test_all_registrations_exist(self):
1029
"""All registered request_handlers can be found."""
1030
# If there's a typo in a register_lazy call, this loop will fail with
1031
# an AttributeError.
1032
for key, item in smart.request.request_handlers.iteritems():
1035
def test_registered_methods(self):
1036
"""Test that known methods are registered to the correct object."""
1038
smart.request.request_handlers.get('Branch.get_config_file'),
1039
smart.branch.SmartServerBranchGetConfigFile)
1041
smart.request.request_handlers.get('Branch.lock_write'),
1042
smart.branch.SmartServerBranchRequestLockWrite)
1044
smart.request.request_handlers.get('Branch.last_revision_info'),
1045
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1047
smart.request.request_handlers.get('Branch.revision_history'),
1048
smart.branch.SmartServerRequestRevisionHistory)
1050
smart.request.request_handlers.get('Branch.set_last_revision'),
1051
smart.branch.SmartServerBranchRequestSetLastRevision)
1053
smart.request.request_handlers.get('Branch.set_last_revision_info'),
1054
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1056
smart.request.request_handlers.get('Branch.unlock'),
1057
smart.branch.SmartServerBranchRequestUnlock)
1059
smart.request.request_handlers.get('BzrDir.find_repository'),
1060
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1062
smart.request.request_handlers.get('BzrDir.find_repositoryV2'),
1063
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1065
smart.request.request_handlers.get('BzrDirFormat.initialize'),
1066
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1068
smart.request.request_handlers.get('BzrDir.open_branch'),
1069
smart.bzrdir.SmartServerRequestOpenBranch)
1071
smart.request.request_handlers.get('Repository.gather_stats'),
1072
smart.repository.SmartServerRepositoryGatherStats)
1074
smart.request.request_handlers.get('Repository.get_parent_map'),
1075
smart.repository.SmartServerRepositoryGetParentMap)
1077
smart.request.request_handlers.get(
1078
'Repository.get_revision_graph'),
1079
smart.repository.SmartServerRepositoryGetRevisionGraph)
1081
smart.request.request_handlers.get('Repository.has_revision'),
1082
smart.repository.SmartServerRequestHasRevision)
1084
smart.request.request_handlers.get('Repository.is_shared'),
1085
smart.repository.SmartServerRepositoryIsShared)
1087
smart.request.request_handlers.get('Repository.lock_write'),
1088
smart.repository.SmartServerRepositoryLockWrite)
1090
smart.request.request_handlers.get('Repository.tarball'),
1091
smart.repository.SmartServerRepositoryTarball)
1093
smart.request.request_handlers.get('Repository.unlock'),
1094
smart.repository.SmartServerRepositoryUnlock)
1096
smart.request.request_handlers.get('Transport.is_readonly'),
1097
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
