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
39
from bzrlib.branch import Branch, BranchReferenceFormat
40
import bzrlib.smart.branch
41
import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir
42
import bzrlib.smart.packrepository
43
import bzrlib.smart.repository
44
from bzrlib.smart.request import (
45
FailedSmartServerResponse,
48
SuccessfulSmartServerResponse,
50
from bzrlib.tests import (
53
from bzrlib.transport import chroot, get_transport
54
from bzrlib.util import bencode
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('default')()
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 (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
251
self._request_class):
252
return SuccessfulSmartServerResponse(
253
('ok', '', rich_root, subtrees, 'no',
254
repo._format.network_name()))
255
elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
256
self._request_class):
257
# All tests so far are on formats, and for non-external
259
return SuccessfulSmartServerResponse(
260
('ok', '', rich_root, subtrees, 'no'))
262
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
264
def test_shared_repository(self):
265
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
266
backing = self.get_transport()
267
request = self._request_class(backing)
268
result = self._make_repository_and_result(shared=True)
269
self.assertEqual(result, request.execute(''))
270
self.make_bzrdir('subdir')
271
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
272
self.assertEqual(result2,
273
request.execute('subdir'))
274
self.make_bzrdir('subdir/deeper')
275
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
276
self.assertEqual(result3,
277
request.execute('subdir/deeper'))
279
def test_rich_root_and_subtree_encoding(self):
280
"""Test for the format attributes for rich root and subtree support."""
281
backing = self.get_transport()
282
request = self._request_class(backing)
283
result = self._make_repository_and_result(format='dirstate-with-subtree')
284
# check the test will be valid
285
self.assertEqual('yes', result.args[2])
286
self.assertEqual('yes', result.args[3])
287
self.assertEqual(result, request.execute(''))
289
def test_supports_external_lookups_no_v2(self):
290
"""Test for the supports_external_lookups attribute."""
291
backing = self.get_transport()
292
request = self._request_class(backing)
293
result = self._make_repository_and_result(format='dirstate-with-subtree')
294
# check the test will be valid
295
self.assertEqual('no', result.args[4])
296
self.assertEqual(result, request.execute(''))
299
class TestSmartServerBzrDirRequestGetConfigFile(
300
tests.TestCaseWithMemoryTransport):
301
"""Tests for BzrDir.get_config_file."""
303
def test_present(self):
304
backing = self.get_transport()
305
dir = self.make_bzrdir('.')
306
dir.get_config().set_default_stack_on("/")
307
local_result = dir._get_config()._get_config_file().read()
308
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
309
request = request_class(backing)
310
expected = SuccessfulSmartServerResponse((), local_result)
311
self.assertEqual(expected, request.execute(''))
313
def test_missing(self):
314
backing = self.get_transport()
315
dir = self.make_bzrdir('.')
316
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
317
request = request_class(backing)
318
expected = SuccessfulSmartServerResponse((), '')
319
self.assertEqual(expected, request.execute(''))
322
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
324
def test_empty_dir(self):
325
"""Initializing an empty dir should succeed and do it."""
326
backing = self.get_transport()
327
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
328
self.assertEqual(SmartServerResponse(('ok', )),
330
made_dir = bzrdir.BzrDir.open_from_transport(backing)
331
# no branch, tree or repository is expected with the current
333
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
334
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
335
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
337
def test_missing_dir(self):
338
"""Initializing a missing directory should fail like the bzrdir api."""
339
backing = self.get_transport()
340
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
341
self.assertRaises(errors.NoSuchFile,
342
request.execute, 'subdir')
344
def test_initialized_dir(self):
345
"""Initializing an extant bzrdir should fail like the bzrdir api."""
346
backing = self.get_transport()
347
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
348
self.make_bzrdir('subdir')
349
self.assertRaises(errors.FileExists,
350
request.execute, 'subdir')
353
class TestSmartServerRequestBzrDirInitializeEx(tests.TestCaseWithMemoryTransport):
354
"""Basic tests for BzrDir.initialize_ex in the smart server.
356
The main unit tests in test_bzrdir exercise the API comprehensively.
359
def test_empty_dir(self):
360
"""Initializing an empty dir should succeed and do it."""
361
backing = self.get_transport()
362
name = self.make_bzrdir('reference')._format.network_name()
363
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
364
self.assertEqual(SmartServerResponse(('', '', '', '', '', '', name,
365
'False', '', '', '')),
366
request.execute(name, '', 'True', 'False', 'False', '', '', '', '',
368
made_dir = bzrdir.BzrDir.open_from_transport(backing)
369
# no branch, tree or repository is expected with the current
371
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
372
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
373
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
375
def test_missing_dir(self):
376
"""Initializing a missing directory should fail like the bzrdir api."""
377
backing = self.get_transport()
378
name = self.make_bzrdir('reference')._format.network_name()
379
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
380
self.assertRaises(errors.NoSuchFile, request.execute, name,
381
'subdir/dir', 'False', 'False', 'False', '', '', '', '', 'False')
383
def test_initialized_dir(self):
384
"""Initializing an extant dirctory should fail like the bzrdir api."""
385
backing = self.get_transport()
386
name = self.make_bzrdir('reference')._format.network_name()
387
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
388
self.make_bzrdir('subdir')
389
self.assertRaises(errors.FileExists, request.execute, name, 'subdir',
390
'False', 'False', 'False', '', '', '', '', 'False')
393
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
395
def test_no_branch(self):
396
"""When there is no branch, ('nobranch', ) is returned."""
397
backing = self.get_transport()
398
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
399
self.make_bzrdir('.')
400
self.assertEqual(SmartServerResponse(('nobranch', )),
403
def test_branch(self):
404
"""When there is a branch, 'ok' is returned."""
405
backing = self.get_transport()
406
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
407
self.make_branch('.')
408
self.assertEqual(SmartServerResponse(('ok', '')),
411
def test_branch_reference(self):
412
"""When there is a branch reference, the reference URL is returned."""
413
backing = self.get_transport()
414
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
415
branch = self.make_branch('branch')
416
checkout = branch.create_checkout('reference',lightweight=True)
417
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
418
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
419
self.assertEqual(SmartServerResponse(('ok', reference_url)),
420
request.execute('reference'))
423
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
425
def test_no_branch(self):
426
"""When there is no branch, ('nobranch', ) is returned."""
427
backing = self.get_transport()
428
self.make_bzrdir('.')
429
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
430
self.assertEqual(SmartServerResponse(('nobranch', )),
433
def test_branch(self):
434
"""When there is a branch, 'ok' is returned."""
435
backing = self.get_transport()
436
expected = self.make_branch('.')._format.network_name()
437
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
438
self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),
441
def test_branch_reference(self):
442
"""When there is a branch reference, the reference URL is returned."""
443
backing = self.get_transport()
444
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
445
branch = self.make_branch('branch')
446
checkout = branch.create_checkout('reference',lightweight=True)
447
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
448
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
449
self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
450
request.execute('reference'))
452
def test_stacked_branch(self):
453
"""Opening a stacked branch does not open the stacked-on branch."""
454
trunk = self.make_branch('trunk')
455
feature = self.make_branch('feature', format='1.9')
456
feature.set_stacked_on_url(trunk.base)
458
Branch.hooks.install_named_hook('open', opened_branches.append, None)
459
backing = self.get_transport()
460
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
463
response = request.execute('feature')
465
request.teardown_jail()
466
expected_format = feature._format.network_name()
468
SuccessfulSmartServerResponse(('branch', expected_format)),
470
self.assertLength(1, opened_branches)
473
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
475
def test_empty(self):
476
"""For an empty branch, the body is empty."""
477
backing = self.get_transport()
478
request = smart.branch.SmartServerRequestRevisionHistory(backing)
479
self.make_branch('.')
480
self.assertEqual(SmartServerResponse(('ok', ), ''),
483
def test_not_empty(self):
484
"""For a non-empty branch, the body is empty."""
485
backing = self.get_transport()
486
request = smart.branch.SmartServerRequestRevisionHistory(backing)
487
tree = self.make_branch_and_memory_tree('.')
490
r1 = tree.commit('1st commit')
491
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
494
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
498
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
500
def test_no_branch(self):
501
"""When there is a bzrdir and no branch, NotBranchError is raised."""
502
backing = self.get_transport()
503
request = smart.branch.SmartServerBranchRequest(backing)
504
self.make_bzrdir('.')
505
self.assertRaises(errors.NotBranchError,
508
def test_branch_reference(self):
509
"""When there is a branch reference, NotBranchError is raised."""
510
backing = self.get_transport()
511
request = smart.branch.SmartServerBranchRequest(backing)
512
branch = self.make_branch('branch')
513
checkout = branch.create_checkout('reference',lightweight=True)
514
self.assertRaises(errors.NotBranchError,
515
request.execute, 'checkout')
518
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
520
def test_empty(self):
521
"""For an empty branch, the result is ('ok', '0', 'null:')."""
522
backing = self.get_transport()
523
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
524
self.make_branch('.')
525
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
528
def test_not_empty(self):
529
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
530
backing = self.get_transport()
531
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
532
tree = self.make_branch_and_memory_tree('.')
535
rev_id_utf8 = u'\xc8'.encode('utf-8')
536
r1 = tree.commit('1st commit')
537
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
540
SmartServerResponse(('ok', '2', rev_id_utf8)),
544
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
546
def test_default(self):
547
"""With no file, we get empty content."""
548
backing = self.get_transport()
549
request = smart.branch.SmartServerBranchGetConfigFile(backing)
550
branch = self.make_branch('.')
551
# there should be no file by default
553
self.assertEqual(SmartServerResponse(('ok', ), content),
556
def test_with_content(self):
557
# SmartServerBranchGetConfigFile should return the content from
558
# branch.control_files.get('branch.conf') for now - in the future it may
559
# perform more complex processing.
560
backing = self.get_transport()
561
request = smart.branch.SmartServerBranchGetConfigFile(backing)
562
branch = self.make_branch('.')
563
branch._transport.put_bytes('branch.conf', 'foo bar baz')
564
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
568
class TestLockedBranch(tests.TestCaseWithMemoryTransport):
570
def get_lock_tokens(self, branch):
571
branch_token = branch.lock_write()
572
repo_token = branch.repository.lock_write()
573
branch.repository.unlock()
574
return branch_token, repo_token
577
class TestSmartServerBranchRequestSetConfigOption(TestLockedBranch):
579
def test_value_name(self):
580
branch = self.make_branch('.')
581
request = smart.branch.SmartServerBranchRequestSetConfigOption(
582
branch.bzrdir.root_transport)
583
branch_token, repo_token = self.get_lock_tokens(branch)
584
config = branch._get_config()
585
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
587
self.assertEqual(SuccessfulSmartServerResponse(()), result)
588
self.assertEqual('bar', config.get_option('foo'))
592
def test_value_name_section(self):
593
branch = self.make_branch('.')
594
request = smart.branch.SmartServerBranchRequestSetConfigOption(
595
branch.bzrdir.root_transport)
596
branch_token, repo_token = self.get_lock_tokens(branch)
597
config = branch._get_config()
598
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
600
self.assertEqual(SuccessfulSmartServerResponse(()), result)
601
self.assertEqual('bar', config.get_option('foo', 'gam'))
606
class SetLastRevisionTestBase(TestLockedBranch):
607
"""Base test case for verbs that implement set_last_revision."""
610
tests.TestCaseWithMemoryTransport.setUp(self)
611
backing_transport = self.get_transport()
612
self.request = self.request_class(backing_transport)
613
self.tree = self.make_branch_and_memory_tree('.')
615
def lock_branch(self):
616
return self.get_lock_tokens(self.tree.branch)
618
def unlock_branch(self):
619
self.tree.branch.unlock()
621
def set_last_revision(self, revision_id, revno):
622
branch_token, repo_token = self.lock_branch()
623
response = self._set_last_revision(
624
revision_id, revno, branch_token, repo_token)
628
def assertRequestSucceeds(self, revision_id, revno):
629
response = self.set_last_revision(revision_id, revno)
630
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
633
class TestSetLastRevisionVerbMixin(object):
634
"""Mixin test case for verbs that implement set_last_revision."""
636
def test_set_null_to_null(self):
637
"""An empty branch can have its last revision set to 'null:'."""
638
self.assertRequestSucceeds('null:', 0)
640
def test_NoSuchRevision(self):
641
"""If the revision_id is not present, the verb returns NoSuchRevision.
643
revision_id = 'non-existent revision'
645
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
646
self.set_last_revision(revision_id, 1))
648
def make_tree_with_two_commits(self):
649
self.tree.lock_write()
651
rev_id_utf8 = u'\xc8'.encode('utf-8')
652
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
653
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
656
def test_branch_last_revision_info_is_updated(self):
657
"""A branch's tip can be set to a revision that is present in its
660
# Make a branch with an empty revision history, but two revisions in
662
self.make_tree_with_two_commits()
663
rev_id_utf8 = u'\xc8'.encode('utf-8')
664
self.tree.branch.set_revision_history([])
666
(0, 'null:'), self.tree.branch.last_revision_info())
667
# We can update the branch to a revision that is present in the
669
self.assertRequestSucceeds(rev_id_utf8, 1)
671
(1, rev_id_utf8), self.tree.branch.last_revision_info())
673
def test_branch_last_revision_info_rewind(self):
674
"""A branch's tip can be set to a revision that is an ancestor of the
677
self.make_tree_with_two_commits()
678
rev_id_utf8 = u'\xc8'.encode('utf-8')
680
(2, 'rev-2'), self.tree.branch.last_revision_info())
681
self.assertRequestSucceeds(rev_id_utf8, 1)
683
(1, rev_id_utf8), self.tree.branch.last_revision_info())
685
def test_TipChangeRejected(self):
686
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
687
returns TipChangeRejected.
689
rejection_message = u'rejection message\N{INTERROBANG}'
690
def hook_that_rejects(params):
691
raise errors.TipChangeRejected(rejection_message)
692
Branch.hooks.install_named_hook(
693
'pre_change_branch_tip', hook_that_rejects, None)
695
FailedSmartServerResponse(
696
('TipChangeRejected', rejection_message.encode('utf-8'))),
697
self.set_last_revision('null:', 0))
700
class TestSmartServerBranchRequestSetLastRevision(
701
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
702
"""Tests for Branch.set_last_revision verb."""
704
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
706
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
707
return self.request.execute(
708
'', branch_token, repo_token, revision_id)
711
class TestSmartServerBranchRequestSetLastRevisionInfo(
712
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
713
"""Tests for Branch.set_last_revision_info verb."""
715
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
717
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
718
return self.request.execute(
719
'', branch_token, repo_token, revno, revision_id)
721
def test_NoSuchRevision(self):
722
"""Branch.set_last_revision_info does not have to return
723
NoSuchRevision if the revision_id is absent.
725
raise tests.TestNotApplicable()
728
class TestSmartServerBranchRequestSetLastRevisionEx(
729
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
730
"""Tests for Branch.set_last_revision_ex verb."""
732
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
734
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
735
return self.request.execute(
736
'', branch_token, repo_token, revision_id, 0, 0)
738
def assertRequestSucceeds(self, revision_id, revno):
739
response = self.set_last_revision(revision_id, revno)
741
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
744
def test_branch_last_revision_info_rewind(self):
745
"""A branch's tip can be set to a revision that is an ancestor of the
746
current tip, but only if allow_overwrite_descendant is passed.
748
self.make_tree_with_two_commits()
749
rev_id_utf8 = u'\xc8'.encode('utf-8')
751
(2, 'rev-2'), self.tree.branch.last_revision_info())
752
# If allow_overwrite_descendant flag is 0, then trying to set the tip
753
# to an older revision ID has no effect.
754
branch_token, repo_token = self.lock_branch()
755
response = self.request.execute(
756
'', branch_token, repo_token, rev_id_utf8, 0, 0)
758
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
761
(2, 'rev-2'), self.tree.branch.last_revision_info())
763
# If allow_overwrite_descendant flag is 1, then setting the tip to an
765
response = self.request.execute(
766
'', branch_token, repo_token, rev_id_utf8, 0, 1)
768
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
772
(1, rev_id_utf8), self.tree.branch.last_revision_info())
774
def make_branch_with_divergent_history(self):
775
"""Make a branch with divergent history in its repo.
777
The branch's tip will be 'child-2', and the repo will also contain
778
'child-1', which diverges from a common base revision.
780
self.tree.lock_write()
782
r1 = self.tree.commit('1st commit')
783
revno_1, revid_1 = self.tree.branch.last_revision_info()
784
r2 = self.tree.commit('2nd commit', rev_id='child-1')
785
# Undo the second commit
786
self.tree.branch.set_last_revision_info(revno_1, revid_1)
787
self.tree.set_parent_ids([revid_1])
788
# Make a new second commit, child-2. child-2 has diverged from
790
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
793
def test_not_allow_diverged(self):
794
"""If allow_diverged is not passed, then setting a divergent history
795
returns a Diverged error.
797
self.make_branch_with_divergent_history()
799
FailedSmartServerResponse(('Diverged',)),
800
self.set_last_revision('child-1', 2))
801
# The branch tip was not changed.
802
self.assertEqual('child-2', self.tree.branch.last_revision())
804
def test_allow_diverged(self):
805
"""If allow_diverged is passed, then setting a divergent history
808
self.make_branch_with_divergent_history()
809
branch_token, repo_token = self.lock_branch()
810
response = self.request.execute(
811
'', branch_token, repo_token, 'child-1', 1, 0)
813
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
816
# The branch tip was changed.
817
self.assertEqual('child-1', self.tree.branch.last_revision())
820
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
822
def test_get_parent_none(self):
823
base_branch = self.make_branch('base')
824
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
825
response = request.execute('base')
827
SuccessfulSmartServerResponse(('',)), response)
829
def test_get_parent_something(self):
830
base_branch = self.make_branch('base')
831
base_branch.set_parent(self.get_url('foo'))
832
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
833
response = request.execute('base')
835
SuccessfulSmartServerResponse(("../foo",)),
839
class TestSmartServerBranchRequestSetParent(tests.TestCaseWithMemoryTransport):
841
def test_set_parent_none(self):
842
branch = self.make_branch('base', format="1.9")
844
branch._set_parent_location('foo')
846
request = smart.branch.SmartServerBranchRequestSetParentLocation(
847
self.get_transport())
848
branch_token = branch.lock_write()
849
repo_token = branch.repository.lock_write()
851
response = request.execute('base', branch_token, repo_token, '')
853
branch.repository.unlock()
855
self.assertEqual(SuccessfulSmartServerResponse(()), response)
856
self.assertEqual(None, branch.get_parent())
858
def test_set_parent_something(self):
859
branch = self.make_branch('base', format="1.9")
860
request = smart.branch.SmartServerBranchRequestSetParentLocation(
861
self.get_transport())
862
branch_token = branch.lock_write()
863
repo_token = branch.repository.lock_write()
865
response = request.execute('base', branch_token, repo_token,
868
branch.repository.unlock()
870
self.assertEqual(SuccessfulSmartServerResponse(()), response)
871
self.assertEqual('http://bar/', branch.get_parent())
874
class TestSmartServerBranchRequestGetTagsBytes(tests.TestCaseWithMemoryTransport):
875
# Only called when the branch format and tags match [yay factory
876
# methods] so only need to test straight forward cases.
878
def test_get_bytes(self):
879
base_branch = self.make_branch('base')
880
request = smart.branch.SmartServerBranchGetTagsBytes(
881
self.get_transport())
882
response = request.execute('base')
884
SuccessfulSmartServerResponse(('',)), response)
887
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
889
def test_get_stacked_on_url(self):
890
base_branch = self.make_branch('base', format='1.6')
891
stacked_branch = self.make_branch('stacked', format='1.6')
892
# typically should be relative
893
stacked_branch.set_stacked_on_url('../base')
894
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
895
self.get_transport())
896
response = request.execute('stacked')
898
SmartServerResponse(('ok', '../base')),
902
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
905
tests.TestCaseWithMemoryTransport.setUp(self)
907
def test_lock_write_on_unlocked_branch(self):
908
backing = self.get_transport()
909
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
910
branch = self.make_branch('.', format='knit')
911
repository = branch.repository
912
response = request.execute('')
913
branch_nonce = branch.control_files._lock.peek().get('nonce')
914
repository_nonce = repository.control_files._lock.peek().get('nonce')
916
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
918
# The branch (and associated repository) is now locked. Verify that
919
# with a new branch object.
920
new_branch = repository.bzrdir.open_branch()
921
self.assertRaises(errors.LockContention, new_branch.lock_write)
923
request = smart.branch.SmartServerBranchRequestUnlock(backing)
924
response = request.execute('', branch_nonce, repository_nonce)
926
def test_lock_write_on_locked_branch(self):
927
backing = self.get_transport()
928
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
929
branch = self.make_branch('.')
930
branch_token = branch.lock_write()
931
branch.leave_lock_in_place()
933
response = request.execute('')
935
SmartServerResponse(('LockContention',)), response)
937
branch.lock_write(branch_token)
938
branch.dont_leave_lock_in_place()
941
def test_lock_write_with_tokens_on_locked_branch(self):
942
backing = self.get_transport()
943
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
944
branch = self.make_branch('.', format='knit')
945
branch_token = branch.lock_write()
946
repo_token = branch.repository.lock_write()
947
branch.repository.unlock()
948
branch.leave_lock_in_place()
949
branch.repository.leave_lock_in_place()
951
response = request.execute('',
952
branch_token, repo_token)
954
SmartServerResponse(('ok', branch_token, repo_token)), response)
956
branch.repository.lock_write(repo_token)
957
branch.repository.dont_leave_lock_in_place()
958
branch.repository.unlock()
959
branch.lock_write(branch_token)
960
branch.dont_leave_lock_in_place()
963
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
964
backing = self.get_transport()
965
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
966
branch = self.make_branch('.', format='knit')
967
branch_token = branch.lock_write()
968
repo_token = branch.repository.lock_write()
969
branch.repository.unlock()
970
branch.leave_lock_in_place()
971
branch.repository.leave_lock_in_place()
973
response = request.execute('',
974
branch_token+'xxx', repo_token)
976
SmartServerResponse(('TokenMismatch',)), response)
978
branch.repository.lock_write(repo_token)
979
branch.repository.dont_leave_lock_in_place()
980
branch.repository.unlock()
981
branch.lock_write(branch_token)
982
branch.dont_leave_lock_in_place()
985
def test_lock_write_on_locked_repo(self):
986
backing = self.get_transport()
987
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
988
branch = self.make_branch('.', format='knit')
989
repo = branch.repository
990
repo_token = repo.lock_write()
991
repo.leave_lock_in_place()
993
response = request.execute('')
995
SmartServerResponse(('LockContention',)), response)
997
repo.lock_write(repo_token)
998
repo.dont_leave_lock_in_place()
1001
def test_lock_write_on_readonly_transport(self):
1002
backing = self.get_readonly_transport()
1003
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1004
branch = self.make_branch('.')
1005
root = self.get_transport().clone('/')
1006
path = urlutils.relative_url(root.base, self.get_transport().base)
1007
response = request.execute(path)
1008
error_name, lock_str, why_str = response.args
1009
self.assertFalse(response.is_successful())
1010
self.assertEqual('LockFailed', error_name)
1013
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
1016
tests.TestCaseWithMemoryTransport.setUp(self)
1018
def test_unlock_on_locked_branch_and_repo(self):
1019
backing = self.get_transport()
1020
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1021
branch = self.make_branch('.', format='knit')
1023
branch_token = branch.lock_write()
1024
repo_token = branch.repository.lock_write()
1025
branch.repository.unlock()
1026
# Unlock the branch (and repo) object, leaving the physical locks
1028
branch.leave_lock_in_place()
1029
branch.repository.leave_lock_in_place()
1031
response = request.execute('',
1032
branch_token, repo_token)
1034
SmartServerResponse(('ok',)), response)
1035
# The branch is now unlocked. Verify that with a new branch
1037
new_branch = branch.bzrdir.open_branch()
1038
new_branch.lock_write()
1041
def test_unlock_on_unlocked_branch_unlocked_repo(self):
1042
backing = self.get_transport()
1043
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1044
branch = self.make_branch('.', format='knit')
1045
response = request.execute(
1046
'', 'branch token', 'repo token')
1048
SmartServerResponse(('TokenMismatch',)), response)
1050
def test_unlock_on_unlocked_branch_locked_repo(self):
1051
backing = self.get_transport()
1052
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1053
branch = self.make_branch('.', format='knit')
1054
# Lock the repository.
1055
repo_token = branch.repository.lock_write()
1056
branch.repository.leave_lock_in_place()
1057
branch.repository.unlock()
1058
# Issue branch lock_write request on the unlocked branch (with locked
1060
response = request.execute(
1061
'', 'branch token', repo_token)
1063
SmartServerResponse(('TokenMismatch',)), response)
1065
branch.repository.lock_write(repo_token)
1066
branch.repository.dont_leave_lock_in_place()
1067
branch.repository.unlock()
1070
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
1072
def test_no_repository(self):
1073
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
1074
# we test this using a shared repository above the named path,
1075
# thus checking the right search logic is used - that is, that
1076
# its the exact path being looked at and the server is not
1078
backing = self.get_transport()
1079
request = smart.repository.SmartServerRepositoryRequest(backing)
1080
self.make_repository('.', shared=True)
1081
self.make_bzrdir('subdir')
1082
self.assertRaises(errors.NoRepositoryPresent,
1083
request.execute, 'subdir')
1086
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
1088
def test_trivial_bzipped(self):
1089
# This tests that the wire encoding is actually bzipped
1090
backing = self.get_transport()
1091
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1092
tree = self.make_branch_and_memory_tree('.')
1094
self.assertEqual(None,
1095
request.execute('', 'missing-id'))
1096
# Note that it returns a body that is bzipped.
1098
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
1099
request.do_body('\n\n0\n'))
1101
def test_trivial_include_missing(self):
1102
backing = self.get_transport()
1103
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1104
tree = self.make_branch_and_memory_tree('.')
1106
self.assertEqual(None,
1107
request.execute('', 'missing-id', 'include-missing:'))
1109
SuccessfulSmartServerResponse(('ok', ),
1110
bz2.compress('missing:missing-id')),
1111
request.do_body('\n\n0\n'))
1114
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
1116
def test_none_argument(self):
1117
backing = self.get_transport()
1118
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1119
tree = self.make_branch_and_memory_tree('.')
1122
r1 = tree.commit('1st commit')
1123
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1126
# the lines of revision_id->revision_parent_list has no guaranteed
1127
# order coming out of a dict, so sort both our test and response
1128
lines = sorted([' '.join([r2, r1]), r1])
1129
response = request.execute('', '')
1130
response.body = '\n'.join(sorted(response.body.split('\n')))
1133
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
1135
def test_specific_revision_argument(self):
1136
backing = self.get_transport()
1137
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1138
tree = self.make_branch_and_memory_tree('.')
1141
rev_id_utf8 = u'\xc9'.encode('utf-8')
1142
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
1143
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1146
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
1147
request.execute('', rev_id_utf8))
1149
def test_no_such_revision(self):
1150
backing = self.get_transport()
1151
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1152
tree = self.make_branch_and_memory_tree('.')
1155
r1 = tree.commit('1st commit')
1158
# Note that it still returns body (of zero bytes).
1160
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
1161
request.execute('', 'missingrevision'))
1164
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
1166
def make_two_commit_repo(self):
1167
tree = self.make_branch_and_memory_tree('.')
1170
r1 = tree.commit('1st commit')
1171
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1173
repo = tree.branch.repository
1176
def test_ancestry_of(self):
1177
"""The search argument may be a 'ancestry-of' some heads'."""
1178
backing = self.get_transport()
1179
request = smart.repository.SmartServerRepositoryGetStream(backing)
1180
repo, r1, r2 = self.make_two_commit_repo()
1181
fetch_spec = ['ancestry-of', r2]
1182
lines = '\n'.join(fetch_spec)
1183
request.execute('', repo._format.network_name())
1184
response = request.do_body(lines)
1185
self.assertEqual(('ok',), response.args)
1186
stream_bytes = ''.join(response.body_stream)
1187
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1189
def test_search(self):
1190
"""The search argument may be a 'search' of some explicit keys."""
1191
backing = self.get_transport()
1192
request = smart.repository.SmartServerRepositoryGetStream(backing)
1193
repo, r1, r2 = self.make_two_commit_repo()
1194
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1195
lines = '\n'.join(fetch_spec)
1196
request.execute('', repo._format.network_name())
1197
response = request.do_body(lines)
1198
self.assertEqual(('ok',), response.args)
1199
stream_bytes = ''.join(response.body_stream)
1200
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1203
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1205
def test_missing_revision(self):
1206
"""For a missing revision, ('no', ) is returned."""
1207
backing = self.get_transport()
1208
request = smart.repository.SmartServerRequestHasRevision(backing)
1209
self.make_repository('.')
1210
self.assertEqual(SmartServerResponse(('no', )),
1211
request.execute('', 'revid'))
1213
def test_present_revision(self):
1214
"""For a present revision, ('yes', ) is returned."""
1215
backing = self.get_transport()
1216
request = smart.repository.SmartServerRequestHasRevision(backing)
1217
tree = self.make_branch_and_memory_tree('.')
1220
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1221
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1223
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1224
self.assertEqual(SmartServerResponse(('yes', )),
1225
request.execute('', rev_id_utf8))
1228
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1230
def test_empty_revid(self):
1231
"""With an empty revid, we get only size an number and revisions"""
1232
backing = self.get_transport()
1233
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1234
repository = self.make_repository('.')
1235
stats = repository.gather_stats()
1236
expected_body = 'revisions: 0\n'
1237
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1238
request.execute('', '', 'no'))
1240
def test_revid_with_committers(self):
1241
"""For a revid we get more infos."""
1242
backing = self.get_transport()
1243
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1244
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1245
tree = self.make_branch_and_memory_tree('.')
1248
# Let's build a predictable result
1249
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1250
tree.commit('a commit', timestamp=654321.4, timezone=0,
1254
stats = tree.branch.repository.gather_stats()
1255
expected_body = ('firstrev: 123456.200 3600\n'
1256
'latestrev: 654321.400 0\n'
1258
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1262
def test_not_empty_repository_with_committers(self):
1263
"""For a revid and requesting committers we get the whole thing."""
1264
backing = self.get_transport()
1265
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1266
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1267
tree = self.make_branch_and_memory_tree('.')
1270
# Let's build a predictable result
1271
tree.commit('a commit', timestamp=123456.2, timezone=3600,
1273
tree.commit('a commit', timestamp=654321.4, timezone=0,
1274
committer='bar', rev_id=rev_id_utf8)
1276
stats = tree.branch.repository.gather_stats()
1278
expected_body = ('committers: 2\n'
1279
'firstrev: 123456.200 3600\n'
1280
'latestrev: 654321.400 0\n'
1282
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1284
rev_id_utf8, 'yes'))
1287
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
1289
def test_is_shared(self):
1290
"""For a shared repository, ('yes', ) is returned."""
1291
backing = self.get_transport()
1292
request = smart.repository.SmartServerRepositoryIsShared(backing)
1293
self.make_repository('.', shared=True)
1294
self.assertEqual(SmartServerResponse(('yes', )),
1295
request.execute('', ))
1297
def test_is_not_shared(self):
1298
"""For a shared repository, ('no', ) is returned."""
1299
backing = self.get_transport()
1300
request = smart.repository.SmartServerRepositoryIsShared(backing)
1301
self.make_repository('.', shared=False)
1302
self.assertEqual(SmartServerResponse(('no', )),
1303
request.execute('', ))
1306
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
1308
def test_lock_write_on_unlocked_repo(self):
1309
backing = self.get_transport()
1310
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1311
repository = self.make_repository('.', format='knit')
1312
response = request.execute('')
1313
nonce = repository.control_files._lock.peek().get('nonce')
1314
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
1315
# The repository is now locked. Verify that with a new repository
1317
new_repo = repository.bzrdir.open_repository()
1318
self.assertRaises(errors.LockContention, new_repo.lock_write)
1320
request = smart.repository.SmartServerRepositoryUnlock(backing)
1321
response = request.execute('', nonce)
1323
def test_lock_write_on_locked_repo(self):
1324
backing = self.get_transport()
1325
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1326
repository = self.make_repository('.', format='knit')
1327
repo_token = repository.lock_write()
1328
repository.leave_lock_in_place()
1330
response = request.execute('')
1332
SmartServerResponse(('LockContention',)), response)
1334
repository.lock_write(repo_token)
1335
repository.dont_leave_lock_in_place()
1338
def test_lock_write_on_readonly_transport(self):
1339
backing = self.get_readonly_transport()
1340
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1341
repository = self.make_repository('.', format='knit')
1342
response = request.execute('')
1343
self.assertFalse(response.is_successful())
1344
self.assertEqual('LockFailed', response.args[0])
1347
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
1349
def make_empty_byte_stream(self, repo):
1350
byte_stream = smart.repository._stream_to_byte_stream([], repo._format)
1351
return ''.join(byte_stream)
1354
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
1356
def test_insert_stream_empty(self):
1357
backing = self.get_transport()
1358
request = smart.repository.SmartServerRepositoryInsertStream(backing)
1359
repository = self.make_repository('.')
1360
response = request.execute('', '')
1361
self.assertEqual(None, response)
1362
response = request.do_chunk(self.make_empty_byte_stream(repository))
1363
self.assertEqual(None, response)
1364
response = request.do_end()
1365
self.assertEqual(SmartServerResponse(('ok', )), response)
1368
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
1370
def test_insert_stream_empty(self):
1371
backing = self.get_transport()
1372
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1374
repository = self.make_repository('.', format='knit')
1375
lock_token = repository.lock_write()
1376
response = request.execute('', '', lock_token)
1377
self.assertEqual(None, response)
1378
response = request.do_chunk(self.make_empty_byte_stream(repository))
1379
self.assertEqual(None, response)
1380
response = request.do_end()
1381
self.assertEqual(SmartServerResponse(('ok', )), response)
1384
def test_insert_stream_with_wrong_lock_token(self):
1385
backing = self.get_transport()
1386
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1388
repository = self.make_repository('.', format='knit')
1389
lock_token = repository.lock_write()
1391
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
1395
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1398
tests.TestCaseWithMemoryTransport.setUp(self)
1400
def test_unlock_on_locked_repo(self):
1401
backing = self.get_transport()
1402
request = smart.repository.SmartServerRepositoryUnlock(backing)
1403
repository = self.make_repository('.', format='knit')
1404
token = repository.lock_write()
1405
repository.leave_lock_in_place()
1407
response = request.execute('', token)
1409
SmartServerResponse(('ok',)), response)
1410
# The repository is now unlocked. Verify that with a new repository
1412
new_repo = repository.bzrdir.open_repository()
1413
new_repo.lock_write()
1416
def test_unlock_on_unlocked_repo(self):
1417
backing = self.get_transport()
1418
request = smart.repository.SmartServerRepositoryUnlock(backing)
1419
repository = self.make_repository('.', format='knit')
1420
response = request.execute('', 'some token')
1422
SmartServerResponse(('TokenMismatch',)), response)
1425
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1427
def test_is_readonly_no(self):
1428
backing = self.get_transport()
1429
request = smart.request.SmartServerIsReadonly(backing)
1430
response = request.execute()
1432
SmartServerResponse(('no',)), response)
1434
def test_is_readonly_yes(self):
1435
backing = self.get_readonly_transport()
1436
request = smart.request.SmartServerIsReadonly(backing)
1437
response = request.execute()
1439
SmartServerResponse(('yes',)), response)
1442
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1444
def test_set_false(self):
1445
backing = self.get_transport()
1446
repo = self.make_repository('.', shared=True)
1447
repo.set_make_working_trees(True)
1448
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1449
request = request_class(backing)
1450
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1451
request.execute('', 'False'))
1452
repo = repo.bzrdir.open_repository()
1453
self.assertFalse(repo.make_working_trees())
1455
def test_set_true(self):
1456
backing = self.get_transport()
1457
repo = self.make_repository('.', shared=True)
1458
repo.set_make_working_trees(False)
1459
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1460
request = request_class(backing)
1461
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1462
request.execute('', 'True'))
1463
repo = repo.bzrdir.open_repository()
1464
self.assertTrue(repo.make_working_trees())
1467
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1469
def make_repo_needing_autopacking(self, path='.'):
1470
# Make a repo in need of autopacking.
1471
tree = self.make_branch_and_tree('.', format='pack-0.92')
1472
repo = tree.branch.repository
1473
# monkey-patch the pack collection to disable autopacking
1474
repo._pack_collection._max_pack_count = lambda count: count
1476
tree.commit('commit %s' % x)
1477
self.assertEqual(10, len(repo._pack_collection.names()))
1478
del repo._pack_collection._max_pack_count
1481
def test_autopack_needed(self):
1482
repo = self.make_repo_needing_autopacking()
1484
self.addCleanup(repo.unlock)
1485
backing = self.get_transport()
1486
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1488
response = request.execute('')
1489
self.assertEqual(SmartServerResponse(('ok',)), response)
1490
repo._pack_collection.reload_pack_names()
1491
self.assertEqual(1, len(repo._pack_collection.names()))
1493
def test_autopack_not_needed(self):
1494
tree = self.make_branch_and_tree('.', format='pack-0.92')
1495
repo = tree.branch.repository
1497
self.addCleanup(repo.unlock)
1499
tree.commit('commit %s' % x)
1500
backing = self.get_transport()
1501
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1503
response = request.execute('')
1504
self.assertEqual(SmartServerResponse(('ok',)), response)
1505
repo._pack_collection.reload_pack_names()
1506
self.assertEqual(9, len(repo._pack_collection.names()))
1508
def test_autopack_on_nonpack_format(self):
1509
"""A request to autopack a non-pack repo is a no-op."""
1510
repo = self.make_repository('.', format='knit')
1511
backing = self.get_transport()
1512
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1514
response = request.execute('')
1515
self.assertEqual(SmartServerResponse(('ok',)), response)
1518
class TestHandlers(tests.TestCase):
1519
"""Tests for the request.request_handlers object."""
1521
def test_all_registrations_exist(self):
1522
"""All registered request_handlers can be found."""
1523
# If there's a typo in a register_lazy call, this loop will fail with
1524
# an AttributeError.
1525
for key, item in smart.request.request_handlers.iteritems():
1528
def assertHandlerEqual(self, verb, handler):
1529
self.assertEqual(smart.request.request_handlers.get(verb), handler)
1531
def test_registered_methods(self):
1532
"""Test that known methods are registered to the correct object."""
1533
self.assertHandlerEqual('Branch.get_config_file',
1534
smart.branch.SmartServerBranchGetConfigFile)
1535
self.assertHandlerEqual('Branch.get_parent',
1536
smart.branch.SmartServerBranchGetParent)
1537
self.assertHandlerEqual('Branch.get_tags_bytes',
1538
smart.branch.SmartServerBranchGetTagsBytes)
1539
self.assertHandlerEqual('Branch.lock_write',
1540
smart.branch.SmartServerBranchRequestLockWrite)
1541
self.assertHandlerEqual('Branch.last_revision_info',
1542
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1543
self.assertHandlerEqual('Branch.revision_history',
1544
smart.branch.SmartServerRequestRevisionHistory)
1545
self.assertHandlerEqual('Branch.set_config_option',
1546
smart.branch.SmartServerBranchRequestSetConfigOption)
1547
self.assertHandlerEqual('Branch.set_last_revision',
1548
smart.branch.SmartServerBranchRequestSetLastRevision)
1549
self.assertHandlerEqual('Branch.set_last_revision_info',
1550
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1551
self.assertHandlerEqual('Branch.set_last_revision_ex',
1552
smart.branch.SmartServerBranchRequestSetLastRevisionEx)
1553
self.assertHandlerEqual('Branch.set_parent_location',
1554
smart.branch.SmartServerBranchRequestSetParentLocation)
1555
self.assertHandlerEqual('Branch.unlock',
1556
smart.branch.SmartServerBranchRequestUnlock)
1557
self.assertHandlerEqual('BzrDir.find_repository',
1558
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1559
self.assertHandlerEqual('BzrDir.find_repositoryV2',
1560
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1561
self.assertHandlerEqual('BzrDirFormat.initialize',
1562
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1563
self.assertHandlerEqual('BzrDirFormat.initialize_ex',
1564
smart.bzrdir.SmartServerRequestBzrDirInitializeEx)
1565
self.assertHandlerEqual('BzrDir.cloning_metadir',
1566
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1567
self.assertHandlerEqual('BzrDir.get_config_file',
1568
smart.bzrdir.SmartServerBzrDirRequestConfigFile)
1569
self.assertHandlerEqual('BzrDir.open_branch',
1570
smart.bzrdir.SmartServerRequestOpenBranch)
1571
self.assertHandlerEqual('BzrDir.open_branchV2',
1572
smart.bzrdir.SmartServerRequestOpenBranchV2)
1573
self.assertHandlerEqual('PackRepository.autopack',
1574
smart.packrepository.SmartServerPackRepositoryAutopack)
1575
self.assertHandlerEqual('Repository.gather_stats',
1576
smart.repository.SmartServerRepositoryGatherStats)
1577
self.assertHandlerEqual('Repository.get_parent_map',
1578
smart.repository.SmartServerRepositoryGetParentMap)
1579
self.assertHandlerEqual('Repository.get_revision_graph',
1580
smart.repository.SmartServerRepositoryGetRevisionGraph)
1581
self.assertHandlerEqual('Repository.get_stream',
1582
smart.repository.SmartServerRepositoryGetStream)
1583
self.assertHandlerEqual('Repository.has_revision',
1584
smart.repository.SmartServerRequestHasRevision)
1585
self.assertHandlerEqual('Repository.insert_stream',
1586
smart.repository.SmartServerRepositoryInsertStream)
1587
self.assertHandlerEqual('Repository.insert_stream_locked',
1588
smart.repository.SmartServerRepositoryInsertStreamLocked)
1589
self.assertHandlerEqual('Repository.is_shared',
1590
smart.repository.SmartServerRepositoryIsShared)
1591
self.assertHandlerEqual('Repository.lock_write',
1592
smart.repository.SmartServerRepositoryLockWrite)
1593
self.assertHandlerEqual('Repository.tarball',
1594
smart.repository.SmartServerRepositoryTarball)
1595
self.assertHandlerEqual('Repository.unlock',
1596
smart.repository.SmartServerRepositoryUnlock)
1597
self.assertHandlerEqual('Transport.is_readonly',
1598
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
