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}),
68
("find_repositoryV3", {
69
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV3}),
71
to_adapt, result = split_suite_by_re(standard_tests,
72
"TestSmartServerRequestFindRepository")
73
v2_only, v1_and_2 = split_suite_by_re(to_adapt,
75
for test in iter_suite_tests(v1_and_2):
76
result.addTests(applier.adapt(test))
77
del applier.scenarios[0]
78
for test in iter_suite_tests(v2_only):
79
result.addTests(applier.adapt(test))
83
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
86
tests.TestCaseWithTransport.setUp(self)
87
self._chroot_server = None
89
def get_transport(self, relpath=None):
90
if self._chroot_server is None:
91
backing_transport = tests.TestCaseWithTransport.get_transport(self)
92
self._chroot_server = chroot.ChrootServer(backing_transport)
93
self._chroot_server.setUp()
94
self.addCleanup(self._chroot_server.tearDown)
95
t = get_transport(self._chroot_server.get_url())
96
if relpath is not None:
101
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
104
super(TestCaseWithSmartMedium, self).setUp()
105
# We're allowed to set the transport class here, so that we don't use
106
# the default or a parameterized class, but rather use the
107
# TestCaseWithTransport infrastructure to set up a smart server and
109
self.transport_server = self.make_transport_server
111
def make_transport_server(self):
112
return smart.server.SmartTCPServer_for_testing('-' + self.id())
114
def get_smart_medium(self):
115
"""Get a smart medium to use in tests."""
116
return self.get_transport().get_smart_medium()
119
class TestSmartServerResponse(tests.TestCase):
121
def test__eq__(self):
122
self.assertEqual(SmartServerResponse(('ok', )),
123
SmartServerResponse(('ok', )))
124
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
125
SmartServerResponse(('ok', ), 'body'))
126
self.assertNotEqual(SmartServerResponse(('ok', )),
127
SmartServerResponse(('notok', )))
128
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
129
SmartServerResponse(('ok', )))
130
self.assertNotEqual(None,
131
SmartServerResponse(('ok', )))
133
def test__str__(self):
134
"""SmartServerResponses can be stringified."""
136
"<SuccessfulSmartServerResponse args=('args',) body='body'>",
137
str(SuccessfulSmartServerResponse(('args',), 'body')))
139
"<FailedSmartServerResponse args=('args',) body='body'>",
140
str(FailedSmartServerResponse(('args',), 'body')))
143
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
145
def test_translate_client_path(self):
146
transport = self.get_transport()
147
request = SmartServerRequest(transport, 'foo/')
148
self.assertEqual('./', request.translate_client_path('foo/'))
150
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
152
errors.PathNotChild, request.translate_client_path, '/')
154
errors.PathNotChild, request.translate_client_path, 'bar/')
155
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
157
def test_transport_from_client_path(self):
158
transport = self.get_transport()
159
request = SmartServerRequest(transport, 'foo/')
162
request.transport_from_client_path('foo/').base)
165
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
166
"""Tests for BzrDir.create_repository."""
168
def test_makes_repository(self):
169
"""When there is a bzrdir present, the call succeeds."""
170
backing = self.get_transport()
171
self.make_bzrdir('.')
172
request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
173
request = request_class(backing)
174
reference_bzrdir_format = bzrdir.format_registry.get('default')()
175
reference_format = reference_bzrdir_format.repository_format
176
network_name = reference_format.network_name()
177
expected = SuccessfulSmartServerResponse(
178
('ok', 'no', 'no', 'no', network_name))
179
self.assertEqual(expected, request.execute('', network_name, 'True'))
182
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
183
"""Tests for BzrDir.find_repository."""
185
def test_no_repository(self):
186
"""When there is no repository to be found, ('norepository', ) is returned."""
187
backing = self.get_transport()
188
request = self._request_class(backing)
189
self.make_bzrdir('.')
190
self.assertEqual(SmartServerResponse(('norepository', )),
193
def test_nonshared_repository(self):
194
# nonshared repositorys only allow 'find' to return a handle when the
195
# path the repository is being searched on is the same as that that
196
# the repository is at.
197
backing = self.get_transport()
198
request = self._request_class(backing)
199
result = self._make_repository_and_result()
200
self.assertEqual(result, request.execute(''))
201
self.make_bzrdir('subdir')
202
self.assertEqual(SmartServerResponse(('norepository', )),
203
request.execute('subdir'))
205
def _make_repository_and_result(self, shared=False, format=None):
206
"""Convenience function to setup a repository.
208
:result: The SmartServerResponse to expect when opening it.
210
repo = self.make_repository('.', shared=shared, format=format)
211
if repo.supports_rich_root():
215
if repo._format.supports_tree_reference:
219
if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
220
self._request_class):
221
return SuccessfulSmartServerResponse(
222
('ok', '', rich_root, subtrees, 'no',
223
repo._format.network_name()))
224
elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
225
self._request_class):
226
# All tests so far are on formats, and for non-external
228
return SuccessfulSmartServerResponse(
229
('ok', '', rich_root, subtrees, 'no'))
231
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
233
def test_shared_repository(self):
234
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
235
backing = self.get_transport()
236
request = self._request_class(backing)
237
result = self._make_repository_and_result(shared=True)
238
self.assertEqual(result, request.execute(''))
239
self.make_bzrdir('subdir')
240
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
241
self.assertEqual(result2,
242
request.execute('subdir'))
243
self.make_bzrdir('subdir/deeper')
244
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
245
self.assertEqual(result3,
246
request.execute('subdir/deeper'))
248
def test_rich_root_and_subtree_encoding(self):
249
"""Test for the format attributes for rich root and subtree support."""
250
backing = self.get_transport()
251
request = self._request_class(backing)
252
result = self._make_repository_and_result(format='dirstate-with-subtree')
253
# check the test will be valid
254
self.assertEqual('yes', result.args[2])
255
self.assertEqual('yes', result.args[3])
256
self.assertEqual(result, request.execute(''))
258
def test_supports_external_lookups_no_v2(self):
259
"""Test for the supports_external_lookups attribute."""
260
backing = self.get_transport()
261
request = self._request_class(backing)
262
result = self._make_repository_and_result(format='dirstate-with-subtree')
263
# check the test will be valid
264
self.assertEqual('no', result.args[4])
265
self.assertEqual(result, request.execute(''))
268
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
270
def test_empty_dir(self):
271
"""Initializing an empty dir should succeed and do it."""
272
backing = self.get_transport()
273
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
274
self.assertEqual(SmartServerResponse(('ok', )),
276
made_dir = bzrdir.BzrDir.open_from_transport(backing)
277
# no branch, tree or repository is expected with the current
279
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
280
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
281
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
283
def test_missing_dir(self):
284
"""Initializing a missing directory should fail like the bzrdir api."""
285
backing = self.get_transport()
286
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
287
self.assertRaises(errors.NoSuchFile,
288
request.execute, 'subdir')
290
def test_initialized_dir(self):
291
"""Initializing an extant bzrdir should fail like the bzrdir api."""
292
backing = self.get_transport()
293
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
294
self.make_bzrdir('subdir')
295
self.assertRaises(errors.FileExists,
296
request.execute, 'subdir')
299
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
301
def test_no_branch(self):
302
"""When there is no branch, ('nobranch', ) is returned."""
303
backing = self.get_transport()
304
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
305
self.make_bzrdir('.')
306
self.assertEqual(SmartServerResponse(('nobranch', )),
309
def test_branch(self):
310
"""When there is a branch, 'ok' is returned."""
311
backing = self.get_transport()
312
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
313
self.make_branch('.')
314
self.assertEqual(SmartServerResponse(('ok', '')),
317
def test_branch_reference(self):
318
"""When there is a branch reference, the reference URL is returned."""
319
backing = self.get_transport()
320
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
321
branch = self.make_branch('branch')
322
checkout = branch.create_checkout('reference',lightweight=True)
323
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
324
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
325
self.assertEqual(SmartServerResponse(('ok', reference_url)),
326
request.execute('reference'))
329
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
331
def test_empty(self):
332
"""For an empty branch, the body is empty."""
333
backing = self.get_transport()
334
request = smart.branch.SmartServerRequestRevisionHistory(backing)
335
self.make_branch('.')
336
self.assertEqual(SmartServerResponse(('ok', ), ''),
339
def test_not_empty(self):
340
"""For a non-empty branch, the body is empty."""
341
backing = self.get_transport()
342
request = smart.branch.SmartServerRequestRevisionHistory(backing)
343
tree = self.make_branch_and_memory_tree('.')
346
r1 = tree.commit('1st commit')
347
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
350
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
354
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
356
def test_no_branch(self):
357
"""When there is a bzrdir and no branch, NotBranchError is raised."""
358
backing = self.get_transport()
359
request = smart.branch.SmartServerBranchRequest(backing)
360
self.make_bzrdir('.')
361
self.assertRaises(errors.NotBranchError,
364
def test_branch_reference(self):
365
"""When there is a branch reference, NotBranchError is raised."""
366
backing = self.get_transport()
367
request = smart.branch.SmartServerBranchRequest(backing)
368
branch = self.make_branch('branch')
369
checkout = branch.create_checkout('reference',lightweight=True)
370
self.assertRaises(errors.NotBranchError,
371
request.execute, 'checkout')
374
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
376
def test_empty(self):
377
"""For an empty branch, the result is ('ok', '0', 'null:')."""
378
backing = self.get_transport()
379
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
380
self.make_branch('.')
381
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
384
def test_not_empty(self):
385
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
386
backing = self.get_transport()
387
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
388
tree = self.make_branch_and_memory_tree('.')
391
rev_id_utf8 = u'\xc8'.encode('utf-8')
392
r1 = tree.commit('1st commit')
393
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
396
SmartServerResponse(('ok', '2', rev_id_utf8)),
400
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
402
def test_default(self):
403
"""With no file, we get empty content."""
404
backing = self.get_transport()
405
request = smart.branch.SmartServerBranchGetConfigFile(backing)
406
branch = self.make_branch('.')
407
# there should be no file by default
409
self.assertEqual(SmartServerResponse(('ok', ), content),
412
def test_with_content(self):
413
# SmartServerBranchGetConfigFile should return the content from
414
# branch.control_files.get('branch.conf') for now - in the future it may
415
# perform more complex processing.
416
backing = self.get_transport()
417
request = smart.branch.SmartServerBranchGetConfigFile(backing)
418
branch = self.make_branch('.')
419
branch._transport.put_bytes('branch.conf', 'foo bar baz')
420
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
424
class SetLastRevisionTestBase(tests.TestCaseWithMemoryTransport):
425
"""Base test case for verbs that implement set_last_revision."""
428
tests.TestCaseWithMemoryTransport.setUp(self)
429
backing_transport = self.get_transport()
430
self.request = self.request_class(backing_transport)
431
self.tree = self.make_branch_and_memory_tree('.')
433
def lock_branch(self):
435
branch_token = b.lock_write()
436
repo_token = b.repository.lock_write()
437
b.repository.unlock()
438
return branch_token, repo_token
440
def unlock_branch(self):
441
self.tree.branch.unlock()
443
def set_last_revision(self, revision_id, revno):
444
branch_token, repo_token = self.lock_branch()
445
response = self._set_last_revision(
446
revision_id, revno, branch_token, repo_token)
450
def assertRequestSucceeds(self, revision_id, revno):
451
response = self.set_last_revision(revision_id, revno)
452
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
455
class TestSetLastRevisionVerbMixin(object):
456
"""Mixin test case for verbs that implement set_last_revision."""
458
def test_set_null_to_null(self):
459
"""An empty branch can have its last revision set to 'null:'."""
460
self.assertRequestSucceeds('null:', 0)
462
def test_NoSuchRevision(self):
463
"""If the revision_id is not present, the verb returns NoSuchRevision.
465
revision_id = 'non-existent revision'
467
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
468
self.set_last_revision(revision_id, 1))
470
def make_tree_with_two_commits(self):
471
self.tree.lock_write()
473
rev_id_utf8 = u'\xc8'.encode('utf-8')
474
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
475
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
478
def test_branch_last_revision_info_is_updated(self):
479
"""A branch's tip can be set to a revision that is present in its
482
# Make a branch with an empty revision history, but two revisions in
484
self.make_tree_with_two_commits()
485
rev_id_utf8 = u'\xc8'.encode('utf-8')
486
self.tree.branch.set_revision_history([])
488
(0, 'null:'), self.tree.branch.last_revision_info())
489
# We can update the branch to a revision that is present in the
491
self.assertRequestSucceeds(rev_id_utf8, 1)
493
(1, rev_id_utf8), self.tree.branch.last_revision_info())
495
def test_branch_last_revision_info_rewind(self):
496
"""A branch's tip can be set to a revision that is an ancestor of the
499
self.make_tree_with_two_commits()
500
rev_id_utf8 = u'\xc8'.encode('utf-8')
502
(2, 'rev-2'), self.tree.branch.last_revision_info())
503
self.assertRequestSucceeds(rev_id_utf8, 1)
505
(1, rev_id_utf8), self.tree.branch.last_revision_info())
507
def test_TipChangeRejected(self):
508
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
509
returns TipChangeRejected.
511
rejection_message = u'rejection message\N{INTERROBANG}'
512
def hook_that_rejects(params):
513
raise errors.TipChangeRejected(rejection_message)
514
Branch.hooks.install_named_hook(
515
'pre_change_branch_tip', hook_that_rejects, None)
517
FailedSmartServerResponse(
518
('TipChangeRejected', rejection_message.encode('utf-8'))),
519
self.set_last_revision('null:', 0))
522
class TestSmartServerBranchRequestSetLastRevision(
523
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
524
"""Tests for Branch.set_last_revision verb."""
526
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
528
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
529
return self.request.execute(
530
'', branch_token, repo_token, revision_id)
533
class TestSmartServerBranchRequestSetLastRevisionInfo(
534
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
535
"""Tests for Branch.set_last_revision_info verb."""
537
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
539
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
540
return self.request.execute(
541
'', branch_token, repo_token, revno, revision_id)
543
def test_NoSuchRevision(self):
544
"""Branch.set_last_revision_info does not have to return
545
NoSuchRevision if the revision_id is absent.
547
raise tests.TestNotApplicable()
550
class TestSmartServerBranchRequestSetLastRevisionEx(
551
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
552
"""Tests for Branch.set_last_revision_ex verb."""
554
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
556
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
557
return self.request.execute(
558
'', branch_token, repo_token, revision_id, 0, 0)
560
def assertRequestSucceeds(self, revision_id, revno):
561
response = self.set_last_revision(revision_id, revno)
563
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
566
def test_branch_last_revision_info_rewind(self):
567
"""A branch's tip can be set to a revision that is an ancestor of the
568
current tip, but only if allow_overwrite_descendant is passed.
570
self.make_tree_with_two_commits()
571
rev_id_utf8 = u'\xc8'.encode('utf-8')
573
(2, 'rev-2'), self.tree.branch.last_revision_info())
574
# If allow_overwrite_descendant flag is 0, then trying to set the tip
575
# to an older revision ID has no effect.
576
branch_token, repo_token = self.lock_branch()
577
response = self.request.execute(
578
'', branch_token, repo_token, rev_id_utf8, 0, 0)
580
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
583
(2, 'rev-2'), self.tree.branch.last_revision_info())
585
# If allow_overwrite_descendant flag is 1, then setting the tip to an
587
response = self.request.execute(
588
'', branch_token, repo_token, rev_id_utf8, 0, 1)
590
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
594
(1, rev_id_utf8), self.tree.branch.last_revision_info())
596
def make_branch_with_divergent_history(self):
597
"""Make a branch with divergent history in its repo.
599
The branch's tip will be 'child-2', and the repo will also contain
600
'child-1', which diverges from a common base revision.
602
self.tree.lock_write()
604
r1 = self.tree.commit('1st commit')
605
revno_1, revid_1 = self.tree.branch.last_revision_info()
606
r2 = self.tree.commit('2nd commit', rev_id='child-1')
607
# Undo the second commit
608
self.tree.branch.set_last_revision_info(revno_1, revid_1)
609
self.tree.set_parent_ids([revid_1])
610
# Make a new second commit, child-2. child-2 has diverged from
612
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
615
def test_not_allow_diverged(self):
616
"""If allow_diverged is not passed, then setting a divergent history
617
returns a Diverged error.
619
self.make_branch_with_divergent_history()
621
FailedSmartServerResponse(('Diverged',)),
622
self.set_last_revision('child-1', 2))
623
# The branch tip was not changed.
624
self.assertEqual('child-2', self.tree.branch.last_revision())
626
def test_allow_diverged(self):
627
"""If allow_diverged is passed, then setting a divergent history
630
self.make_branch_with_divergent_history()
631
branch_token, repo_token = self.lock_branch()
632
response = self.request.execute(
633
'', branch_token, repo_token, 'child-1', 1, 0)
635
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
638
# The branch tip was changed.
639
self.assertEqual('child-1', self.tree.branch.last_revision())
642
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
644
def test_get_stacked_on_url(self):
645
base_branch = self.make_branch('base', format='1.6')
646
stacked_branch = self.make_branch('stacked', format='1.6')
647
# typically should be relative
648
stacked_branch.set_stacked_on_url('../base')
649
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
650
self.get_transport())
651
response = request.execute('stacked')
653
SmartServerResponse(('ok', '../base')),
657
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
660
tests.TestCaseWithMemoryTransport.setUp(self)
662
def test_lock_write_on_unlocked_branch(self):
663
backing = self.get_transport()
664
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
665
branch = self.make_branch('.', format='knit')
666
repository = branch.repository
667
response = request.execute('')
668
branch_nonce = branch.control_files._lock.peek().get('nonce')
669
repository_nonce = repository.control_files._lock.peek().get('nonce')
671
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
673
# The branch (and associated repository) is now locked. Verify that
674
# with a new branch object.
675
new_branch = repository.bzrdir.open_branch()
676
self.assertRaises(errors.LockContention, new_branch.lock_write)
678
def test_lock_write_on_locked_branch(self):
679
backing = self.get_transport()
680
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
681
branch = self.make_branch('.')
683
branch.leave_lock_in_place()
685
response = request.execute('')
687
SmartServerResponse(('LockContention',)), response)
689
def test_lock_write_with_tokens_on_locked_branch(self):
690
backing = self.get_transport()
691
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
692
branch = self.make_branch('.', format='knit')
693
branch_token = branch.lock_write()
694
repo_token = branch.repository.lock_write()
695
branch.repository.unlock()
696
branch.leave_lock_in_place()
697
branch.repository.leave_lock_in_place()
699
response = request.execute('',
700
branch_token, repo_token)
702
SmartServerResponse(('ok', branch_token, repo_token)), response)
704
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
705
backing = self.get_transport()
706
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
707
branch = self.make_branch('.', format='knit')
708
branch_token = branch.lock_write()
709
repo_token = branch.repository.lock_write()
710
branch.repository.unlock()
711
branch.leave_lock_in_place()
712
branch.repository.leave_lock_in_place()
714
response = request.execute('',
715
branch_token+'xxx', repo_token)
717
SmartServerResponse(('TokenMismatch',)), response)
719
def test_lock_write_on_locked_repo(self):
720
backing = self.get_transport()
721
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
722
branch = self.make_branch('.', format='knit')
723
branch.repository.lock_write()
724
branch.repository.leave_lock_in_place()
725
branch.repository.unlock()
726
response = request.execute('')
728
SmartServerResponse(('LockContention',)), response)
730
def test_lock_write_on_readonly_transport(self):
731
backing = self.get_readonly_transport()
732
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
733
branch = self.make_branch('.')
734
root = self.get_transport().clone('/')
735
path = urlutils.relative_url(root.base, self.get_transport().base)
736
response = request.execute(path)
737
error_name, lock_str, why_str = response.args
738
self.assertFalse(response.is_successful())
739
self.assertEqual('LockFailed', error_name)
742
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
745
tests.TestCaseWithMemoryTransport.setUp(self)
747
def test_unlock_on_locked_branch_and_repo(self):
748
backing = self.get_transport()
749
request = smart.branch.SmartServerBranchRequestUnlock(backing)
750
branch = self.make_branch('.', format='knit')
752
branch_token = branch.lock_write()
753
repo_token = branch.repository.lock_write()
754
branch.repository.unlock()
755
# Unlock the branch (and repo) object, leaving the physical locks
757
branch.leave_lock_in_place()
758
branch.repository.leave_lock_in_place()
760
response = request.execute('',
761
branch_token, repo_token)
763
SmartServerResponse(('ok',)), response)
764
# The branch is now unlocked. Verify that with a new branch
766
new_branch = branch.bzrdir.open_branch()
767
new_branch.lock_write()
770
def test_unlock_on_unlocked_branch_unlocked_repo(self):
771
backing = self.get_transport()
772
request = smart.branch.SmartServerBranchRequestUnlock(backing)
773
branch = self.make_branch('.', format='knit')
774
response = request.execute(
775
'', 'branch token', 'repo token')
777
SmartServerResponse(('TokenMismatch',)), response)
779
def test_unlock_on_unlocked_branch_locked_repo(self):
780
backing = self.get_transport()
781
request = smart.branch.SmartServerBranchRequestUnlock(backing)
782
branch = self.make_branch('.', format='knit')
783
# Lock the repository.
784
repo_token = branch.repository.lock_write()
785
branch.repository.leave_lock_in_place()
786
branch.repository.unlock()
787
# Issue branch lock_write request on the unlocked branch (with locked
789
response = request.execute(
790
'', 'branch token', repo_token)
792
SmartServerResponse(('TokenMismatch',)), response)
795
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
797
def test_no_repository(self):
798
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
799
# we test this using a shared repository above the named path,
800
# thus checking the right search logic is used - that is, that
801
# its the exact path being looked at and the server is not
803
backing = self.get_transport()
804
request = smart.repository.SmartServerRepositoryRequest(backing)
805
self.make_repository('.', shared=True)
806
self.make_bzrdir('subdir')
807
self.assertRaises(errors.NoRepositoryPresent,
808
request.execute, 'subdir')
811
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
813
def test_trivial_bzipped(self):
814
# This tests that the wire encoding is actually bzipped
815
backing = self.get_transport()
816
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
817
tree = self.make_branch_and_memory_tree('.')
819
self.assertEqual(None,
820
request.execute('', 'missing-id'))
821
# Note that it returns a body (of '' bzipped).
823
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
824
request.do_body('\n\n0\n'))
827
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
829
def test_none_argument(self):
830
backing = self.get_transport()
831
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
832
tree = self.make_branch_and_memory_tree('.')
835
r1 = tree.commit('1st commit')
836
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
839
# the lines of revision_id->revision_parent_list has no guaranteed
840
# order coming out of a dict, so sort both our test and response
841
lines = sorted([' '.join([r2, r1]), r1])
842
response = request.execute('', '')
843
response.body = '\n'.join(sorted(response.body.split('\n')))
846
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
848
def test_specific_revision_argument(self):
849
backing = self.get_transport()
850
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
851
tree = self.make_branch_and_memory_tree('.')
854
rev_id_utf8 = u'\xc9'.encode('utf-8')
855
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
856
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
859
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
860
request.execute('', rev_id_utf8))
862
def test_no_such_revision(self):
863
backing = self.get_transport()
864
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
865
tree = self.make_branch_and_memory_tree('.')
868
r1 = tree.commit('1st commit')
871
# Note that it still returns body (of zero bytes).
873
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
874
request.execute('', 'missingrevision'))
877
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
879
def test_missing_revision(self):
880
"""For a missing revision, ('no', ) is returned."""
881
backing = self.get_transport()
882
request = smart.repository.SmartServerRequestHasRevision(backing)
883
self.make_repository('.')
884
self.assertEqual(SmartServerResponse(('no', )),
885
request.execute('', 'revid'))
887
def test_present_revision(self):
888
"""For a present revision, ('yes', ) is returned."""
889
backing = self.get_transport()
890
request = smart.repository.SmartServerRequestHasRevision(backing)
891
tree = self.make_branch_and_memory_tree('.')
894
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
895
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
897
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
898
self.assertEqual(SmartServerResponse(('yes', )),
899
request.execute('', rev_id_utf8))
902
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
904
def test_empty_revid(self):
905
"""With an empty revid, we get only size an number and revisions"""
906
backing = self.get_transport()
907
request = smart.repository.SmartServerRepositoryGatherStats(backing)
908
repository = self.make_repository('.')
909
stats = repository.gather_stats()
910
expected_body = 'revisions: 0\n'
911
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
912
request.execute('', '', 'no'))
914
def test_revid_with_committers(self):
915
"""For a revid we get more infos."""
916
backing = self.get_transport()
917
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
918
request = smart.repository.SmartServerRepositoryGatherStats(backing)
919
tree = self.make_branch_and_memory_tree('.')
922
# Let's build a predictable result
923
tree.commit('a commit', timestamp=123456.2, timezone=3600)
924
tree.commit('a commit', timestamp=654321.4, timezone=0,
928
stats = tree.branch.repository.gather_stats()
929
expected_body = ('firstrev: 123456.200 3600\n'
930
'latestrev: 654321.400 0\n'
932
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
936
def test_not_empty_repository_with_committers(self):
937
"""For a revid and requesting committers we get the whole thing."""
938
backing = self.get_transport()
939
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
940
request = smart.repository.SmartServerRepositoryGatherStats(backing)
941
tree = self.make_branch_and_memory_tree('.')
944
# Let's build a predictable result
945
tree.commit('a commit', timestamp=123456.2, timezone=3600,
947
tree.commit('a commit', timestamp=654321.4, timezone=0,
948
committer='bar', rev_id=rev_id_utf8)
950
stats = tree.branch.repository.gather_stats()
952
expected_body = ('committers: 2\n'
953
'firstrev: 123456.200 3600\n'
954
'latestrev: 654321.400 0\n'
956
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
961
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
963
def test_is_shared(self):
964
"""For a shared repository, ('yes', ) is returned."""
965
backing = self.get_transport()
966
request = smart.repository.SmartServerRepositoryIsShared(backing)
967
self.make_repository('.', shared=True)
968
self.assertEqual(SmartServerResponse(('yes', )),
969
request.execute('', ))
971
def test_is_not_shared(self):
972
"""For a shared repository, ('no', ) is returned."""
973
backing = self.get_transport()
974
request = smart.repository.SmartServerRepositoryIsShared(backing)
975
self.make_repository('.', shared=False)
976
self.assertEqual(SmartServerResponse(('no', )),
977
request.execute('', ))
980
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
983
tests.TestCaseWithMemoryTransport.setUp(self)
985
def test_lock_write_on_unlocked_repo(self):
986
backing = self.get_transport()
987
request = smart.repository.SmartServerRepositoryLockWrite(backing)
988
repository = self.make_repository('.', format='knit')
989
response = request.execute('')
990
nonce = repository.control_files._lock.peek().get('nonce')
991
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
992
# The repository is now locked. Verify that with a new repository
994
new_repo = repository.bzrdir.open_repository()
995
self.assertRaises(errors.LockContention, new_repo.lock_write)
997
def test_lock_write_on_locked_repo(self):
998
backing = self.get_transport()
999
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1000
repository = self.make_repository('.', format='knit')
1001
repository.lock_write()
1002
repository.leave_lock_in_place()
1004
response = request.execute('')
1006
SmartServerResponse(('LockContention',)), response)
1008
def test_lock_write_on_readonly_transport(self):
1009
backing = self.get_readonly_transport()
1010
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1011
repository = self.make_repository('.', format='knit')
1012
response = request.execute('')
1013
self.assertFalse(response.is_successful())
1014
self.assertEqual('LockFailed', response.args[0])
1017
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1020
tests.TestCaseWithMemoryTransport.setUp(self)
1022
def test_unlock_on_locked_repo(self):
1023
backing = self.get_transport()
1024
request = smart.repository.SmartServerRepositoryUnlock(backing)
1025
repository = self.make_repository('.', format='knit')
1026
token = repository.lock_write()
1027
repository.leave_lock_in_place()
1029
response = request.execute('', token)
1031
SmartServerResponse(('ok',)), response)
1032
# The repository is now unlocked. Verify that with a new repository
1034
new_repo = repository.bzrdir.open_repository()
1035
new_repo.lock_write()
1038
def test_unlock_on_unlocked_repo(self):
1039
backing = self.get_transport()
1040
request = smart.repository.SmartServerRepositoryUnlock(backing)
1041
repository = self.make_repository('.', format='knit')
1042
response = request.execute('', 'some token')
1044
SmartServerResponse(('TokenMismatch',)), response)
1047
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1049
def test_is_readonly_no(self):
1050
backing = self.get_transport()
1051
request = smart.request.SmartServerIsReadonly(backing)
1052
response = request.execute()
1054
SmartServerResponse(('no',)), response)
1056
def test_is_readonly_yes(self):
1057
backing = self.get_readonly_transport()
1058
request = smart.request.SmartServerIsReadonly(backing)
1059
response = request.execute()
1061
SmartServerResponse(('yes',)), response)
1064
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1066
def test_set_false(self):
1067
backing = self.get_transport()
1068
repo = self.make_repository('.', shared=True)
1069
repo.set_make_working_trees(True)
1070
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1071
request = request_class(backing)
1072
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1073
request.execute('', 'False'))
1074
repo = repo.bzrdir.open_repository()
1075
self.assertFalse(repo.make_working_trees())
1077
def test_set_true(self):
1078
backing = self.get_transport()
1079
repo = self.make_repository('.', shared=True)
1080
repo.set_make_working_trees(False)
1081
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1082
request = request_class(backing)
1083
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1084
request.execute('', 'True'))
1085
repo = repo.bzrdir.open_repository()
1086
self.assertTrue(repo.make_working_trees())
1089
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1091
def make_repo_needing_autopacking(self, path='.'):
1092
# Make a repo in need of autopacking.
1093
tree = self.make_branch_and_tree('.', format='pack-0.92')
1094
repo = tree.branch.repository
1095
# monkey-patch the pack collection to disable autopacking
1096
repo._pack_collection._max_pack_count = lambda count: count
1098
tree.commit('commit %s' % x)
1099
self.assertEqual(10, len(repo._pack_collection.names()))
1100
del repo._pack_collection._max_pack_count
1103
def test_autopack_needed(self):
1104
repo = self.make_repo_needing_autopacking()
1105
backing = self.get_transport()
1106
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1108
response = request.execute('')
1109
self.assertEqual(SmartServerResponse(('ok',)), response)
1110
repo._pack_collection.reload_pack_names()
1111
self.assertEqual(1, len(repo._pack_collection.names()))
1113
def test_autopack_not_needed(self):
1114
tree = self.make_branch_and_tree('.', format='pack-0.92')
1115
repo = tree.branch.repository
1117
tree.commit('commit %s' % x)
1118
backing = self.get_transport()
1119
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1121
response = request.execute('')
1122
self.assertEqual(SmartServerResponse(('ok',)), response)
1123
repo._pack_collection.reload_pack_names()
1124
self.assertEqual(9, len(repo._pack_collection.names()))
1126
def test_autopack_on_nonpack_format(self):
1127
"""A request to autopack a non-pack repo is a no-op."""
1128
repo = self.make_repository('.', format='knit')
1129
backing = self.get_transport()
1130
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1132
response = request.execute('')
1133
self.assertEqual(SmartServerResponse(('ok',)), response)
1136
class TestHandlers(tests.TestCase):
1137
"""Tests for the request.request_handlers object."""
1139
def test_all_registrations_exist(self):
1140
"""All registered request_handlers can be found."""
1141
# If there's a typo in a register_lazy call, this loop will fail with
1142
# an AttributeError.
1143
for key, item in smart.request.request_handlers.iteritems():
1146
def test_registered_methods(self):
1147
"""Test that known methods are registered to the correct object."""
1149
smart.request.request_handlers.get('Branch.get_config_file'),
1150
smart.branch.SmartServerBranchGetConfigFile)
1152
smart.request.request_handlers.get('Branch.lock_write'),
1153
smart.branch.SmartServerBranchRequestLockWrite)
1155
smart.request.request_handlers.get('Branch.last_revision_info'),
1156
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1158
smart.request.request_handlers.get('Branch.revision_history'),
1159
smart.branch.SmartServerRequestRevisionHistory)
1161
smart.request.request_handlers.get('Branch.set_last_revision'),
1162
smart.branch.SmartServerBranchRequestSetLastRevision)
1164
smart.request.request_handlers.get('Branch.set_last_revision_info'),
1165
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1167
smart.request.request_handlers.get('Branch.unlock'),
1168
smart.branch.SmartServerBranchRequestUnlock)
1170
smart.request.request_handlers.get('BzrDir.find_repository'),
1171
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1173
smart.request.request_handlers.get('BzrDir.find_repositoryV2'),
1174
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1176
smart.request.request_handlers.get('BzrDirFormat.initialize'),
1177
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1179
smart.request.request_handlers.get('BzrDir.open_branch'),
1180
smart.bzrdir.SmartServerRequestOpenBranch)
1182
smart.request.request_handlers.get('PackRepository.autopack'),
1183
smart.packrepository.SmartServerPackRepositoryAutopack)
1185
smart.request.request_handlers.get('Repository.gather_stats'),
1186
smart.repository.SmartServerRepositoryGatherStats)
1188
smart.request.request_handlers.get('Repository.get_parent_map'),
1189
smart.repository.SmartServerRepositoryGetParentMap)
1191
smart.request.request_handlers.get(
1192
'Repository.get_revision_graph'),
1193
smart.repository.SmartServerRepositoryGetRevisionGraph)
1195
smart.request.request_handlers.get('Repository.has_revision'),
1196
smart.repository.SmartServerRequestHasRevision)
1198
smart.request.request_handlers.get('Repository.is_shared'),
1199
smart.repository.SmartServerRepositoryIsShared)
1201
smart.request.request_handlers.get('Repository.lock_write'),
1202
smart.repository.SmartServerRepositoryLockWrite)
1204
smart.request.request_handlers.get('Repository.get_stream'),
1205
smart.repository.SmartServerRepositoryGetStream)
1207
smart.request.request_handlers.get('Repository.tarball'),
1208
smart.repository.SmartServerRepositoryTarball)
1210
smart.request.request_handlers.get('Repository.unlock'),
1211
smart.repository.SmartServerRepositoryUnlock)
1213
smart.request.request_handlers.get('Transport.is_readonly'),
1214
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
