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')
1194
parent.add([''], ['TREE_ROOT'])
1195
r1 = parent.commit(message='first commit')
1196
r2 = parent.commit(message='second commit')
1198
local = self.make_branch_and_memory_tree('local', format='1.9')
1199
local.branch.pull(parent.branch)
1200
local.set_parent_ids([r2])
1201
r3 = local.commit(message='local commit')
1202
local.branch.create_clone_on_transport(
1203
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
1205
SmartServerResponse(('history-incomplete', 2, r2)),
1206
request.execute('stacked', 1, (3, r3)))
1208
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
1210
def make_two_commit_repo(self):
1211
tree = self.make_branch_and_memory_tree('.')
1214
r1 = tree.commit('1st commit')
1215
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1217
repo = tree.branch.repository
1220
def test_ancestry_of(self):
1221
"""The search argument may be a 'ancestry-of' some heads'."""
1222
backing = self.get_transport()
1223
request = smart.repository.SmartServerRepositoryGetStream(backing)
1224
repo, r1, r2 = self.make_two_commit_repo()
1225
fetch_spec = ['ancestry-of', r2]
1226
lines = '\n'.join(fetch_spec)
1227
request.execute('', repo._format.network_name())
1228
response = request.do_body(lines)
1229
self.assertEqual(('ok',), response.args)
1230
stream_bytes = ''.join(response.body_stream)
1231
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1233
def test_search(self):
1234
"""The search argument may be a 'search' of some explicit keys."""
1235
backing = self.get_transport()
1236
request = smart.repository.SmartServerRepositoryGetStream(backing)
1237
repo, r1, r2 = self.make_two_commit_repo()
1238
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1239
lines = '\n'.join(fetch_spec)
1240
request.execute('', repo._format.network_name())
1241
response = request.do_body(lines)
1242
self.assertEqual(('ok',), response.args)
1243
stream_bytes = ''.join(response.body_stream)
1244
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1247
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1249
def test_missing_revision(self):
1250
"""For a missing revision, ('no', ) is returned."""
1251
backing = self.get_transport()
1252
request = smart.repository.SmartServerRequestHasRevision(backing)
1253
self.make_repository('.')
1254
self.assertEqual(SmartServerResponse(('no', )),
1255
request.execute('', 'revid'))
1257
def test_present_revision(self):
1258
"""For a present revision, ('yes', ) is returned."""
1259
backing = self.get_transport()
1260
request = smart.repository.SmartServerRequestHasRevision(backing)
1261
tree = self.make_branch_and_memory_tree('.')
1264
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1265
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1267
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1268
self.assertEqual(SmartServerResponse(('yes', )),
1269
request.execute('', rev_id_utf8))
1272
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1274
def test_empty_revid(self):
1275
"""With an empty revid, we get only size an number and revisions"""
1276
backing = self.get_transport()
1277
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1278
repository = self.make_repository('.')
1279
stats = repository.gather_stats()
1280
expected_body = 'revisions: 0\n'
1281
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1282
request.execute('', '', 'no'))
1284
def test_revid_with_committers(self):
1285
"""For a revid we get more infos."""
1286
backing = self.get_transport()
1287
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1288
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1289
tree = self.make_branch_and_memory_tree('.')
1292
# Let's build a predictable result
1293
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1294
tree.commit('a commit', timestamp=654321.4, timezone=0,
1298
stats = tree.branch.repository.gather_stats()
1299
expected_body = ('firstrev: 123456.200 3600\n'
1300
'latestrev: 654321.400 0\n'
1302
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1306
def test_not_empty_repository_with_committers(self):
1307
"""For a revid and requesting committers we get the whole thing."""
1308
backing = self.get_transport()
1309
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1310
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1311
tree = self.make_branch_and_memory_tree('.')
1314
# Let's build a predictable result
1315
tree.commit('a commit', timestamp=123456.2, timezone=3600,
1317
tree.commit('a commit', timestamp=654321.4, timezone=0,
1318
committer='bar', rev_id=rev_id_utf8)
1320
stats = tree.branch.repository.gather_stats()
1322
expected_body = ('committers: 2\n'
1323
'firstrev: 123456.200 3600\n'
1324
'latestrev: 654321.400 0\n'
1326
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1328
rev_id_utf8, 'yes'))
1331
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
1333
def test_is_shared(self):
1334
"""For a shared repository, ('yes', ) is returned."""
1335
backing = self.get_transport()
1336
request = smart.repository.SmartServerRepositoryIsShared(backing)
1337
self.make_repository('.', shared=True)
1338
self.assertEqual(SmartServerResponse(('yes', )),
1339
request.execute('', ))
1341
def test_is_not_shared(self):
1342
"""For a shared repository, ('no', ) is returned."""
1343
backing = self.get_transport()
1344
request = smart.repository.SmartServerRepositoryIsShared(backing)
1345
self.make_repository('.', shared=False)
1346
self.assertEqual(SmartServerResponse(('no', )),
1347
request.execute('', ))
1350
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
1352
def test_lock_write_on_unlocked_repo(self):
1353
backing = self.get_transport()
1354
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1355
repository = self.make_repository('.', format='knit')
1356
response = request.execute('')
1357
nonce = repository.control_files._lock.peek().get('nonce')
1358
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
1359
# The repository is now locked. Verify that with a new repository
1361
new_repo = repository.bzrdir.open_repository()
1362
self.assertRaises(errors.LockContention, new_repo.lock_write)
1364
request = smart.repository.SmartServerRepositoryUnlock(backing)
1365
response = request.execute('', nonce)
1367
def test_lock_write_on_locked_repo(self):
1368
backing = self.get_transport()
1369
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1370
repository = self.make_repository('.', format='knit')
1371
repo_token = repository.lock_write()
1372
repository.leave_lock_in_place()
1374
response = request.execute('')
1376
SmartServerResponse(('LockContention',)), response)
1378
repository.lock_write(repo_token)
1379
repository.dont_leave_lock_in_place()
1382
def test_lock_write_on_readonly_transport(self):
1383
backing = self.get_readonly_transport()
1384
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1385
repository = self.make_repository('.', format='knit')
1386
response = request.execute('')
1387
self.assertFalse(response.is_successful())
1388
self.assertEqual('LockFailed', response.args[0])
1391
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
1393
def make_empty_byte_stream(self, repo):
1394
byte_stream = smart.repository._stream_to_byte_stream([], repo._format)
1395
return ''.join(byte_stream)
1398
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
1400
def test_insert_stream_empty(self):
1401
backing = self.get_transport()
1402
request = smart.repository.SmartServerRepositoryInsertStream(backing)
1403
repository = self.make_repository('.')
1404
response = request.execute('', '')
1405
self.assertEqual(None, response)
1406
response = request.do_chunk(self.make_empty_byte_stream(repository))
1407
self.assertEqual(None, response)
1408
response = request.do_end()
1409
self.assertEqual(SmartServerResponse(('ok', )), response)
1412
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
1414
def test_insert_stream_empty(self):
1415
backing = self.get_transport()
1416
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1418
repository = self.make_repository('.', format='knit')
1419
lock_token = repository.lock_write()
1420
response = request.execute('', '', lock_token)
1421
self.assertEqual(None, response)
1422
response = request.do_chunk(self.make_empty_byte_stream(repository))
1423
self.assertEqual(None, response)
1424
response = request.do_end()
1425
self.assertEqual(SmartServerResponse(('ok', )), response)
1428
def test_insert_stream_with_wrong_lock_token(self):
1429
backing = self.get_transport()
1430
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1432
repository = self.make_repository('.', format='knit')
1433
lock_token = repository.lock_write()
1435
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
1439
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1442
tests.TestCaseWithMemoryTransport.setUp(self)
1444
def test_unlock_on_locked_repo(self):
1445
backing = self.get_transport()
1446
request = smart.repository.SmartServerRepositoryUnlock(backing)
1447
repository = self.make_repository('.', format='knit')
1448
token = repository.lock_write()
1449
repository.leave_lock_in_place()
1451
response = request.execute('', token)
1453
SmartServerResponse(('ok',)), response)
1454
# The repository is now unlocked. Verify that with a new repository
1456
new_repo = repository.bzrdir.open_repository()
1457
new_repo.lock_write()
1460
def test_unlock_on_unlocked_repo(self):
1461
backing = self.get_transport()
1462
request = smart.repository.SmartServerRepositoryUnlock(backing)
1463
repository = self.make_repository('.', format='knit')
1464
response = request.execute('', 'some token')
1466
SmartServerResponse(('TokenMismatch',)), response)
1469
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1471
def test_is_readonly_no(self):
1472
backing = self.get_transport()
1473
request = smart.request.SmartServerIsReadonly(backing)
1474
response = request.execute()
1476
SmartServerResponse(('no',)), response)
1478
def test_is_readonly_yes(self):
1479
backing = self.get_readonly_transport()
1480
request = smart.request.SmartServerIsReadonly(backing)
1481
response = request.execute()
1483
SmartServerResponse(('yes',)), response)
1486
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1488
def test_set_false(self):
1489
backing = self.get_transport()
1490
repo = self.make_repository('.', shared=True)
1491
repo.set_make_working_trees(True)
1492
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1493
request = request_class(backing)
1494
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1495
request.execute('', 'False'))
1496
repo = repo.bzrdir.open_repository()
1497
self.assertFalse(repo.make_working_trees())
1499
def test_set_true(self):
1500
backing = self.get_transport()
1501
repo = self.make_repository('.', shared=True)
1502
repo.set_make_working_trees(False)
1503
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1504
request = request_class(backing)
1505
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1506
request.execute('', 'True'))
1507
repo = repo.bzrdir.open_repository()
1508
self.assertTrue(repo.make_working_trees())
1511
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1513
def make_repo_needing_autopacking(self, path='.'):
1514
# Make a repo in need of autopacking.
1515
tree = self.make_branch_and_tree('.', format='pack-0.92')
1516
repo = tree.branch.repository
1517
# monkey-patch the pack collection to disable autopacking
1518
repo._pack_collection._max_pack_count = lambda count: count
1520
tree.commit('commit %s' % x)
1521
self.assertEqual(10, len(repo._pack_collection.names()))
1522
del repo._pack_collection._max_pack_count
1525
def test_autopack_needed(self):
1526
repo = self.make_repo_needing_autopacking()
1528
self.addCleanup(repo.unlock)
1529
backing = self.get_transport()
1530
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1532
response = request.execute('')
1533
self.assertEqual(SmartServerResponse(('ok',)), response)
1534
repo._pack_collection.reload_pack_names()
1535
self.assertEqual(1, len(repo._pack_collection.names()))
1537
def test_autopack_not_needed(self):
1538
tree = self.make_branch_and_tree('.', format='pack-0.92')
1539
repo = tree.branch.repository
1541
self.addCleanup(repo.unlock)
1543
tree.commit('commit %s' % x)
1544
backing = self.get_transport()
1545
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1547
response = request.execute('')
1548
self.assertEqual(SmartServerResponse(('ok',)), response)
1549
repo._pack_collection.reload_pack_names()
1550
self.assertEqual(9, len(repo._pack_collection.names()))
1552
def test_autopack_on_nonpack_format(self):
1553
"""A request to autopack a non-pack repo is a no-op."""
1554
repo = self.make_repository('.', format='knit')
1555
backing = self.get_transport()
1556
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1558
response = request.execute('')
1559
self.assertEqual(SmartServerResponse(('ok',)), response)
1562
class TestHandlers(tests.TestCase):
1563
"""Tests for the request.request_handlers object."""
1565
def test_all_registrations_exist(self):
1566
"""All registered request_handlers can be found."""
1567
# If there's a typo in a register_lazy call, this loop will fail with
1568
# an AttributeError.
1569
for key, item in smart.request.request_handlers.iteritems():
1572
def assertHandlerEqual(self, verb, handler):
1573
self.assertEqual(smart.request.request_handlers.get(verb), handler)
1575
def test_registered_methods(self):
1576
"""Test that known methods are registered to the correct object."""
1577
self.assertHandlerEqual('Branch.get_config_file',
1578
smart.branch.SmartServerBranchGetConfigFile)
1579
self.assertHandlerEqual('Branch.get_parent',
1580
smart.branch.SmartServerBranchGetParent)
1581
self.assertHandlerEqual('Branch.get_tags_bytes',
1582
smart.branch.SmartServerBranchGetTagsBytes)
1583
self.assertHandlerEqual('Branch.lock_write',
1584
smart.branch.SmartServerBranchRequestLockWrite)
1585
self.assertHandlerEqual('Branch.last_revision_info',
1586
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1587
self.assertHandlerEqual('Branch.revision_history',
1588
smart.branch.SmartServerRequestRevisionHistory)
1589
self.assertHandlerEqual('Branch.set_config_option',
1590
smart.branch.SmartServerBranchRequestSetConfigOption)
1591
self.assertHandlerEqual('Branch.set_last_revision',
1592
smart.branch.SmartServerBranchRequestSetLastRevision)
1593
self.assertHandlerEqual('Branch.set_last_revision_info',
1594
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1595
self.assertHandlerEqual('Branch.set_last_revision_ex',
1596
smart.branch.SmartServerBranchRequestSetLastRevisionEx)
1597
self.assertHandlerEqual('Branch.set_parent_location',
1598
smart.branch.SmartServerBranchRequestSetParentLocation)
1599
self.assertHandlerEqual('Branch.unlock',
1600
smart.branch.SmartServerBranchRequestUnlock)
1601
self.assertHandlerEqual('BzrDir.find_repository',
1602
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1603
self.assertHandlerEqual('BzrDir.find_repositoryV2',
1604
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1605
self.assertHandlerEqual('BzrDirFormat.initialize',
1606
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1607
self.assertHandlerEqual('BzrDirFormat.initialize_ex_1.16',
1608
smart.bzrdir.SmartServerRequestBzrDirInitializeEx)
1609
self.assertHandlerEqual('BzrDir.cloning_metadir',
1610
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1611
self.assertHandlerEqual('BzrDir.get_config_file',
1612
smart.bzrdir.SmartServerBzrDirRequestConfigFile)
1613
self.assertHandlerEqual('BzrDir.open_branch',
1614
smart.bzrdir.SmartServerRequestOpenBranch)
1615
self.assertHandlerEqual('BzrDir.open_branchV2',
1616
smart.bzrdir.SmartServerRequestOpenBranchV2)
1617
self.assertHandlerEqual('PackRepository.autopack',
1618
smart.packrepository.SmartServerPackRepositoryAutopack)
1619
self.assertHandlerEqual('Repository.gather_stats',
1620
smart.repository.SmartServerRepositoryGatherStats)
1621
self.assertHandlerEqual('Repository.get_parent_map',
1622
smart.repository.SmartServerRepositoryGetParentMap)
1623
self.assertHandlerEqual('Repository.get_rev_id_for_revno',
1624
smart.repository.SmartServerRepositoryGetRevIdForRevno)
1625
self.assertHandlerEqual('Repository.get_revision_graph',
1626
smart.repository.SmartServerRepositoryGetRevisionGraph)
1627
self.assertHandlerEqual('Repository.get_stream',
1628
smart.repository.SmartServerRepositoryGetStream)
1629
self.assertHandlerEqual('Repository.has_revision',
1630
smart.repository.SmartServerRequestHasRevision)
1631
self.assertHandlerEqual('Repository.insert_stream',
1632
smart.repository.SmartServerRepositoryInsertStream)
1633
self.assertHandlerEqual('Repository.insert_stream_locked',
1634
smart.repository.SmartServerRepositoryInsertStreamLocked)
1635
self.assertHandlerEqual('Repository.is_shared',
1636
smart.repository.SmartServerRepositoryIsShared)
1637
self.assertHandlerEqual('Repository.lock_write',
1638
smart.repository.SmartServerRepositoryLockWrite)
1639
self.assertHandlerEqual('Repository.tarball',
1640
smart.repository.SmartServerRepositoryTarball)
1641
self.assertHandlerEqual('Repository.unlock',
1642
smart.repository.SmartServerRepositoryUnlock)
1643
self.assertHandlerEqual('Transport.is_readonly',
1644
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
