14
14
# along with this program; if not, write to the Free Software
15
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
17
"""Tests for the smart wire/domain protocol."""
19
from StringIO import StringIO
39
from bzrlib.branch import Branch, BranchReferenceFormat
23
from bzrlib import bzrdir, errors, pack, smart, tests
24
from bzrlib.smart.request import SmartServerResponse
25
import bzrlib.smart.bzrdir
40
26
import bzrlib.smart.branch
41
import bzrlib.smart.bzrdir
42
27
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
28
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
31
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
128
57
self.assertNotEqual(None,
129
58
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):
61
class TestSmartServerRequestFindRepository(tests.TestCaseWithTransport):
164
62
"""Tests for BzrDir.find_repository."""
166
64
def test_no_repository(self):
167
65
"""When there is no repository to be found, ('norepository', ) is returned."""
168
66
backing = self.get_transport()
169
request = self._request_class(backing)
67
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
170
68
self.make_bzrdir('.')
171
69
self.assertEqual(SmartServerResponse(('norepository', )),
70
request.execute(backing.local_abspath('')))
174
72
def test_nonshared_repository(self):
175
73
# nonshared repositorys only allow 'find' to return a handle when the
176
74
# path the repository is being searched on is the same as that that
177
75
# the repository is at.
178
76
backing = self.get_transport()
179
request = self._request_class(backing)
77
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
180
78
result = self._make_repository_and_result()
181
self.assertEqual(result, request.execute(''))
79
self.assertEqual(result, request.execute(backing.local_abspath('')))
182
80
self.make_bzrdir('subdir')
183
81
self.assertEqual(SmartServerResponse(('norepository', )),
184
request.execute('subdir'))
82
request.execute(backing.local_abspath('subdir')))
186
84
def _make_repository_and_result(self, shared=False, format=None):
187
85
"""Convenience function to setup a repository.
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))
98
return SmartServerResponse(('ok', '', rich_root, subtrees))
209
100
def test_shared_repository(self):
210
101
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
211
102
backing = self.get_transport()
212
request = self._request_class(backing)
103
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
213
104
result = self._make_repository_and_result(shared=True)
214
self.assertEqual(result, request.execute(''))
105
self.assertEqual(result, request.execute(backing.local_abspath('')))
215
106
self.make_bzrdir('subdir')
216
107
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
217
108
self.assertEqual(result2,
218
request.execute('subdir'))
109
request.execute(backing.local_abspath('subdir')))
219
110
self.make_bzrdir('subdir/deeper')
220
111
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
221
112
self.assertEqual(result3,
222
request.execute('subdir/deeper'))
113
request.execute(backing.local_abspath('subdir/deeper')))
224
115
def test_rich_root_and_subtree_encoding(self):
225
116
"""Test for the format attributes for rich root and subtree support."""
226
117
backing = self.get_transport()
227
request = self._request_class(backing)
118
request = smart.bzrdir.SmartServerRequestFindRepository(backing)
228
119
result = self._make_repository_and_result(format='dirstate-with-subtree')
229
120
# check the test will be valid
230
121
self.assertEqual('yes', result.args[2])
231
122
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):
123
self.assertEqual(result, request.execute(backing.local_abspath('')))
126
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithTransport):
246
128
def test_empty_dir(self):
247
129
"""Initializing an empty dir should succeed and do it."""
248
130
backing = self.get_transport()
249
131
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
250
132
self.assertEqual(SmartServerResponse(('ok', )),
133
request.execute(backing.local_abspath('.')))
252
134
made_dir = bzrdir.BzrDir.open_from_transport(backing)
253
135
# no branch, tree or repository is expected with the current
254
136
# default formart.
392
276
backing = self.get_transport()
393
277
request = smart.branch.SmartServerBranchGetConfigFile(backing)
394
278
branch = self.make_branch('.')
395
branch._transport.put_bytes('branch.conf', 'foo bar baz')
279
branch.control_files.put_utf8('branch.conf', 'foo bar baz')
396
280
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)
281
request.execute(backing.local_abspath('')))
284
class TestSmartServerBranchRequestSetLastRevision(tests.TestCaseWithTransport):
286
def test_empty(self):
287
backing = self.get_transport()
288
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
289
b = self.make_branch('.')
290
branch_token = b.lock_write()
291
repo_token = b.repository.lock_write()
292
b.repository.unlock()
294
self.assertEqual(SmartServerResponse(('ok',)),
296
backing.local_abspath(''), branch_token, repo_token,
301
def test_not_present_revision_id(self):
302
backing = self.get_transport()
303
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
304
b = self.make_branch('.')
305
branch_token = b.lock_write()
306
repo_token = b.repository.lock_write()
307
b.repository.unlock()
309
revision_id = 'non-existent revision'
311
SmartServerResponse(('NoSuchRevision', revision_id)),
313
backing.local_abspath(''), branch_token, repo_token,
318
def test_revision_id_present(self):
319
backing = self.get_transport()
320
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
321
tree = self.make_branch_and_memory_tree('.')
324
rev_id_utf8 = u'\xc8'.encode('utf-8')
325
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
326
r2 = tree.commit('2nd commit')
328
branch_token = tree.branch.lock_write()
329
repo_token = tree.branch.repository.lock_write()
330
tree.branch.repository.unlock()
333
SmartServerResponse(('ok',)),
335
backing.local_abspath(''), branch_token, repo_token,
337
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
341
def test_revision_id_present2(self):
342
backing = self.get_transport()
343
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
344
tree = self.make_branch_and_memory_tree('.')
347
rev_id_utf8 = u'\xc8'.encode('utf-8')
348
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
349
r2 = tree.commit('2nd commit')
351
tree.branch.set_revision_history([])
352
branch_token = tree.branch.lock_write()
353
repo_token = tree.branch.repository.lock_write()
354
tree.branch.repository.unlock()
357
SmartServerResponse(('ok',)),
359
backing.local_abspath(''), branch_token, repo_token,
361
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
366
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithTransport):
369
tests.TestCaseWithTransport.setUp(self)
370
self.reduceLockdirTimeout()
623
372
def test_lock_write_on_unlocked_branch(self):
624
373
backing = self.get_transport()
625
374
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
626
branch = self.make_branch('.', format='knit')
375
branch = self.make_branch('.')
627
376
repository = branch.repository
628
response = request.execute('')
377
response = request.execute(backing.local_abspath(''))
629
378
branch_nonce = branch.control_files._lock.peek().get('nonce')
630
379
repository_nonce = repository.control_files._lock.peek().get('nonce')
631
380
self.assertEqual(
999
738
def test_unlock_on_unlocked_repo(self):
1000
739
backing = self.get_transport()
1001
740
request = smart.repository.SmartServerRepositoryUnlock(backing)
1002
repository = self.make_repository('.', format='knit')
1003
response = request.execute('', 'some token')
741
repository = self.make_repository('.')
742
response = request.execute(backing.local_abspath(''), 'some token')
1004
743
self.assertEqual(
1005
744
SmartServerResponse(('TokenMismatch',)), response)
1008
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
747
class TestSmartServerRepositoryTarball(tests.TestCaseWithTransport):
749
def test_repository_tarball(self):
750
backing = self.get_transport()
751
request = smart.repository.SmartServerRepositoryTarball(backing)
752
repository = self.make_repository('.')
753
# make some extraneous junk in the repository directory which should
755
self.build_tree(['.bzr/repository/extra-junk'])
756
response = request.execute(backing.local_abspath(''), 'bz2')
757
self.assertEqual(('ok',), response.args)
758
# body should be a tbz2
759
body_file = StringIO(response.body)
760
body_tar = tarfile.open('body_tar.tbz2', fileobj=body_file,
762
# let's make sure there are some key repository components inside it.
763
# the tarfile returns directories with trailing slashes...
764
names = set([n.rstrip('/') for n in body_tar.getnames()])
765
self.assertTrue('.bzr/repository/lock' in names)
766
self.assertTrue('.bzr/repository/format' in names)
767
self.assertTrue('.bzr/repository/extra-junk' not in names,
768
"extraneous file present in tar file")
771
class TestSmartServerRepositoryStreamKnitData(tests.TestCaseWithTransport):
773
def test_fetch_revisions(self):
774
backing = self.get_transport()
775
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
776
tree = self.make_branch_and_memory_tree('.')
779
rev_id1_utf8 = u'\xc8'.encode('utf-8')
780
rev_id2_utf8 = u'\xc9'.encode('utf-8')
781
r1 = tree.commit('1st commit', rev_id=rev_id1_utf8)
782
r1 = tree.commit('2nd commit', rev_id=rev_id2_utf8)
785
response = request.execute(backing.local_abspath(''), rev_id2_utf8)
786
self.assertEqual(('ok',), response.args)
787
from cStringIO import StringIO
788
unpacker = pack.ContainerReader(StringIO(response.body))
790
for [name], read_bytes in unpacker.iter_records():
792
bytes = read_bytes(None)
793
# The bytes should be a valid bencoded string.
794
bencode.bdecode(bytes)
795
# XXX: assert that the bencoded knit records have the right
798
def test_no_such_revision_error(self):
799
backing = self.get_transport()
800
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
801
repo = self.make_repository('.')
802
rev_id1_utf8 = u'\xc8'.encode('utf-8')
803
response = request.execute(backing.local_abspath(''), rev_id1_utf8)
805
SmartServerResponse(('NoSuchRevision', rev_id1_utf8)),
809
class TestSmartServerIsReadonly(tests.TestCaseWithTransport):
1010
811
def test_is_readonly_no(self):
1011
812
backing = self.get_transport()
added
removed
bzrlib/tests/test_smart.py
