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 BranchReferenceFormat
40
import bzrlib.smart.branch
41
import bzrlib.smart.bzrdir
42
import bzrlib.smart.repository
43
from bzrlib.smart.request import (
44
FailedSmartServerResponse,
47
SuccessfulSmartServerResponse,
49
from bzrlib.tests import (
54
from bzrlib.transport import chroot, get_transport
55
from bzrlib.util import bencode
58
def load_tests(standard_tests, module, loader):
59
"""Multiply tests version and protocol consistency."""
60
# FindRepository tests.
61
bzrdir_mod = bzrlib.smart.bzrdir
62
applier = TestScenarioApplier()
65
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),
66
("find_repositoryV2", {
67
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),
69
to_adapt, result = split_suite_by_re(standard_tests,
70
"TestSmartServerRequestFindRepository")
71
v2_only, v1_and_2 = split_suite_by_re(to_adapt,
73
for test in iter_suite_tests(v1_and_2):
74
result.addTests(applier.adapt(test))
75
del applier.scenarios[0]
76
for test in iter_suite_tests(v2_only):
77
result.addTests(applier.adapt(test))
81
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
84
tests.TestCaseWithTransport.setUp(self)
85
self._chroot_server = None
87
def get_transport(self, relpath=None):
88
if self._chroot_server is None:
89
backing_transport = tests.TestCaseWithTransport.get_transport(self)
90
self._chroot_server = chroot.ChrootServer(backing_transport)
91
self._chroot_server.setUp()
92
self.addCleanup(self._chroot_server.tearDown)
93
t = get_transport(self._chroot_server.get_url())
94
if relpath is not None:
99
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
102
super(TestCaseWithSmartMedium, self).setUp()
103
# We're allowed to set the transport class here, so that we don't use
104
# the default or a parameterized class, but rather use the
105
# TestCaseWithTransport infrastructure to set up a smart server and
107
self.transport_server = self.make_transport_server
109
def make_transport_server(self):
110
return smart.server.SmartTCPServer_for_testing('-' + self.id())
112
def get_smart_medium(self):
113
"""Get a smart medium to use in tests."""
114
return self.get_transport().get_smart_medium()
117
class TestSmartServerResponse(tests.TestCase):
119
def test__eq__(self):
120
self.assertEqual(SmartServerResponse(('ok', )),
121
SmartServerResponse(('ok', )))
122
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
123
SmartServerResponse(('ok', ), 'body'))
124
self.assertNotEqual(SmartServerResponse(('ok', )),
125
SmartServerResponse(('notok', )))
126
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
127
SmartServerResponse(('ok', )))
128
self.assertNotEqual(None,
129
SmartServerResponse(('ok', )))
131
def test__str__(self):
132
"""SmartServerResponses can be stringified."""
134
"<SmartServerResponse status=OK args=('args',) body='body'>",
135
str(SuccessfulSmartServerResponse(('args',), 'body')))
137
"<SmartServerResponse status=ERR args=('args',) body='body'>",
138
str(FailedSmartServerResponse(('args',), 'body')))
141
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
143
def test_translate_client_path(self):
144
transport = self.get_transport()
145
request = SmartServerRequest(transport, 'foo/')
146
self.assertEqual('./', request.translate_client_path('foo/'))
148
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
150
errors.PathNotChild, request.translate_client_path, '/')
152
errors.PathNotChild, request.translate_client_path, 'bar/')
153
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
155
def test_transport_from_client_path(self):
156
transport = self.get_transport()
157
request = SmartServerRequest(transport, 'foo/')
160
request.transport_from_client_path('foo/').base)
163
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
164
"""Tests for BzrDir.find_repository."""
166
def test_no_repository(self):
167
"""When there is no repository to be found, ('norepository', ) is returned."""
168
backing = self.get_transport()
169
request = self._request_class(backing)
170
self.make_bzrdir('.')
171
self.assertEqual(SmartServerResponse(('norepository', )),
174
def test_nonshared_repository(self):
175
# nonshared repositorys only allow 'find' to return a handle when the
176
# path the repository is being searched on is the same as that that
177
# the repository is at.
178
backing = self.get_transport()
179
request = self._request_class(backing)
180
result = self._make_repository_and_result()
181
self.assertEqual(result, request.execute(''))
182
self.make_bzrdir('subdir')
183
self.assertEqual(SmartServerResponse(('norepository', )),
184
request.execute('subdir'))
186
def _make_repository_and_result(self, shared=False, format=None):
187
"""Convenience function to setup a repository.
189
:result: The SmartServerResponse to expect when opening it.
191
repo = self.make_repository('.', shared=shared, format=format)
192
if repo.supports_rich_root():
196
if repo._format.supports_tree_reference:
200
if (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
201
self._request_class):
202
# All tests so far are on formats, and for non-external
204
return SuccessfulSmartServerResponse(
205
('ok', '', rich_root, subtrees, 'no'))
207
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
209
def test_shared_repository(self):
210
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
211
backing = self.get_transport()
212
request = self._request_class(backing)
213
result = self._make_repository_and_result(shared=True)
214
self.assertEqual(result, request.execute(''))
215
self.make_bzrdir('subdir')
216
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
217
self.assertEqual(result2,
218
request.execute('subdir'))
219
self.make_bzrdir('subdir/deeper')
220
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
221
self.assertEqual(result3,
222
request.execute('subdir/deeper'))
224
def test_rich_root_and_subtree_encoding(self):
225
"""Test for the format attributes for rich root and subtree support."""
226
backing = self.get_transport()
227
request = self._request_class(backing)
228
result = self._make_repository_and_result(format='dirstate-with-subtree')
229
# check the test will be valid
230
self.assertEqual('yes', result.args[2])
231
self.assertEqual('yes', result.args[3])
232
self.assertEqual(result, request.execute(''))
234
def test_supports_external_lookups_no_v2(self):
235
"""Test for the supports_external_lookups attribute."""
236
backing = self.get_transport()
237
request = self._request_class(backing)
238
result = self._make_repository_and_result(format='dirstate-with-subtree')
239
# check the test will be valid
240
self.assertEqual('no', result.args[4])
241
self.assertEqual(result, request.execute(''))
244
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
246
def test_empty_dir(self):
247
"""Initializing an empty dir should succeed and do it."""
248
backing = self.get_transport()
249
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
250
self.assertEqual(SmartServerResponse(('ok', )),
252
made_dir = bzrdir.BzrDir.open_from_transport(backing)
253
# no branch, tree or repository is expected with the current
255
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
256
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
257
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
259
def test_missing_dir(self):
260
"""Initializing a missing directory should fail like the bzrdir api."""
261
backing = self.get_transport()
262
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
263
self.assertRaises(errors.NoSuchFile,
264
request.execute, 'subdir')
266
def test_initialized_dir(self):
267
"""Initializing an extant bzrdir should fail like the bzrdir api."""
268
backing = self.get_transport()
269
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
270
self.make_bzrdir('subdir')
271
self.assertRaises(errors.FileExists,
272
request.execute, 'subdir')
275
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
277
def test_no_branch(self):
278
"""When there is no branch, ('nobranch', ) is returned."""
279
backing = self.get_transport()
280
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
281
self.make_bzrdir('.')
282
self.assertEqual(SmartServerResponse(('nobranch', )),
285
def test_branch(self):
286
"""When there is a branch, 'ok' is returned."""
287
backing = self.get_transport()
288
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
289
self.make_branch('.')
290
self.assertEqual(SmartServerResponse(('ok', '')),
293
def test_branch_reference(self):
294
"""When there is a branch reference, the reference URL is returned."""
295
backing = self.get_transport()
296
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
297
branch = self.make_branch('branch')
298
checkout = branch.create_checkout('reference',lightweight=True)
299
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
300
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
301
self.assertEqual(SmartServerResponse(('ok', reference_url)),
302
request.execute('reference'))
305
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
307
def test_empty(self):
308
"""For an empty branch, the body is empty."""
309
backing = self.get_transport()
310
request = smart.branch.SmartServerRequestRevisionHistory(backing)
311
self.make_branch('.')
312
self.assertEqual(SmartServerResponse(('ok', ), ''),
315
def test_not_empty(self):
316
"""For a non-empty branch, the body is empty."""
317
backing = self.get_transport()
318
request = smart.branch.SmartServerRequestRevisionHistory(backing)
319
tree = self.make_branch_and_memory_tree('.')
322
r1 = tree.commit('1st commit')
323
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
326
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
330
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
332
def test_no_branch(self):
333
"""When there is a bzrdir and no branch, NotBranchError is raised."""
334
backing = self.get_transport()
335
request = smart.branch.SmartServerBranchRequest(backing)
336
self.make_bzrdir('.')
337
self.assertRaises(errors.NotBranchError,
340
def test_branch_reference(self):
341
"""When there is a branch reference, NotBranchError is raised."""
342
backing = self.get_transport()
343
request = smart.branch.SmartServerBranchRequest(backing)
344
branch = self.make_branch('branch')
345
checkout = branch.create_checkout('reference',lightweight=True)
346
self.assertRaises(errors.NotBranchError,
347
request.execute, 'checkout')
350
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
352
def test_empty(self):
353
"""For an empty branch, the result is ('ok', '0', 'null:')."""
354
backing = self.get_transport()
355
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
356
self.make_branch('.')
357
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
360
def test_not_empty(self):
361
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
362
backing = self.get_transport()
363
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
364
tree = self.make_branch_and_memory_tree('.')
367
rev_id_utf8 = u'\xc8'.encode('utf-8')
368
r1 = tree.commit('1st commit')
369
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
372
SmartServerResponse(('ok', '2', rev_id_utf8)),
376
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
378
def test_default(self):
379
"""With no file, we get empty content."""
380
backing = self.get_transport()
381
request = smart.branch.SmartServerBranchGetConfigFile(backing)
382
branch = self.make_branch('.')
383
# there should be no file by default
385
self.assertEqual(SmartServerResponse(('ok', ), content),
388
def test_with_content(self):
389
# SmartServerBranchGetConfigFile should return the content from
390
# branch.control_files.get('branch.conf') for now - in the future it may
391
# perform more complex processing.
392
backing = self.get_transport()
393
request = smart.branch.SmartServerBranchGetConfigFile(backing)
394
branch = self.make_branch('.')
395
branch._transport.put_bytes('branch.conf', 'foo bar baz')
396
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
400
class TestSmartServerBranchRequestSetLastRevision(tests.TestCaseWithMemoryTransport):
402
def test_empty(self):
403
backing = self.get_transport()
404
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
405
b = self.make_branch('.')
406
branch_token = b.lock_write()
407
repo_token = b.repository.lock_write()
408
b.repository.unlock()
410
self.assertEqual(SmartServerResponse(('ok',)),
412
'', branch_token, repo_token,
417
def test_not_present_revision_id(self):
418
backing = self.get_transport()
419
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
420
b = self.make_branch('.')
421
branch_token = b.lock_write()
422
repo_token = b.repository.lock_write()
423
b.repository.unlock()
425
revision_id = 'non-existent revision'
427
SmartServerResponse(('NoSuchRevision', revision_id)),
429
'', branch_token, repo_token,
434
def test_revision_id_present(self):
435
backing = self.get_transport()
436
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
437
tree = self.make_branch_and_memory_tree('.')
440
rev_id_utf8 = u'\xc8'.encode('utf-8')
441
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
442
r2 = tree.commit('2nd commit')
444
branch_token = tree.branch.lock_write()
445
repo_token = tree.branch.repository.lock_write()
446
tree.branch.repository.unlock()
449
SmartServerResponse(('ok',)),
451
'', branch_token, repo_token,
453
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
457
def test_revision_id_present2(self):
458
backing = self.get_transport()
459
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
460
tree = self.make_branch_and_memory_tree('.')
463
rev_id_utf8 = u'\xc8'.encode('utf-8')
464
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
465
r2 = tree.commit('2nd commit')
467
tree.branch.set_revision_history([])
468
branch_token = tree.branch.lock_write()
469
repo_token = tree.branch.repository.lock_write()
470
tree.branch.repository.unlock()
473
SmartServerResponse(('ok',)),
475
'', branch_token, repo_token,
477
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
482
class TestSmartServerBranchRequestSetLastRevisionInfo(tests.TestCaseWithTransport):
484
def lock_branch(self, branch):
485
branch_token = branch.lock_write()
486
repo_token = branch.repository.lock_write()
487
branch.repository.unlock()
488
self.addCleanup(branch.unlock)
489
return branch_token, repo_token
491
def make_locked_branch(self, format=None):
492
branch = self.make_branch('.', format=format)
493
branch_token, repo_token = self.lock_branch(branch)
494
return branch, branch_token, repo_token
496
def test_empty(self):
497
"""An empty branch can have its last revision set to 'null:'."""
498
b, branch_token, repo_token = self.make_locked_branch()
499
backing = self.get_transport()
500
request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(
502
response = request.execute('', branch_token, repo_token, '0', 'null:')
503
self.assertEqual(SmartServerResponse(('ok',)), response)
505
def assertBranchLastRevisionInfo(self, expected_info, branch_relpath):
506
branch = bzrdir.BzrDir.open(branch_relpath).open_branch()
507
self.assertEqual(expected_info, branch.last_revision_info())
509
def test_branch_revision_info_is_updated(self):
510
"""This method really does update the branch last revision info."""
511
tree = self.make_branch_and_memory_tree('.')
514
tree.commit('First commit', rev_id='revision-1')
515
tree.commit('Second commit', rev_id='revision-2')
519
branch_token, repo_token = self.lock_branch(branch)
520
backing = self.get_transport()
521
request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(
523
self.assertBranchLastRevisionInfo((2, 'revision-2'), '.')
524
response = request.execute(
525
'', branch_token, repo_token, '1', 'revision-1')
526
self.assertEqual(SmartServerResponse(('ok',)), response)
527
self.assertBranchLastRevisionInfo((1, 'revision-1'), '.')
529
def test_not_present_revid(self):
530
"""Some branch formats will check that the revision is present in the
531
repository. When that check fails, a NoSuchRevision error is returned
534
# Make a knit format branch, because that format checks the values
535
# given to set_last_revision_info.
536
b, branch_token, repo_token = self.make_locked_branch(format='knit')
537
backing = self.get_transport()
538
request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(
540
response = request.execute(
541
'', branch_token, repo_token, '1', 'not-present')
543
SmartServerResponse(('NoSuchRevision', 'not-present')), response)
546
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
549
tests.TestCaseWithMemoryTransport.setUp(self)
551
def test_lock_write_on_unlocked_branch(self):
552
backing = self.get_transport()
553
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
554
branch = self.make_branch('.', format='knit')
555
repository = branch.repository
556
response = request.execute('')
557
branch_nonce = branch.control_files._lock.peek().get('nonce')
558
repository_nonce = repository.control_files._lock.peek().get('nonce')
560
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
562
# The branch (and associated repository) is now locked. Verify that
563
# with a new branch object.
564
new_branch = repository.bzrdir.open_branch()
565
self.assertRaises(errors.LockContention, new_branch.lock_write)
567
def test_lock_write_on_locked_branch(self):
568
backing = self.get_transport()
569
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
570
branch = self.make_branch('.')
572
branch.leave_lock_in_place()
574
response = request.execute('')
576
SmartServerResponse(('LockContention',)), response)
578
def test_lock_write_with_tokens_on_locked_branch(self):
579
backing = self.get_transport()
580
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
581
branch = self.make_branch('.', format='knit')
582
branch_token = branch.lock_write()
583
repo_token = branch.repository.lock_write()
584
branch.repository.unlock()
585
branch.leave_lock_in_place()
586
branch.repository.leave_lock_in_place()
588
response = request.execute('',
589
branch_token, repo_token)
591
SmartServerResponse(('ok', branch_token, repo_token)), response)
593
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
594
backing = self.get_transport()
595
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
596
branch = self.make_branch('.', format='knit')
597
branch_token = branch.lock_write()
598
repo_token = branch.repository.lock_write()
599
branch.repository.unlock()
600
branch.leave_lock_in_place()
601
branch.repository.leave_lock_in_place()
603
response = request.execute('',
604
branch_token+'xxx', repo_token)
606
SmartServerResponse(('TokenMismatch',)), response)
608
def test_lock_write_on_locked_repo(self):
609
backing = self.get_transport()
610
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
611
branch = self.make_branch('.', format='knit')
612
branch.repository.lock_write()
613
branch.repository.leave_lock_in_place()
614
branch.repository.unlock()
615
response = request.execute('')
617
SmartServerResponse(('LockContention',)), response)
619
def test_lock_write_on_readonly_transport(self):
620
backing = self.get_readonly_transport()
621
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
622
branch = self.make_branch('.')
623
root = self.get_transport().clone('/')
624
path = urlutils.relative_url(root.base, self.get_transport().base)
625
response = request.execute(path)
626
error_name, lock_str, why_str = response.args
627
self.assertFalse(response.is_successful())
628
self.assertEqual('LockFailed', error_name)
631
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
634
tests.TestCaseWithMemoryTransport.setUp(self)
636
def test_unlock_on_locked_branch_and_repo(self):
637
backing = self.get_transport()
638
request = smart.branch.SmartServerBranchRequestUnlock(backing)
639
branch = self.make_branch('.', format='knit')
641
branch_token = branch.lock_write()
642
repo_token = branch.repository.lock_write()
643
branch.repository.unlock()
644
# Unlock the branch (and repo) object, leaving the physical locks
646
branch.leave_lock_in_place()
647
branch.repository.leave_lock_in_place()
649
response = request.execute('',
650
branch_token, repo_token)
652
SmartServerResponse(('ok',)), response)
653
# The branch is now unlocked. Verify that with a new branch
655
new_branch = branch.bzrdir.open_branch()
656
new_branch.lock_write()
659
def test_unlock_on_unlocked_branch_unlocked_repo(self):
660
backing = self.get_transport()
661
request = smart.branch.SmartServerBranchRequestUnlock(backing)
662
branch = self.make_branch('.', format='knit')
663
response = request.execute(
664
'', 'branch token', 'repo token')
666
SmartServerResponse(('TokenMismatch',)), response)
668
def test_unlock_on_unlocked_branch_locked_repo(self):
669
backing = self.get_transport()
670
request = smart.branch.SmartServerBranchRequestUnlock(backing)
671
branch = self.make_branch('.', format='knit')
672
# Lock the repository.
673
repo_token = branch.repository.lock_write()
674
branch.repository.leave_lock_in_place()
675
branch.repository.unlock()
676
# Issue branch lock_write request on the unlocked branch (with locked
678
response = request.execute(
679
'', 'branch token', repo_token)
681
SmartServerResponse(('TokenMismatch',)), response)
684
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
686
def test_no_repository(self):
687
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
688
# we test this using a shared repository above the named path,
689
# thus checking the right search logic is used - that is, that
690
# its the exact path being looked at and the server is not
692
backing = self.get_transport()
693
request = smart.repository.SmartServerRepositoryRequest(backing)
694
self.make_repository('.', shared=True)
695
self.make_bzrdir('subdir')
696
self.assertRaises(errors.NoRepositoryPresent,
697
request.execute, 'subdir')
700
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithTransport):
702
def test_trivial_bzipped(self):
703
# This tests that the wire encoding is actually bzipped
704
backing = self.get_transport()
705
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
706
tree = self.make_branch_and_memory_tree('.')
708
self.assertEqual(None,
709
request.execute('', 'missing-id'))
710
# Note that it returns a body (of '' bzipped).
712
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
713
request.do_body('\n\n0\n'))
716
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
718
def test_none_argument(self):
719
backing = self.get_transport()
720
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
721
tree = self.make_branch_and_memory_tree('.')
724
r1 = tree.commit('1st commit')
725
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
728
# the lines of revision_id->revision_parent_list has no guaranteed
729
# order coming out of a dict, so sort both our test and response
730
lines = sorted([' '.join([r2, r1]), r1])
731
response = request.execute('', '')
732
response.body = '\n'.join(sorted(response.body.split('\n')))
735
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
737
def test_specific_revision_argument(self):
738
backing = self.get_transport()
739
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
740
tree = self.make_branch_and_memory_tree('.')
743
rev_id_utf8 = u'\xc9'.encode('utf-8')
744
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
745
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
748
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
749
request.execute('', rev_id_utf8))
751
def test_no_such_revision(self):
752
backing = self.get_transport()
753
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
754
tree = self.make_branch_and_memory_tree('.')
757
r1 = tree.commit('1st commit')
760
# Note that it still returns body (of zero bytes).
762
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
763
request.execute('', 'missingrevision'))
766
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
768
def test_missing_revision(self):
769
"""For a missing revision, ('no', ) is returned."""
770
backing = self.get_transport()
771
request = smart.repository.SmartServerRequestHasRevision(backing)
772
self.make_repository('.')
773
self.assertEqual(SmartServerResponse(('no', )),
774
request.execute('', 'revid'))
776
def test_present_revision(self):
777
"""For a present revision, ('yes', ) is returned."""
778
backing = self.get_transport()
779
request = smart.repository.SmartServerRequestHasRevision(backing)
780
tree = self.make_branch_and_memory_tree('.')
783
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
784
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
786
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
787
self.assertEqual(SmartServerResponse(('yes', )),
788
request.execute('', rev_id_utf8))
791
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
793
def test_empty_revid(self):
794
"""With an empty revid, we get only size an number and revisions"""
795
backing = self.get_transport()
796
request = smart.repository.SmartServerRepositoryGatherStats(backing)
797
repository = self.make_repository('.')
798
stats = repository.gather_stats()
799
expected_body = 'revisions: 0\n'
800
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
801
request.execute('', '', 'no'))
803
def test_revid_with_committers(self):
804
"""For a revid we get more infos."""
805
backing = self.get_transport()
806
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
807
request = smart.repository.SmartServerRepositoryGatherStats(backing)
808
tree = self.make_branch_and_memory_tree('.')
811
# Let's build a predictable result
812
tree.commit('a commit', timestamp=123456.2, timezone=3600)
813
tree.commit('a commit', timestamp=654321.4, timezone=0,
817
stats = tree.branch.repository.gather_stats()
818
expected_body = ('firstrev: 123456.200 3600\n'
819
'latestrev: 654321.400 0\n'
821
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
825
def test_not_empty_repository_with_committers(self):
826
"""For a revid and requesting committers we get the whole thing."""
827
backing = self.get_transport()
828
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
829
request = smart.repository.SmartServerRepositoryGatherStats(backing)
830
tree = self.make_branch_and_memory_tree('.')
833
# Let's build a predictable result
834
tree.commit('a commit', timestamp=123456.2, timezone=3600,
836
tree.commit('a commit', timestamp=654321.4, timezone=0,
837
committer='bar', rev_id=rev_id_utf8)
839
stats = tree.branch.repository.gather_stats()
841
expected_body = ('committers: 2\n'
842
'firstrev: 123456.200 3600\n'
843
'latestrev: 654321.400 0\n'
845
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
850
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
852
def test_is_shared(self):
853
"""For a shared repository, ('yes', ) is returned."""
854
backing = self.get_transport()
855
request = smart.repository.SmartServerRepositoryIsShared(backing)
856
self.make_repository('.', shared=True)
857
self.assertEqual(SmartServerResponse(('yes', )),
858
request.execute('', ))
860
def test_is_not_shared(self):
861
"""For a shared repository, ('no', ) is returned."""
862
backing = self.get_transport()
863
request = smart.repository.SmartServerRepositoryIsShared(backing)
864
self.make_repository('.', shared=False)
865
self.assertEqual(SmartServerResponse(('no', )),
866
request.execute('', ))
869
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
872
tests.TestCaseWithMemoryTransport.setUp(self)
874
def test_lock_write_on_unlocked_repo(self):
875
backing = self.get_transport()
876
request = smart.repository.SmartServerRepositoryLockWrite(backing)
877
repository = self.make_repository('.', format='knit')
878
response = request.execute('')
879
nonce = repository.control_files._lock.peek().get('nonce')
880
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
881
# The repository is now locked. Verify that with a new repository
883
new_repo = repository.bzrdir.open_repository()
884
self.assertRaises(errors.LockContention, new_repo.lock_write)
886
def test_lock_write_on_locked_repo(self):
887
backing = self.get_transport()
888
request = smart.repository.SmartServerRepositoryLockWrite(backing)
889
repository = self.make_repository('.', format='knit')
890
repository.lock_write()
891
repository.leave_lock_in_place()
893
response = request.execute('')
895
SmartServerResponse(('LockContention',)), response)
897
def test_lock_write_on_readonly_transport(self):
898
backing = self.get_readonly_transport()
899
request = smart.repository.SmartServerRepositoryLockWrite(backing)
900
repository = self.make_repository('.', format='knit')
901
response = request.execute('')
902
self.assertFalse(response.is_successful())
903
self.assertEqual('LockFailed', response.args[0])
906
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
909
tests.TestCaseWithMemoryTransport.setUp(self)
911
def test_unlock_on_locked_repo(self):
912
backing = self.get_transport()
913
request = smart.repository.SmartServerRepositoryUnlock(backing)
914
repository = self.make_repository('.', format='knit')
915
token = repository.lock_write()
916
repository.leave_lock_in_place()
918
response = request.execute('', token)
920
SmartServerResponse(('ok',)), response)
921
# The repository is now unlocked. Verify that with a new repository
923
new_repo = repository.bzrdir.open_repository()
924
new_repo.lock_write()
927
def test_unlock_on_unlocked_repo(self):
928
backing = self.get_transport()
929
request = smart.repository.SmartServerRepositoryUnlock(backing)
930
repository = self.make_repository('.', format='knit')
931
response = request.execute('', 'some token')
933
SmartServerResponse(('TokenMismatch',)), response)
936
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
938
def test_is_readonly_no(self):
939
backing = self.get_transport()
940
request = smart.request.SmartServerIsReadonly(backing)
941
response = request.execute()
943
SmartServerResponse(('no',)), response)
945
def test_is_readonly_yes(self):
946
backing = self.get_readonly_transport()
947
request = smart.request.SmartServerIsReadonly(backing)
948
response = request.execute()
950
SmartServerResponse(('yes',)), response)
953
class TestHandlers(tests.TestCase):
954
"""Tests for the request.request_handlers object."""
956
def test_registered_methods(self):
957
"""Test that known methods are registered to the correct object."""
959
smart.request.request_handlers.get('Branch.get_config_file'),
960
smart.branch.SmartServerBranchGetConfigFile)
962
smart.request.request_handlers.get('Branch.lock_write'),
963
smart.branch.SmartServerBranchRequestLockWrite)
965
smart.request.request_handlers.get('Branch.last_revision_info'),
966
smart.branch.SmartServerBranchRequestLastRevisionInfo)
968
smart.request.request_handlers.get('Branch.revision_history'),
969
smart.branch.SmartServerRequestRevisionHistory)
971
smart.request.request_handlers.get('Branch.set_last_revision'),
972
smart.branch.SmartServerBranchRequestSetLastRevision)
974
smart.request.request_handlers.get('Branch.set_last_revision_info'),
975
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
977
smart.request.request_handlers.get('Branch.unlock'),
978
smart.branch.SmartServerBranchRequestUnlock)
980
smart.request.request_handlers.get('BzrDir.find_repository'),
981
smart.bzrdir.SmartServerRequestFindRepositoryV1)
983
smart.request.request_handlers.get('BzrDir.find_repositoryV2'),
984
smart.bzrdir.SmartServerRequestFindRepositoryV2)
986
smart.request.request_handlers.get('BzrDirFormat.initialize'),
987
smart.bzrdir.SmartServerRequestInitializeBzrDir)
989
smart.request.request_handlers.get('BzrDir.open_branch'),
990
smart.bzrdir.SmartServerRequestOpenBranch)
992
smart.request.request_handlers.get('Repository.gather_stats'),
993
smart.repository.SmartServerRepositoryGatherStats)
995
smart.request.request_handlers.get('Repository.get_parent_map'),
996
smart.repository.SmartServerRepositoryGetParentMap)
998
smart.request.request_handlers.get(
999
'Repository.get_revision_graph'),
1000
smart.repository.SmartServerRepositoryGetRevisionGraph)
1002
smart.request.request_handlers.get('Repository.has_revision'),
1003
smart.repository.SmartServerRequestHasRevision)
1005
smart.request.request_handlers.get('Repository.is_shared'),
1006
smart.repository.SmartServerRepositoryIsShared)
1008
smart.request.request_handlers.get('Repository.lock_write'),
1009
smart.repository.SmartServerRepositoryLockWrite)
1011
smart.request.request_handlers.get('Repository.tarball'),
1012
smart.repository.SmartServerRepositoryTarball)
1014
smart.request.request_handlers.get('Repository.unlock'),
1015
smart.repository.SmartServerRepositoryUnlock)
1017
smart.request.request_handlers.get('Transport.is_readonly'),
1018
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
