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 = smart.server.SmartTCPServer_for_testing
109
def get_smart_medium(self):
110
"""Get a smart medium to use in tests."""
111
return self.get_transport().get_smart_medium()
114
class TestSmartServerResponse(tests.TestCase):
116
def test__eq__(self):
117
self.assertEqual(SmartServerResponse(('ok', )),
118
SmartServerResponse(('ok', )))
119
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
120
SmartServerResponse(('ok', ), 'body'))
121
self.assertNotEqual(SmartServerResponse(('ok', )),
122
SmartServerResponse(('notok', )))
123
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
124
SmartServerResponse(('ok', )))
125
self.assertNotEqual(None,
126
SmartServerResponse(('ok', )))
128
def test__str__(self):
129
"""SmartServerResponses can be stringified."""
131
"<SmartServerResponse status=OK args=('args',) body='body'>",
132
str(SuccessfulSmartServerResponse(('args',), 'body')))
134
"<SmartServerResponse status=ERR args=('args',) body='body'>",
135
str(FailedSmartServerResponse(('args',), 'body')))
138
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
140
def test_translate_client_path(self):
141
transport = self.get_transport()
142
request = SmartServerRequest(transport, 'foo/')
143
self.assertEqual('./', request.translate_client_path('foo/'))
145
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
147
errors.PathNotChild, request.translate_client_path, '/')
149
errors.PathNotChild, request.translate_client_path, 'bar/')
150
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
152
def test_transport_from_client_path(self):
153
transport = self.get_transport()
154
request = SmartServerRequest(transport, 'foo/')
157
request.transport_from_client_path('foo/').base)
160
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
161
"""Tests for BzrDir.find_repository."""
163
def test_no_repository(self):
164
"""When there is no repository to be found, ('norepository', ) is returned."""
165
backing = self.get_transport()
166
request = self._request_class(backing)
167
self.make_bzrdir('.')
168
self.assertEqual(SmartServerResponse(('norepository', )),
171
def test_nonshared_repository(self):
172
# nonshared repositorys only allow 'find' to return a handle when the
173
# path the repository is being searched on is the same as that that
174
# the repository is at.
175
backing = self.get_transport()
176
request = self._request_class(backing)
177
result = self._make_repository_and_result()
178
self.assertEqual(result, request.execute(''))
179
self.make_bzrdir('subdir')
180
self.assertEqual(SmartServerResponse(('norepository', )),
181
request.execute('subdir'))
183
def _make_repository_and_result(self, shared=False, format=None):
184
"""Convenience function to setup a repository.
186
:result: The SmartServerResponse to expect when opening it.
188
repo = self.make_repository('.', shared=shared, format=format)
189
if repo.supports_rich_root():
193
if repo._format.supports_tree_reference:
197
if (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
198
self._request_class):
199
# All tests so far are on formats, and for non-external
201
return SuccessfulSmartServerResponse(
202
('ok', '', rich_root, subtrees, 'no'))
204
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
206
def test_shared_repository(self):
207
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
208
backing = self.get_transport()
209
request = self._request_class(backing)
210
result = self._make_repository_and_result(shared=True)
211
self.assertEqual(result, request.execute(''))
212
self.make_bzrdir('subdir')
213
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
214
self.assertEqual(result2,
215
request.execute('subdir'))
216
self.make_bzrdir('subdir/deeper')
217
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
218
self.assertEqual(result3,
219
request.execute('subdir/deeper'))
221
def test_rich_root_and_subtree_encoding(self):
222
"""Test for the format attributes for rich root and subtree support."""
223
backing = self.get_transport()
224
request = self._request_class(backing)
225
result = self._make_repository_and_result(format='dirstate-with-subtree')
226
# check the test will be valid
227
self.assertEqual('yes', result.args[2])
228
self.assertEqual('yes', result.args[3])
229
self.assertEqual(result, request.execute(''))
231
def test_supports_external_lookups_no_v2(self):
232
"""Test for the supports_external_lookups attribute."""
233
backing = self.get_transport()
234
request = self._request_class(backing)
235
result = self._make_repository_and_result(format='dirstate-with-subtree')
236
# check the test will be valid
237
self.assertEqual('no', result.args[4])
238
self.assertEqual(result, request.execute(''))
241
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
243
def test_empty_dir(self):
244
"""Initializing an empty dir should succeed and do it."""
245
backing = self.get_transport()
246
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
247
self.assertEqual(SmartServerResponse(('ok', )),
249
made_dir = bzrdir.BzrDir.open_from_transport(backing)
250
# no branch, tree or repository is expected with the current
252
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
253
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
254
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
256
def test_missing_dir(self):
257
"""Initializing a missing directory should fail like the bzrdir api."""
258
backing = self.get_transport()
259
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
260
self.assertRaises(errors.NoSuchFile,
261
request.execute, 'subdir')
263
def test_initialized_dir(self):
264
"""Initializing an extant bzrdir should fail like the bzrdir api."""
265
backing = self.get_transport()
266
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
267
self.make_bzrdir('subdir')
268
self.assertRaises(errors.FileExists,
269
request.execute, 'subdir')
272
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
274
def test_no_branch(self):
275
"""When there is no branch, ('nobranch', ) is returned."""
276
backing = self.get_transport()
277
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
278
self.make_bzrdir('.')
279
self.assertEqual(SmartServerResponse(('nobranch', )),
282
def test_branch(self):
283
"""When there is a branch, 'ok' is returned."""
284
backing = self.get_transport()
285
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
286
self.make_branch('.')
287
self.assertEqual(SmartServerResponse(('ok', '')),
290
def test_branch_reference(self):
291
"""When there is a branch reference, the reference URL is returned."""
292
backing = self.get_transport()
293
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
294
branch = self.make_branch('branch')
295
checkout = branch.create_checkout('reference',lightweight=True)
296
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
297
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
298
self.assertEqual(SmartServerResponse(('ok', reference_url)),
299
request.execute('reference'))
302
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
304
def test_empty(self):
305
"""For an empty branch, the body is empty."""
306
backing = self.get_transport()
307
request = smart.branch.SmartServerRequestRevisionHistory(backing)
308
self.make_branch('.')
309
self.assertEqual(SmartServerResponse(('ok', ), ''),
312
def test_not_empty(self):
313
"""For a non-empty branch, the body is empty."""
314
backing = self.get_transport()
315
request = smart.branch.SmartServerRequestRevisionHistory(backing)
316
tree = self.make_branch_and_memory_tree('.')
319
r1 = tree.commit('1st commit')
320
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
323
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
327
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
329
def test_no_branch(self):
330
"""When there is a bzrdir and no branch, NotBranchError is raised."""
331
backing = self.get_transport()
332
request = smart.branch.SmartServerBranchRequest(backing)
333
self.make_bzrdir('.')
334
self.assertRaises(errors.NotBranchError,
337
def test_branch_reference(self):
338
"""When there is a branch reference, NotBranchError is raised."""
339
backing = self.get_transport()
340
request = smart.branch.SmartServerBranchRequest(backing)
341
branch = self.make_branch('branch')
342
checkout = branch.create_checkout('reference',lightweight=True)
343
self.assertRaises(errors.NotBranchError,
344
request.execute, 'checkout')
347
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
349
def test_empty(self):
350
"""For an empty branch, the result is ('ok', '0', 'null:')."""
351
backing = self.get_transport()
352
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
353
self.make_branch('.')
354
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
357
def test_not_empty(self):
358
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
359
backing = self.get_transport()
360
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
361
tree = self.make_branch_and_memory_tree('.')
364
rev_id_utf8 = u'\xc8'.encode('utf-8')
365
r1 = tree.commit('1st commit')
366
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
369
SmartServerResponse(('ok', '2', rev_id_utf8)),
373
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
375
def test_default(self):
376
"""With no file, we get empty content."""
377
backing = self.get_transport()
378
request = smart.branch.SmartServerBranchGetConfigFile(backing)
379
branch = self.make_branch('.')
380
# there should be no file by default
382
self.assertEqual(SmartServerResponse(('ok', ), content),
385
def test_with_content(self):
386
# SmartServerBranchGetConfigFile should return the content from
387
# branch.control_files.get('branch.conf') for now - in the future it may
388
# perform more complex processing.
389
backing = self.get_transport()
390
request = smart.branch.SmartServerBranchGetConfigFile(backing)
391
branch = self.make_branch('.')
392
branch.control_files.put_utf8('branch.conf', 'foo bar baz')
393
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
397
class TestSmartServerBranchRequestSetLastRevision(tests.TestCaseWithMemoryTransport):
399
def test_empty(self):
400
backing = self.get_transport()
401
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
402
b = self.make_branch('.')
403
branch_token = b.lock_write()
404
repo_token = b.repository.lock_write()
405
b.repository.unlock()
407
self.assertEqual(SmartServerResponse(('ok',)),
409
'', branch_token, repo_token,
414
def test_not_present_revision_id(self):
415
backing = self.get_transport()
416
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
417
b = self.make_branch('.')
418
branch_token = b.lock_write()
419
repo_token = b.repository.lock_write()
420
b.repository.unlock()
422
revision_id = 'non-existent revision'
424
SmartServerResponse(('NoSuchRevision', revision_id)),
426
'', branch_token, repo_token,
431
def test_revision_id_present(self):
432
backing = self.get_transport()
433
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
434
tree = self.make_branch_and_memory_tree('.')
437
rev_id_utf8 = u'\xc8'.encode('utf-8')
438
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
439
r2 = tree.commit('2nd commit')
441
branch_token = tree.branch.lock_write()
442
repo_token = tree.branch.repository.lock_write()
443
tree.branch.repository.unlock()
446
SmartServerResponse(('ok',)),
448
'', branch_token, repo_token,
450
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
454
def test_revision_id_present2(self):
455
backing = self.get_transport()
456
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
457
tree = self.make_branch_and_memory_tree('.')
460
rev_id_utf8 = u'\xc8'.encode('utf-8')
461
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
462
r2 = tree.commit('2nd commit')
464
tree.branch.set_revision_history([])
465
branch_token = tree.branch.lock_write()
466
repo_token = tree.branch.repository.lock_write()
467
tree.branch.repository.unlock()
470
SmartServerResponse(('ok',)),
472
'', branch_token, repo_token,
474
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
479
class TestSmartServerBranchRequestSetLastRevisionInfo(tests.TestCaseWithTransport):
481
def test_revision_id_present(self):
482
backing = self.get_transport()
483
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
484
tree = self.make_branch_and_memory_tree('.')
487
rev_id_utf8 = u'\xc8'.encode('utf-8')
488
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
489
r2 = tree.commit('2nd commit')
491
branch_token = tree.branch.lock_write()
492
repo_token = tree.branch.repository.lock_write()
493
tree.branch.repository.unlock()
496
SmartServerResponse(('ok',)),
498
'', branch_token, repo_token,
500
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
505
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
508
tests.TestCaseWithMemoryTransport.setUp(self)
510
def test_lock_write_on_unlocked_branch(self):
511
backing = self.get_transport()
512
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
513
branch = self.make_branch('.', format='knit')
514
repository = branch.repository
515
response = request.execute('')
516
branch_nonce = branch.control_files._lock.peek().get('nonce')
517
repository_nonce = repository.control_files._lock.peek().get('nonce')
519
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
521
# The branch (and associated repository) is now locked. Verify that
522
# with a new branch object.
523
new_branch = repository.bzrdir.open_branch()
524
self.assertRaises(errors.LockContention, new_branch.lock_write)
526
def test_lock_write_on_locked_branch(self):
527
backing = self.get_transport()
528
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
529
branch = self.make_branch('.')
531
branch.leave_lock_in_place()
533
response = request.execute('')
535
SmartServerResponse(('LockContention',)), response)
537
def test_lock_write_with_tokens_on_locked_branch(self):
538
backing = self.get_transport()
539
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
540
branch = self.make_branch('.', format='knit')
541
branch_token = branch.lock_write()
542
repo_token = branch.repository.lock_write()
543
branch.repository.unlock()
544
branch.leave_lock_in_place()
545
branch.repository.leave_lock_in_place()
547
response = request.execute('',
548
branch_token, repo_token)
550
SmartServerResponse(('ok', branch_token, repo_token)), response)
552
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
553
backing = self.get_transport()
554
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
555
branch = self.make_branch('.', format='knit')
556
branch_token = branch.lock_write()
557
repo_token = branch.repository.lock_write()
558
branch.repository.unlock()
559
branch.leave_lock_in_place()
560
branch.repository.leave_lock_in_place()
562
response = request.execute('',
563
branch_token+'xxx', repo_token)
565
SmartServerResponse(('TokenMismatch',)), response)
567
def test_lock_write_on_locked_repo(self):
568
backing = self.get_transport()
569
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
570
branch = self.make_branch('.', format='knit')
571
branch.repository.lock_write()
572
branch.repository.leave_lock_in_place()
573
branch.repository.unlock()
574
response = request.execute('')
576
SmartServerResponse(('LockContention',)), response)
578
def test_lock_write_on_readonly_transport(self):
579
backing = self.get_readonly_transport()
580
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
581
branch = self.make_branch('.')
582
root = self.get_transport().clone('/')
583
path = urlutils.relative_url(root.base, self.get_transport().base)
584
response = request.execute(path)
585
error_name, lock_str, why_str = response.args
586
self.assertFalse(response.is_successful())
587
self.assertEqual('LockFailed', error_name)
590
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
593
tests.TestCaseWithMemoryTransport.setUp(self)
595
def test_unlock_on_locked_branch_and_repo(self):
596
backing = self.get_transport()
597
request = smart.branch.SmartServerBranchRequestUnlock(backing)
598
branch = self.make_branch('.', format='knit')
600
branch_token = branch.lock_write()
601
repo_token = branch.repository.lock_write()
602
branch.repository.unlock()
603
# Unlock the branch (and repo) object, leaving the physical locks
605
branch.leave_lock_in_place()
606
branch.repository.leave_lock_in_place()
608
response = request.execute('',
609
branch_token, repo_token)
611
SmartServerResponse(('ok',)), response)
612
# The branch is now unlocked. Verify that with a new branch
614
new_branch = branch.bzrdir.open_branch()
615
new_branch.lock_write()
618
def test_unlock_on_unlocked_branch_unlocked_repo(self):
619
backing = self.get_transport()
620
request = smart.branch.SmartServerBranchRequestUnlock(backing)
621
branch = self.make_branch('.', format='knit')
622
response = request.execute(
623
'', 'branch token', 'repo token')
625
SmartServerResponse(('TokenMismatch',)), response)
627
def test_unlock_on_unlocked_branch_locked_repo(self):
628
backing = self.get_transport()
629
request = smart.branch.SmartServerBranchRequestUnlock(backing)
630
branch = self.make_branch('.', format='knit')
631
# Lock the repository.
632
repo_token = branch.repository.lock_write()
633
branch.repository.leave_lock_in_place()
634
branch.repository.unlock()
635
# Issue branch lock_write request on the unlocked branch (with locked
637
response = request.execute(
638
'', 'branch token', repo_token)
640
SmartServerResponse(('TokenMismatch',)), response)
643
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
645
def test_no_repository(self):
646
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
647
# we test this using a shared repository above the named path,
648
# thus checking the right search logic is used - that is, that
649
# its the exact path being looked at and the server is not
651
backing = self.get_transport()
652
request = smart.repository.SmartServerRepositoryRequest(backing)
653
self.make_repository('.', shared=True)
654
self.make_bzrdir('subdir')
655
self.assertRaises(errors.NoRepositoryPresent,
656
request.execute, 'subdir')
659
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithTransport):
661
def test_trivial_bzipped(self):
662
# This tests that the wire encoding is actually bzipped
663
backing = self.get_transport()
664
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
665
tree = self.make_branch_and_memory_tree('.')
667
self.assertEqual(None,
668
request.execute('', 'missing-id'))
669
# Note that it returns a body (of '' bzipped).
671
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
672
request.do_body('\n\n0\n'))
675
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
677
def test_none_argument(self):
678
backing = self.get_transport()
679
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
680
tree = self.make_branch_and_memory_tree('.')
683
r1 = tree.commit('1st commit')
684
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
687
# the lines of revision_id->revision_parent_list has no guaranteed
688
# order coming out of a dict, so sort both our test and response
689
lines = sorted([' '.join([r2, r1]), r1])
690
response = request.execute('', '')
691
response.body = '\n'.join(sorted(response.body.split('\n')))
694
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
696
def test_specific_revision_argument(self):
697
backing = self.get_transport()
698
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
699
tree = self.make_branch_and_memory_tree('.')
702
rev_id_utf8 = u'\xc9'.encode('utf-8')
703
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
704
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
707
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
708
request.execute('', rev_id_utf8))
710
def test_no_such_revision(self):
711
backing = self.get_transport()
712
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
713
tree = self.make_branch_and_memory_tree('.')
716
r1 = tree.commit('1st commit')
719
# Note that it still returns body (of zero bytes).
721
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
722
request.execute('', 'missingrevision'))
725
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
727
def test_missing_revision(self):
728
"""For a missing revision, ('no', ) is returned."""
729
backing = self.get_transport()
730
request = smart.repository.SmartServerRequestHasRevision(backing)
731
self.make_repository('.')
732
self.assertEqual(SmartServerResponse(('no', )),
733
request.execute('', 'revid'))
735
def test_present_revision(self):
736
"""For a present revision, ('yes', ) is returned."""
737
backing = self.get_transport()
738
request = smart.repository.SmartServerRequestHasRevision(backing)
739
tree = self.make_branch_and_memory_tree('.')
742
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
743
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
745
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
746
self.assertEqual(SmartServerResponse(('yes', )),
747
request.execute('', rev_id_utf8))
750
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
752
def test_empty_revid(self):
753
"""With an empty revid, we get only size an number and revisions"""
754
backing = self.get_transport()
755
request = smart.repository.SmartServerRepositoryGatherStats(backing)
756
repository = self.make_repository('.')
757
stats = repository.gather_stats()
759
expected_body = 'revisions: 0\nsize: %d\n' % size
760
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
761
request.execute('', '', 'no'))
763
def test_revid_with_committers(self):
764
"""For a revid we get more infos."""
765
backing = self.get_transport()
766
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
767
request = smart.repository.SmartServerRepositoryGatherStats(backing)
768
tree = self.make_branch_and_memory_tree('.')
771
# Let's build a predictable result
772
tree.commit('a commit', timestamp=123456.2, timezone=3600)
773
tree.commit('a commit', timestamp=654321.4, timezone=0,
777
stats = tree.branch.repository.gather_stats()
779
expected_body = ('firstrev: 123456.200 3600\n'
780
'latestrev: 654321.400 0\n'
783
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
787
def test_not_empty_repository_with_committers(self):
788
"""For a revid and requesting committers we get the whole thing."""
789
backing = self.get_transport()
790
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
791
request = smart.repository.SmartServerRepositoryGatherStats(backing)
792
tree = self.make_branch_and_memory_tree('.')
795
# Let's build a predictable result
796
tree.commit('a commit', timestamp=123456.2, timezone=3600,
798
tree.commit('a commit', timestamp=654321.4, timezone=0,
799
committer='bar', rev_id=rev_id_utf8)
801
stats = tree.branch.repository.gather_stats()
804
expected_body = ('committers: 2\n'
805
'firstrev: 123456.200 3600\n'
806
'latestrev: 654321.400 0\n'
809
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
814
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
816
def test_is_shared(self):
817
"""For a shared repository, ('yes', ) is returned."""
818
backing = self.get_transport()
819
request = smart.repository.SmartServerRepositoryIsShared(backing)
820
self.make_repository('.', shared=True)
821
self.assertEqual(SmartServerResponse(('yes', )),
822
request.execute('', ))
824
def test_is_not_shared(self):
825
"""For a shared repository, ('no', ) is returned."""
826
backing = self.get_transport()
827
request = smart.repository.SmartServerRepositoryIsShared(backing)
828
self.make_repository('.', shared=False)
829
self.assertEqual(SmartServerResponse(('no', )),
830
request.execute('', ))
833
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
836
tests.TestCaseWithMemoryTransport.setUp(self)
838
def test_lock_write_on_unlocked_repo(self):
839
backing = self.get_transport()
840
request = smart.repository.SmartServerRepositoryLockWrite(backing)
841
repository = self.make_repository('.', format='knit')
842
response = request.execute('')
843
nonce = repository.control_files._lock.peek().get('nonce')
844
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
845
# The repository is now locked. Verify that with a new repository
847
new_repo = repository.bzrdir.open_repository()
848
self.assertRaises(errors.LockContention, new_repo.lock_write)
850
def test_lock_write_on_locked_repo(self):
851
backing = self.get_transport()
852
request = smart.repository.SmartServerRepositoryLockWrite(backing)
853
repository = self.make_repository('.', format='knit')
854
repository.lock_write()
855
repository.leave_lock_in_place()
857
response = request.execute('')
859
SmartServerResponse(('LockContention',)), response)
861
def test_lock_write_on_readonly_transport(self):
862
backing = self.get_readonly_transport()
863
request = smart.repository.SmartServerRepositoryLockWrite(backing)
864
repository = self.make_repository('.', format='knit')
865
response = request.execute('')
866
self.assertFalse(response.is_successful())
867
self.assertEqual('LockFailed', response.args[0])
870
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
873
tests.TestCaseWithMemoryTransport.setUp(self)
875
def test_unlock_on_locked_repo(self):
876
backing = self.get_transport()
877
request = smart.repository.SmartServerRepositoryUnlock(backing)
878
repository = self.make_repository('.', format='knit')
879
token = repository.lock_write()
880
repository.leave_lock_in_place()
882
response = request.execute('', token)
884
SmartServerResponse(('ok',)), response)
885
# The repository is now unlocked. Verify that with a new repository
887
new_repo = repository.bzrdir.open_repository()
888
new_repo.lock_write()
891
def test_unlock_on_unlocked_repo(self):
892
backing = self.get_transport()
893
request = smart.repository.SmartServerRepositoryUnlock(backing)
894
repository = self.make_repository('.', format='knit')
895
response = request.execute('', 'some token')
897
SmartServerResponse(('TokenMismatch',)), response)
900
class TestSmartServerRepositoryTarball(tests.TestCaseWithTransport):
902
def test_repository_tarball(self):
903
backing = self.get_transport()
904
request = smart.repository.SmartServerRepositoryTarball(backing)
905
repository = self.make_repository('.')
906
# make some extraneous junk in the repository directory which should
908
self.build_tree(['.bzr/repository/extra-junk'])
909
response = request.execute('', 'bz2')
910
self.assertEqual(('ok',), response.args)
911
# body should be a tbz2
912
body_file = StringIO(response.body)
913
body_tar = tarfile.open('body_tar.tbz2', fileobj=body_file,
915
# let's make sure there are some key repository components inside it.
916
# the tarfile returns directories with trailing slashes...
917
names = set([n.rstrip('/') for n in body_tar.getnames()])
918
self.assertTrue('.bzr/repository/lock' in names)
919
self.assertTrue('.bzr/repository/format' in names)
920
self.assertTrue('.bzr/repository/extra-junk' not in names,
921
"extraneous file present in tar file")
924
class TestSmartServerRepositoryStreamKnitData(tests.TestCaseWithMemoryTransport):
926
def test_fetch_revisions(self):
927
backing = self.get_transport()
928
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
929
tree = self.make_branch_and_memory_tree('.')
932
rev_id1_utf8 = u'\xc8'.encode('utf-8')
933
rev_id2_utf8 = u'\xc9'.encode('utf-8')
934
r1 = tree.commit('1st commit', rev_id=rev_id1_utf8)
935
r1 = tree.commit('2nd commit', rev_id=rev_id2_utf8)
938
response = request.execute('', rev_id2_utf8)
939
self.assertEqual(('ok',), response.args)
940
unpacker = pack.ContainerReader(StringIO(response.body))
942
for [name], read_bytes in unpacker.iter_records():
944
bytes = read_bytes(None)
945
# The bytes should be a valid bencoded string.
946
bencode.bdecode(bytes)
947
# XXX: assert that the bencoded knit records have the right
950
def test_no_such_revision_error(self):
951
backing = self.get_transport()
952
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
953
repo = self.make_repository('.')
954
rev_id1_utf8 = u'\xc8'.encode('utf-8')
955
response = request.execute('', rev_id1_utf8)
957
SmartServerResponse(('NoSuchRevision', rev_id1_utf8)),
961
class TestSmartServerRepositoryStreamRevisionsChunked(tests.TestCaseWithMemoryTransport):
963
def test_fetch_revisions(self):
964
backing = self.get_transport()
965
request = smart.repository.SmartServerRepositoryStreamRevisionsChunked(
967
tree = self.make_branch_and_memory_tree('.')
970
rev_id1_utf8 = u'\xc8'.encode('utf-8')
971
rev_id2_utf8 = u'\xc9'.encode('utf-8')
972
tree.commit('1st commit', rev_id=rev_id1_utf8)
973
tree.commit('2nd commit', rev_id=rev_id2_utf8)
976
response = request.execute('')
977
self.assertEqual(None, response)
978
response = request.do_body("%s\n%s\n1" % (rev_id2_utf8, rev_id1_utf8))
979
self.assertEqual(('ok',), response.args)
980
parser = pack.ContainerPushParser()
982
for stream_bytes in response.body_stream:
983
parser.accept_bytes(stream_bytes)
984
for [name], record_bytes in parser.read_pending_records():
986
# The bytes should be a valid bencoded string.
987
bencode.bdecode(record_bytes)
988
# XXX: assert that the bencoded knit records have the right
991
def test_no_such_revision_error(self):
992
backing = self.get_transport()
993
request = smart.repository.SmartServerRepositoryStreamRevisionsChunked(
995
repo = self.make_repository('.')
996
rev_id1_utf8 = u'\xc8'.encode('utf-8')
997
response = request.execute('')
998
self.assertEqual(None, response)
999
response = request.do_body("%s\n\n1" % (rev_id1_utf8,))
1001
FailedSmartServerResponse(('NoSuchRevision', )),
1005
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1007
def test_is_readonly_no(self):
1008
backing = self.get_transport()
1009
request = smart.request.SmartServerIsReadonly(backing)
1010
response = request.execute()
1012
SmartServerResponse(('no',)), response)
1014
def test_is_readonly_yes(self):
1015
backing = self.get_readonly_transport()
1016
request = smart.request.SmartServerIsReadonly(backing)
1017
response = request.execute()
1019
SmartServerResponse(('yes',)), response)
1022
class TestHandlers(tests.TestCase):
1023
"""Tests for the request.request_handlers object."""
1025
def test_registered_methods(self):
1026
"""Test that known methods are registered to the correct object."""
1028
smart.request.request_handlers.get('Branch.get_config_file'),
1029
smart.branch.SmartServerBranchGetConfigFile)
1031
smart.request.request_handlers.get('Branch.lock_write'),
1032
smart.branch.SmartServerBranchRequestLockWrite)
1034
smart.request.request_handlers.get('Branch.last_revision_info'),
1035
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1037
smart.request.request_handlers.get('Branch.revision_history'),
1038
smart.branch.SmartServerRequestRevisionHistory)
1040
smart.request.request_handlers.get('Branch.set_last_revision'),
1041
smart.branch.SmartServerBranchRequestSetLastRevision)
1043
smart.request.request_handlers.get('Branch.set_last_revision_info'),
1044
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1046
smart.request.request_handlers.get('Branch.unlock'),
1047
smart.branch.SmartServerBranchRequestUnlock)
1049
smart.request.request_handlers.get('BzrDir.find_repository'),
1050
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1052
smart.request.request_handlers.get('BzrDir.find_repositoryV2'),
1053
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1055
smart.request.request_handlers.get('BzrDirFormat.initialize'),
1056
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1058
smart.request.request_handlers.get('BzrDir.open_branch'),
1059
smart.bzrdir.SmartServerRequestOpenBranch)
1061
smart.request.request_handlers.get('Repository.gather_stats'),
1062
smart.repository.SmartServerRepositoryGatherStats)
1064
smart.request.request_handlers.get('Repository.get_parent_map'),
1065
smart.repository.SmartServerRepositoryGetParentMap)
1067
smart.request.request_handlers.get(
1068
'Repository.get_revision_graph'),
1069
smart.repository.SmartServerRepositoryGetRevisionGraph)
1071
smart.request.request_handlers.get('Repository.has_revision'),
1072
smart.repository.SmartServerRequestHasRevision)
1074
smart.request.request_handlers.get('Repository.is_shared'),
1075
smart.repository.SmartServerRepositoryIsShared)
1077
smart.request.request_handlers.get('Repository.lock_write'),
1078
smart.repository.SmartServerRepositoryLockWrite)
1080
smart.request.request_handlers.get('Repository.tarball'),
1081
smart.repository.SmartServerRepositoryTarball)
1083
smart.request.request_handlers.get('Repository.unlock'),
1084
smart.repository.SmartServerRepositoryUnlock)
1086
smart.request.request_handlers.get('Transport.is_readonly'),
1087
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
