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 TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
164
"""Tests for BzrDir.create_repository."""
166
def test_makes_repository(self):
167
"""When there is a bzrdir present, the call succeeds."""
168
backing = self.get_transport()
169
self.make_bzrdir('.')
170
request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
171
request = request_class(backing)
172
reference_bzrdir_format = bzrdir.format_registry.get('default')()
173
reference_format = reference_bzrdir_format.repository_format
174
network_name = reference_format.network_name()
175
expected = SuccessfulSmartServerResponse(
176
('ok', 'no', 'no', 'no', network_name))
177
self.assertEqual(expected, request.execute('', network_name, 'True'))
180
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
181
"""Tests for BzrDir.find_repository."""
183
def test_no_repository(self):
184
"""When there is no repository to be found, ('norepository', ) is returned."""
185
backing = self.get_transport()
186
request = self._request_class(backing)
187
self.make_bzrdir('.')
188
self.assertEqual(SmartServerResponse(('norepository', )),
191
def test_nonshared_repository(self):
192
# nonshared repositorys only allow 'find' to return a handle when the
193
# path the repository is being searched on is the same as that that
194
# the repository is at.
195
backing = self.get_transport()
196
request = self._request_class(backing)
197
result = self._make_repository_and_result()
198
self.assertEqual(result, request.execute(''))
199
self.make_bzrdir('subdir')
200
self.assertEqual(SmartServerResponse(('norepository', )),
201
request.execute('subdir'))
203
def _make_repository_and_result(self, shared=False, format=None):
204
"""Convenience function to setup a repository.
206
:result: The SmartServerResponse to expect when opening it.
208
repo = self.make_repository('.', shared=shared, format=format)
209
if repo.supports_rich_root():
213
if repo._format.supports_tree_reference:
217
if (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
218
self._request_class):
219
# All tests so far are on formats, and for non-external
221
return SuccessfulSmartServerResponse(
222
('ok', '', rich_root, subtrees, 'no'))
224
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
226
def test_shared_repository(self):
227
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
228
backing = self.get_transport()
229
request = self._request_class(backing)
230
result = self._make_repository_and_result(shared=True)
231
self.assertEqual(result, request.execute(''))
232
self.make_bzrdir('subdir')
233
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
234
self.assertEqual(result2,
235
request.execute('subdir'))
236
self.make_bzrdir('subdir/deeper')
237
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
238
self.assertEqual(result3,
239
request.execute('subdir/deeper'))
241
def test_rich_root_and_subtree_encoding(self):
242
"""Test for the format attributes for rich root and subtree support."""
243
backing = self.get_transport()
244
request = self._request_class(backing)
245
result = self._make_repository_and_result(format='dirstate-with-subtree')
246
# check the test will be valid
247
self.assertEqual('yes', result.args[2])
248
self.assertEqual('yes', result.args[3])
249
self.assertEqual(result, request.execute(''))
251
def test_supports_external_lookups_no_v2(self):
252
"""Test for the supports_external_lookups attribute."""
253
backing = self.get_transport()
254
request = self._request_class(backing)
255
result = self._make_repository_and_result(format='dirstate-with-subtree')
256
# check the test will be valid
257
self.assertEqual('no', result.args[4])
258
self.assertEqual(result, request.execute(''))
261
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
263
def test_empty_dir(self):
264
"""Initializing an empty dir should succeed and do it."""
265
backing = self.get_transport()
266
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
267
self.assertEqual(SmartServerResponse(('ok', )),
269
made_dir = bzrdir.BzrDir.open_from_transport(backing)
270
# no branch, tree or repository is expected with the current
272
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
273
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
274
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
276
def test_missing_dir(self):
277
"""Initializing a missing directory should fail like the bzrdir api."""
278
backing = self.get_transport()
279
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
280
self.assertRaises(errors.NoSuchFile,
281
request.execute, 'subdir')
283
def test_initialized_dir(self):
284
"""Initializing an extant bzrdir should fail like the bzrdir api."""
285
backing = self.get_transport()
286
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
287
self.make_bzrdir('subdir')
288
self.assertRaises(errors.FileExists,
289
request.execute, 'subdir')
292
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
294
def test_no_branch(self):
295
"""When there is no branch, ('nobranch', ) is returned."""
296
backing = self.get_transport()
297
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
298
self.make_bzrdir('.')
299
self.assertEqual(SmartServerResponse(('nobranch', )),
302
def test_branch(self):
303
"""When there is a branch, 'ok' is returned."""
304
backing = self.get_transport()
305
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
306
self.make_branch('.')
307
self.assertEqual(SmartServerResponse(('ok', '')),
310
def test_branch_reference(self):
311
"""When there is a branch reference, the reference URL is returned."""
312
backing = self.get_transport()
313
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
314
branch = self.make_branch('branch')
315
checkout = branch.create_checkout('reference',lightweight=True)
316
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
317
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
318
self.assertEqual(SmartServerResponse(('ok', reference_url)),
319
request.execute('reference'))
322
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
324
def test_empty(self):
325
"""For an empty branch, the body is empty."""
326
backing = self.get_transport()
327
request = smart.branch.SmartServerRequestRevisionHistory(backing)
328
self.make_branch('.')
329
self.assertEqual(SmartServerResponse(('ok', ), ''),
332
def test_not_empty(self):
333
"""For a non-empty branch, the body is empty."""
334
backing = self.get_transport()
335
request = smart.branch.SmartServerRequestRevisionHistory(backing)
336
tree = self.make_branch_and_memory_tree('.')
339
r1 = tree.commit('1st commit')
340
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
343
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
347
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
349
def test_no_branch(self):
350
"""When there is a bzrdir and no branch, NotBranchError is raised."""
351
backing = self.get_transport()
352
request = smart.branch.SmartServerBranchRequest(backing)
353
self.make_bzrdir('.')
354
self.assertRaises(errors.NotBranchError,
357
def test_branch_reference(self):
358
"""When there is a branch reference, NotBranchError is raised."""
359
backing = self.get_transport()
360
request = smart.branch.SmartServerBranchRequest(backing)
361
branch = self.make_branch('branch')
362
checkout = branch.create_checkout('reference',lightweight=True)
363
self.assertRaises(errors.NotBranchError,
364
request.execute, 'checkout')
367
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
369
def test_empty(self):
370
"""For an empty branch, the result is ('ok', '0', 'null:')."""
371
backing = self.get_transport()
372
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
373
self.make_branch('.')
374
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
377
def test_not_empty(self):
378
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
379
backing = self.get_transport()
380
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
381
tree = self.make_branch_and_memory_tree('.')
384
rev_id_utf8 = u'\xc8'.encode('utf-8')
385
r1 = tree.commit('1st commit')
386
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
389
SmartServerResponse(('ok', '2', rev_id_utf8)),
393
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
395
def test_default(self):
396
"""With no file, we get empty content."""
397
backing = self.get_transport()
398
request = smart.branch.SmartServerBranchGetConfigFile(backing)
399
branch = self.make_branch('.')
400
# there should be no file by default
402
self.assertEqual(SmartServerResponse(('ok', ), content),
405
def test_with_content(self):
406
# SmartServerBranchGetConfigFile should return the content from
407
# branch.control_files.get('branch.conf') for now - in the future it may
408
# perform more complex processing.
409
backing = self.get_transport()
410
request = smart.branch.SmartServerBranchGetConfigFile(backing)
411
branch = self.make_branch('.')
412
branch._transport.put_bytes('branch.conf', 'foo bar baz')
413
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
417
class SetLastRevisionTestBase(tests.TestCaseWithMemoryTransport):
418
"""Base test case for verbs that implement set_last_revision."""
421
tests.TestCaseWithMemoryTransport.setUp(self)
422
backing_transport = self.get_transport()
423
self.request = self.request_class(backing_transport)
424
self.tree = self.make_branch_and_memory_tree('.')
426
def lock_branch(self):
428
branch_token = b.lock_write()
429
repo_token = b.repository.lock_write()
430
b.repository.unlock()
431
return branch_token, repo_token
433
def unlock_branch(self):
434
self.tree.branch.unlock()
436
def set_last_revision(self, revision_id, revno):
437
branch_token, repo_token = self.lock_branch()
438
response = self._set_last_revision(
439
revision_id, revno, branch_token, repo_token)
443
def assertRequestSucceeds(self, revision_id, revno):
444
response = self.set_last_revision(revision_id, revno)
445
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
448
class TestSetLastRevisionVerbMixin(object):
449
"""Mixin test case for verbs that implement set_last_revision."""
451
def test_set_null_to_null(self):
452
"""An empty branch can have its last revision set to 'null:'."""
453
self.assertRequestSucceeds('null:', 0)
455
def test_NoSuchRevision(self):
456
"""If the revision_id is not present, the verb returns NoSuchRevision.
458
revision_id = 'non-existent revision'
460
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
461
self.set_last_revision(revision_id, 1))
463
def make_tree_with_two_commits(self):
464
self.tree.lock_write()
466
rev_id_utf8 = u'\xc8'.encode('utf-8')
467
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
468
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
471
def test_branch_last_revision_info_is_updated(self):
472
"""A branch's tip can be set to a revision that is present in its
475
# Make a branch with an empty revision history, but two revisions in
477
self.make_tree_with_two_commits()
478
rev_id_utf8 = u'\xc8'.encode('utf-8')
479
self.tree.branch.set_revision_history([])
481
(0, 'null:'), self.tree.branch.last_revision_info())
482
# We can update the branch to a revision that is present in the
484
self.assertRequestSucceeds(rev_id_utf8, 1)
486
(1, rev_id_utf8), self.tree.branch.last_revision_info())
488
def test_branch_last_revision_info_rewind(self):
489
"""A branch's tip can be set to a revision that is an ancestor of the
492
self.make_tree_with_two_commits()
493
rev_id_utf8 = u'\xc8'.encode('utf-8')
495
(2, 'rev-2'), self.tree.branch.last_revision_info())
496
self.assertRequestSucceeds(rev_id_utf8, 1)
498
(1, rev_id_utf8), self.tree.branch.last_revision_info())
500
def test_TipChangeRejected(self):
501
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
502
returns TipChangeRejected.
504
rejection_message = u'rejection message\N{INTERROBANG}'
505
def hook_that_rejects(params):
506
raise errors.TipChangeRejected(rejection_message)
507
Branch.hooks.install_named_hook(
508
'pre_change_branch_tip', hook_that_rejects, None)
510
FailedSmartServerResponse(
511
('TipChangeRejected', rejection_message.encode('utf-8'))),
512
self.set_last_revision('null:', 0))
515
class TestSmartServerBranchRequestSetLastRevision(
516
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
517
"""Tests for Branch.set_last_revision verb."""
519
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
521
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
522
return self.request.execute(
523
'', branch_token, repo_token, revision_id)
526
class TestSmartServerBranchRequestSetLastRevisionInfo(
527
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
528
"""Tests for Branch.set_last_revision_info verb."""
530
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
532
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
533
return self.request.execute(
534
'', branch_token, repo_token, revno, revision_id)
536
def test_NoSuchRevision(self):
537
"""Branch.set_last_revision_info does not have to return
538
NoSuchRevision if the revision_id is absent.
540
raise tests.TestNotApplicable()
543
class TestSmartServerBranchRequestSetLastRevisionEx(
544
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
545
"""Tests for Branch.set_last_revision_ex verb."""
547
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
549
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
550
return self.request.execute(
551
'', branch_token, repo_token, revision_id, 0, 0)
553
def assertRequestSucceeds(self, revision_id, revno):
554
response = self.set_last_revision(revision_id, revno)
556
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
559
def test_branch_last_revision_info_rewind(self):
560
"""A branch's tip can be set to a revision that is an ancestor of the
561
current tip, but only if allow_overwrite_descendant is passed.
563
self.make_tree_with_two_commits()
564
rev_id_utf8 = u'\xc8'.encode('utf-8')
566
(2, 'rev-2'), self.tree.branch.last_revision_info())
567
# If allow_overwrite_descendant flag is 0, then trying to set the tip
568
# to an older revision ID has no effect.
569
branch_token, repo_token = self.lock_branch()
570
response = self.request.execute(
571
'', branch_token, repo_token, rev_id_utf8, 0, 0)
573
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
576
(2, 'rev-2'), self.tree.branch.last_revision_info())
578
# If allow_overwrite_descendant flag is 1, then setting the tip to an
580
response = self.request.execute(
581
'', branch_token, repo_token, rev_id_utf8, 0, 1)
583
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
587
(1, rev_id_utf8), self.tree.branch.last_revision_info())
589
def make_branch_with_divergent_history(self):
590
"""Make a branch with divergent history in its repo.
592
The branch's tip will be 'child-2', and the repo will also contain
593
'child-1', which diverges from a common base revision.
595
self.tree.lock_write()
597
r1 = self.tree.commit('1st commit')
598
revno_1, revid_1 = self.tree.branch.last_revision_info()
599
r2 = self.tree.commit('2nd commit', rev_id='child-1')
600
# Undo the second commit
601
self.tree.branch.set_last_revision_info(revno_1, revid_1)
602
self.tree.set_parent_ids([revid_1])
603
# Make a new second commit, child-2. child-2 has diverged from
605
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
608
def test_not_allow_diverged(self):
609
"""If allow_diverged is not passed, then setting a divergent history
610
returns a Diverged error.
612
self.make_branch_with_divergent_history()
614
FailedSmartServerResponse(('Diverged',)),
615
self.set_last_revision('child-1', 2))
616
# The branch tip was not changed.
617
self.assertEqual('child-2', self.tree.branch.last_revision())
619
def test_allow_diverged(self):
620
"""If allow_diverged is passed, then setting a divergent history
623
self.make_branch_with_divergent_history()
624
branch_token, repo_token = self.lock_branch()
625
response = self.request.execute(
626
'', branch_token, repo_token, 'child-1', 1, 0)
628
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
631
# The branch tip was changed.
632
self.assertEqual('child-1', self.tree.branch.last_revision())
635
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
637
def test_get_stacked_on_url(self):
638
base_branch = self.make_branch('base', format='1.6')
639
stacked_branch = self.make_branch('stacked', format='1.6')
640
# typically should be relative
641
stacked_branch.set_stacked_on_url('../base')
642
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
643
self.get_transport())
644
response = request.execute('stacked')
646
SmartServerResponse(('ok', '../base')),
650
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
653
tests.TestCaseWithMemoryTransport.setUp(self)
655
def test_lock_write_on_unlocked_branch(self):
656
backing = self.get_transport()
657
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
658
branch = self.make_branch('.', format='knit')
659
repository = branch.repository
660
response = request.execute('')
661
branch_nonce = branch.control_files._lock.peek().get('nonce')
662
repository_nonce = repository.control_files._lock.peek().get('nonce')
664
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
666
# The branch (and associated repository) is now locked. Verify that
667
# with a new branch object.
668
new_branch = repository.bzrdir.open_branch()
669
self.assertRaises(errors.LockContention, new_branch.lock_write)
671
def test_lock_write_on_locked_branch(self):
672
backing = self.get_transport()
673
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
674
branch = self.make_branch('.')
676
branch.leave_lock_in_place()
678
response = request.execute('')
680
SmartServerResponse(('LockContention',)), response)
682
def test_lock_write_with_tokens_on_locked_branch(self):
683
backing = self.get_transport()
684
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
685
branch = self.make_branch('.', format='knit')
686
branch_token = branch.lock_write()
687
repo_token = branch.repository.lock_write()
688
branch.repository.unlock()
689
branch.leave_lock_in_place()
690
branch.repository.leave_lock_in_place()
692
response = request.execute('',
693
branch_token, repo_token)
695
SmartServerResponse(('ok', branch_token, repo_token)), response)
697
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
698
backing = self.get_transport()
699
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
700
branch = self.make_branch('.', format='knit')
701
branch_token = branch.lock_write()
702
repo_token = branch.repository.lock_write()
703
branch.repository.unlock()
704
branch.leave_lock_in_place()
705
branch.repository.leave_lock_in_place()
707
response = request.execute('',
708
branch_token+'xxx', repo_token)
710
SmartServerResponse(('TokenMismatch',)), response)
712
def test_lock_write_on_locked_repo(self):
713
backing = self.get_transport()
714
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
715
branch = self.make_branch('.', format='knit')
716
branch.repository.lock_write()
717
branch.repository.leave_lock_in_place()
718
branch.repository.unlock()
719
response = request.execute('')
721
SmartServerResponse(('LockContention',)), response)
723
def test_lock_write_on_readonly_transport(self):
724
backing = self.get_readonly_transport()
725
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
726
branch = self.make_branch('.')
727
root = self.get_transport().clone('/')
728
path = urlutils.relative_url(root.base, self.get_transport().base)
729
response = request.execute(path)
730
error_name, lock_str, why_str = response.args
731
self.assertFalse(response.is_successful())
732
self.assertEqual('LockFailed', error_name)
735
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
738
tests.TestCaseWithMemoryTransport.setUp(self)
740
def test_unlock_on_locked_branch_and_repo(self):
741
backing = self.get_transport()
742
request = smart.branch.SmartServerBranchRequestUnlock(backing)
743
branch = self.make_branch('.', format='knit')
745
branch_token = branch.lock_write()
746
repo_token = branch.repository.lock_write()
747
branch.repository.unlock()
748
# Unlock the branch (and repo) object, leaving the physical locks
750
branch.leave_lock_in_place()
751
branch.repository.leave_lock_in_place()
753
response = request.execute('',
754
branch_token, repo_token)
756
SmartServerResponse(('ok',)), response)
757
# The branch is now unlocked. Verify that with a new branch
759
new_branch = branch.bzrdir.open_branch()
760
new_branch.lock_write()
763
def test_unlock_on_unlocked_branch_unlocked_repo(self):
764
backing = self.get_transport()
765
request = smart.branch.SmartServerBranchRequestUnlock(backing)
766
branch = self.make_branch('.', format='knit')
767
response = request.execute(
768
'', 'branch token', 'repo token')
770
SmartServerResponse(('TokenMismatch',)), response)
772
def test_unlock_on_unlocked_branch_locked_repo(self):
773
backing = self.get_transport()
774
request = smart.branch.SmartServerBranchRequestUnlock(backing)
775
branch = self.make_branch('.', format='knit')
776
# Lock the repository.
777
repo_token = branch.repository.lock_write()
778
branch.repository.leave_lock_in_place()
779
branch.repository.unlock()
780
# Issue branch lock_write request on the unlocked branch (with locked
782
response = request.execute(
783
'', 'branch token', repo_token)
785
SmartServerResponse(('TokenMismatch',)), response)
788
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
790
def test_no_repository(self):
791
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
792
# we test this using a shared repository above the named path,
793
# thus checking the right search logic is used - that is, that
794
# its the exact path being looked at and the server is not
796
backing = self.get_transport()
797
request = smart.repository.SmartServerRepositoryRequest(backing)
798
self.make_repository('.', shared=True)
799
self.make_bzrdir('subdir')
800
self.assertRaises(errors.NoRepositoryPresent,
801
request.execute, 'subdir')
804
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
806
def test_trivial_bzipped(self):
807
# This tests that the wire encoding is actually bzipped
808
backing = self.get_transport()
809
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
810
tree = self.make_branch_and_memory_tree('.')
812
self.assertEqual(None,
813
request.execute('', 'missing-id'))
814
# Note that it returns a body (of '' bzipped).
816
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
817
request.do_body('\n\n0\n'))
820
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
822
def test_none_argument(self):
823
backing = self.get_transport()
824
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
825
tree = self.make_branch_and_memory_tree('.')
828
r1 = tree.commit('1st commit')
829
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
832
# the lines of revision_id->revision_parent_list has no guaranteed
833
# order coming out of a dict, so sort both our test and response
834
lines = sorted([' '.join([r2, r1]), r1])
835
response = request.execute('', '')
836
response.body = '\n'.join(sorted(response.body.split('\n')))
839
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
841
def test_specific_revision_argument(self):
842
backing = self.get_transport()
843
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
844
tree = self.make_branch_and_memory_tree('.')
847
rev_id_utf8 = u'\xc9'.encode('utf-8')
848
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
849
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
852
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
853
request.execute('', rev_id_utf8))
855
def test_no_such_revision(self):
856
backing = self.get_transport()
857
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
858
tree = self.make_branch_and_memory_tree('.')
861
r1 = tree.commit('1st commit')
864
# Note that it still returns body (of zero bytes).
866
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
867
request.execute('', 'missingrevision'))
870
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
872
def test_missing_revision(self):
873
"""For a missing revision, ('no', ) is returned."""
874
backing = self.get_transport()
875
request = smart.repository.SmartServerRequestHasRevision(backing)
876
self.make_repository('.')
877
self.assertEqual(SmartServerResponse(('no', )),
878
request.execute('', 'revid'))
880
def test_present_revision(self):
881
"""For a present revision, ('yes', ) is returned."""
882
backing = self.get_transport()
883
request = smart.repository.SmartServerRequestHasRevision(backing)
884
tree = self.make_branch_and_memory_tree('.')
887
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
888
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
890
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
891
self.assertEqual(SmartServerResponse(('yes', )),
892
request.execute('', rev_id_utf8))
895
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
897
def test_empty_revid(self):
898
"""With an empty revid, we get only size an number and revisions"""
899
backing = self.get_transport()
900
request = smart.repository.SmartServerRepositoryGatherStats(backing)
901
repository = self.make_repository('.')
902
stats = repository.gather_stats()
903
expected_body = 'revisions: 0\n'
904
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
905
request.execute('', '', 'no'))
907
def test_revid_with_committers(self):
908
"""For a revid we get more infos."""
909
backing = self.get_transport()
910
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
911
request = smart.repository.SmartServerRepositoryGatherStats(backing)
912
tree = self.make_branch_and_memory_tree('.')
915
# Let's build a predictable result
916
tree.commit('a commit', timestamp=123456.2, timezone=3600)
917
tree.commit('a commit', timestamp=654321.4, timezone=0,
921
stats = tree.branch.repository.gather_stats()
922
expected_body = ('firstrev: 123456.200 3600\n'
923
'latestrev: 654321.400 0\n'
925
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
929
def test_not_empty_repository_with_committers(self):
930
"""For a revid and requesting committers we get the whole thing."""
931
backing = self.get_transport()
932
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
933
request = smart.repository.SmartServerRepositoryGatherStats(backing)
934
tree = self.make_branch_and_memory_tree('.')
937
# Let's build a predictable result
938
tree.commit('a commit', timestamp=123456.2, timezone=3600,
940
tree.commit('a commit', timestamp=654321.4, timezone=0,
941
committer='bar', rev_id=rev_id_utf8)
943
stats = tree.branch.repository.gather_stats()
945
expected_body = ('committers: 2\n'
946
'firstrev: 123456.200 3600\n'
947
'latestrev: 654321.400 0\n'
949
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
954
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
956
def test_is_shared(self):
957
"""For a shared repository, ('yes', ) is returned."""
958
backing = self.get_transport()
959
request = smart.repository.SmartServerRepositoryIsShared(backing)
960
self.make_repository('.', shared=True)
961
self.assertEqual(SmartServerResponse(('yes', )),
962
request.execute('', ))
964
def test_is_not_shared(self):
965
"""For a shared repository, ('no', ) is returned."""
966
backing = self.get_transport()
967
request = smart.repository.SmartServerRepositoryIsShared(backing)
968
self.make_repository('.', shared=False)
969
self.assertEqual(SmartServerResponse(('no', )),
970
request.execute('', ))
973
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
976
tests.TestCaseWithMemoryTransport.setUp(self)
978
def test_lock_write_on_unlocked_repo(self):
979
backing = self.get_transport()
980
request = smart.repository.SmartServerRepositoryLockWrite(backing)
981
repository = self.make_repository('.', format='knit')
982
response = request.execute('')
983
nonce = repository.control_files._lock.peek().get('nonce')
984
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
985
# The repository is now locked. Verify that with a new repository
987
new_repo = repository.bzrdir.open_repository()
988
self.assertRaises(errors.LockContention, new_repo.lock_write)
990
def test_lock_write_on_locked_repo(self):
991
backing = self.get_transport()
992
request = smart.repository.SmartServerRepositoryLockWrite(backing)
993
repository = self.make_repository('.', format='knit')
994
repository.lock_write()
995
repository.leave_lock_in_place()
997
response = request.execute('')
999
SmartServerResponse(('LockContention',)), response)
1001
def test_lock_write_on_readonly_transport(self):
1002
backing = self.get_readonly_transport()
1003
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1004
repository = self.make_repository('.', format='knit')
1005
response = request.execute('')
1006
self.assertFalse(response.is_successful())
1007
self.assertEqual('LockFailed', response.args[0])
1010
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1013
tests.TestCaseWithMemoryTransport.setUp(self)
1015
def test_unlock_on_locked_repo(self):
1016
backing = self.get_transport()
1017
request = smart.repository.SmartServerRepositoryUnlock(backing)
1018
repository = self.make_repository('.', format='knit')
1019
token = repository.lock_write()
1020
repository.leave_lock_in_place()
1022
response = request.execute('', token)
1024
SmartServerResponse(('ok',)), response)
1025
# The repository is now unlocked. Verify that with a new repository
1027
new_repo = repository.bzrdir.open_repository()
1028
new_repo.lock_write()
1031
def test_unlock_on_unlocked_repo(self):
1032
backing = self.get_transport()
1033
request = smart.repository.SmartServerRepositoryUnlock(backing)
1034
repository = self.make_repository('.', format='knit')
1035
response = request.execute('', 'some token')
1037
SmartServerResponse(('TokenMismatch',)), response)
1040
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1042
def test_is_readonly_no(self):
1043
backing = self.get_transport()
1044
request = smart.request.SmartServerIsReadonly(backing)
1045
response = request.execute()
1047
SmartServerResponse(('no',)), response)
1049
def test_is_readonly_yes(self):
1050
backing = self.get_readonly_transport()
1051
request = smart.request.SmartServerIsReadonly(backing)
1052
response = request.execute()
1054
SmartServerResponse(('yes',)), response)
1057
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1059
def test_set_false(self):
1060
backing = self.get_transport()
1061
repo = self.make_repository('.', shared=True)
1062
repo.set_make_working_trees(True)
1063
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1064
request = request_class(backing)
1065
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1066
request.execute('', 'False'))
1067
repo = repo.bzrdir.open_repository()
1068
self.assertFalse(repo.make_working_trees())
1070
def test_set_true(self):
1071
backing = self.get_transport()
1072
repo = self.make_repository('.', shared=True)
1073
repo.set_make_working_trees(False)
1074
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1075
request = request_class(backing)
1076
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1077
request.execute('', 'True'))
1078
repo = repo.bzrdir.open_repository()
1079
self.assertTrue(repo.make_working_trees())
1082
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1084
def make_repo_needing_autopacking(self, path='.'):
1085
# Make a repo in need of autopacking.
1086
tree = self.make_branch_and_tree('.', format='pack-0.92')
1087
repo = tree.branch.repository
1088
# monkey-patch the pack collection to disable autopacking
1089
repo._pack_collection._max_pack_count = lambda count: count
1091
tree.commit('commit %s' % x)
1092
self.assertEqual(10, len(repo._pack_collection.names()))
1093
del repo._pack_collection._max_pack_count
1096
def test_autopack_needed(self):
1097
repo = self.make_repo_needing_autopacking()
1098
backing = self.get_transport()
1099
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1101
response = request.execute('')
1102
self.assertEqual(SmartServerResponse(('ok',)), response)
1103
repo._pack_collection.reload_pack_names()
1104
self.assertEqual(1, len(repo._pack_collection.names()))
1106
def test_autopack_not_needed(self):
1107
tree = self.make_branch_and_tree('.', format='pack-0.92')
1108
repo = tree.branch.repository
1110
tree.commit('commit %s' % x)
1111
backing = self.get_transport()
1112
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1114
response = request.execute('')
1115
self.assertEqual(SmartServerResponse(('ok',)), response)
1116
repo._pack_collection.reload_pack_names()
1117
self.assertEqual(9, len(repo._pack_collection.names()))
1119
def test_autopack_on_nonpack_format(self):
1120
"""A request to autopack a non-pack repo is a no-op."""
1121
repo = self.make_repository('.', format='knit')
1122
backing = self.get_transport()
1123
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1125
response = request.execute('')
1126
self.assertEqual(SmartServerResponse(('ok',)), response)
1129
class TestHandlers(tests.TestCase):
1130
"""Tests for the request.request_handlers object."""
1132
def test_all_registrations_exist(self):
1133
"""All registered request_handlers can be found."""
1134
# If there's a typo in a register_lazy call, this loop will fail with
1135
# an AttributeError.
1136
for key, item in smart.request.request_handlers.iteritems():
1139
def test_registered_methods(self):
1140
"""Test that known methods are registered to the correct object."""
1142
smart.request.request_handlers.get('Branch.get_config_file'),
1143
smart.branch.SmartServerBranchGetConfigFile)
1145
smart.request.request_handlers.get('Branch.lock_write'),
1146
smart.branch.SmartServerBranchRequestLockWrite)
1148
smart.request.request_handlers.get('Branch.last_revision_info'),
1149
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1151
smart.request.request_handlers.get('Branch.revision_history'),
1152
smart.branch.SmartServerRequestRevisionHistory)
1154
smart.request.request_handlers.get('Branch.set_last_revision'),
1155
smart.branch.SmartServerBranchRequestSetLastRevision)
1157
smart.request.request_handlers.get('Branch.set_last_revision_info'),
1158
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1160
smart.request.request_handlers.get('Branch.unlock'),
1161
smart.branch.SmartServerBranchRequestUnlock)
1163
smart.request.request_handlers.get('BzrDir.find_repository'),
1164
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1166
smart.request.request_handlers.get('BzrDir.find_repositoryV2'),
1167
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1169
smart.request.request_handlers.get('BzrDirFormat.initialize'),
1170
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1172
smart.request.request_handlers.get('BzrDir.open_branch'),
1173
smart.bzrdir.SmartServerRequestOpenBranch)
1175
smart.request.request_handlers.get('PackRepository.autopack'),
1176
smart.packrepository.SmartServerPackRepositoryAutopack)
1178
smart.request.request_handlers.get('Repository.gather_stats'),
1179
smart.repository.SmartServerRepositoryGatherStats)
1181
smart.request.request_handlers.get('Repository.get_parent_map'),
1182
smart.repository.SmartServerRepositoryGetParentMap)
1184
smart.request.request_handlers.get(
1185
'Repository.get_revision_graph'),
1186
smart.repository.SmartServerRepositoryGetRevisionGraph)
1188
smart.request.request_handlers.get('Repository.has_revision'),
1189
smart.repository.SmartServerRequestHasRevision)
1191
smart.request.request_handlers.get('Repository.is_shared'),
1192
smart.repository.SmartServerRepositoryIsShared)
1194
smart.request.request_handlers.get('Repository.lock_write'),
1195
smart.repository.SmartServerRepositoryLockWrite)
1197
smart.request.request_handlers.get('Repository.tarball'),
1198
smart.repository.SmartServerRepositoryTarball)
1200
smart.request.request_handlers.get('Repository.unlock'),
1201
smart.repository.SmartServerRepositoryUnlock)
1203
smart.request.request_handlers.get('Transport.is_readonly'),
1204
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
