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
"<SuccessfulSmartServerResponse args=('args',) body='body'>",
135
str(SuccessfulSmartServerResponse(('args',), 'body')))
137
"<FailedSmartServerResponse 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 TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
620
def test_get_stacked_on_url(self):
621
base_branch = self.make_branch('base', format='1.6')
622
stacked_branch = self.make_branch('stacked', format='1.6')
623
# typically should be relative
624
stacked_branch.set_stacked_on_url('../base')
625
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
626
self.get_transport())
627
response = request.execute('stacked')
629
SmartServerResponse(('ok', '../base')),
633
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
636
tests.TestCaseWithMemoryTransport.setUp(self)
638
def test_lock_write_on_unlocked_branch(self):
639
backing = self.get_transport()
640
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
641
branch = self.make_branch('.', format='knit')
642
repository = branch.repository
643
response = request.execute('')
644
branch_nonce = branch.control_files._lock.peek().get('nonce')
645
repository_nonce = repository.control_files._lock.peek().get('nonce')
647
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
649
# The branch (and associated repository) is now locked. Verify that
650
# with a new branch object.
651
new_branch = repository.bzrdir.open_branch()
652
self.assertRaises(errors.LockContention, new_branch.lock_write)
654
def test_lock_write_on_locked_branch(self):
655
backing = self.get_transport()
656
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
657
branch = self.make_branch('.')
659
branch.leave_lock_in_place()
661
response = request.execute('')
663
SmartServerResponse(('LockContention',)), response)
665
def test_lock_write_with_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, repo_token)
678
SmartServerResponse(('ok', branch_token, repo_token)), response)
680
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
681
backing = self.get_transport()
682
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
683
branch = self.make_branch('.', format='knit')
684
branch_token = branch.lock_write()
685
repo_token = branch.repository.lock_write()
686
branch.repository.unlock()
687
branch.leave_lock_in_place()
688
branch.repository.leave_lock_in_place()
690
response = request.execute('',
691
branch_token+'xxx', repo_token)
693
SmartServerResponse(('TokenMismatch',)), response)
695
def test_lock_write_on_locked_repo(self):
696
backing = self.get_transport()
697
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
698
branch = self.make_branch('.', format='knit')
699
branch.repository.lock_write()
700
branch.repository.leave_lock_in_place()
701
branch.repository.unlock()
702
response = request.execute('')
704
SmartServerResponse(('LockContention',)), response)
706
def test_lock_write_on_readonly_transport(self):
707
backing = self.get_readonly_transport()
708
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
709
branch = self.make_branch('.')
710
root = self.get_transport().clone('/')
711
path = urlutils.relative_url(root.base, self.get_transport().base)
712
response = request.execute(path)
713
error_name, lock_str, why_str = response.args
714
self.assertFalse(response.is_successful())
715
self.assertEqual('LockFailed', error_name)
718
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
721
tests.TestCaseWithMemoryTransport.setUp(self)
723
def test_unlock_on_locked_branch_and_repo(self):
724
backing = self.get_transport()
725
request = smart.branch.SmartServerBranchRequestUnlock(backing)
726
branch = self.make_branch('.', format='knit')
728
branch_token = branch.lock_write()
729
repo_token = branch.repository.lock_write()
730
branch.repository.unlock()
731
# Unlock the branch (and repo) object, leaving the physical locks
733
branch.leave_lock_in_place()
734
branch.repository.leave_lock_in_place()
736
response = request.execute('',
737
branch_token, repo_token)
739
SmartServerResponse(('ok',)), response)
740
# The branch is now unlocked. Verify that with a new branch
742
new_branch = branch.bzrdir.open_branch()
743
new_branch.lock_write()
746
def test_unlock_on_unlocked_branch_unlocked_repo(self):
747
backing = self.get_transport()
748
request = smart.branch.SmartServerBranchRequestUnlock(backing)
749
branch = self.make_branch('.', format='knit')
750
response = request.execute(
751
'', 'branch token', 'repo token')
753
SmartServerResponse(('TokenMismatch',)), response)
755
def test_unlock_on_unlocked_branch_locked_repo(self):
756
backing = self.get_transport()
757
request = smart.branch.SmartServerBranchRequestUnlock(backing)
758
branch = self.make_branch('.', format='knit')
759
# Lock the repository.
760
repo_token = branch.repository.lock_write()
761
branch.repository.leave_lock_in_place()
762
branch.repository.unlock()
763
# Issue branch lock_write request on the unlocked branch (with locked
765
response = request.execute(
766
'', 'branch token', repo_token)
768
SmartServerResponse(('TokenMismatch',)), response)
771
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
773
def test_no_repository(self):
774
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
775
# we test this using a shared repository above the named path,
776
# thus checking the right search logic is used - that is, that
777
# its the exact path being looked at and the server is not
779
backing = self.get_transport()
780
request = smart.repository.SmartServerRepositoryRequest(backing)
781
self.make_repository('.', shared=True)
782
self.make_bzrdir('subdir')
783
self.assertRaises(errors.NoRepositoryPresent,
784
request.execute, 'subdir')
787
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
789
def test_trivial_bzipped(self):
790
# This tests that the wire encoding is actually bzipped
791
backing = self.get_transport()
792
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
793
tree = self.make_branch_and_memory_tree('.')
795
self.assertEqual(None,
796
request.execute('', 'missing-id'))
797
# Note that it returns a body (of '' bzipped).
799
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
800
request.do_body('\n\n0\n'))
803
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
805
def test_none_argument(self):
806
backing = self.get_transport()
807
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
808
tree = self.make_branch_and_memory_tree('.')
811
r1 = tree.commit('1st commit')
812
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
815
# the lines of revision_id->revision_parent_list has no guaranteed
816
# order coming out of a dict, so sort both our test and response
817
lines = sorted([' '.join([r2, r1]), r1])
818
response = request.execute('', '')
819
response.body = '\n'.join(sorted(response.body.split('\n')))
822
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
824
def test_specific_revision_argument(self):
825
backing = self.get_transport()
826
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
827
tree = self.make_branch_and_memory_tree('.')
830
rev_id_utf8 = u'\xc9'.encode('utf-8')
831
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
832
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
835
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
836
request.execute('', rev_id_utf8))
838
def test_no_such_revision(self):
839
backing = self.get_transport()
840
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
841
tree = self.make_branch_and_memory_tree('.')
844
r1 = tree.commit('1st commit')
847
# Note that it still returns body (of zero bytes).
849
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
850
request.execute('', 'missingrevision'))
853
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
855
def test_missing_revision(self):
856
"""For a missing revision, ('no', ) is returned."""
857
backing = self.get_transport()
858
request = smart.repository.SmartServerRequestHasRevision(backing)
859
self.make_repository('.')
860
self.assertEqual(SmartServerResponse(('no', )),
861
request.execute('', 'revid'))
863
def test_present_revision(self):
864
"""For a present revision, ('yes', ) is returned."""
865
backing = self.get_transport()
866
request = smart.repository.SmartServerRequestHasRevision(backing)
867
tree = self.make_branch_and_memory_tree('.')
870
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
871
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
873
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
874
self.assertEqual(SmartServerResponse(('yes', )),
875
request.execute('', rev_id_utf8))
878
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
880
def test_empty_revid(self):
881
"""With an empty revid, we get only size an number and revisions"""
882
backing = self.get_transport()
883
request = smart.repository.SmartServerRepositoryGatherStats(backing)
884
repository = self.make_repository('.')
885
stats = repository.gather_stats()
886
expected_body = 'revisions: 0\n'
887
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
888
request.execute('', '', 'no'))
890
def test_revid_with_committers(self):
891
"""For a revid we get more infos."""
892
backing = self.get_transport()
893
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
894
request = smart.repository.SmartServerRepositoryGatherStats(backing)
895
tree = self.make_branch_and_memory_tree('.')
898
# Let's build a predictable result
899
tree.commit('a commit', timestamp=123456.2, timezone=3600)
900
tree.commit('a commit', timestamp=654321.4, timezone=0,
904
stats = tree.branch.repository.gather_stats()
905
expected_body = ('firstrev: 123456.200 3600\n'
906
'latestrev: 654321.400 0\n'
908
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
912
def test_not_empty_repository_with_committers(self):
913
"""For a revid and requesting committers we get the whole thing."""
914
backing = self.get_transport()
915
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
916
request = smart.repository.SmartServerRepositoryGatherStats(backing)
917
tree = self.make_branch_and_memory_tree('.')
920
# Let's build a predictable result
921
tree.commit('a commit', timestamp=123456.2, timezone=3600,
923
tree.commit('a commit', timestamp=654321.4, timezone=0,
924
committer='bar', rev_id=rev_id_utf8)
926
stats = tree.branch.repository.gather_stats()
928
expected_body = ('committers: 2\n'
929
'firstrev: 123456.200 3600\n'
930
'latestrev: 654321.400 0\n'
932
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
937
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
939
def test_is_shared(self):
940
"""For a shared repository, ('yes', ) is returned."""
941
backing = self.get_transport()
942
request = smart.repository.SmartServerRepositoryIsShared(backing)
943
self.make_repository('.', shared=True)
944
self.assertEqual(SmartServerResponse(('yes', )),
945
request.execute('', ))
947
def test_is_not_shared(self):
948
"""For a shared repository, ('no', ) is returned."""
949
backing = self.get_transport()
950
request = smart.repository.SmartServerRepositoryIsShared(backing)
951
self.make_repository('.', shared=False)
952
self.assertEqual(SmartServerResponse(('no', )),
953
request.execute('', ))
956
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
959
tests.TestCaseWithMemoryTransport.setUp(self)
961
def test_lock_write_on_unlocked_repo(self):
962
backing = self.get_transport()
963
request = smart.repository.SmartServerRepositoryLockWrite(backing)
964
repository = self.make_repository('.', format='knit')
965
response = request.execute('')
966
nonce = repository.control_files._lock.peek().get('nonce')
967
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
968
# The repository is now locked. Verify that with a new repository
970
new_repo = repository.bzrdir.open_repository()
971
self.assertRaises(errors.LockContention, new_repo.lock_write)
973
def test_lock_write_on_locked_repo(self):
974
backing = self.get_transport()
975
request = smart.repository.SmartServerRepositoryLockWrite(backing)
976
repository = self.make_repository('.', format='knit')
977
repository.lock_write()
978
repository.leave_lock_in_place()
980
response = request.execute('')
982
SmartServerResponse(('LockContention',)), response)
984
def test_lock_write_on_readonly_transport(self):
985
backing = self.get_readonly_transport()
986
request = smart.repository.SmartServerRepositoryLockWrite(backing)
987
repository = self.make_repository('.', format='knit')
988
response = request.execute('')
989
self.assertFalse(response.is_successful())
990
self.assertEqual('LockFailed', response.args[0])
993
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
996
tests.TestCaseWithMemoryTransport.setUp(self)
998
def test_unlock_on_locked_repo(self):
999
backing = self.get_transport()
1000
request = smart.repository.SmartServerRepositoryUnlock(backing)
1001
repository = self.make_repository('.', format='knit')
1002
token = repository.lock_write()
1003
repository.leave_lock_in_place()
1005
response = request.execute('', token)
1007
SmartServerResponse(('ok',)), response)
1008
# The repository is now unlocked. Verify that with a new repository
1010
new_repo = repository.bzrdir.open_repository()
1011
new_repo.lock_write()
1014
def test_unlock_on_unlocked_repo(self):
1015
backing = self.get_transport()
1016
request = smart.repository.SmartServerRepositoryUnlock(backing)
1017
repository = self.make_repository('.', format='knit')
1018
response = request.execute('', 'some token')
1020
SmartServerResponse(('TokenMismatch',)), response)
1023
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1025
def test_is_readonly_no(self):
1026
backing = self.get_transport()
1027
request = smart.request.SmartServerIsReadonly(backing)
1028
response = request.execute()
1030
SmartServerResponse(('no',)), response)
1032
def test_is_readonly_yes(self):
1033
backing = self.get_readonly_transport()
1034
request = smart.request.SmartServerIsReadonly(backing)
1035
response = request.execute()
1037
SmartServerResponse(('yes',)), response)
1040
class TestHandlers(tests.TestCase):
1041
"""Tests for the request.request_handlers object."""
1043
def test_all_registrations_exist(self):
1044
"""All registered request_handlers can be found."""
1045
# If there's a typo in a register_lazy call, this loop will fail with
1046
# an AttributeError.
1047
for key, item in smart.request.request_handlers.iteritems():
1050
def test_registered_methods(self):
1051
"""Test that known methods are registered to the correct object."""
1053
smart.request.request_handlers.get('Branch.get_config_file'),
1054
smart.branch.SmartServerBranchGetConfigFile)
1056
smart.request.request_handlers.get('Branch.lock_write'),
1057
smart.branch.SmartServerBranchRequestLockWrite)
1059
smart.request.request_handlers.get('Branch.last_revision_info'),
1060
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1062
smart.request.request_handlers.get('Branch.revision_history'),
1063
smart.branch.SmartServerRequestRevisionHistory)
1065
smart.request.request_handlers.get('Branch.set_last_revision'),
1066
smart.branch.SmartServerBranchRequestSetLastRevision)
1068
smart.request.request_handlers.get('Branch.set_last_revision_info'),
1069
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1071
smart.request.request_handlers.get('Branch.unlock'),
1072
smart.branch.SmartServerBranchRequestUnlock)
1074
smart.request.request_handlers.get('BzrDir.find_repository'),
1075
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1077
smart.request.request_handlers.get('BzrDir.find_repositoryV2'),
1078
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1080
smart.request.request_handlers.get('BzrDirFormat.initialize'),
1081
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1083
smart.request.request_handlers.get('BzrDir.open_branch'),
1084
smart.bzrdir.SmartServerRequestOpenBranch)
1086
smart.request.request_handlers.get('Repository.gather_stats'),
1087
smart.repository.SmartServerRepositoryGatherStats)
1089
smart.request.request_handlers.get('Repository.get_parent_map'),
1090
smart.repository.SmartServerRepositoryGetParentMap)
1092
smart.request.request_handlers.get(
1093
'Repository.get_revision_graph'),
1094
smart.repository.SmartServerRepositoryGetRevisionGraph)
1096
smart.request.request_handlers.get('Repository.has_revision'),
1097
smart.repository.SmartServerRequestHasRevision)
1099
smart.request.request_handlers.get('Repository.is_shared'),
1100
smart.repository.SmartServerRepositoryIsShared)
1102
smart.request.request_handlers.get('Repository.lock_write'),
1103
smart.repository.SmartServerRepositoryLockWrite)
1105
smart.request.request_handlers.get('Repository.tarball'),
1106
smart.repository.SmartServerRepositoryTarball)
1108
smart.request.request_handlers.get('Repository.unlock'),
1109
smart.repository.SmartServerRepositoryUnlock)
1111
smart.request.request_handlers.get('Transport.is_readonly'),
1112
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
