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.start_server(self._chroot_server)
91
t = get_transport(self._chroot_server.get_url())
92
if relpath is not None:
97
class TestCaseWithSmartMedium(tests.TestCaseWithTransport):
100
super(TestCaseWithSmartMedium, self).setUp()
101
# We're allowed to set the transport class here, so that we don't use
102
# the default or a parameterized class, but rather use the
103
# TestCaseWithTransport infrastructure to set up a smart server and
105
self.transport_server = self.make_transport_server
107
def make_transport_server(self):
108
return smart.server.SmartTCPServer_for_testing('-' + self.id())
110
def get_smart_medium(self):
111
"""Get a smart medium to use in tests."""
112
return self.get_transport().get_smart_medium()
115
class TestSmartServerResponse(tests.TestCase):
117
def test__eq__(self):
118
self.assertEqual(SmartServerResponse(('ok', )),
119
SmartServerResponse(('ok', )))
120
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
121
SmartServerResponse(('ok', ), 'body'))
122
self.assertNotEqual(SmartServerResponse(('ok', )),
123
SmartServerResponse(('notok', )))
124
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
125
SmartServerResponse(('ok', )))
126
self.assertNotEqual(None,
127
SmartServerResponse(('ok', )))
129
def test__str__(self):
130
"""SmartServerResponses can be stringified."""
132
"<SuccessfulSmartServerResponse args=('args',) body='body'>",
133
str(SuccessfulSmartServerResponse(('args',), 'body')))
135
"<FailedSmartServerResponse args=('args',) body='body'>",
136
str(FailedSmartServerResponse(('args',), 'body')))
139
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
141
def test_translate_client_path(self):
142
transport = self.get_transport()
143
request = SmartServerRequest(transport, 'foo/')
144
self.assertEqual('./', request.translate_client_path('foo/'))
146
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
148
errors.PathNotChild, request.translate_client_path, '/')
150
errors.PathNotChild, request.translate_client_path, 'bar/')
151
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
153
def test_transport_from_client_path(self):
154
transport = self.get_transport()
155
request = SmartServerRequest(transport, 'foo/')
158
request.transport_from_client_path('foo/').base)
161
class TestSmartServerBzrDirRequestCloningMetaDir(
162
tests.TestCaseWithMemoryTransport):
163
"""Tests for BzrDir.cloning_metadir."""
165
def test_cloning_metadir(self):
166
"""When there is a bzrdir present, the call succeeds."""
167
backing = self.get_transport()
168
dir = self.make_bzrdir('.')
169
local_result = dir.cloning_metadir()
170
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
171
request = request_class(backing)
172
expected = SuccessfulSmartServerResponse(
173
(local_result.network_name(),
174
local_result.repository_format.network_name(),
175
('branch', local_result.get_branch_format().network_name())))
176
self.assertEqual(expected, request.execute('', 'False'))
178
def test_cloning_metadir_reference(self):
179
"""The request fails when bzrdir contains a branch reference."""
180
backing = self.get_transport()
181
referenced_branch = self.make_branch('referenced')
182
dir = self.make_bzrdir('.')
183
local_result = dir.cloning_metadir()
184
reference = BranchReferenceFormat().initialize(dir, referenced_branch)
185
reference_url = BranchReferenceFormat().get_reference(dir)
186
# The server shouldn't try to follow the branch reference, so it's fine
187
# if the referenced branch isn't reachable.
188
backing.rename('referenced', 'moved')
189
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
190
request = request_class(backing)
191
expected = FailedSmartServerResponse(('BranchReference',))
192
self.assertEqual(expected, request.execute('', 'False'))
195
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
196
"""Tests for BzrDir.create_repository."""
198
def test_makes_repository(self):
199
"""When there is a bzrdir present, the call succeeds."""
200
backing = self.get_transport()
201
self.make_bzrdir('.')
202
request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
203
request = request_class(backing)
204
reference_bzrdir_format = bzrdir.format_registry.get('pack-0.92')()
205
reference_format = reference_bzrdir_format.repository_format
206
network_name = reference_format.network_name()
207
expected = SuccessfulSmartServerResponse(
208
('ok', 'no', 'no', 'no', network_name))
209
self.assertEqual(expected, request.execute('', network_name, 'True'))
212
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
213
"""Tests for BzrDir.find_repository."""
215
def test_no_repository(self):
216
"""When there is no repository to be found, ('norepository', ) is returned."""
217
backing = self.get_transport()
218
request = self._request_class(backing)
219
self.make_bzrdir('.')
220
self.assertEqual(SmartServerResponse(('norepository', )),
223
def test_nonshared_repository(self):
224
# nonshared repositorys only allow 'find' to return a handle when the
225
# path the repository is being searched on is the same as that that
226
# the repository is at.
227
backing = self.get_transport()
228
request = self._request_class(backing)
229
result = self._make_repository_and_result()
230
self.assertEqual(result, request.execute(''))
231
self.make_bzrdir('subdir')
232
self.assertEqual(SmartServerResponse(('norepository', )),
233
request.execute('subdir'))
235
def _make_repository_and_result(self, shared=False, format=None):
236
"""Convenience function to setup a repository.
238
:result: The SmartServerResponse to expect when opening it.
240
repo = self.make_repository('.', shared=shared, format=format)
241
if repo.supports_rich_root():
245
if repo._format.supports_tree_reference:
249
if repo._format.supports_external_lookups:
253
if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
254
self._request_class):
255
return SuccessfulSmartServerResponse(
256
('ok', '', rich_root, subtrees, external,
257
repo._format.network_name()))
258
elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
259
self._request_class):
260
# All tests so far are on formats, and for non-external
262
return SuccessfulSmartServerResponse(
263
('ok', '', rich_root, subtrees, external))
265
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
267
def test_shared_repository(self):
268
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
269
backing = self.get_transport()
270
request = self._request_class(backing)
271
result = self._make_repository_and_result(shared=True)
272
self.assertEqual(result, request.execute(''))
273
self.make_bzrdir('subdir')
274
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
275
self.assertEqual(result2,
276
request.execute('subdir'))
277
self.make_bzrdir('subdir/deeper')
278
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
279
self.assertEqual(result3,
280
request.execute('subdir/deeper'))
282
def test_rich_root_and_subtree_encoding(self):
283
"""Test for the format attributes for rich root and subtree support."""
284
backing = self.get_transport()
285
request = self._request_class(backing)
286
result = self._make_repository_and_result(format='dirstate-with-subtree')
287
# check the test will be valid
288
self.assertEqual('yes', result.args[2])
289
self.assertEqual('yes', result.args[3])
290
self.assertEqual(result, request.execute(''))
292
def test_supports_external_lookups_no_v2(self):
293
"""Test for the supports_external_lookups attribute."""
294
backing = self.get_transport()
295
request = self._request_class(backing)
296
result = self._make_repository_and_result(format='dirstate-with-subtree')
297
# check the test will be valid
298
self.assertEqual('no', result.args[4])
299
self.assertEqual(result, request.execute(''))
302
class TestSmartServerBzrDirRequestGetConfigFile(
303
tests.TestCaseWithMemoryTransport):
304
"""Tests for BzrDir.get_config_file."""
306
def test_present(self):
307
backing = self.get_transport()
308
dir = self.make_bzrdir('.')
309
dir.get_config().set_default_stack_on("/")
310
local_result = dir._get_config()._get_config_file().read()
311
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
312
request = request_class(backing)
313
expected = SuccessfulSmartServerResponse((), local_result)
314
self.assertEqual(expected, request.execute(''))
316
def test_missing(self):
317
backing = self.get_transport()
318
dir = self.make_bzrdir('.')
319
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
320
request = request_class(backing)
321
expected = SuccessfulSmartServerResponse((), '')
322
self.assertEqual(expected, request.execute(''))
325
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
327
def test_empty_dir(self):
328
"""Initializing an empty dir should succeed and do it."""
329
backing = self.get_transport()
330
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
331
self.assertEqual(SmartServerResponse(('ok', )),
333
made_dir = bzrdir.BzrDir.open_from_transport(backing)
334
# no branch, tree or repository is expected with the current
336
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
337
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
338
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
340
def test_missing_dir(self):
341
"""Initializing a missing directory should fail like the bzrdir api."""
342
backing = self.get_transport()
343
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
344
self.assertRaises(errors.NoSuchFile,
345
request.execute, 'subdir')
347
def test_initialized_dir(self):
348
"""Initializing an extant bzrdir should fail like the bzrdir api."""
349
backing = self.get_transport()
350
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
351
self.make_bzrdir('subdir')
352
self.assertRaises(errors.FileExists,
353
request.execute, 'subdir')
356
class TestSmartServerRequestBzrDirInitializeEx(tests.TestCaseWithMemoryTransport):
357
"""Basic tests for BzrDir.initialize_ex_1.16 in the smart server.
359
The main unit tests in test_bzrdir exercise the API comprehensively.
362
def test_empty_dir(self):
363
"""Initializing an empty dir should succeed and do it."""
364
backing = self.get_transport()
365
name = self.make_bzrdir('reference')._format.network_name()
366
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
367
self.assertEqual(SmartServerResponse(('', '', '', '', '', '', name,
368
'False', '', '', '')),
369
request.execute(name, '', 'True', 'False', 'False', '', '', '', '',
371
made_dir = bzrdir.BzrDir.open_from_transport(backing)
372
# no branch, tree or repository is expected with the current
374
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
375
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
376
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
378
def test_missing_dir(self):
379
"""Initializing a missing directory should fail like the bzrdir api."""
380
backing = self.get_transport()
381
name = self.make_bzrdir('reference')._format.network_name()
382
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
383
self.assertRaises(errors.NoSuchFile, request.execute, name,
384
'subdir/dir', 'False', 'False', 'False', '', '', '', '', 'False')
386
def test_initialized_dir(self):
387
"""Initializing an extant directory should fail like the bzrdir api."""
388
backing = self.get_transport()
389
name = self.make_bzrdir('reference')._format.network_name()
390
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
391
self.make_bzrdir('subdir')
392
self.assertRaises(errors.FileExists, request.execute, name, 'subdir',
393
'False', 'False', 'False', '', '', '', '', 'False')
396
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
398
def test_no_branch(self):
399
"""When there is no branch, ('nobranch', ) is returned."""
400
backing = self.get_transport()
401
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
402
self.make_bzrdir('.')
403
self.assertEqual(SmartServerResponse(('nobranch', )),
406
def test_branch(self):
407
"""When there is a branch, 'ok' is returned."""
408
backing = self.get_transport()
409
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
410
self.make_branch('.')
411
self.assertEqual(SmartServerResponse(('ok', '')),
414
def test_branch_reference(self):
415
"""When there is a branch reference, the reference URL is returned."""
416
backing = self.get_transport()
417
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
418
branch = self.make_branch('branch')
419
checkout = branch.create_checkout('reference',lightweight=True)
420
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
421
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
422
self.assertEqual(SmartServerResponse(('ok', reference_url)),
423
request.execute('reference'))
426
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
428
def test_no_branch(self):
429
"""When there is no branch, ('nobranch', ) is returned."""
430
backing = self.get_transport()
431
self.make_bzrdir('.')
432
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
433
self.assertEqual(SmartServerResponse(('nobranch', )),
436
def test_branch(self):
437
"""When there is a branch, 'ok' is returned."""
438
backing = self.get_transport()
439
expected = self.make_branch('.')._format.network_name()
440
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
441
self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),
444
def test_branch_reference(self):
445
"""When there is a branch reference, the reference URL is returned."""
446
backing = self.get_transport()
447
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
448
branch = self.make_branch('branch')
449
checkout = branch.create_checkout('reference',lightweight=True)
450
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
451
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
452
self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
453
request.execute('reference'))
455
def test_stacked_branch(self):
456
"""Opening a stacked branch does not open the stacked-on branch."""
457
trunk = self.make_branch('trunk')
458
feature = self.make_branch('feature')
459
feature.set_stacked_on_url(trunk.base)
461
Branch.hooks.install_named_hook('open', opened_branches.append, None)
462
backing = self.get_transport()
463
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
466
response = request.execute('feature')
468
request.teardown_jail()
469
expected_format = feature._format.network_name()
471
SuccessfulSmartServerResponse(('branch', expected_format)),
473
self.assertLength(1, opened_branches)
476
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
478
def test_empty(self):
479
"""For an empty branch, the body is empty."""
480
backing = self.get_transport()
481
request = smart.branch.SmartServerRequestRevisionHistory(backing)
482
self.make_branch('.')
483
self.assertEqual(SmartServerResponse(('ok', ), ''),
486
def test_not_empty(self):
487
"""For a non-empty branch, the body is empty."""
488
backing = self.get_transport()
489
request = smart.branch.SmartServerRequestRevisionHistory(backing)
490
tree = self.make_branch_and_memory_tree('.')
493
r1 = tree.commit('1st commit')
494
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
497
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
501
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
503
def test_no_branch(self):
504
"""When there is a bzrdir and no branch, NotBranchError is raised."""
505
backing = self.get_transport()
506
request = smart.branch.SmartServerBranchRequest(backing)
507
self.make_bzrdir('.')
508
self.assertRaises(errors.NotBranchError,
511
def test_branch_reference(self):
512
"""When there is a branch reference, NotBranchError is raised."""
513
backing = self.get_transport()
514
request = smart.branch.SmartServerBranchRequest(backing)
515
branch = self.make_branch('branch')
516
checkout = branch.create_checkout('reference',lightweight=True)
517
self.assertRaises(errors.NotBranchError,
518
request.execute, 'checkout')
521
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
523
def test_empty(self):
524
"""For an empty branch, the result is ('ok', '0', 'null:')."""
525
backing = self.get_transport()
526
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
527
self.make_branch('.')
528
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
531
def test_not_empty(self):
532
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
533
backing = self.get_transport()
534
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
535
tree = self.make_branch_and_memory_tree('.')
538
rev_id_utf8 = u'\xc8'.encode('utf-8')
539
r1 = tree.commit('1st commit')
540
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
543
SmartServerResponse(('ok', '2', rev_id_utf8)),
547
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
549
def test_default(self):
550
"""With no file, we get empty content."""
551
backing = self.get_transport()
552
request = smart.branch.SmartServerBranchGetConfigFile(backing)
553
branch = self.make_branch('.')
554
# there should be no file by default
556
self.assertEqual(SmartServerResponse(('ok', ), content),
559
def test_with_content(self):
560
# SmartServerBranchGetConfigFile should return the content from
561
# branch.control_files.get('branch.conf') for now - in the future it may
562
# perform more complex processing.
563
backing = self.get_transport()
564
request = smart.branch.SmartServerBranchGetConfigFile(backing)
565
branch = self.make_branch('.')
566
branch._transport.put_bytes('branch.conf', 'foo bar baz')
567
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
571
class TestLockedBranch(tests.TestCaseWithMemoryTransport):
573
def get_lock_tokens(self, branch):
574
branch_token = branch.lock_write()
575
repo_token = branch.repository.lock_write()
576
branch.repository.unlock()
577
return branch_token, repo_token
580
class TestSmartServerBranchRequestSetConfigOption(TestLockedBranch):
582
def test_value_name(self):
583
branch = self.make_branch('.')
584
request = smart.branch.SmartServerBranchRequestSetConfigOption(
585
branch.bzrdir.root_transport)
586
branch_token, repo_token = self.get_lock_tokens(branch)
587
config = branch._get_config()
588
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
590
self.assertEqual(SuccessfulSmartServerResponse(()), result)
591
self.assertEqual('bar', config.get_option('foo'))
595
def test_value_name_section(self):
596
branch = self.make_branch('.')
597
request = smart.branch.SmartServerBranchRequestSetConfigOption(
598
branch.bzrdir.root_transport)
599
branch_token, repo_token = self.get_lock_tokens(branch)
600
config = branch._get_config()
601
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
603
self.assertEqual(SuccessfulSmartServerResponse(()), result)
604
self.assertEqual('bar', config.get_option('foo', 'gam'))
609
class TestSmartServerBranchRequestSetTagsBytes(TestLockedBranch):
610
# Only called when the branch format and tags match [yay factory
611
# methods] so only need to test straight forward cases.
613
def test_set_bytes(self):
614
base_branch = self.make_branch('base')
615
tag_bytes = base_branch._get_tags_bytes()
616
# get_lock_tokens takes out a lock.
617
branch_token, repo_token = self.get_lock_tokens(base_branch)
618
request = smart.branch.SmartServerBranchSetTagsBytes(
619
self.get_transport())
620
response = request.execute('base', branch_token, repo_token)
621
self.assertEqual(None, response)
622
response = request.do_chunk(tag_bytes)
623
self.assertEqual(None, response)
624
response = request.do_end()
626
SuccessfulSmartServerResponse(()), response)
629
def test_lock_failed(self):
630
base_branch = self.make_branch('base')
631
base_branch.lock_write()
632
tag_bytes = base_branch._get_tags_bytes()
633
request = smart.branch.SmartServerBranchSetTagsBytes(
634
self.get_transport())
635
self.assertRaises(errors.TokenMismatch, request.execute,
636
'base', 'wrong token', 'wrong token')
637
# The request handler will keep processing the message parts, so even
638
# if the request fails immediately do_chunk and do_end are still
640
request.do_chunk(tag_bytes)
646
class SetLastRevisionTestBase(TestLockedBranch):
647
"""Base test case for verbs that implement set_last_revision."""
650
tests.TestCaseWithMemoryTransport.setUp(self)
651
backing_transport = self.get_transport()
652
self.request = self.request_class(backing_transport)
653
self.tree = self.make_branch_and_memory_tree('.')
655
def lock_branch(self):
656
return self.get_lock_tokens(self.tree.branch)
658
def unlock_branch(self):
659
self.tree.branch.unlock()
661
def set_last_revision(self, revision_id, revno):
662
branch_token, repo_token = self.lock_branch()
663
response = self._set_last_revision(
664
revision_id, revno, branch_token, repo_token)
668
def assertRequestSucceeds(self, revision_id, revno):
669
response = self.set_last_revision(revision_id, revno)
670
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
673
class TestSetLastRevisionVerbMixin(object):
674
"""Mixin test case for verbs that implement set_last_revision."""
676
def test_set_null_to_null(self):
677
"""An empty branch can have its last revision set to 'null:'."""
678
self.assertRequestSucceeds('null:', 0)
680
def test_NoSuchRevision(self):
681
"""If the revision_id is not present, the verb returns NoSuchRevision.
683
revision_id = 'non-existent revision'
685
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
686
self.set_last_revision(revision_id, 1))
688
def make_tree_with_two_commits(self):
689
self.tree.lock_write()
691
rev_id_utf8 = u'\xc8'.encode('utf-8')
692
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
693
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
696
def test_branch_last_revision_info_is_updated(self):
697
"""A branch's tip can be set to a revision that is present in its
700
# Make a branch with an empty revision history, but two revisions in
702
self.make_tree_with_two_commits()
703
rev_id_utf8 = u'\xc8'.encode('utf-8')
704
self.tree.branch.set_revision_history([])
706
(0, 'null:'), self.tree.branch.last_revision_info())
707
# We can update the branch to a revision that is present in the
709
self.assertRequestSucceeds(rev_id_utf8, 1)
711
(1, rev_id_utf8), self.tree.branch.last_revision_info())
713
def test_branch_last_revision_info_rewind(self):
714
"""A branch's tip can be set to a revision that is an ancestor of the
717
self.make_tree_with_two_commits()
718
rev_id_utf8 = u'\xc8'.encode('utf-8')
720
(2, 'rev-2'), self.tree.branch.last_revision_info())
721
self.assertRequestSucceeds(rev_id_utf8, 1)
723
(1, rev_id_utf8), self.tree.branch.last_revision_info())
725
def test_TipChangeRejected(self):
726
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
727
returns TipChangeRejected.
729
rejection_message = u'rejection message\N{INTERROBANG}'
730
def hook_that_rejects(params):
731
raise errors.TipChangeRejected(rejection_message)
732
Branch.hooks.install_named_hook(
733
'pre_change_branch_tip', hook_that_rejects, None)
735
FailedSmartServerResponse(
736
('TipChangeRejected', rejection_message.encode('utf-8'))),
737
self.set_last_revision('null:', 0))
740
class TestSmartServerBranchRequestSetLastRevision(
741
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
742
"""Tests for Branch.set_last_revision verb."""
744
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
746
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
747
return self.request.execute(
748
'', branch_token, repo_token, revision_id)
751
class TestSmartServerBranchRequestSetLastRevisionInfo(
752
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
753
"""Tests for Branch.set_last_revision_info verb."""
755
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
757
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
758
return self.request.execute(
759
'', branch_token, repo_token, revno, revision_id)
761
def test_NoSuchRevision(self):
762
"""Branch.set_last_revision_info does not have to return
763
NoSuchRevision if the revision_id is absent.
765
raise tests.TestNotApplicable()
768
class TestSmartServerBranchRequestSetLastRevisionEx(
769
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
770
"""Tests for Branch.set_last_revision_ex verb."""
772
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
774
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
775
return self.request.execute(
776
'', branch_token, repo_token, revision_id, 0, 0)
778
def assertRequestSucceeds(self, revision_id, revno):
779
response = self.set_last_revision(revision_id, revno)
781
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
784
def test_branch_last_revision_info_rewind(self):
785
"""A branch's tip can be set to a revision that is an ancestor of the
786
current tip, but only if allow_overwrite_descendant is passed.
788
self.make_tree_with_two_commits()
789
rev_id_utf8 = u'\xc8'.encode('utf-8')
791
(2, 'rev-2'), self.tree.branch.last_revision_info())
792
# If allow_overwrite_descendant flag is 0, then trying to set the tip
793
# to an older revision ID has no effect.
794
branch_token, repo_token = self.lock_branch()
795
response = self.request.execute(
796
'', branch_token, repo_token, rev_id_utf8, 0, 0)
798
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
801
(2, 'rev-2'), self.tree.branch.last_revision_info())
803
# If allow_overwrite_descendant flag is 1, then setting the tip to an
805
response = self.request.execute(
806
'', branch_token, repo_token, rev_id_utf8, 0, 1)
808
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
812
(1, rev_id_utf8), self.tree.branch.last_revision_info())
814
def make_branch_with_divergent_history(self):
815
"""Make a branch with divergent history in its repo.
817
The branch's tip will be 'child-2', and the repo will also contain
818
'child-1', which diverges from a common base revision.
820
self.tree.lock_write()
822
r1 = self.tree.commit('1st commit')
823
revno_1, revid_1 = self.tree.branch.last_revision_info()
824
r2 = self.tree.commit('2nd commit', rev_id='child-1')
825
# Undo the second commit
826
self.tree.branch.set_last_revision_info(revno_1, revid_1)
827
self.tree.set_parent_ids([revid_1])
828
# Make a new second commit, child-2. child-2 has diverged from
830
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
833
def test_not_allow_diverged(self):
834
"""If allow_diverged is not passed, then setting a divergent history
835
returns a Diverged error.
837
self.make_branch_with_divergent_history()
839
FailedSmartServerResponse(('Diverged',)),
840
self.set_last_revision('child-1', 2))
841
# The branch tip was not changed.
842
self.assertEqual('child-2', self.tree.branch.last_revision())
844
def test_allow_diverged(self):
845
"""If allow_diverged is passed, then setting a divergent history
848
self.make_branch_with_divergent_history()
849
branch_token, repo_token = self.lock_branch()
850
response = self.request.execute(
851
'', branch_token, repo_token, 'child-1', 1, 0)
853
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
856
# The branch tip was changed.
857
self.assertEqual('child-1', self.tree.branch.last_revision())
860
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
862
def test_get_parent_none(self):
863
base_branch = self.make_branch('base')
864
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
865
response = request.execute('base')
867
SuccessfulSmartServerResponse(('',)), response)
869
def test_get_parent_something(self):
870
base_branch = self.make_branch('base')
871
base_branch.set_parent(self.get_url('foo'))
872
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
873
response = request.execute('base')
875
SuccessfulSmartServerResponse(("../foo",)),
879
class TestSmartServerBranchRequestSetParent(tests.TestCaseWithMemoryTransport):
881
def test_set_parent_none(self):
882
branch = self.make_branch('base', format="1.9")
884
branch._set_parent_location('foo')
886
request = smart.branch.SmartServerBranchRequestSetParentLocation(
887
self.get_transport())
888
branch_token = branch.lock_write()
889
repo_token = branch.repository.lock_write()
891
response = request.execute('base', branch_token, repo_token, '')
893
branch.repository.unlock()
895
self.assertEqual(SuccessfulSmartServerResponse(()), response)
896
self.assertEqual(None, branch.get_parent())
898
def test_set_parent_something(self):
899
branch = self.make_branch('base', format="1.9")
900
request = smart.branch.SmartServerBranchRequestSetParentLocation(
901
self.get_transport())
902
branch_token = branch.lock_write()
903
repo_token = branch.repository.lock_write()
905
response = request.execute('base', branch_token, repo_token,
908
branch.repository.unlock()
910
self.assertEqual(SuccessfulSmartServerResponse(()), response)
911
self.assertEqual('http://bar/', branch.get_parent())
914
class TestSmartServerBranchRequestGetTagsBytes(tests.TestCaseWithMemoryTransport):
915
# Only called when the branch format and tags match [yay factory
916
# methods] so only need to test straight forward cases.
918
def test_get_bytes(self):
919
base_branch = self.make_branch('base')
920
request = smart.branch.SmartServerBranchGetTagsBytes(
921
self.get_transport())
922
response = request.execute('base')
924
SuccessfulSmartServerResponse(('',)), response)
927
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
929
def test_get_stacked_on_url(self):
930
base_branch = self.make_branch('base', format='1.6')
931
stacked_branch = self.make_branch('stacked', format='1.6')
932
# typically should be relative
933
stacked_branch.set_stacked_on_url('../base')
934
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
935
self.get_transport())
936
response = request.execute('stacked')
938
SmartServerResponse(('ok', '../base')),
942
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
945
tests.TestCaseWithMemoryTransport.setUp(self)
947
def test_lock_write_on_unlocked_branch(self):
948
backing = self.get_transport()
949
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
950
branch = self.make_branch('.', format='knit')
951
repository = branch.repository
952
response = request.execute('')
953
branch_nonce = branch.control_files._lock.peek().get('nonce')
954
repository_nonce = repository.control_files._lock.peek().get('nonce')
956
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
958
# The branch (and associated repository) is now locked. Verify that
959
# with a new branch object.
960
new_branch = repository.bzrdir.open_branch()
961
self.assertRaises(errors.LockContention, new_branch.lock_write)
963
request = smart.branch.SmartServerBranchRequestUnlock(backing)
964
response = request.execute('', branch_nonce, repository_nonce)
966
def test_lock_write_on_locked_branch(self):
967
backing = self.get_transport()
968
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
969
branch = self.make_branch('.')
970
branch_token = branch.lock_write()
971
branch.leave_lock_in_place()
973
response = request.execute('')
975
SmartServerResponse(('LockContention',)), response)
977
branch.lock_write(branch_token)
978
branch.dont_leave_lock_in_place()
981
def test_lock_write_with_tokens_on_locked_branch(self):
982
backing = self.get_transport()
983
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
984
branch = self.make_branch('.', format='knit')
985
branch_token = branch.lock_write()
986
repo_token = branch.repository.lock_write()
987
branch.repository.unlock()
988
branch.leave_lock_in_place()
989
branch.repository.leave_lock_in_place()
991
response = request.execute('',
992
branch_token, repo_token)
994
SmartServerResponse(('ok', branch_token, repo_token)), response)
996
branch.repository.lock_write(repo_token)
997
branch.repository.dont_leave_lock_in_place()
998
branch.repository.unlock()
999
branch.lock_write(branch_token)
1000
branch.dont_leave_lock_in_place()
1003
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
1004
backing = self.get_transport()
1005
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1006
branch = self.make_branch('.', format='knit')
1007
branch_token = branch.lock_write()
1008
repo_token = branch.repository.lock_write()
1009
branch.repository.unlock()
1010
branch.leave_lock_in_place()
1011
branch.repository.leave_lock_in_place()
1013
response = request.execute('',
1014
branch_token+'xxx', repo_token)
1016
SmartServerResponse(('TokenMismatch',)), response)
1018
branch.repository.lock_write(repo_token)
1019
branch.repository.dont_leave_lock_in_place()
1020
branch.repository.unlock()
1021
branch.lock_write(branch_token)
1022
branch.dont_leave_lock_in_place()
1025
def test_lock_write_on_locked_repo(self):
1026
backing = self.get_transport()
1027
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1028
branch = self.make_branch('.', format='knit')
1029
repo = branch.repository
1030
repo_token = repo.lock_write()
1031
repo.leave_lock_in_place()
1033
response = request.execute('')
1035
SmartServerResponse(('LockContention',)), response)
1037
repo.lock_write(repo_token)
1038
repo.dont_leave_lock_in_place()
1041
def test_lock_write_on_readonly_transport(self):
1042
backing = self.get_readonly_transport()
1043
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1044
branch = self.make_branch('.')
1045
root = self.get_transport().clone('/')
1046
path = urlutils.relative_url(root.base, self.get_transport().base)
1047
response = request.execute(path)
1048
error_name, lock_str, why_str = response.args
1049
self.assertFalse(response.is_successful())
1050
self.assertEqual('LockFailed', error_name)
1053
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
1056
tests.TestCaseWithMemoryTransport.setUp(self)
1058
def test_unlock_on_locked_branch_and_repo(self):
1059
backing = self.get_transport()
1060
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1061
branch = self.make_branch('.', format='knit')
1063
branch_token = branch.lock_write()
1064
repo_token = branch.repository.lock_write()
1065
branch.repository.unlock()
1066
# Unlock the branch (and repo) object, leaving the physical locks
1068
branch.leave_lock_in_place()
1069
branch.repository.leave_lock_in_place()
1071
response = request.execute('',
1072
branch_token, repo_token)
1074
SmartServerResponse(('ok',)), response)
1075
# The branch is now unlocked. Verify that with a new branch
1077
new_branch = branch.bzrdir.open_branch()
1078
new_branch.lock_write()
1081
def test_unlock_on_unlocked_branch_unlocked_repo(self):
1082
backing = self.get_transport()
1083
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1084
branch = self.make_branch('.', format='knit')
1085
response = request.execute(
1086
'', 'branch token', 'repo token')
1088
SmartServerResponse(('TokenMismatch',)), response)
1090
def test_unlock_on_unlocked_branch_locked_repo(self):
1091
backing = self.get_transport()
1092
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1093
branch = self.make_branch('.', format='knit')
1094
# Lock the repository.
1095
repo_token = branch.repository.lock_write()
1096
branch.repository.leave_lock_in_place()
1097
branch.repository.unlock()
1098
# Issue branch lock_write request on the unlocked branch (with locked
1100
response = request.execute(
1101
'', 'branch token', repo_token)
1103
SmartServerResponse(('TokenMismatch',)), response)
1105
branch.repository.lock_write(repo_token)
1106
branch.repository.dont_leave_lock_in_place()
1107
branch.repository.unlock()
1110
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
1112
def test_no_repository(self):
1113
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
1114
# we test this using a shared repository above the named path,
1115
# thus checking the right search logic is used - that is, that
1116
# its the exact path being looked at and the server is not
1118
backing = self.get_transport()
1119
request = smart.repository.SmartServerRepositoryRequest(backing)
1120
self.make_repository('.', shared=True)
1121
self.make_bzrdir('subdir')
1122
self.assertRaises(errors.NoRepositoryPresent,
1123
request.execute, 'subdir')
1126
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
1128
def test_trivial_bzipped(self):
1129
# This tests that the wire encoding is actually bzipped
1130
backing = self.get_transport()
1131
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1132
tree = self.make_branch_and_memory_tree('.')
1134
self.assertEqual(None,
1135
request.execute('', 'missing-id'))
1136
# Note that it returns a body that is bzipped.
1138
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
1139
request.do_body('\n\n0\n'))
1141
def test_trivial_include_missing(self):
1142
backing = self.get_transport()
1143
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1144
tree = self.make_branch_and_memory_tree('.')
1146
self.assertEqual(None,
1147
request.execute('', 'missing-id', 'include-missing:'))
1149
SuccessfulSmartServerResponse(('ok', ),
1150
bz2.compress('missing:missing-id')),
1151
request.do_body('\n\n0\n'))
1154
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
1156
def test_none_argument(self):
1157
backing = self.get_transport()
1158
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1159
tree = self.make_branch_and_memory_tree('.')
1162
r1 = tree.commit('1st commit')
1163
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1166
# the lines of revision_id->revision_parent_list has no guaranteed
1167
# order coming out of a dict, so sort both our test and response
1168
lines = sorted([' '.join([r2, r1]), r1])
1169
response = request.execute('', '')
1170
response.body = '\n'.join(sorted(response.body.split('\n')))
1173
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
1175
def test_specific_revision_argument(self):
1176
backing = self.get_transport()
1177
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1178
tree = self.make_branch_and_memory_tree('.')
1181
rev_id_utf8 = u'\xc9'.encode('utf-8')
1182
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
1183
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1186
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
1187
request.execute('', rev_id_utf8))
1189
def test_no_such_revision(self):
1190
backing = self.get_transport()
1191
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1192
tree = self.make_branch_and_memory_tree('.')
1195
r1 = tree.commit('1st commit')
1198
# Note that it still returns body (of zero bytes).
1200
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
1201
request.execute('', 'missingrevision'))
1204
class TestSmartServerRepositoryGetRevIdForRevno(tests.TestCaseWithMemoryTransport):
1206
def test_revno_found(self):
1207
backing = self.get_transport()
1208
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1209
tree = self.make_branch_and_memory_tree('.')
1212
rev1_id_utf8 = u'\xc8'.encode('utf-8')
1213
rev2_id_utf8 = u'\xc9'.encode('utf-8')
1214
tree.commit('1st commit', rev_id=rev1_id_utf8)
1215
tree.commit('2nd commit', rev_id=rev2_id_utf8)
1218
self.assertEqual(SmartServerResponse(('ok', rev1_id_utf8)),
1219
request.execute('', 1, (2, rev2_id_utf8)))
1221
def test_known_revid_missing(self):
1222
backing = self.get_transport()
1223
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1224
repo = self.make_repository('.')
1226
FailedSmartServerResponse(('nosuchrevision', 'ghost')),
1227
request.execute('', 1, (2, 'ghost')))
1229
def test_history_incomplete(self):
1230
backing = self.get_transport()
1231
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1232
parent = self.make_branch_and_memory_tree('parent', format='1.9')
1234
parent.add([''], ['TREE_ROOT'])
1235
r1 = parent.commit(message='first commit')
1236
r2 = parent.commit(message='second commit')
1238
local = self.make_branch_and_memory_tree('local', format='1.9')
1239
local.branch.pull(parent.branch)
1240
local.set_parent_ids([r2])
1241
r3 = local.commit(message='local commit')
1242
local.branch.create_clone_on_transport(
1243
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
1245
SmartServerResponse(('history-incomplete', 2, r2)),
1246
request.execute('stacked', 1, (3, r3)))
1249
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
1251
def make_two_commit_repo(self):
1252
tree = self.make_branch_and_memory_tree('.')
1255
r1 = tree.commit('1st commit')
1256
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1258
repo = tree.branch.repository
1261
def test_ancestry_of(self):
1262
"""The search argument may be a 'ancestry-of' some heads'."""
1263
backing = self.get_transport()
1264
request = smart.repository.SmartServerRepositoryGetStream(backing)
1265
repo, r1, r2 = self.make_two_commit_repo()
1266
fetch_spec = ['ancestry-of', r2]
1267
lines = '\n'.join(fetch_spec)
1268
request.execute('', repo._format.network_name())
1269
response = request.do_body(lines)
1270
self.assertEqual(('ok',), response.args)
1271
stream_bytes = ''.join(response.body_stream)
1272
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1274
def test_search(self):
1275
"""The search argument may be a 'search' of some explicit keys."""
1276
backing = self.get_transport()
1277
request = smart.repository.SmartServerRepositoryGetStream(backing)
1278
repo, r1, r2 = self.make_two_commit_repo()
1279
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1280
lines = '\n'.join(fetch_spec)
1281
request.execute('', repo._format.network_name())
1282
response = request.do_body(lines)
1283
self.assertEqual(('ok',), response.args)
1284
stream_bytes = ''.join(response.body_stream)
1285
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1288
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1290
def test_missing_revision(self):
1291
"""For a missing revision, ('no', ) is returned."""
1292
backing = self.get_transport()
1293
request = smart.repository.SmartServerRequestHasRevision(backing)
1294
self.make_repository('.')
1295
self.assertEqual(SmartServerResponse(('no', )),
1296
request.execute('', 'revid'))
1298
def test_present_revision(self):
1299
"""For a present revision, ('yes', ) is returned."""
1300
backing = self.get_transport()
1301
request = smart.repository.SmartServerRequestHasRevision(backing)
1302
tree = self.make_branch_and_memory_tree('.')
1305
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1306
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1308
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1309
self.assertEqual(SmartServerResponse(('yes', )),
1310
request.execute('', rev_id_utf8))
1313
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1315
def test_empty_revid(self):
1316
"""With an empty revid, we get only size an number and revisions"""
1317
backing = self.get_transport()
1318
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1319
repository = self.make_repository('.')
1320
stats = repository.gather_stats()
1321
expected_body = 'revisions: 0\n'
1322
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1323
request.execute('', '', 'no'))
1325
def test_revid_with_committers(self):
1326
"""For a revid we get more infos."""
1327
backing = self.get_transport()
1328
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1329
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1330
tree = self.make_branch_and_memory_tree('.')
1333
# Let's build a predictable result
1334
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1335
tree.commit('a commit', timestamp=654321.4, timezone=0,
1339
stats = tree.branch.repository.gather_stats()
1340
expected_body = ('firstrev: 123456.200 3600\n'
1341
'latestrev: 654321.400 0\n'
1343
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1347
def test_not_empty_repository_with_committers(self):
1348
"""For a revid and requesting committers we get the whole thing."""
1349
backing = self.get_transport()
1350
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1351
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1352
tree = self.make_branch_and_memory_tree('.')
1355
# Let's build a predictable result
1356
tree.commit('a commit', timestamp=123456.2, timezone=3600,
1358
tree.commit('a commit', timestamp=654321.4, timezone=0,
1359
committer='bar', rev_id=rev_id_utf8)
1361
stats = tree.branch.repository.gather_stats()
1363
expected_body = ('committers: 2\n'
1364
'firstrev: 123456.200 3600\n'
1365
'latestrev: 654321.400 0\n'
1367
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1369
rev_id_utf8, 'yes'))
1372
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
1374
def test_is_shared(self):
1375
"""For a shared repository, ('yes', ) is returned."""
1376
backing = self.get_transport()
1377
request = smart.repository.SmartServerRepositoryIsShared(backing)
1378
self.make_repository('.', shared=True)
1379
self.assertEqual(SmartServerResponse(('yes', )),
1380
request.execute('', ))
1382
def test_is_not_shared(self):
1383
"""For a shared repository, ('no', ) is returned."""
1384
backing = self.get_transport()
1385
request = smart.repository.SmartServerRepositoryIsShared(backing)
1386
self.make_repository('.', shared=False)
1387
self.assertEqual(SmartServerResponse(('no', )),
1388
request.execute('', ))
1391
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
1393
def test_lock_write_on_unlocked_repo(self):
1394
backing = self.get_transport()
1395
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1396
repository = self.make_repository('.', format='knit')
1397
response = request.execute('')
1398
nonce = repository.control_files._lock.peek().get('nonce')
1399
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
1400
# The repository is now locked. Verify that with a new repository
1402
new_repo = repository.bzrdir.open_repository()
1403
self.assertRaises(errors.LockContention, new_repo.lock_write)
1405
request = smart.repository.SmartServerRepositoryUnlock(backing)
1406
response = request.execute('', nonce)
1408
def test_lock_write_on_locked_repo(self):
1409
backing = self.get_transport()
1410
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1411
repository = self.make_repository('.', format='knit')
1412
repo_token = repository.lock_write()
1413
repository.leave_lock_in_place()
1415
response = request.execute('')
1417
SmartServerResponse(('LockContention',)), response)
1419
repository.lock_write(repo_token)
1420
repository.dont_leave_lock_in_place()
1423
def test_lock_write_on_readonly_transport(self):
1424
backing = self.get_readonly_transport()
1425
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1426
repository = self.make_repository('.', format='knit')
1427
response = request.execute('')
1428
self.assertFalse(response.is_successful())
1429
self.assertEqual('LockFailed', response.args[0])
1432
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
1434
def make_empty_byte_stream(self, repo):
1435
byte_stream = smart.repository._stream_to_byte_stream([], repo._format)
1436
return ''.join(byte_stream)
1439
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
1441
def test_insert_stream_empty(self):
1442
backing = self.get_transport()
1443
request = smart.repository.SmartServerRepositoryInsertStream(backing)
1444
repository = self.make_repository('.')
1445
response = request.execute('', '')
1446
self.assertEqual(None, response)
1447
response = request.do_chunk(self.make_empty_byte_stream(repository))
1448
self.assertEqual(None, response)
1449
response = request.do_end()
1450
self.assertEqual(SmartServerResponse(('ok', )), response)
1453
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
1455
def test_insert_stream_empty(self):
1456
backing = self.get_transport()
1457
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1459
repository = self.make_repository('.', format='knit')
1460
lock_token = repository.lock_write()
1461
response = request.execute('', '', lock_token)
1462
self.assertEqual(None, response)
1463
response = request.do_chunk(self.make_empty_byte_stream(repository))
1464
self.assertEqual(None, response)
1465
response = request.do_end()
1466
self.assertEqual(SmartServerResponse(('ok', )), response)
1469
def test_insert_stream_with_wrong_lock_token(self):
1470
backing = self.get_transport()
1471
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1473
repository = self.make_repository('.', format='knit')
1474
lock_token = repository.lock_write()
1476
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
1480
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1483
tests.TestCaseWithMemoryTransport.setUp(self)
1485
def test_unlock_on_locked_repo(self):
1486
backing = self.get_transport()
1487
request = smart.repository.SmartServerRepositoryUnlock(backing)
1488
repository = self.make_repository('.', format='knit')
1489
token = repository.lock_write()
1490
repository.leave_lock_in_place()
1492
response = request.execute('', token)
1494
SmartServerResponse(('ok',)), response)
1495
# The repository is now unlocked. Verify that with a new repository
1497
new_repo = repository.bzrdir.open_repository()
1498
new_repo.lock_write()
1501
def test_unlock_on_unlocked_repo(self):
1502
backing = self.get_transport()
1503
request = smart.repository.SmartServerRepositoryUnlock(backing)
1504
repository = self.make_repository('.', format='knit')
1505
response = request.execute('', 'some token')
1507
SmartServerResponse(('TokenMismatch',)), response)
1510
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1512
def test_is_readonly_no(self):
1513
backing = self.get_transport()
1514
request = smart.request.SmartServerIsReadonly(backing)
1515
response = request.execute()
1517
SmartServerResponse(('no',)), response)
1519
def test_is_readonly_yes(self):
1520
backing = self.get_readonly_transport()
1521
request = smart.request.SmartServerIsReadonly(backing)
1522
response = request.execute()
1524
SmartServerResponse(('yes',)), response)
1527
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1529
def test_set_false(self):
1530
backing = self.get_transport()
1531
repo = self.make_repository('.', shared=True)
1532
repo.set_make_working_trees(True)
1533
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1534
request = request_class(backing)
1535
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1536
request.execute('', 'False'))
1537
repo = repo.bzrdir.open_repository()
1538
self.assertFalse(repo.make_working_trees())
1540
def test_set_true(self):
1541
backing = self.get_transport()
1542
repo = self.make_repository('.', shared=True)
1543
repo.set_make_working_trees(False)
1544
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1545
request = request_class(backing)
1546
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1547
request.execute('', 'True'))
1548
repo = repo.bzrdir.open_repository()
1549
self.assertTrue(repo.make_working_trees())
1552
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1554
def make_repo_needing_autopacking(self, path='.'):
1555
# Make a repo in need of autopacking.
1556
tree = self.make_branch_and_tree('.', format='pack-0.92')
1557
repo = tree.branch.repository
1558
# monkey-patch the pack collection to disable autopacking
1559
repo._pack_collection._max_pack_count = lambda count: count
1561
tree.commit('commit %s' % x)
1562
self.assertEqual(10, len(repo._pack_collection.names()))
1563
del repo._pack_collection._max_pack_count
1566
def test_autopack_needed(self):
1567
repo = self.make_repo_needing_autopacking()
1569
self.addCleanup(repo.unlock)
1570
backing = self.get_transport()
1571
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1573
response = request.execute('')
1574
self.assertEqual(SmartServerResponse(('ok',)), response)
1575
repo._pack_collection.reload_pack_names()
1576
self.assertEqual(1, len(repo._pack_collection.names()))
1578
def test_autopack_not_needed(self):
1579
tree = self.make_branch_and_tree('.', format='pack-0.92')
1580
repo = tree.branch.repository
1582
self.addCleanup(repo.unlock)
1584
tree.commit('commit %s' % x)
1585
backing = self.get_transport()
1586
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1588
response = request.execute('')
1589
self.assertEqual(SmartServerResponse(('ok',)), response)
1590
repo._pack_collection.reload_pack_names()
1591
self.assertEqual(9, len(repo._pack_collection.names()))
1593
def test_autopack_on_nonpack_format(self):
1594
"""A request to autopack a non-pack repo is a no-op."""
1595
repo = self.make_repository('.', format='knit')
1596
backing = self.get_transport()
1597
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1599
response = request.execute('')
1600
self.assertEqual(SmartServerResponse(('ok',)), response)
1603
class TestHandlers(tests.TestCase):
1604
"""Tests for the request.request_handlers object."""
1606
def test_all_registrations_exist(self):
1607
"""All registered request_handlers can be found."""
1608
# If there's a typo in a register_lazy call, this loop will fail with
1609
# an AttributeError.
1610
for key, item in smart.request.request_handlers.iteritems():
1613
def assertHandlerEqual(self, verb, handler):
1614
self.assertEqual(smart.request.request_handlers.get(verb), handler)
1616
def test_registered_methods(self):
1617
"""Test that known methods are registered to the correct object."""
1618
self.assertHandlerEqual('Branch.get_config_file',
1619
smart.branch.SmartServerBranchGetConfigFile)
1620
self.assertHandlerEqual('Branch.get_parent',
1621
smart.branch.SmartServerBranchGetParent)
1622
self.assertHandlerEqual('Branch.get_tags_bytes',
1623
smart.branch.SmartServerBranchGetTagsBytes)
1624
self.assertHandlerEqual('Branch.lock_write',
1625
smart.branch.SmartServerBranchRequestLockWrite)
1626
self.assertHandlerEqual('Branch.last_revision_info',
1627
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1628
self.assertHandlerEqual('Branch.revision_history',
1629
smart.branch.SmartServerRequestRevisionHistory)
1630
self.assertHandlerEqual('Branch.set_config_option',
1631
smart.branch.SmartServerBranchRequestSetConfigOption)
1632
self.assertHandlerEqual('Branch.set_last_revision',
1633
smart.branch.SmartServerBranchRequestSetLastRevision)
1634
self.assertHandlerEqual('Branch.set_last_revision_info',
1635
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1636
self.assertHandlerEqual('Branch.set_last_revision_ex',
1637
smart.branch.SmartServerBranchRequestSetLastRevisionEx)
1638
self.assertHandlerEqual('Branch.set_parent_location',
1639
smart.branch.SmartServerBranchRequestSetParentLocation)
1640
self.assertHandlerEqual('Branch.unlock',
1641
smart.branch.SmartServerBranchRequestUnlock)
1642
self.assertHandlerEqual('BzrDir.find_repository',
1643
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1644
self.assertHandlerEqual('BzrDir.find_repositoryV2',
1645
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1646
self.assertHandlerEqual('BzrDirFormat.initialize',
1647
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1648
self.assertHandlerEqual('BzrDirFormat.initialize_ex_1.16',
1649
smart.bzrdir.SmartServerRequestBzrDirInitializeEx)
1650
self.assertHandlerEqual('BzrDir.cloning_metadir',
1651
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1652
self.assertHandlerEqual('BzrDir.get_config_file',
1653
smart.bzrdir.SmartServerBzrDirRequestConfigFile)
1654
self.assertHandlerEqual('BzrDir.open_branch',
1655
smart.bzrdir.SmartServerRequestOpenBranch)
1656
self.assertHandlerEqual('BzrDir.open_branchV2',
1657
smart.bzrdir.SmartServerRequestOpenBranchV2)
1658
self.assertHandlerEqual('PackRepository.autopack',
1659
smart.packrepository.SmartServerPackRepositoryAutopack)
1660
self.assertHandlerEqual('Repository.gather_stats',
1661
smart.repository.SmartServerRepositoryGatherStats)
1662
self.assertHandlerEqual('Repository.get_parent_map',
1663
smart.repository.SmartServerRepositoryGetParentMap)
1664
self.assertHandlerEqual('Repository.get_rev_id_for_revno',
1665
smart.repository.SmartServerRepositoryGetRevIdForRevno)
1666
self.assertHandlerEqual('Repository.get_revision_graph',
1667
smart.repository.SmartServerRepositoryGetRevisionGraph)
1668
self.assertHandlerEqual('Repository.get_stream',
1669
smart.repository.SmartServerRepositoryGetStream)
1670
self.assertHandlerEqual('Repository.has_revision',
1671
smart.repository.SmartServerRequestHasRevision)
1672
self.assertHandlerEqual('Repository.insert_stream',
1673
smart.repository.SmartServerRepositoryInsertStream)
1674
self.assertHandlerEqual('Repository.insert_stream_locked',
1675
smart.repository.SmartServerRepositoryInsertStreamLocked)
1676
self.assertHandlerEqual('Repository.is_shared',
1677
smart.repository.SmartServerRepositoryIsShared)
1678
self.assertHandlerEqual('Repository.lock_write',
1679
smart.repository.SmartServerRepositoryLockWrite)
1680
self.assertHandlerEqual('Repository.tarball',
1681
smart.repository.SmartServerRepositoryTarball)
1682
self.assertHandlerEqual('Repository.unlock',
1683
smart.repository.SmartServerRepositoryUnlock)
1684
self.assertHandlerEqual('Transport.is_readonly',
1685
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
