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('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_1.16 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 directory 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 TestSmartServerRepositoryGetRevIdForRevno(tests.TestCaseWithMemoryTransport):
1166
def test_revno_found(self):
1167
backing = self.get_transport()
1168
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1169
tree = self.make_branch_and_memory_tree('.')
1172
rev1_id_utf8 = u'\xc8'.encode('utf-8')
1173
rev2_id_utf8 = u'\xc9'.encode('utf-8')
1174
tree.commit('1st commit', rev_id=rev1_id_utf8)
1175
tree.commit('2nd commit', rev_id=rev2_id_utf8)
1178
self.assertEqual(SmartServerResponse(('ok', rev1_id_utf8)),
1179
request.execute('', 1, (2, rev2_id_utf8)))
1181
def test_known_revid_missing(self):
1182
backing = self.get_transport()
1183
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1184
repo = self.make_repository('.')
1186
FailedSmartServerResponse(('nosuchrevision', 'ghost')),
1187
request.execute('', 1, (2, 'ghost')))
1189
def test_history_incomplete(self):
1190
backing = self.get_transport()
1191
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1192
parent = self.make_branch_and_memory_tree('parent', format='1.9')
1193
r1 = parent.commit(message='first commit')
1194
r2 = parent.commit(message='second commit')
1195
local = self.make_branch_and_memory_tree('local', format='1.9')
1196
local.branch.pull(parent.branch)
1197
local.set_parent_ids([r2])
1198
r3 = local.commit(message='local commit')
1199
local.branch.create_clone_on_transport(
1200
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
1202
SmartServerResponse(('history-incomplete', 2, r2)),
1203
request.execute('stacked', 1, (3, r3)))
1205
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
1207
def make_two_commit_repo(self):
1208
tree = self.make_branch_and_memory_tree('.')
1211
r1 = tree.commit('1st commit')
1212
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1214
repo = tree.branch.repository
1217
def test_ancestry_of(self):
1218
"""The search argument may be a 'ancestry-of' some heads'."""
1219
backing = self.get_transport()
1220
request = smart.repository.SmartServerRepositoryGetStream(backing)
1221
repo, r1, r2 = self.make_two_commit_repo()
1222
fetch_spec = ['ancestry-of', r2]
1223
lines = '\n'.join(fetch_spec)
1224
request.execute('', repo._format.network_name())
1225
response = request.do_body(lines)
1226
self.assertEqual(('ok',), response.args)
1227
stream_bytes = ''.join(response.body_stream)
1228
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1230
def test_search(self):
1231
"""The search argument may be a 'search' of some explicit keys."""
1232
backing = self.get_transport()
1233
request = smart.repository.SmartServerRepositoryGetStream(backing)
1234
repo, r1, r2 = self.make_two_commit_repo()
1235
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1236
lines = '\n'.join(fetch_spec)
1237
request.execute('', repo._format.network_name())
1238
response = request.do_body(lines)
1239
self.assertEqual(('ok',), response.args)
1240
stream_bytes = ''.join(response.body_stream)
1241
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1244
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1246
def test_missing_revision(self):
1247
"""For a missing revision, ('no', ) is returned."""
1248
backing = self.get_transport()
1249
request = smart.repository.SmartServerRequestHasRevision(backing)
1250
self.make_repository('.')
1251
self.assertEqual(SmartServerResponse(('no', )),
1252
request.execute('', 'revid'))
1254
def test_present_revision(self):
1255
"""For a present revision, ('yes', ) is returned."""
1256
backing = self.get_transport()
1257
request = smart.repository.SmartServerRequestHasRevision(backing)
1258
tree = self.make_branch_and_memory_tree('.')
1261
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1262
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1264
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1265
self.assertEqual(SmartServerResponse(('yes', )),
1266
request.execute('', rev_id_utf8))
1269
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1271
def test_empty_revid(self):
1272
"""With an empty revid, we get only size an number and revisions"""
1273
backing = self.get_transport()
1274
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1275
repository = self.make_repository('.')
1276
stats = repository.gather_stats()
1277
expected_body = 'revisions: 0\n'
1278
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1279
request.execute('', '', 'no'))
1281
def test_revid_with_committers(self):
1282
"""For a revid we get more infos."""
1283
backing = self.get_transport()
1284
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1285
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1286
tree = self.make_branch_and_memory_tree('.')
1289
# Let's build a predictable result
1290
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1291
tree.commit('a commit', timestamp=654321.4, timezone=0,
1295
stats = tree.branch.repository.gather_stats()
1296
expected_body = ('firstrev: 123456.200 3600\n'
1297
'latestrev: 654321.400 0\n'
1299
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1303
def test_not_empty_repository_with_committers(self):
1304
"""For a revid and requesting committers we get the whole thing."""
1305
backing = self.get_transport()
1306
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1307
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1308
tree = self.make_branch_and_memory_tree('.')
1311
# Let's build a predictable result
1312
tree.commit('a commit', timestamp=123456.2, timezone=3600,
1314
tree.commit('a commit', timestamp=654321.4, timezone=0,
1315
committer='bar', rev_id=rev_id_utf8)
1317
stats = tree.branch.repository.gather_stats()
1319
expected_body = ('committers: 2\n'
1320
'firstrev: 123456.200 3600\n'
1321
'latestrev: 654321.400 0\n'
1323
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1325
rev_id_utf8, 'yes'))
1328
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
1330
def test_is_shared(self):
1331
"""For a shared repository, ('yes', ) is returned."""
1332
backing = self.get_transport()
1333
request = smart.repository.SmartServerRepositoryIsShared(backing)
1334
self.make_repository('.', shared=True)
1335
self.assertEqual(SmartServerResponse(('yes', )),
1336
request.execute('', ))
1338
def test_is_not_shared(self):
1339
"""For a shared repository, ('no', ) is returned."""
1340
backing = self.get_transport()
1341
request = smart.repository.SmartServerRepositoryIsShared(backing)
1342
self.make_repository('.', shared=False)
1343
self.assertEqual(SmartServerResponse(('no', )),
1344
request.execute('', ))
1347
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
1349
def test_lock_write_on_unlocked_repo(self):
1350
backing = self.get_transport()
1351
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1352
repository = self.make_repository('.', format='knit')
1353
response = request.execute('')
1354
nonce = repository.control_files._lock.peek().get('nonce')
1355
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
1356
# The repository is now locked. Verify that with a new repository
1358
new_repo = repository.bzrdir.open_repository()
1359
self.assertRaises(errors.LockContention, new_repo.lock_write)
1361
request = smart.repository.SmartServerRepositoryUnlock(backing)
1362
response = request.execute('', nonce)
1364
def test_lock_write_on_locked_repo(self):
1365
backing = self.get_transport()
1366
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1367
repository = self.make_repository('.', format='knit')
1368
repo_token = repository.lock_write()
1369
repository.leave_lock_in_place()
1371
response = request.execute('')
1373
SmartServerResponse(('LockContention',)), response)
1375
repository.lock_write(repo_token)
1376
repository.dont_leave_lock_in_place()
1379
def test_lock_write_on_readonly_transport(self):
1380
backing = self.get_readonly_transport()
1381
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1382
repository = self.make_repository('.', format='knit')
1383
response = request.execute('')
1384
self.assertFalse(response.is_successful())
1385
self.assertEqual('LockFailed', response.args[0])
1388
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
1390
def make_empty_byte_stream(self, repo):
1391
byte_stream = smart.repository._stream_to_byte_stream([], repo._format)
1392
return ''.join(byte_stream)
1395
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
1397
def test_insert_stream_empty(self):
1398
backing = self.get_transport()
1399
request = smart.repository.SmartServerRepositoryInsertStream(backing)
1400
repository = self.make_repository('.')
1401
response = request.execute('', '')
1402
self.assertEqual(None, response)
1403
response = request.do_chunk(self.make_empty_byte_stream(repository))
1404
self.assertEqual(None, response)
1405
response = request.do_end()
1406
self.assertEqual(SmartServerResponse(('ok', )), response)
1409
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
1411
def test_insert_stream_empty(self):
1412
backing = self.get_transport()
1413
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1415
repository = self.make_repository('.', format='knit')
1416
lock_token = repository.lock_write()
1417
response = request.execute('', '', lock_token)
1418
self.assertEqual(None, response)
1419
response = request.do_chunk(self.make_empty_byte_stream(repository))
1420
self.assertEqual(None, response)
1421
response = request.do_end()
1422
self.assertEqual(SmartServerResponse(('ok', )), response)
1425
def test_insert_stream_with_wrong_lock_token(self):
1426
backing = self.get_transport()
1427
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1429
repository = self.make_repository('.', format='knit')
1430
lock_token = repository.lock_write()
1432
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
1436
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1439
tests.TestCaseWithMemoryTransport.setUp(self)
1441
def test_unlock_on_locked_repo(self):
1442
backing = self.get_transport()
1443
request = smart.repository.SmartServerRepositoryUnlock(backing)
1444
repository = self.make_repository('.', format='knit')
1445
token = repository.lock_write()
1446
repository.leave_lock_in_place()
1448
response = request.execute('', token)
1450
SmartServerResponse(('ok',)), response)
1451
# The repository is now unlocked. Verify that with a new repository
1453
new_repo = repository.bzrdir.open_repository()
1454
new_repo.lock_write()
1457
def test_unlock_on_unlocked_repo(self):
1458
backing = self.get_transport()
1459
request = smart.repository.SmartServerRepositoryUnlock(backing)
1460
repository = self.make_repository('.', format='knit')
1461
response = request.execute('', 'some token')
1463
SmartServerResponse(('TokenMismatch',)), response)
1466
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1468
def test_is_readonly_no(self):
1469
backing = self.get_transport()
1470
request = smart.request.SmartServerIsReadonly(backing)
1471
response = request.execute()
1473
SmartServerResponse(('no',)), response)
1475
def test_is_readonly_yes(self):
1476
backing = self.get_readonly_transport()
1477
request = smart.request.SmartServerIsReadonly(backing)
1478
response = request.execute()
1480
SmartServerResponse(('yes',)), response)
1483
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1485
def test_set_false(self):
1486
backing = self.get_transport()
1487
repo = self.make_repository('.', shared=True)
1488
repo.set_make_working_trees(True)
1489
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1490
request = request_class(backing)
1491
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1492
request.execute('', 'False'))
1493
repo = repo.bzrdir.open_repository()
1494
self.assertFalse(repo.make_working_trees())
1496
def test_set_true(self):
1497
backing = self.get_transport()
1498
repo = self.make_repository('.', shared=True)
1499
repo.set_make_working_trees(False)
1500
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1501
request = request_class(backing)
1502
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1503
request.execute('', 'True'))
1504
repo = repo.bzrdir.open_repository()
1505
self.assertTrue(repo.make_working_trees())
1508
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1510
def make_repo_needing_autopacking(self, path='.'):
1511
# Make a repo in need of autopacking.
1512
tree = self.make_branch_and_tree('.', format='pack-0.92')
1513
repo = tree.branch.repository
1514
# monkey-patch the pack collection to disable autopacking
1515
repo._pack_collection._max_pack_count = lambda count: count
1517
tree.commit('commit %s' % x)
1518
self.assertEqual(10, len(repo._pack_collection.names()))
1519
del repo._pack_collection._max_pack_count
1522
def test_autopack_needed(self):
1523
repo = self.make_repo_needing_autopacking()
1525
self.addCleanup(repo.unlock)
1526
backing = self.get_transport()
1527
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1529
response = request.execute('')
1530
self.assertEqual(SmartServerResponse(('ok',)), response)
1531
repo._pack_collection.reload_pack_names()
1532
self.assertEqual(1, len(repo._pack_collection.names()))
1534
def test_autopack_not_needed(self):
1535
tree = self.make_branch_and_tree('.', format='pack-0.92')
1536
repo = tree.branch.repository
1538
self.addCleanup(repo.unlock)
1540
tree.commit('commit %s' % x)
1541
backing = self.get_transport()
1542
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1544
response = request.execute('')
1545
self.assertEqual(SmartServerResponse(('ok',)), response)
1546
repo._pack_collection.reload_pack_names()
1547
self.assertEqual(9, len(repo._pack_collection.names()))
1549
def test_autopack_on_nonpack_format(self):
1550
"""A request to autopack a non-pack repo is a no-op."""
1551
repo = self.make_repository('.', format='knit')
1552
backing = self.get_transport()
1553
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1555
response = request.execute('')
1556
self.assertEqual(SmartServerResponse(('ok',)), response)
1559
class TestHandlers(tests.TestCase):
1560
"""Tests for the request.request_handlers object."""
1562
def test_all_registrations_exist(self):
1563
"""All registered request_handlers can be found."""
1564
# If there's a typo in a register_lazy call, this loop will fail with
1565
# an AttributeError.
1566
for key, item in smart.request.request_handlers.iteritems():
1569
def assertHandlerEqual(self, verb, handler):
1570
self.assertEqual(smart.request.request_handlers.get(verb), handler)
1572
def test_registered_methods(self):
1573
"""Test that known methods are registered to the correct object."""
1574
self.assertHandlerEqual('Branch.get_config_file',
1575
smart.branch.SmartServerBranchGetConfigFile)
1576
self.assertHandlerEqual('Branch.get_parent',
1577
smart.branch.SmartServerBranchGetParent)
1578
self.assertHandlerEqual('Branch.get_tags_bytes',
1579
smart.branch.SmartServerBranchGetTagsBytes)
1580
self.assertHandlerEqual('Branch.lock_write',
1581
smart.branch.SmartServerBranchRequestLockWrite)
1582
self.assertHandlerEqual('Branch.last_revision_info',
1583
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1584
self.assertHandlerEqual('Branch.revision_history',
1585
smart.branch.SmartServerRequestRevisionHistory)
1586
self.assertHandlerEqual('Branch.set_config_option',
1587
smart.branch.SmartServerBranchRequestSetConfigOption)
1588
self.assertHandlerEqual('Branch.set_last_revision',
1589
smart.branch.SmartServerBranchRequestSetLastRevision)
1590
self.assertHandlerEqual('Branch.set_last_revision_info',
1591
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1592
self.assertHandlerEqual('Branch.set_last_revision_ex',
1593
smart.branch.SmartServerBranchRequestSetLastRevisionEx)
1594
self.assertHandlerEqual('Branch.set_parent_location',
1595
smart.branch.SmartServerBranchRequestSetParentLocation)
1596
self.assertHandlerEqual('Branch.unlock',
1597
smart.branch.SmartServerBranchRequestUnlock)
1598
self.assertHandlerEqual('BzrDir.find_repository',
1599
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1600
self.assertHandlerEqual('BzrDir.find_repositoryV2',
1601
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1602
self.assertHandlerEqual('BzrDirFormat.initialize',
1603
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1604
self.assertHandlerEqual('BzrDirFormat.initialize_ex_1.16',
1605
smart.bzrdir.SmartServerRequestBzrDirInitializeEx)
1606
self.assertHandlerEqual('BzrDir.cloning_metadir',
1607
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1608
self.assertHandlerEqual('BzrDir.get_config_file',
1609
smart.bzrdir.SmartServerBzrDirRequestConfigFile)
1610
self.assertHandlerEqual('BzrDir.open_branch',
1611
smart.bzrdir.SmartServerRequestOpenBranch)
1612
self.assertHandlerEqual('BzrDir.open_branchV2',
1613
smart.bzrdir.SmartServerRequestOpenBranchV2)
1614
self.assertHandlerEqual('PackRepository.autopack',
1615
smart.packrepository.SmartServerPackRepositoryAutopack)
1616
self.assertHandlerEqual('Repository.gather_stats',
1617
smart.repository.SmartServerRepositoryGatherStats)
1618
self.assertHandlerEqual('Repository.get_parent_map',
1619
smart.repository.SmartServerRepositoryGetParentMap)
1620
self.assertHandlerEqual('Repository.get_rev_id_for_revno',
1621
smart.repository.SmartServerRepositoryGetRevIdForRevno)
1622
self.assertHandlerEqual('Repository.get_revision_graph',
1623
smart.repository.SmartServerRepositoryGetRevisionGraph)
1624
self.assertHandlerEqual('Repository.get_stream',
1625
smart.repository.SmartServerRepositoryGetStream)
1626
self.assertHandlerEqual('Repository.has_revision',
1627
smart.repository.SmartServerRequestHasRevision)
1628
self.assertHandlerEqual('Repository.insert_stream',
1629
smart.repository.SmartServerRepositoryInsertStream)
1630
self.assertHandlerEqual('Repository.insert_stream_locked',
1631
smart.repository.SmartServerRepositoryInsertStreamLocked)
1632
self.assertHandlerEqual('Repository.is_shared',
1633
smart.repository.SmartServerRepositoryIsShared)
1634
self.assertHandlerEqual('Repository.lock_write',
1635
smart.repository.SmartServerRepositoryLockWrite)
1636
self.assertHandlerEqual('Repository.tarball',
1637
smart.repository.SmartServerRepositoryTarball)
1638
self.assertHandlerEqual('Repository.unlock',
1639
smart.repository.SmartServerRepositoryUnlock)
1640
self.assertHandlerEqual('Transport.is_readonly',
1641
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
