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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 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
40
from bzrlib.branch import Branch, BranchReferenceFormat
41
import bzrlib.smart.branch
42
import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir
43
import bzrlib.smart.packrepository
44
import bzrlib.smart.repository
45
from bzrlib.smart.request import (
46
FailedSmartServerResponse,
49
SuccessfulSmartServerResponse,
51
from bzrlib.tests import (
54
from bzrlib.transport import chroot, get_transport
57
def load_tests(standard_tests, module, loader):
58
"""Multiply tests version and protocol consistency."""
59
# FindRepository tests.
60
bzrdir_mod = bzrlib.smart.bzrdir
63
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),
64
("find_repositoryV2", {
65
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),
66
("find_repositoryV3", {
67
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV3}),
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
tests.multiply_tests(v1_and_2, scenarios, result)
74
# The first scenario is only applicable to v1 protocols, it is deleted
76
tests.multiply_tests(v2_only, scenarios[1:], result)
80
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
83
tests.TestCaseWithTransport.setUp(self)
84
self._chroot_server = None
86
def get_transport(self, relpath=None):
87
if self._chroot_server is None:
88
backing_transport = tests.TestCaseWithTransport.get_transport(self)
89
self._chroot_server = chroot.ChrootServer(backing_transport)
90
self._chroot_server.setUp()
91
self.addCleanup(self._chroot_server.tearDown)
92
t = get_transport(self._chroot_server.get_url())
93
if relpath is not None:
98
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
101
super(TestCaseWithSmartMedium, self).setUp()
102
# We're allowed to set the transport class here, so that we don't use
103
# the default or a parameterized class, but rather use the
104
# TestCaseWithTransport infrastructure to set up a smart server and
106
self.transport_server = self.make_transport_server
108
def make_transport_server(self):
109
return smart.server.SmartTCPServer_for_testing('-' + self.id())
111
def get_smart_medium(self):
112
"""Get a smart medium to use in tests."""
113
return self.get_transport().get_smart_medium()
116
class TestSmartServerResponse(tests.TestCase):
118
def test__eq__(self):
119
self.assertEqual(SmartServerResponse(('ok', )),
120
SmartServerResponse(('ok', )))
121
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
122
SmartServerResponse(('ok', ), 'body'))
123
self.assertNotEqual(SmartServerResponse(('ok', )),
124
SmartServerResponse(('notok', )))
125
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
126
SmartServerResponse(('ok', )))
127
self.assertNotEqual(None,
128
SmartServerResponse(('ok', )))
130
def test__str__(self):
131
"""SmartServerResponses can be stringified."""
133
"<SuccessfulSmartServerResponse args=('args',) body='body'>",
134
str(SuccessfulSmartServerResponse(('args',), 'body')))
136
"<FailedSmartServerResponse args=('args',) body='body'>",
137
str(FailedSmartServerResponse(('args',), 'body')))
140
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
142
def test_translate_client_path(self):
143
transport = self.get_transport()
144
request = SmartServerRequest(transport, 'foo/')
145
self.assertEqual('./', request.translate_client_path('foo/'))
147
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
149
errors.PathNotChild, request.translate_client_path, '/')
151
errors.PathNotChild, request.translate_client_path, 'bar/')
152
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
154
def test_transport_from_client_path(self):
155
transport = self.get_transport()
156
request = SmartServerRequest(transport, 'foo/')
159
request.transport_from_client_path('foo/').base)
162
class TestSmartServerBzrDirRequestCloningMetaDir(
163
tests.TestCaseWithMemoryTransport):
164
"""Tests for BzrDir.cloning_metadir."""
166
def test_cloning_metadir(self):
167
"""When there is a bzrdir present, the call succeeds."""
168
backing = self.get_transport()
169
dir = self.make_bzrdir('.')
170
local_result = dir.cloning_metadir()
171
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
172
request = request_class(backing)
173
expected = SuccessfulSmartServerResponse(
174
(local_result.network_name(),
175
local_result.repository_format.network_name(),
176
('branch', local_result.get_branch_format().network_name())))
177
self.assertEqual(expected, request.execute('', 'False'))
179
def test_cloning_metadir_reference(self):
180
"""The request fails when bzrdir contains a branch reference."""
181
backing = self.get_transport()
182
referenced_branch = self.make_branch('referenced')
183
dir = self.make_bzrdir('.')
184
local_result = dir.cloning_metadir()
185
reference = BranchReferenceFormat().initialize(dir, referenced_branch)
186
reference_url = BranchReferenceFormat().get_reference(dir)
187
# The server shouldn't try to follow the branch reference, so it's fine
188
# if the referenced branch isn't reachable.
189
backing.rename('referenced', 'moved')
190
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
191
request = request_class(backing)
192
expected = FailedSmartServerResponse(('BranchReference',))
193
self.assertEqual(expected, request.execute('', 'False'))
196
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
197
"""Tests for BzrDir.create_repository."""
199
def test_makes_repository(self):
200
"""When there is a bzrdir present, the call succeeds."""
201
backing = self.get_transport()
202
self.make_bzrdir('.')
203
request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
204
request = request_class(backing)
205
reference_bzrdir_format = bzrdir.format_registry.get('pack-0.92')()
206
reference_format = reference_bzrdir_format.repository_format
207
network_name = reference_format.network_name()
208
expected = SuccessfulSmartServerResponse(
209
('ok', 'no', 'no', 'no', network_name))
210
self.assertEqual(expected, request.execute('', network_name, 'True'))
213
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
214
"""Tests for BzrDir.find_repository."""
216
def test_no_repository(self):
217
"""When there is no repository to be found, ('norepository', ) is returned."""
218
backing = self.get_transport()
219
request = self._request_class(backing)
220
self.make_bzrdir('.')
221
self.assertEqual(SmartServerResponse(('norepository', )),
224
def test_nonshared_repository(self):
225
# nonshared repositorys only allow 'find' to return a handle when the
226
# path the repository is being searched on is the same as that that
227
# the repository is at.
228
backing = self.get_transport()
229
request = self._request_class(backing)
230
result = self._make_repository_and_result()
231
self.assertEqual(result, request.execute(''))
232
self.make_bzrdir('subdir')
233
self.assertEqual(SmartServerResponse(('norepository', )),
234
request.execute('subdir'))
236
def _make_repository_and_result(self, shared=False, format=None):
237
"""Convenience function to setup a repository.
239
:result: The SmartServerResponse to expect when opening it.
241
repo = self.make_repository('.', shared=shared, format=format)
242
if repo.supports_rich_root():
246
if repo._format.supports_tree_reference:
250
if repo._format.supports_external_lookups:
254
if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
255
self._request_class):
256
return SuccessfulSmartServerResponse(
257
('ok', '', rich_root, subtrees, external,
258
repo._format.network_name()))
259
elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
260
self._request_class):
261
# All tests so far are on formats, and for non-external
263
return SuccessfulSmartServerResponse(
264
('ok', '', rich_root, subtrees, external))
266
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
268
def test_shared_repository(self):
269
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
270
backing = self.get_transport()
271
request = self._request_class(backing)
272
result = self._make_repository_and_result(shared=True)
273
self.assertEqual(result, request.execute(''))
274
self.make_bzrdir('subdir')
275
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
276
self.assertEqual(result2,
277
request.execute('subdir'))
278
self.make_bzrdir('subdir/deeper')
279
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
280
self.assertEqual(result3,
281
request.execute('subdir/deeper'))
283
def test_rich_root_and_subtree_encoding(self):
284
"""Test for the format attributes for rich root and subtree support."""
285
backing = self.get_transport()
286
request = self._request_class(backing)
287
result = self._make_repository_and_result(format='dirstate-with-subtree')
288
# check the test will be valid
289
self.assertEqual('yes', result.args[2])
290
self.assertEqual('yes', result.args[3])
291
self.assertEqual(result, request.execute(''))
293
def test_supports_external_lookups_no_v2(self):
294
"""Test for the supports_external_lookups attribute."""
295
backing = self.get_transport()
296
request = self._request_class(backing)
297
result = self._make_repository_and_result(format='dirstate-with-subtree')
298
# check the test will be valid
299
self.assertEqual('no', result.args[4])
300
self.assertEqual(result, request.execute(''))
303
class TestSmartServerBzrDirRequestGetConfigFile(
304
tests.TestCaseWithMemoryTransport):
305
"""Tests for BzrDir.get_config_file."""
307
def test_present(self):
308
backing = self.get_transport()
309
dir = self.make_bzrdir('.')
310
dir.get_config().set_default_stack_on("/")
311
local_result = dir._get_config()._get_config_file().read()
312
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
313
request = request_class(backing)
314
expected = SuccessfulSmartServerResponse((), local_result)
315
self.assertEqual(expected, request.execute(''))
317
def test_missing(self):
318
backing = self.get_transport()
319
dir = self.make_bzrdir('.')
320
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
321
request = request_class(backing)
322
expected = SuccessfulSmartServerResponse((), '')
323
self.assertEqual(expected, request.execute(''))
326
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
328
def test_empty_dir(self):
329
"""Initializing an empty dir should succeed and do it."""
330
backing = self.get_transport()
331
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
332
self.assertEqual(SmartServerResponse(('ok', )),
334
made_dir = bzrdir.BzrDir.open_from_transport(backing)
335
# no branch, tree or repository is expected with the current
337
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
338
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
339
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
341
def test_missing_dir(self):
342
"""Initializing a missing directory should fail like the bzrdir api."""
343
backing = self.get_transport()
344
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
345
self.assertRaises(errors.NoSuchFile,
346
request.execute, 'subdir')
348
def test_initialized_dir(self):
349
"""Initializing an extant bzrdir should fail like the bzrdir api."""
350
backing = self.get_transport()
351
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
352
self.make_bzrdir('subdir')
353
self.assertRaises(errors.FileExists,
354
request.execute, 'subdir')
357
class TestSmartServerRequestBzrDirInitializeEx(tests.TestCaseWithMemoryTransport):
358
"""Basic tests for BzrDir.initialize_ex_1.16 in the smart server.
360
The main unit tests in test_bzrdir exercise the API comprehensively.
363
def test_empty_dir(self):
364
"""Initializing an empty dir should succeed and do it."""
365
backing = self.get_transport()
366
name = self.make_bzrdir('reference')._format.network_name()
367
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
368
self.assertEqual(SmartServerResponse(('', '', '', '', '', '', name,
369
'False', '', '', '')),
370
request.execute(name, '', 'True', 'False', 'False', '', '', '', '',
372
made_dir = bzrdir.BzrDir.open_from_transport(backing)
373
# no branch, tree or repository is expected with the current
375
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
376
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
377
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
379
def test_missing_dir(self):
380
"""Initializing a missing directory should fail like the bzrdir api."""
381
backing = self.get_transport()
382
name = self.make_bzrdir('reference')._format.network_name()
383
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
384
self.assertRaises(errors.NoSuchFile, request.execute, name,
385
'subdir/dir', 'False', 'False', 'False', '', '', '', '', 'False')
387
def test_initialized_dir(self):
388
"""Initializing an extant directory should fail like the bzrdir api."""
389
backing = self.get_transport()
390
name = self.make_bzrdir('reference')._format.network_name()
391
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
392
self.make_bzrdir('subdir')
393
self.assertRaises(errors.FileExists, request.execute, name, 'subdir',
394
'False', 'False', 'False', '', '', '', '', 'False')
397
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
399
def test_no_branch(self):
400
"""When there is no branch, ('nobranch', ) is returned."""
401
backing = self.get_transport()
402
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
403
self.make_bzrdir('.')
404
self.assertEqual(SmartServerResponse(('nobranch', )),
407
def test_branch(self):
408
"""When there is a branch, 'ok' is returned."""
409
backing = self.get_transport()
410
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
411
self.make_branch('.')
412
self.assertEqual(SmartServerResponse(('ok', '')),
415
def test_branch_reference(self):
416
"""When there is a branch reference, the reference URL is returned."""
417
backing = self.get_transport()
418
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
419
branch = self.make_branch('branch')
420
checkout = branch.create_checkout('reference',lightweight=True)
421
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
422
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
423
self.assertEqual(SmartServerResponse(('ok', reference_url)),
424
request.execute('reference'))
427
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
429
def test_no_branch(self):
430
"""When there is no branch, ('nobranch', ) is returned."""
431
backing = self.get_transport()
432
self.make_bzrdir('.')
433
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
434
self.assertEqual(SmartServerResponse(('nobranch', )),
437
def test_branch(self):
438
"""When there is a branch, 'ok' is returned."""
439
backing = self.get_transport()
440
expected = self.make_branch('.')._format.network_name()
441
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
442
self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),
445
def test_branch_reference(self):
446
"""When there is a branch reference, the reference URL is returned."""
447
backing = self.get_transport()
448
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
449
branch = self.make_branch('branch')
450
checkout = branch.create_checkout('reference',lightweight=True)
451
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
452
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
453
self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
454
request.execute('reference'))
456
def test_stacked_branch(self):
457
"""Opening a stacked branch does not open the stacked-on branch."""
458
trunk = self.make_branch('trunk')
459
feature = self.make_branch('feature')
460
feature.set_stacked_on_url(trunk.base)
462
Branch.hooks.install_named_hook('open', opened_branches.append, None)
463
backing = self.get_transport()
464
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
467
response = request.execute('feature')
469
request.teardown_jail()
470
expected_format = feature._format.network_name()
472
SuccessfulSmartServerResponse(('branch', expected_format)),
474
self.assertLength(1, opened_branches)
477
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
479
def test_empty(self):
480
"""For an empty branch, the body is empty."""
481
backing = self.get_transport()
482
request = smart.branch.SmartServerRequestRevisionHistory(backing)
483
self.make_branch('.')
484
self.assertEqual(SmartServerResponse(('ok', ), ''),
487
def test_not_empty(self):
488
"""For a non-empty branch, the body is empty."""
489
backing = self.get_transport()
490
request = smart.branch.SmartServerRequestRevisionHistory(backing)
491
tree = self.make_branch_and_memory_tree('.')
494
r1 = tree.commit('1st commit')
495
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
498
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
502
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
504
def test_no_branch(self):
505
"""When there is a bzrdir and no branch, NotBranchError is raised."""
506
backing = self.get_transport()
507
request = smart.branch.SmartServerBranchRequest(backing)
508
self.make_bzrdir('.')
509
self.assertRaises(errors.NotBranchError,
512
def test_branch_reference(self):
513
"""When there is a branch reference, NotBranchError is raised."""
514
backing = self.get_transport()
515
request = smart.branch.SmartServerBranchRequest(backing)
516
branch = self.make_branch('branch')
517
checkout = branch.create_checkout('reference',lightweight=True)
518
self.assertRaises(errors.NotBranchError,
519
request.execute, 'checkout')
522
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
524
def test_empty(self):
525
"""For an empty branch, the result is ('ok', '0', 'null:')."""
526
backing = self.get_transport()
527
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
528
self.make_branch('.')
529
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
532
def test_not_empty(self):
533
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
534
backing = self.get_transport()
535
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
536
tree = self.make_branch_and_memory_tree('.')
539
rev_id_utf8 = u'\xc8'.encode('utf-8')
540
r1 = tree.commit('1st commit')
541
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
544
SmartServerResponse(('ok', '2', rev_id_utf8)),
548
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
550
def test_default(self):
551
"""With no file, we get empty content."""
552
backing = self.get_transport()
553
request = smart.branch.SmartServerBranchGetConfigFile(backing)
554
branch = self.make_branch('.')
555
# there should be no file by default
557
self.assertEqual(SmartServerResponse(('ok', ), content),
560
def test_with_content(self):
561
# SmartServerBranchGetConfigFile should return the content from
562
# branch.control_files.get('branch.conf') for now - in the future it may
563
# perform more complex processing.
564
backing = self.get_transport()
565
request = smart.branch.SmartServerBranchGetConfigFile(backing)
566
branch = self.make_branch('.')
567
branch._transport.put_bytes('branch.conf', 'foo bar baz')
568
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
572
class TestLockedBranch(tests.TestCaseWithMemoryTransport):
574
def get_lock_tokens(self, branch):
575
branch_token = branch.lock_write()
576
repo_token = branch.repository.lock_write()
577
branch.repository.unlock()
578
return branch_token, repo_token
581
class TestSmartServerBranchRequestSetConfigOption(TestLockedBranch):
583
def test_value_name(self):
584
branch = self.make_branch('.')
585
request = smart.branch.SmartServerBranchRequestSetConfigOption(
586
branch.bzrdir.root_transport)
587
branch_token, repo_token = self.get_lock_tokens(branch)
588
config = branch._get_config()
589
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
591
self.assertEqual(SuccessfulSmartServerResponse(()), result)
592
self.assertEqual('bar', config.get_option('foo'))
596
def test_value_name_section(self):
597
branch = self.make_branch('.')
598
request = smart.branch.SmartServerBranchRequestSetConfigOption(
599
branch.bzrdir.root_transport)
600
branch_token, repo_token = self.get_lock_tokens(branch)
601
config = branch._get_config()
602
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
604
self.assertEqual(SuccessfulSmartServerResponse(()), result)
605
self.assertEqual('bar', config.get_option('foo', 'gam'))
610
class TestSmartServerBranchRequestSetTagsBytes(TestLockedBranch):
611
# Only called when the branch format and tags match [yay factory
612
# methods] so only need to test straight forward cases.
614
def test_set_bytes(self):
615
base_branch = self.make_branch('base')
616
tag_bytes = base_branch._get_tags_bytes()
617
# get_lock_tokens takes out a lock.
618
branch_token, repo_token = self.get_lock_tokens(base_branch)
619
request = smart.branch.SmartServerBranchSetTagsBytes(
620
self.get_transport())
621
response = request.execute('base', branch_token, repo_token)
622
self.assertEqual(None, response)
623
response = request.do_chunk(tag_bytes)
624
self.assertEqual(None, response)
625
response = request.do_end()
627
SuccessfulSmartServerResponse(()), response)
630
def test_lock_failed(self):
631
base_branch = self.make_branch('base')
632
base_branch.lock_write()
633
tag_bytes = base_branch._get_tags_bytes()
634
request = smart.branch.SmartServerBranchSetTagsBytes(
635
self.get_transport())
636
self.assertRaises(errors.TokenMismatch, request.execute,
637
'base', 'wrong token', 'wrong token')
638
# The request handler will keep processing the message parts, so even
639
# if the request fails immediately do_chunk and do_end are still
641
request.do_chunk(tag_bytes)
647
class SetLastRevisionTestBase(TestLockedBranch):
648
"""Base test case for verbs that implement set_last_revision."""
651
tests.TestCaseWithMemoryTransport.setUp(self)
652
backing_transport = self.get_transport()
653
self.request = self.request_class(backing_transport)
654
self.tree = self.make_branch_and_memory_tree('.')
656
def lock_branch(self):
657
return self.get_lock_tokens(self.tree.branch)
659
def unlock_branch(self):
660
self.tree.branch.unlock()
662
def set_last_revision(self, revision_id, revno):
663
branch_token, repo_token = self.lock_branch()
664
response = self._set_last_revision(
665
revision_id, revno, branch_token, repo_token)
669
def assertRequestSucceeds(self, revision_id, revno):
670
response = self.set_last_revision(revision_id, revno)
671
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
674
class TestSetLastRevisionVerbMixin(object):
675
"""Mixin test case for verbs that implement set_last_revision."""
677
def test_set_null_to_null(self):
678
"""An empty branch can have its last revision set to 'null:'."""
679
self.assertRequestSucceeds('null:', 0)
681
def test_NoSuchRevision(self):
682
"""If the revision_id is not present, the verb returns NoSuchRevision.
684
revision_id = 'non-existent revision'
686
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
687
self.set_last_revision(revision_id, 1))
689
def make_tree_with_two_commits(self):
690
self.tree.lock_write()
692
rev_id_utf8 = u'\xc8'.encode('utf-8')
693
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
694
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
697
def test_branch_last_revision_info_is_updated(self):
698
"""A branch's tip can be set to a revision that is present in its
701
# Make a branch with an empty revision history, but two revisions in
703
self.make_tree_with_two_commits()
704
rev_id_utf8 = u'\xc8'.encode('utf-8')
705
self.tree.branch.set_revision_history([])
707
(0, 'null:'), self.tree.branch.last_revision_info())
708
# We can update the branch to a revision that is present in the
710
self.assertRequestSucceeds(rev_id_utf8, 1)
712
(1, rev_id_utf8), self.tree.branch.last_revision_info())
714
def test_branch_last_revision_info_rewind(self):
715
"""A branch's tip can be set to a revision that is an ancestor of the
718
self.make_tree_with_two_commits()
719
rev_id_utf8 = u'\xc8'.encode('utf-8')
721
(2, 'rev-2'), self.tree.branch.last_revision_info())
722
self.assertRequestSucceeds(rev_id_utf8, 1)
724
(1, rev_id_utf8), self.tree.branch.last_revision_info())
726
def test_TipChangeRejected(self):
727
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
728
returns TipChangeRejected.
730
rejection_message = u'rejection message\N{INTERROBANG}'
731
def hook_that_rejects(params):
732
raise errors.TipChangeRejected(rejection_message)
733
Branch.hooks.install_named_hook(
734
'pre_change_branch_tip', hook_that_rejects, None)
736
FailedSmartServerResponse(
737
('TipChangeRejected', rejection_message.encode('utf-8'))),
738
self.set_last_revision('null:', 0))
741
class TestSmartServerBranchRequestSetLastRevision(
742
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
743
"""Tests for Branch.set_last_revision verb."""
745
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
747
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
748
return self.request.execute(
749
'', branch_token, repo_token, revision_id)
752
class TestSmartServerBranchRequestSetLastRevisionInfo(
753
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
754
"""Tests for Branch.set_last_revision_info verb."""
756
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
758
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
759
return self.request.execute(
760
'', branch_token, repo_token, revno, revision_id)
762
def test_NoSuchRevision(self):
763
"""Branch.set_last_revision_info does not have to return
764
NoSuchRevision if the revision_id is absent.
766
raise tests.TestNotApplicable()
769
class TestSmartServerBranchRequestSetLastRevisionEx(
770
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
771
"""Tests for Branch.set_last_revision_ex verb."""
773
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
775
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
776
return self.request.execute(
777
'', branch_token, repo_token, revision_id, 0, 0)
779
def assertRequestSucceeds(self, revision_id, revno):
780
response = self.set_last_revision(revision_id, revno)
782
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
785
def test_branch_last_revision_info_rewind(self):
786
"""A branch's tip can be set to a revision that is an ancestor of the
787
current tip, but only if allow_overwrite_descendant is passed.
789
self.make_tree_with_two_commits()
790
rev_id_utf8 = u'\xc8'.encode('utf-8')
792
(2, 'rev-2'), self.tree.branch.last_revision_info())
793
# If allow_overwrite_descendant flag is 0, then trying to set the tip
794
# to an older revision ID has no effect.
795
branch_token, repo_token = self.lock_branch()
796
response = self.request.execute(
797
'', branch_token, repo_token, rev_id_utf8, 0, 0)
799
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
802
(2, 'rev-2'), self.tree.branch.last_revision_info())
804
# If allow_overwrite_descendant flag is 1, then setting the tip to an
806
response = self.request.execute(
807
'', branch_token, repo_token, rev_id_utf8, 0, 1)
809
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
813
(1, rev_id_utf8), self.tree.branch.last_revision_info())
815
def make_branch_with_divergent_history(self):
816
"""Make a branch with divergent history in its repo.
818
The branch's tip will be 'child-2', and the repo will also contain
819
'child-1', which diverges from a common base revision.
821
self.tree.lock_write()
823
r1 = self.tree.commit('1st commit')
824
revno_1, revid_1 = self.tree.branch.last_revision_info()
825
r2 = self.tree.commit('2nd commit', rev_id='child-1')
826
# Undo the second commit
827
self.tree.branch.set_last_revision_info(revno_1, revid_1)
828
self.tree.set_parent_ids([revid_1])
829
# Make a new second commit, child-2. child-2 has diverged from
831
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
834
def test_not_allow_diverged(self):
835
"""If allow_diverged is not passed, then setting a divergent history
836
returns a Diverged error.
838
self.make_branch_with_divergent_history()
840
FailedSmartServerResponse(('Diverged',)),
841
self.set_last_revision('child-1', 2))
842
# The branch tip was not changed.
843
self.assertEqual('child-2', self.tree.branch.last_revision())
845
def test_allow_diverged(self):
846
"""If allow_diverged is passed, then setting a divergent history
849
self.make_branch_with_divergent_history()
850
branch_token, repo_token = self.lock_branch()
851
response = self.request.execute(
852
'', branch_token, repo_token, 'child-1', 1, 0)
854
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
857
# The branch tip was changed.
858
self.assertEqual('child-1', self.tree.branch.last_revision())
861
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
863
def test_get_parent_none(self):
864
base_branch = self.make_branch('base')
865
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
866
response = request.execute('base')
868
SuccessfulSmartServerResponse(('',)), response)
870
def test_get_parent_something(self):
871
base_branch = self.make_branch('base')
872
base_branch.set_parent(self.get_url('foo'))
873
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
874
response = request.execute('base')
876
SuccessfulSmartServerResponse(("../foo",)),
880
class TestSmartServerBranchRequestSetParent(tests.TestCaseWithMemoryTransport):
882
def test_set_parent_none(self):
883
branch = self.make_branch('base', format="1.9")
885
branch._set_parent_location('foo')
887
request = smart.branch.SmartServerBranchRequestSetParentLocation(
888
self.get_transport())
889
branch_token = branch.lock_write()
890
repo_token = branch.repository.lock_write()
892
response = request.execute('base', branch_token, repo_token, '')
894
branch.repository.unlock()
896
self.assertEqual(SuccessfulSmartServerResponse(()), response)
897
self.assertEqual(None, branch.get_parent())
899
def test_set_parent_something(self):
900
branch = self.make_branch('base', format="1.9")
901
request = smart.branch.SmartServerBranchRequestSetParentLocation(
902
self.get_transport())
903
branch_token = branch.lock_write()
904
repo_token = branch.repository.lock_write()
906
response = request.execute('base', branch_token, repo_token,
909
branch.repository.unlock()
911
self.assertEqual(SuccessfulSmartServerResponse(()), response)
912
self.assertEqual('http://bar/', branch.get_parent())
915
class TestSmartServerBranchRequestGetTagsBytes(tests.TestCaseWithMemoryTransport):
916
# Only called when the branch format and tags match [yay factory
917
# methods] so only need to test straight forward cases.
919
def test_get_bytes(self):
920
base_branch = self.make_branch('base')
921
request = smart.branch.SmartServerBranchGetTagsBytes(
922
self.get_transport())
923
response = request.execute('base')
925
SuccessfulSmartServerResponse(('',)), response)
928
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
930
def test_get_stacked_on_url(self):
931
base_branch = self.make_branch('base', format='1.6')
932
stacked_branch = self.make_branch('stacked', format='1.6')
933
# typically should be relative
934
stacked_branch.set_stacked_on_url('../base')
935
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
936
self.get_transport())
937
response = request.execute('stacked')
939
SmartServerResponse(('ok', '../base')),
943
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
946
tests.TestCaseWithMemoryTransport.setUp(self)
948
def test_lock_write_on_unlocked_branch(self):
949
backing = self.get_transport()
950
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
951
branch = self.make_branch('.', format='knit')
952
repository = branch.repository
953
response = request.execute('')
954
branch_nonce = branch.control_files._lock.peek().get('nonce')
955
repository_nonce = repository.control_files._lock.peek().get('nonce')
957
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
959
# The branch (and associated repository) is now locked. Verify that
960
# with a new branch object.
961
new_branch = repository.bzrdir.open_branch()
962
self.assertRaises(errors.LockContention, new_branch.lock_write)
964
request = smart.branch.SmartServerBranchRequestUnlock(backing)
965
response = request.execute('', branch_nonce, repository_nonce)
967
def test_lock_write_on_locked_branch(self):
968
backing = self.get_transport()
969
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
970
branch = self.make_branch('.')
971
branch_token = branch.lock_write()
972
branch.leave_lock_in_place()
974
response = request.execute('')
976
SmartServerResponse(('LockContention',)), response)
978
branch.lock_write(branch_token)
979
branch.dont_leave_lock_in_place()
982
def test_lock_write_with_tokens_on_locked_branch(self):
983
backing = self.get_transport()
984
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
985
branch = self.make_branch('.', format='knit')
986
branch_token = branch.lock_write()
987
repo_token = branch.repository.lock_write()
988
branch.repository.unlock()
989
branch.leave_lock_in_place()
990
branch.repository.leave_lock_in_place()
992
response = request.execute('',
993
branch_token, repo_token)
995
SmartServerResponse(('ok', branch_token, repo_token)), response)
997
branch.repository.lock_write(repo_token)
998
branch.repository.dont_leave_lock_in_place()
999
branch.repository.unlock()
1000
branch.lock_write(branch_token)
1001
branch.dont_leave_lock_in_place()
1004
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
1005
backing = self.get_transport()
1006
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1007
branch = self.make_branch('.', format='knit')
1008
branch_token = branch.lock_write()
1009
repo_token = branch.repository.lock_write()
1010
branch.repository.unlock()
1011
branch.leave_lock_in_place()
1012
branch.repository.leave_lock_in_place()
1014
response = request.execute('',
1015
branch_token+'xxx', repo_token)
1017
SmartServerResponse(('TokenMismatch',)), response)
1019
branch.repository.lock_write(repo_token)
1020
branch.repository.dont_leave_lock_in_place()
1021
branch.repository.unlock()
1022
branch.lock_write(branch_token)
1023
branch.dont_leave_lock_in_place()
1026
def test_lock_write_on_locked_repo(self):
1027
backing = self.get_transport()
1028
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1029
branch = self.make_branch('.', format='knit')
1030
repo = branch.repository
1031
repo_token = repo.lock_write()
1032
repo.leave_lock_in_place()
1034
response = request.execute('')
1036
SmartServerResponse(('LockContention',)), response)
1038
repo.lock_write(repo_token)
1039
repo.dont_leave_lock_in_place()
1042
def test_lock_write_on_readonly_transport(self):
1043
backing = self.get_readonly_transport()
1044
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1045
branch = self.make_branch('.')
1046
root = self.get_transport().clone('/')
1047
path = urlutils.relative_url(root.base, self.get_transport().base)
1048
response = request.execute(path)
1049
error_name, lock_str, why_str = response.args
1050
self.assertFalse(response.is_successful())
1051
self.assertEqual('LockFailed', error_name)
1054
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
1057
tests.TestCaseWithMemoryTransport.setUp(self)
1059
def test_unlock_on_locked_branch_and_repo(self):
1060
backing = self.get_transport()
1061
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1062
branch = self.make_branch('.', format='knit')
1064
branch_token = branch.lock_write()
1065
repo_token = branch.repository.lock_write()
1066
branch.repository.unlock()
1067
# Unlock the branch (and repo) object, leaving the physical locks
1069
branch.leave_lock_in_place()
1070
branch.repository.leave_lock_in_place()
1072
response = request.execute('',
1073
branch_token, repo_token)
1075
SmartServerResponse(('ok',)), response)
1076
# The branch is now unlocked. Verify that with a new branch
1078
new_branch = branch.bzrdir.open_branch()
1079
new_branch.lock_write()
1082
def test_unlock_on_unlocked_branch_unlocked_repo(self):
1083
backing = self.get_transport()
1084
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1085
branch = self.make_branch('.', format='knit')
1086
response = request.execute(
1087
'', 'branch token', 'repo token')
1089
SmartServerResponse(('TokenMismatch',)), response)
1091
def test_unlock_on_unlocked_branch_locked_repo(self):
1092
backing = self.get_transport()
1093
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1094
branch = self.make_branch('.', format='knit')
1095
# Lock the repository.
1096
repo_token = branch.repository.lock_write()
1097
branch.repository.leave_lock_in_place()
1098
branch.repository.unlock()
1099
# Issue branch lock_write request on the unlocked branch (with locked
1101
response = request.execute(
1102
'', 'branch token', repo_token)
1104
SmartServerResponse(('TokenMismatch',)), response)
1106
branch.repository.lock_write(repo_token)
1107
branch.repository.dont_leave_lock_in_place()
1108
branch.repository.unlock()
1111
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
1113
def test_no_repository(self):
1114
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
1115
# we test this using a shared repository above the named path,
1116
# thus checking the right search logic is used - that is, that
1117
# its the exact path being looked at and the server is not
1119
backing = self.get_transport()
1120
request = smart.repository.SmartServerRepositoryRequest(backing)
1121
self.make_repository('.', shared=True)
1122
self.make_bzrdir('subdir')
1123
self.assertRaises(errors.NoRepositoryPresent,
1124
request.execute, 'subdir')
1127
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
1129
def test_trivial_bzipped(self):
1130
# This tests that the wire encoding is actually bzipped
1131
backing = self.get_transport()
1132
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1133
tree = self.make_branch_and_memory_tree('.')
1135
self.assertEqual(None,
1136
request.execute('', 'missing-id'))
1137
# Note that it returns a body that is bzipped.
1139
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
1140
request.do_body('\n\n0\n'))
1142
def test_trivial_include_missing(self):
1143
backing = self.get_transport()
1144
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1145
tree = self.make_branch_and_memory_tree('.')
1147
self.assertEqual(None,
1148
request.execute('', 'missing-id', 'include-missing:'))
1150
SuccessfulSmartServerResponse(('ok', ),
1151
bz2.compress('missing:missing-id')),
1152
request.do_body('\n\n0\n'))
1155
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
1157
def test_none_argument(self):
1158
backing = self.get_transport()
1159
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1160
tree = self.make_branch_and_memory_tree('.')
1163
r1 = tree.commit('1st commit')
1164
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1167
# the lines of revision_id->revision_parent_list has no guaranteed
1168
# order coming out of a dict, so sort both our test and response
1169
lines = sorted([' '.join([r2, r1]), r1])
1170
response = request.execute('', '')
1171
response.body = '\n'.join(sorted(response.body.split('\n')))
1174
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
1176
def test_specific_revision_argument(self):
1177
backing = self.get_transport()
1178
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1179
tree = self.make_branch_and_memory_tree('.')
1182
rev_id_utf8 = u'\xc9'.encode('utf-8')
1183
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
1184
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1187
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
1188
request.execute('', rev_id_utf8))
1190
def test_no_such_revision(self):
1191
backing = self.get_transport()
1192
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1193
tree = self.make_branch_and_memory_tree('.')
1196
r1 = tree.commit('1st commit')
1199
# Note that it still returns body (of zero bytes).
1201
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
1202
request.execute('', 'missingrevision'))
1205
class TestSmartServerRepositoryGetRevIdForRevno(tests.TestCaseWithMemoryTransport):
1207
def test_revno_found(self):
1208
backing = self.get_transport()
1209
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1210
tree = self.make_branch_and_memory_tree('.')
1213
rev1_id_utf8 = u'\xc8'.encode('utf-8')
1214
rev2_id_utf8 = u'\xc9'.encode('utf-8')
1215
tree.commit('1st commit', rev_id=rev1_id_utf8)
1216
tree.commit('2nd commit', rev_id=rev2_id_utf8)
1219
self.assertEqual(SmartServerResponse(('ok', rev1_id_utf8)),
1220
request.execute('', 1, (2, rev2_id_utf8)))
1222
def test_known_revid_missing(self):
1223
backing = self.get_transport()
1224
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1225
repo = self.make_repository('.')
1227
FailedSmartServerResponse(('nosuchrevision', 'ghost')),
1228
request.execute('', 1, (2, 'ghost')))
1230
def test_history_incomplete(self):
1231
backing = self.get_transport()
1232
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1233
parent = self.make_branch_and_memory_tree('parent', format='1.9')
1235
parent.add([''], ['TREE_ROOT'])
1236
r1 = parent.commit(message='first commit')
1237
r2 = parent.commit(message='second commit')
1239
local = self.make_branch_and_memory_tree('local', format='1.9')
1240
local.branch.pull(parent.branch)
1241
local.set_parent_ids([r2])
1242
r3 = local.commit(message='local commit')
1243
local.branch.create_clone_on_transport(
1244
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
1246
SmartServerResponse(('history-incomplete', 2, r2)),
1247
request.execute('stacked', 1, (3, r3)))
1250
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
1252
def make_two_commit_repo(self):
1253
tree = self.make_branch_and_memory_tree('.')
1256
r1 = tree.commit('1st commit')
1257
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1259
repo = tree.branch.repository
1262
def test_ancestry_of(self):
1263
"""The search argument may be a 'ancestry-of' some heads'."""
1264
backing = self.get_transport()
1265
request = smart.repository.SmartServerRepositoryGetStream(backing)
1266
repo, r1, r2 = self.make_two_commit_repo()
1267
fetch_spec = ['ancestry-of', r2]
1268
lines = '\n'.join(fetch_spec)
1269
request.execute('', repo._format.network_name())
1270
response = request.do_body(lines)
1271
self.assertEqual(('ok',), response.args)
1272
stream_bytes = ''.join(response.body_stream)
1273
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1275
def test_search(self):
1276
"""The search argument may be a 'search' of some explicit keys."""
1277
backing = self.get_transport()
1278
request = smart.repository.SmartServerRepositoryGetStream(backing)
1279
repo, r1, r2 = self.make_two_commit_repo()
1280
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1281
lines = '\n'.join(fetch_spec)
1282
request.execute('', repo._format.network_name())
1283
response = request.do_body(lines)
1284
self.assertEqual(('ok',), response.args)
1285
stream_bytes = ''.join(response.body_stream)
1286
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1289
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1291
def test_missing_revision(self):
1292
"""For a missing revision, ('no', ) is returned."""
1293
backing = self.get_transport()
1294
request = smart.repository.SmartServerRequestHasRevision(backing)
1295
self.make_repository('.')
1296
self.assertEqual(SmartServerResponse(('no', )),
1297
request.execute('', 'revid'))
1299
def test_present_revision(self):
1300
"""For a present revision, ('yes', ) is returned."""
1301
backing = self.get_transport()
1302
request = smart.repository.SmartServerRequestHasRevision(backing)
1303
tree = self.make_branch_and_memory_tree('.')
1306
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1307
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1309
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1310
self.assertEqual(SmartServerResponse(('yes', )),
1311
request.execute('', rev_id_utf8))
1314
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1316
def test_empty_revid(self):
1317
"""With an empty revid, we get only size an number and revisions"""
1318
backing = self.get_transport()
1319
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1320
repository = self.make_repository('.')
1321
stats = repository.gather_stats()
1322
expected_body = 'revisions: 0\n'
1323
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1324
request.execute('', '', 'no'))
1326
def test_revid_with_committers(self):
1327
"""For a revid we get more infos."""
1328
backing = self.get_transport()
1329
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1330
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1331
tree = self.make_branch_and_memory_tree('.')
1334
# Let's build a predictable result
1335
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1336
tree.commit('a commit', timestamp=654321.4, timezone=0,
1340
stats = tree.branch.repository.gather_stats()
1341
expected_body = ('firstrev: 123456.200 3600\n'
1342
'latestrev: 654321.400 0\n'
1344
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1348
def test_not_empty_repository_with_committers(self):
1349
"""For a revid and requesting committers we get the whole thing."""
1350
backing = self.get_transport()
1351
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1352
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1353
tree = self.make_branch_and_memory_tree('.')
1356
# Let's build a predictable result
1357
tree.commit('a commit', timestamp=123456.2, timezone=3600,
1359
tree.commit('a commit', timestamp=654321.4, timezone=0,
1360
committer='bar', rev_id=rev_id_utf8)
1362
stats = tree.branch.repository.gather_stats()
1364
expected_body = ('committers: 2\n'
1365
'firstrev: 123456.200 3600\n'
1366
'latestrev: 654321.400 0\n'
1368
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1370
rev_id_utf8, 'yes'))
1373
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
1375
def test_is_shared(self):
1376
"""For a shared repository, ('yes', ) is returned."""
1377
backing = self.get_transport()
1378
request = smart.repository.SmartServerRepositoryIsShared(backing)
1379
self.make_repository('.', shared=True)
1380
self.assertEqual(SmartServerResponse(('yes', )),
1381
request.execute('', ))
1383
def test_is_not_shared(self):
1384
"""For a shared repository, ('no', ) is returned."""
1385
backing = self.get_transport()
1386
request = smart.repository.SmartServerRepositoryIsShared(backing)
1387
self.make_repository('.', shared=False)
1388
self.assertEqual(SmartServerResponse(('no', )),
1389
request.execute('', ))
1392
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
1394
def test_lock_write_on_unlocked_repo(self):
1395
backing = self.get_transport()
1396
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1397
repository = self.make_repository('.', format='knit')
1398
response = request.execute('')
1399
nonce = repository.control_files._lock.peek().get('nonce')
1400
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
1401
# The repository is now locked. Verify that with a new repository
1403
new_repo = repository.bzrdir.open_repository()
1404
self.assertRaises(errors.LockContention, new_repo.lock_write)
1406
request = smart.repository.SmartServerRepositoryUnlock(backing)
1407
response = request.execute('', nonce)
1409
def test_lock_write_on_locked_repo(self):
1410
backing = self.get_transport()
1411
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1412
repository = self.make_repository('.', format='knit')
1413
repo_token = repository.lock_write()
1414
repository.leave_lock_in_place()
1416
response = request.execute('')
1418
SmartServerResponse(('LockContention',)), response)
1420
repository.lock_write(repo_token)
1421
repository.dont_leave_lock_in_place()
1424
def test_lock_write_on_readonly_transport(self):
1425
backing = self.get_readonly_transport()
1426
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1427
repository = self.make_repository('.', format='knit')
1428
response = request.execute('')
1429
self.assertFalse(response.is_successful())
1430
self.assertEqual('LockFailed', response.args[0])
1433
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
1435
def make_empty_byte_stream(self, repo):
1436
byte_stream = smart.repository._stream_to_byte_stream([], repo._format)
1437
return ''.join(byte_stream)
1440
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
1442
def test_insert_stream_empty(self):
1443
backing = self.get_transport()
1444
request = smart.repository.SmartServerRepositoryInsertStream(backing)
1445
repository = self.make_repository('.')
1446
response = request.execute('', '')
1447
self.assertEqual(None, response)
1448
response = request.do_chunk(self.make_empty_byte_stream(repository))
1449
self.assertEqual(None, response)
1450
response = request.do_end()
1451
self.assertEqual(SmartServerResponse(('ok', )), response)
1454
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
1456
def test_insert_stream_empty(self):
1457
backing = self.get_transport()
1458
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1460
repository = self.make_repository('.', format='knit')
1461
lock_token = repository.lock_write()
1462
response = request.execute('', '', lock_token)
1463
self.assertEqual(None, response)
1464
response = request.do_chunk(self.make_empty_byte_stream(repository))
1465
self.assertEqual(None, response)
1466
response = request.do_end()
1467
self.assertEqual(SmartServerResponse(('ok', )), response)
1470
def test_insert_stream_with_wrong_lock_token(self):
1471
backing = self.get_transport()
1472
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1474
repository = self.make_repository('.', format='knit')
1475
lock_token = repository.lock_write()
1477
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
1481
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1484
tests.TestCaseWithMemoryTransport.setUp(self)
1486
def test_unlock_on_locked_repo(self):
1487
backing = self.get_transport()
1488
request = smart.repository.SmartServerRepositoryUnlock(backing)
1489
repository = self.make_repository('.', format='knit')
1490
token = repository.lock_write()
1491
repository.leave_lock_in_place()
1493
response = request.execute('', token)
1495
SmartServerResponse(('ok',)), response)
1496
# The repository is now unlocked. Verify that with a new repository
1498
new_repo = repository.bzrdir.open_repository()
1499
new_repo.lock_write()
1502
def test_unlock_on_unlocked_repo(self):
1503
backing = self.get_transport()
1504
request = smart.repository.SmartServerRepositoryUnlock(backing)
1505
repository = self.make_repository('.', format='knit')
1506
response = request.execute('', 'some token')
1508
SmartServerResponse(('TokenMismatch',)), response)
1511
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1513
def test_is_readonly_no(self):
1514
backing = self.get_transport()
1515
request = smart.request.SmartServerIsReadonly(backing)
1516
response = request.execute()
1518
SmartServerResponse(('no',)), response)
1520
def test_is_readonly_yes(self):
1521
backing = self.get_readonly_transport()
1522
request = smart.request.SmartServerIsReadonly(backing)
1523
response = request.execute()
1525
SmartServerResponse(('yes',)), response)
1528
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1530
def test_set_false(self):
1531
backing = self.get_transport()
1532
repo = self.make_repository('.', shared=True)
1533
repo.set_make_working_trees(True)
1534
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1535
request = request_class(backing)
1536
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1537
request.execute('', 'False'))
1538
repo = repo.bzrdir.open_repository()
1539
self.assertFalse(repo.make_working_trees())
1541
def test_set_true(self):
1542
backing = self.get_transport()
1543
repo = self.make_repository('.', shared=True)
1544
repo.set_make_working_trees(False)
1545
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1546
request = request_class(backing)
1547
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1548
request.execute('', 'True'))
1549
repo = repo.bzrdir.open_repository()
1550
self.assertTrue(repo.make_working_trees())
1553
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1555
def make_repo_needing_autopacking(self, path='.'):
1556
# Make a repo in need of autopacking.
1557
tree = self.make_branch_and_tree('.', format='pack-0.92')
1558
repo = tree.branch.repository
1559
# monkey-patch the pack collection to disable autopacking
1560
repo._pack_collection._max_pack_count = lambda count: count
1562
tree.commit('commit %s' % x)
1563
self.assertEqual(10, len(repo._pack_collection.names()))
1564
del repo._pack_collection._max_pack_count
1567
def test_autopack_needed(self):
1568
repo = self.make_repo_needing_autopacking()
1570
self.addCleanup(repo.unlock)
1571
backing = self.get_transport()
1572
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1574
response = request.execute('')
1575
self.assertEqual(SmartServerResponse(('ok',)), response)
1576
repo._pack_collection.reload_pack_names()
1577
self.assertEqual(1, len(repo._pack_collection.names()))
1579
def test_autopack_not_needed(self):
1580
tree = self.make_branch_and_tree('.', format='pack-0.92')
1581
repo = tree.branch.repository
1583
self.addCleanup(repo.unlock)
1585
tree.commit('commit %s' % x)
1586
backing = self.get_transport()
1587
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1589
response = request.execute('')
1590
self.assertEqual(SmartServerResponse(('ok',)), response)
1591
repo._pack_collection.reload_pack_names()
1592
self.assertEqual(9, len(repo._pack_collection.names()))
1594
def test_autopack_on_nonpack_format(self):
1595
"""A request to autopack a non-pack repo is a no-op."""
1596
repo = self.make_repository('.', format='knit')
1597
backing = self.get_transport()
1598
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1600
response = request.execute('')
1601
self.assertEqual(SmartServerResponse(('ok',)), response)
1604
class TestHandlers(tests.TestCase):
1605
"""Tests for the request.request_handlers object."""
1607
def test_all_registrations_exist(self):
1608
"""All registered request_handlers can be found."""
1609
# If there's a typo in a register_lazy call, this loop will fail with
1610
# an AttributeError.
1611
for key, item in smart.request.request_handlers.iteritems():
1614
def assertHandlerEqual(self, verb, handler):
1615
self.assertEqual(smart.request.request_handlers.get(verb), handler)
1617
def test_registered_methods(self):
1618
"""Test that known methods are registered to the correct object."""
1619
self.assertHandlerEqual('Branch.get_config_file',
1620
smart.branch.SmartServerBranchGetConfigFile)
1621
self.assertHandlerEqual('Branch.get_parent',
1622
smart.branch.SmartServerBranchGetParent)
1623
self.assertHandlerEqual('Branch.get_tags_bytes',
1624
smart.branch.SmartServerBranchGetTagsBytes)
1625
self.assertHandlerEqual('Branch.lock_write',
1626
smart.branch.SmartServerBranchRequestLockWrite)
1627
self.assertHandlerEqual('Branch.last_revision_info',
1628
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1629
self.assertHandlerEqual('Branch.revision_history',
1630
smart.branch.SmartServerRequestRevisionHistory)
1631
self.assertHandlerEqual('Branch.set_config_option',
1632
smart.branch.SmartServerBranchRequestSetConfigOption)
1633
self.assertHandlerEqual('Branch.set_last_revision',
1634
smart.branch.SmartServerBranchRequestSetLastRevision)
1635
self.assertHandlerEqual('Branch.set_last_revision_info',
1636
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1637
self.assertHandlerEqual('Branch.set_last_revision_ex',
1638
smart.branch.SmartServerBranchRequestSetLastRevisionEx)
1639
self.assertHandlerEqual('Branch.set_parent_location',
1640
smart.branch.SmartServerBranchRequestSetParentLocation)
1641
self.assertHandlerEqual('Branch.unlock',
1642
smart.branch.SmartServerBranchRequestUnlock)
1643
self.assertHandlerEqual('BzrDir.find_repository',
1644
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1645
self.assertHandlerEqual('BzrDir.find_repositoryV2',
1646
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1647
self.assertHandlerEqual('BzrDirFormat.initialize',
1648
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1649
self.assertHandlerEqual('BzrDirFormat.initialize_ex_1.16',
1650
smart.bzrdir.SmartServerRequestBzrDirInitializeEx)
1651
self.assertHandlerEqual('BzrDir.cloning_metadir',
1652
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1653
self.assertHandlerEqual('BzrDir.get_config_file',
1654
smart.bzrdir.SmartServerBzrDirRequestConfigFile)
1655
self.assertHandlerEqual('BzrDir.open_branch',
1656
smart.bzrdir.SmartServerRequestOpenBranch)
1657
self.assertHandlerEqual('BzrDir.open_branchV2',
1658
smart.bzrdir.SmartServerRequestOpenBranchV2)
1659
self.assertHandlerEqual('PackRepository.autopack',
1660
smart.packrepository.SmartServerPackRepositoryAutopack)
1661
self.assertHandlerEqual('Repository.gather_stats',
1662
smart.repository.SmartServerRepositoryGatherStats)
1663
self.assertHandlerEqual('Repository.get_parent_map',
1664
smart.repository.SmartServerRepositoryGetParentMap)
1665
self.assertHandlerEqual('Repository.get_rev_id_for_revno',
1666
smart.repository.SmartServerRepositoryGetRevIdForRevno)
1667
self.assertHandlerEqual('Repository.get_revision_graph',
1668
smart.repository.SmartServerRepositoryGetRevisionGraph)
1669
self.assertHandlerEqual('Repository.get_stream',
1670
smart.repository.SmartServerRepositoryGetStream)
1671
self.assertHandlerEqual('Repository.has_revision',
1672
smart.repository.SmartServerRequestHasRevision)
1673
self.assertHandlerEqual('Repository.insert_stream',
1674
smart.repository.SmartServerRepositoryInsertStream)
1675
self.assertHandlerEqual('Repository.insert_stream_locked',
1676
smart.repository.SmartServerRepositoryInsertStreamLocked)
1677
self.assertHandlerEqual('Repository.is_shared',
1678
smart.repository.SmartServerRepositoryIsShared)
1679
self.assertHandlerEqual('Repository.lock_write',
1680
smart.repository.SmartServerRepositoryLockWrite)
1681
self.assertHandlerEqual('Repository.tarball',
1682
smart.repository.SmartServerRepositoryTarball)
1683
self.assertHandlerEqual('Repository.unlock',
1684
smart.repository.SmartServerRepositoryUnlock)
1685
self.assertHandlerEqual('Transport.is_readonly',
1686
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
