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
41
from bzrlib.branch import Branch, BranchReferenceFormat
42
import bzrlib.smart.branch
43
import bzrlib.smart.bzrdir, bzrlib.smart.bzrdir as smart_dir
44
import bzrlib.smart.packrepository
45
import bzrlib.smart.repository
46
from bzrlib.smart.request import (
47
FailedSmartServerResponse,
50
SuccessfulSmartServerResponse,
52
from bzrlib.tests import (
55
from bzrlib.transport import chroot, get_transport
58
def load_tests(standard_tests, module, loader):
59
"""Multiply tests version and protocol consistency."""
60
# FindRepository tests.
61
bzrdir_mod = bzrlib.smart.bzrdir
64
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV1}),
65
("find_repositoryV2", {
66
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV2}),
67
("find_repositoryV3", {
68
"_request_class":bzrdir_mod.SmartServerRequestFindRepositoryV3}),
70
to_adapt, result = split_suite_by_re(standard_tests,
71
"TestSmartServerRequestFindRepository")
72
v2_only, v1_and_2 = split_suite_by_re(to_adapt,
74
tests.multiply_tests(v1_and_2, scenarios, result)
75
# The first scenario is only applicable to v1 protocols, it is deleted
77
tests.multiply_tests(v2_only, scenarios[1:], result)
81
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
84
tests.TestCaseWithTransport.setUp(self)
85
self._chroot_server = None
87
def get_transport(self, relpath=None):
88
if self._chroot_server is None:
89
backing_transport = tests.TestCaseWithTransport.get_transport(self)
90
self._chroot_server = chroot.ChrootServer(backing_transport)
91
self.start_server(self._chroot_server)
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 TestByteStreamToStream(tests.TestCase):
118
def test_repeated_substreams_same_kind_are_one_stream(self):
119
# Make a stream - an iterable of bytestrings.
120
stream = [('text', [versionedfile.FulltextContentFactory(('k1',), None,
121
None, 'foo')]),('text', [
122
versionedfile.FulltextContentFactory(('k2',), None, None, 'bar')])]
123
fmt = bzrdir.format_registry.get('pack-0.92')().repository_format
124
bytes = smart.repository._stream_to_byte_stream(stream, fmt)
126
# Iterate the resulting iterable; checking that we get only one stream
128
fmt, stream = smart.repository._byte_stream_to_stream(bytes)
129
for kind, substream in stream:
130
streams.append((kind, list(substream)))
131
self.assertLength(1, streams)
132
self.assertLength(2, streams[0][1])
135
class TestSmartServerResponse(tests.TestCase):
137
def test__eq__(self):
138
self.assertEqual(SmartServerResponse(('ok', )),
139
SmartServerResponse(('ok', )))
140
self.assertEqual(SmartServerResponse(('ok', ), 'body'),
141
SmartServerResponse(('ok', ), 'body'))
142
self.assertNotEqual(SmartServerResponse(('ok', )),
143
SmartServerResponse(('notok', )))
144
self.assertNotEqual(SmartServerResponse(('ok', ), 'body'),
145
SmartServerResponse(('ok', )))
146
self.assertNotEqual(None,
147
SmartServerResponse(('ok', )))
149
def test__str__(self):
150
"""SmartServerResponses can be stringified."""
152
"<SuccessfulSmartServerResponse args=('args',) body='body'>",
153
str(SuccessfulSmartServerResponse(('args',), 'body')))
155
"<FailedSmartServerResponse args=('args',) body='body'>",
156
str(FailedSmartServerResponse(('args',), 'body')))
159
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
161
def test_translate_client_path(self):
162
transport = self.get_transport()
163
request = SmartServerRequest(transport, 'foo/')
164
self.assertEqual('./', request.translate_client_path('foo/'))
166
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
168
errors.PathNotChild, request.translate_client_path, '/')
170
errors.PathNotChild, request.translate_client_path, 'bar/')
171
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
173
def test_transport_from_client_path(self):
174
transport = self.get_transport()
175
request = SmartServerRequest(transport, 'foo/')
178
request.transport_from_client_path('foo/').base)
181
class TestSmartServerBzrDirRequestCloningMetaDir(
182
tests.TestCaseWithMemoryTransport):
183
"""Tests for BzrDir.cloning_metadir."""
185
def test_cloning_metadir(self):
186
"""When there is a bzrdir present, the call succeeds."""
187
backing = self.get_transport()
188
dir = self.make_bzrdir('.')
189
local_result = dir.cloning_metadir()
190
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
191
request = request_class(backing)
192
expected = SuccessfulSmartServerResponse(
193
(local_result.network_name(),
194
local_result.repository_format.network_name(),
195
('branch', local_result.get_branch_format().network_name())))
196
self.assertEqual(expected, request.execute('', 'False'))
198
def test_cloning_metadir_reference(self):
199
"""The request fails when bzrdir contains a branch reference."""
200
backing = self.get_transport()
201
referenced_branch = self.make_branch('referenced')
202
dir = self.make_bzrdir('.')
203
local_result = dir.cloning_metadir()
204
reference = BranchReferenceFormat().initialize(dir, referenced_branch)
205
reference_url = BranchReferenceFormat().get_reference(dir)
206
# The server shouldn't try to follow the branch reference, so it's fine
207
# if the referenced branch isn't reachable.
208
backing.rename('referenced', 'moved')
209
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
210
request = request_class(backing)
211
expected = FailedSmartServerResponse(('BranchReference',))
212
self.assertEqual(expected, request.execute('', 'False'))
215
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
216
"""Tests for BzrDir.create_repository."""
218
def test_makes_repository(self):
219
"""When there is a bzrdir present, the call succeeds."""
220
backing = self.get_transport()
221
self.make_bzrdir('.')
222
request_class = bzrlib.smart.bzrdir.SmartServerRequestCreateRepository
223
request = request_class(backing)
224
reference_bzrdir_format = bzrdir.format_registry.get('pack-0.92')()
225
reference_format = reference_bzrdir_format.repository_format
226
network_name = reference_format.network_name()
227
expected = SuccessfulSmartServerResponse(
228
('ok', 'no', 'no', 'no', network_name))
229
self.assertEqual(expected, request.execute('', network_name, 'True'))
232
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
233
"""Tests for BzrDir.find_repository."""
235
def test_no_repository(self):
236
"""When there is no repository to be found, ('norepository', ) is returned."""
237
backing = self.get_transport()
238
request = self._request_class(backing)
239
self.make_bzrdir('.')
240
self.assertEqual(SmartServerResponse(('norepository', )),
243
def test_nonshared_repository(self):
244
# nonshared repositorys only allow 'find' to return a handle when the
245
# path the repository is being searched on is the same as that that
246
# the repository is at.
247
backing = self.get_transport()
248
request = self._request_class(backing)
249
result = self._make_repository_and_result()
250
self.assertEqual(result, request.execute(''))
251
self.make_bzrdir('subdir')
252
self.assertEqual(SmartServerResponse(('norepository', )),
253
request.execute('subdir'))
255
def _make_repository_and_result(self, shared=False, format=None):
256
"""Convenience function to setup a repository.
258
:result: The SmartServerResponse to expect when opening it.
260
repo = self.make_repository('.', shared=shared, format=format)
261
if repo.supports_rich_root():
265
if repo._format.supports_tree_reference:
269
if repo._format.supports_external_lookups:
273
if (smart.bzrdir.SmartServerRequestFindRepositoryV3 ==
274
self._request_class):
275
return SuccessfulSmartServerResponse(
276
('ok', '', rich_root, subtrees, external,
277
repo._format.network_name()))
278
elif (smart.bzrdir.SmartServerRequestFindRepositoryV2 ==
279
self._request_class):
280
# All tests so far are on formats, and for non-external
282
return SuccessfulSmartServerResponse(
283
('ok', '', rich_root, subtrees, external))
285
return SuccessfulSmartServerResponse(('ok', '', rich_root, subtrees))
287
def test_shared_repository(self):
288
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
289
backing = self.get_transport()
290
request = self._request_class(backing)
291
result = self._make_repository_and_result(shared=True)
292
self.assertEqual(result, request.execute(''))
293
self.make_bzrdir('subdir')
294
result2 = SmartServerResponse(result.args[0:1] + ('..', ) + result.args[2:])
295
self.assertEqual(result2,
296
request.execute('subdir'))
297
self.make_bzrdir('subdir/deeper')
298
result3 = SmartServerResponse(result.args[0:1] + ('../..', ) + result.args[2:])
299
self.assertEqual(result3,
300
request.execute('subdir/deeper'))
302
def test_rich_root_and_subtree_encoding(self):
303
"""Test for the format attributes for rich root and subtree support."""
304
backing = self.get_transport()
305
request = self._request_class(backing)
306
result = self._make_repository_and_result(format='dirstate-with-subtree')
307
# check the test will be valid
308
self.assertEqual('yes', result.args[2])
309
self.assertEqual('yes', result.args[3])
310
self.assertEqual(result, request.execute(''))
312
def test_supports_external_lookups_no_v2(self):
313
"""Test for the supports_external_lookups attribute."""
314
backing = self.get_transport()
315
request = self._request_class(backing)
316
result = self._make_repository_and_result(format='dirstate-with-subtree')
317
# check the test will be valid
318
self.assertEqual('no', result.args[4])
319
self.assertEqual(result, request.execute(''))
322
class TestSmartServerBzrDirRequestGetConfigFile(
323
tests.TestCaseWithMemoryTransport):
324
"""Tests for BzrDir.get_config_file."""
326
def test_present(self):
327
backing = self.get_transport()
328
dir = self.make_bzrdir('.')
329
dir.get_config().set_default_stack_on("/")
330
local_result = dir._get_config()._get_config_file().read()
331
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
332
request = request_class(backing)
333
expected = SuccessfulSmartServerResponse((), local_result)
334
self.assertEqual(expected, request.execute(''))
336
def test_missing(self):
337
backing = self.get_transport()
338
dir = self.make_bzrdir('.')
339
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
340
request = request_class(backing)
341
expected = SuccessfulSmartServerResponse((), '')
342
self.assertEqual(expected, request.execute(''))
345
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
347
def test_empty_dir(self):
348
"""Initializing an empty dir should succeed and do it."""
349
backing = self.get_transport()
350
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
351
self.assertEqual(SmartServerResponse(('ok', )),
353
made_dir = bzrdir.BzrDir.open_from_transport(backing)
354
# no branch, tree or repository is expected with the current
356
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
357
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
358
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
360
def test_missing_dir(self):
361
"""Initializing a missing directory should fail like the bzrdir api."""
362
backing = self.get_transport()
363
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
364
self.assertRaises(errors.NoSuchFile,
365
request.execute, 'subdir')
367
def test_initialized_dir(self):
368
"""Initializing an extant bzrdir should fail like the bzrdir api."""
369
backing = self.get_transport()
370
request = smart.bzrdir.SmartServerRequestInitializeBzrDir(backing)
371
self.make_bzrdir('subdir')
372
self.assertRaises(errors.FileExists,
373
request.execute, 'subdir')
376
class TestSmartServerRequestBzrDirInitializeEx(tests.TestCaseWithMemoryTransport):
377
"""Basic tests for BzrDir.initialize_ex_1.16 in the smart server.
379
The main unit tests in test_bzrdir exercise the API comprehensively.
382
def test_empty_dir(self):
383
"""Initializing an empty dir should succeed and do it."""
384
backing = self.get_transport()
385
name = self.make_bzrdir('reference')._format.network_name()
386
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
387
self.assertEqual(SmartServerResponse(('', '', '', '', '', '', name,
388
'False', '', '', '')),
389
request.execute(name, '', 'True', 'False', 'False', '', '', '', '',
391
made_dir = bzrdir.BzrDir.open_from_transport(backing)
392
# no branch, tree or repository is expected with the current
394
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
395
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
396
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
398
def test_missing_dir(self):
399
"""Initializing a missing directory should fail like the bzrdir api."""
400
backing = self.get_transport()
401
name = self.make_bzrdir('reference')._format.network_name()
402
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
403
self.assertRaises(errors.NoSuchFile, request.execute, name,
404
'subdir/dir', 'False', 'False', 'False', '', '', '', '', 'False')
406
def test_initialized_dir(self):
407
"""Initializing an extant directory should fail like the bzrdir api."""
408
backing = self.get_transport()
409
name = self.make_bzrdir('reference')._format.network_name()
410
request = smart.bzrdir.SmartServerRequestBzrDirInitializeEx(backing)
411
self.make_bzrdir('subdir')
412
self.assertRaises(errors.FileExists, request.execute, name, 'subdir',
413
'False', 'False', 'False', '', '', '', '', 'False')
416
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
418
def test_no_branch(self):
419
"""When there is no branch, ('nobranch', ) is returned."""
420
backing = self.get_transport()
421
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
422
self.make_bzrdir('.')
423
self.assertEqual(SmartServerResponse(('nobranch', )),
426
def test_branch(self):
427
"""When there is a branch, 'ok' is returned."""
428
backing = self.get_transport()
429
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
430
self.make_branch('.')
431
self.assertEqual(SmartServerResponse(('ok', '')),
434
def test_branch_reference(self):
435
"""When there is a branch reference, the reference URL is returned."""
436
backing = self.get_transport()
437
request = smart.bzrdir.SmartServerRequestOpenBranch(backing)
438
branch = self.make_branch('branch')
439
checkout = branch.create_checkout('reference',lightweight=True)
440
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
441
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
442
self.assertEqual(SmartServerResponse(('ok', reference_url)),
443
request.execute('reference'))
446
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
448
def test_no_branch(self):
449
"""When there is no branch, ('nobranch', ) is returned."""
450
backing = self.get_transport()
451
self.make_bzrdir('.')
452
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
453
self.assertEqual(SmartServerResponse(('nobranch', )),
456
def test_branch(self):
457
"""When there is a branch, 'ok' is returned."""
458
backing = self.get_transport()
459
expected = self.make_branch('.')._format.network_name()
460
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
461
self.assertEqual(SuccessfulSmartServerResponse(('branch', expected)),
464
def test_branch_reference(self):
465
"""When there is a branch reference, the reference URL is returned."""
466
backing = self.get_transport()
467
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
468
branch = self.make_branch('branch')
469
checkout = branch.create_checkout('reference',lightweight=True)
470
reference_url = BranchReferenceFormat().get_reference(checkout.bzrdir)
471
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
472
self.assertEqual(SuccessfulSmartServerResponse(('ref', reference_url)),
473
request.execute('reference'))
475
def test_stacked_branch(self):
476
"""Opening a stacked branch does not open the stacked-on branch."""
477
trunk = self.make_branch('trunk')
478
feature = self.make_branch('feature')
479
feature.set_stacked_on_url(trunk.base)
481
Branch.hooks.install_named_hook('open', opened_branches.append, None)
482
backing = self.get_transport()
483
request = smart.bzrdir.SmartServerRequestOpenBranchV2(backing)
486
response = request.execute('feature')
488
request.teardown_jail()
489
expected_format = feature._format.network_name()
491
SuccessfulSmartServerResponse(('branch', expected_format)),
493
self.assertLength(1, opened_branches)
496
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
498
def test_empty(self):
499
"""For an empty branch, the body is empty."""
500
backing = self.get_transport()
501
request = smart.branch.SmartServerRequestRevisionHistory(backing)
502
self.make_branch('.')
503
self.assertEqual(SmartServerResponse(('ok', ), ''),
506
def test_not_empty(self):
507
"""For a non-empty branch, the body is empty."""
508
backing = self.get_transport()
509
request = smart.branch.SmartServerRequestRevisionHistory(backing)
510
tree = self.make_branch_and_memory_tree('.')
513
r1 = tree.commit('1st commit')
514
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
517
SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
521
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
523
def test_no_branch(self):
524
"""When there is a bzrdir and no branch, NotBranchError is raised."""
525
backing = self.get_transport()
526
request = smart.branch.SmartServerBranchRequest(backing)
527
self.make_bzrdir('.')
528
self.assertRaises(errors.NotBranchError,
531
def test_branch_reference(self):
532
"""When there is a branch reference, NotBranchError is raised."""
533
backing = self.get_transport()
534
request = smart.branch.SmartServerBranchRequest(backing)
535
branch = self.make_branch('branch')
536
checkout = branch.create_checkout('reference',lightweight=True)
537
self.assertRaises(errors.NotBranchError,
538
request.execute, 'checkout')
541
class TestSmartServerBranchRequestLastRevisionInfo(tests.TestCaseWithMemoryTransport):
543
def test_empty(self):
544
"""For an empty branch, the result is ('ok', '0', 'null:')."""
545
backing = self.get_transport()
546
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
547
self.make_branch('.')
548
self.assertEqual(SmartServerResponse(('ok', '0', 'null:')),
551
def test_not_empty(self):
552
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
553
backing = self.get_transport()
554
request = smart.branch.SmartServerBranchRequestLastRevisionInfo(backing)
555
tree = self.make_branch_and_memory_tree('.')
558
rev_id_utf8 = u'\xc8'.encode('utf-8')
559
r1 = tree.commit('1st commit')
560
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
563
SmartServerResponse(('ok', '2', rev_id_utf8)),
567
class TestSmartServerBranchRequestGetConfigFile(tests.TestCaseWithMemoryTransport):
569
def test_default(self):
570
"""With no file, we get empty content."""
571
backing = self.get_transport()
572
request = smart.branch.SmartServerBranchGetConfigFile(backing)
573
branch = self.make_branch('.')
574
# there should be no file by default
576
self.assertEqual(SmartServerResponse(('ok', ), content),
579
def test_with_content(self):
580
# SmartServerBranchGetConfigFile should return the content from
581
# branch.control_files.get('branch.conf') for now - in the future it may
582
# perform more complex processing.
583
backing = self.get_transport()
584
request = smart.branch.SmartServerBranchGetConfigFile(backing)
585
branch = self.make_branch('.')
586
branch._transport.put_bytes('branch.conf', 'foo bar baz')
587
self.assertEqual(SmartServerResponse(('ok', ), 'foo bar baz'),
591
class TestLockedBranch(tests.TestCaseWithMemoryTransport):
593
def get_lock_tokens(self, branch):
594
branch_token = branch.lock_write()
595
repo_token = branch.repository.lock_write()
596
branch.repository.unlock()
597
return branch_token, repo_token
600
class TestSmartServerBranchRequestSetConfigOption(TestLockedBranch):
602
def test_value_name(self):
603
branch = self.make_branch('.')
604
request = smart.branch.SmartServerBranchRequestSetConfigOption(
605
branch.bzrdir.root_transport)
606
branch_token, repo_token = self.get_lock_tokens(branch)
607
config = branch._get_config()
608
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
610
self.assertEqual(SuccessfulSmartServerResponse(()), result)
611
self.assertEqual('bar', config.get_option('foo'))
615
def test_value_name_section(self):
616
branch = self.make_branch('.')
617
request = smart.branch.SmartServerBranchRequestSetConfigOption(
618
branch.bzrdir.root_transport)
619
branch_token, repo_token = self.get_lock_tokens(branch)
620
config = branch._get_config()
621
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
623
self.assertEqual(SuccessfulSmartServerResponse(()), result)
624
self.assertEqual('bar', config.get_option('foo', 'gam'))
629
class TestSmartServerBranchRequestSetTagsBytes(TestLockedBranch):
630
# Only called when the branch format and tags match [yay factory
631
# methods] so only need to test straight forward cases.
633
def test_set_bytes(self):
634
base_branch = self.make_branch('base')
635
tag_bytes = base_branch._get_tags_bytes()
636
# get_lock_tokens takes out a lock.
637
branch_token, repo_token = self.get_lock_tokens(base_branch)
638
request = smart.branch.SmartServerBranchSetTagsBytes(
639
self.get_transport())
640
response = request.execute('base', branch_token, repo_token)
641
self.assertEqual(None, response)
642
response = request.do_chunk(tag_bytes)
643
self.assertEqual(None, response)
644
response = request.do_end()
646
SuccessfulSmartServerResponse(()), response)
649
def test_lock_failed(self):
650
base_branch = self.make_branch('base')
651
base_branch.lock_write()
652
tag_bytes = base_branch._get_tags_bytes()
653
request = smart.branch.SmartServerBranchSetTagsBytes(
654
self.get_transport())
655
self.assertRaises(errors.TokenMismatch, request.execute,
656
'base', 'wrong token', 'wrong token')
657
# The request handler will keep processing the message parts, so even
658
# if the request fails immediately do_chunk and do_end are still
660
request.do_chunk(tag_bytes)
666
class SetLastRevisionTestBase(TestLockedBranch):
667
"""Base test case for verbs that implement set_last_revision."""
670
tests.TestCaseWithMemoryTransport.setUp(self)
671
backing_transport = self.get_transport()
672
self.request = self.request_class(backing_transport)
673
self.tree = self.make_branch_and_memory_tree('.')
675
def lock_branch(self):
676
return self.get_lock_tokens(self.tree.branch)
678
def unlock_branch(self):
679
self.tree.branch.unlock()
681
def set_last_revision(self, revision_id, revno):
682
branch_token, repo_token = self.lock_branch()
683
response = self._set_last_revision(
684
revision_id, revno, branch_token, repo_token)
688
def assertRequestSucceeds(self, revision_id, revno):
689
response = self.set_last_revision(revision_id, revno)
690
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
693
class TestSetLastRevisionVerbMixin(object):
694
"""Mixin test case for verbs that implement set_last_revision."""
696
def test_set_null_to_null(self):
697
"""An empty branch can have its last revision set to 'null:'."""
698
self.assertRequestSucceeds('null:', 0)
700
def test_NoSuchRevision(self):
701
"""If the revision_id is not present, the verb returns NoSuchRevision.
703
revision_id = 'non-existent revision'
705
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
706
self.set_last_revision(revision_id, 1))
708
def make_tree_with_two_commits(self):
709
self.tree.lock_write()
711
rev_id_utf8 = u'\xc8'.encode('utf-8')
712
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
713
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
716
def test_branch_last_revision_info_is_updated(self):
717
"""A branch's tip can be set to a revision that is present in its
720
# Make a branch with an empty revision history, but two revisions in
722
self.make_tree_with_two_commits()
723
rev_id_utf8 = u'\xc8'.encode('utf-8')
724
self.tree.branch.set_revision_history([])
726
(0, 'null:'), self.tree.branch.last_revision_info())
727
# We can update the branch to a revision that is present in the
729
self.assertRequestSucceeds(rev_id_utf8, 1)
731
(1, rev_id_utf8), self.tree.branch.last_revision_info())
733
def test_branch_last_revision_info_rewind(self):
734
"""A branch's tip can be set to a revision that is an ancestor of the
737
self.make_tree_with_two_commits()
738
rev_id_utf8 = u'\xc8'.encode('utf-8')
740
(2, 'rev-2'), self.tree.branch.last_revision_info())
741
self.assertRequestSucceeds(rev_id_utf8, 1)
743
(1, rev_id_utf8), self.tree.branch.last_revision_info())
745
def test_TipChangeRejected(self):
746
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
747
returns TipChangeRejected.
749
rejection_message = u'rejection message\N{INTERROBANG}'
750
def hook_that_rejects(params):
751
raise errors.TipChangeRejected(rejection_message)
752
Branch.hooks.install_named_hook(
753
'pre_change_branch_tip', hook_that_rejects, None)
755
FailedSmartServerResponse(
756
('TipChangeRejected', rejection_message.encode('utf-8'))),
757
self.set_last_revision('null:', 0))
760
class TestSmartServerBranchRequestSetLastRevision(
761
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
762
"""Tests for Branch.set_last_revision verb."""
764
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
766
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
767
return self.request.execute(
768
'', branch_token, repo_token, revision_id)
771
class TestSmartServerBranchRequestSetLastRevisionInfo(
772
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
773
"""Tests for Branch.set_last_revision_info verb."""
775
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
777
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
778
return self.request.execute(
779
'', branch_token, repo_token, revno, revision_id)
781
def test_NoSuchRevision(self):
782
"""Branch.set_last_revision_info does not have to return
783
NoSuchRevision if the revision_id is absent.
785
raise tests.TestNotApplicable()
788
class TestSmartServerBranchRequestSetLastRevisionEx(
789
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
790
"""Tests for Branch.set_last_revision_ex verb."""
792
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
794
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
795
return self.request.execute(
796
'', branch_token, repo_token, revision_id, 0, 0)
798
def assertRequestSucceeds(self, revision_id, revno):
799
response = self.set_last_revision(revision_id, revno)
801
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
804
def test_branch_last_revision_info_rewind(self):
805
"""A branch's tip can be set to a revision that is an ancestor of the
806
current tip, but only if allow_overwrite_descendant is passed.
808
self.make_tree_with_two_commits()
809
rev_id_utf8 = u'\xc8'.encode('utf-8')
811
(2, 'rev-2'), self.tree.branch.last_revision_info())
812
# If allow_overwrite_descendant flag is 0, then trying to set the tip
813
# to an older revision ID has no effect.
814
branch_token, repo_token = self.lock_branch()
815
response = self.request.execute(
816
'', branch_token, repo_token, rev_id_utf8, 0, 0)
818
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
821
(2, 'rev-2'), self.tree.branch.last_revision_info())
823
# If allow_overwrite_descendant flag is 1, then setting the tip to an
825
response = self.request.execute(
826
'', branch_token, repo_token, rev_id_utf8, 0, 1)
828
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
832
(1, rev_id_utf8), self.tree.branch.last_revision_info())
834
def make_branch_with_divergent_history(self):
835
"""Make a branch with divergent history in its repo.
837
The branch's tip will be 'child-2', and the repo will also contain
838
'child-1', which diverges from a common base revision.
840
self.tree.lock_write()
842
r1 = self.tree.commit('1st commit')
843
revno_1, revid_1 = self.tree.branch.last_revision_info()
844
r2 = self.tree.commit('2nd commit', rev_id='child-1')
845
# Undo the second commit
846
self.tree.branch.set_last_revision_info(revno_1, revid_1)
847
self.tree.set_parent_ids([revid_1])
848
# Make a new second commit, child-2. child-2 has diverged from
850
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
853
def test_not_allow_diverged(self):
854
"""If allow_diverged is not passed, then setting a divergent history
855
returns a Diverged error.
857
self.make_branch_with_divergent_history()
859
FailedSmartServerResponse(('Diverged',)),
860
self.set_last_revision('child-1', 2))
861
# The branch tip was not changed.
862
self.assertEqual('child-2', self.tree.branch.last_revision())
864
def test_allow_diverged(self):
865
"""If allow_diverged is passed, then setting a divergent history
868
self.make_branch_with_divergent_history()
869
branch_token, repo_token = self.lock_branch()
870
response = self.request.execute(
871
'', branch_token, repo_token, 'child-1', 1, 0)
873
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
876
# The branch tip was changed.
877
self.assertEqual('child-1', self.tree.branch.last_revision())
880
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
882
def test_get_parent_none(self):
883
base_branch = self.make_branch('base')
884
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
885
response = request.execute('base')
887
SuccessfulSmartServerResponse(('',)), response)
889
def test_get_parent_something(self):
890
base_branch = self.make_branch('base')
891
base_branch.set_parent(self.get_url('foo'))
892
request = smart.branch.SmartServerBranchGetParent(self.get_transport())
893
response = request.execute('base')
895
SuccessfulSmartServerResponse(("../foo",)),
899
class TestSmartServerBranchRequestSetParent(tests.TestCaseWithMemoryTransport):
901
def test_set_parent_none(self):
902
branch = self.make_branch('base', format="1.9")
904
branch._set_parent_location('foo')
906
request = smart.branch.SmartServerBranchRequestSetParentLocation(
907
self.get_transport())
908
branch_token = branch.lock_write()
909
repo_token = branch.repository.lock_write()
911
response = request.execute('base', branch_token, repo_token, '')
913
branch.repository.unlock()
915
self.assertEqual(SuccessfulSmartServerResponse(()), response)
916
self.assertEqual(None, branch.get_parent())
918
def test_set_parent_something(self):
919
branch = self.make_branch('base', format="1.9")
920
request = smart.branch.SmartServerBranchRequestSetParentLocation(
921
self.get_transport())
922
branch_token = branch.lock_write()
923
repo_token = branch.repository.lock_write()
925
response = request.execute('base', branch_token, repo_token,
928
branch.repository.unlock()
930
self.assertEqual(SuccessfulSmartServerResponse(()), response)
931
self.assertEqual('http://bar/', branch.get_parent())
934
class TestSmartServerBranchRequestGetTagsBytes(tests.TestCaseWithMemoryTransport):
935
# Only called when the branch format and tags match [yay factory
936
# methods] so only need to test straight forward cases.
938
def test_get_bytes(self):
939
base_branch = self.make_branch('base')
940
request = smart.branch.SmartServerBranchGetTagsBytes(
941
self.get_transport())
942
response = request.execute('base')
944
SuccessfulSmartServerResponse(('',)), response)
947
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
949
def test_get_stacked_on_url(self):
950
base_branch = self.make_branch('base', format='1.6')
951
stacked_branch = self.make_branch('stacked', format='1.6')
952
# typically should be relative
953
stacked_branch.set_stacked_on_url('../base')
954
request = smart.branch.SmartServerBranchRequestGetStackedOnURL(
955
self.get_transport())
956
response = request.execute('stacked')
958
SmartServerResponse(('ok', '../base')),
962
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
965
tests.TestCaseWithMemoryTransport.setUp(self)
967
def test_lock_write_on_unlocked_branch(self):
968
backing = self.get_transport()
969
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
970
branch = self.make_branch('.', format='knit')
971
repository = branch.repository
972
response = request.execute('')
973
branch_nonce = branch.control_files._lock.peek().get('nonce')
974
repository_nonce = repository.control_files._lock.peek().get('nonce')
976
SmartServerResponse(('ok', branch_nonce, repository_nonce)),
978
# The branch (and associated repository) is now locked. Verify that
979
# with a new branch object.
980
new_branch = repository.bzrdir.open_branch()
981
self.assertRaises(errors.LockContention, new_branch.lock_write)
983
request = smart.branch.SmartServerBranchRequestUnlock(backing)
984
response = request.execute('', branch_nonce, repository_nonce)
986
def test_lock_write_on_locked_branch(self):
987
backing = self.get_transport()
988
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
989
branch = self.make_branch('.')
990
branch_token = branch.lock_write()
991
branch.leave_lock_in_place()
993
response = request.execute('')
995
SmartServerResponse(('LockContention',)), response)
997
branch.lock_write(branch_token)
998
branch.dont_leave_lock_in_place()
1001
def test_lock_write_with_tokens_on_locked_branch(self):
1002
backing = self.get_transport()
1003
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1004
branch = self.make_branch('.', format='knit')
1005
branch_token = branch.lock_write()
1006
repo_token = branch.repository.lock_write()
1007
branch.repository.unlock()
1008
branch.leave_lock_in_place()
1009
branch.repository.leave_lock_in_place()
1011
response = request.execute('',
1012
branch_token, repo_token)
1014
SmartServerResponse(('ok', branch_token, repo_token)), response)
1016
branch.repository.lock_write(repo_token)
1017
branch.repository.dont_leave_lock_in_place()
1018
branch.repository.unlock()
1019
branch.lock_write(branch_token)
1020
branch.dont_leave_lock_in_place()
1023
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
1024
backing = self.get_transport()
1025
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1026
branch = self.make_branch('.', format='knit')
1027
branch_token = branch.lock_write()
1028
repo_token = branch.repository.lock_write()
1029
branch.repository.unlock()
1030
branch.leave_lock_in_place()
1031
branch.repository.leave_lock_in_place()
1033
response = request.execute('',
1034
branch_token+'xxx', repo_token)
1036
SmartServerResponse(('TokenMismatch',)), response)
1038
branch.repository.lock_write(repo_token)
1039
branch.repository.dont_leave_lock_in_place()
1040
branch.repository.unlock()
1041
branch.lock_write(branch_token)
1042
branch.dont_leave_lock_in_place()
1045
def test_lock_write_on_locked_repo(self):
1046
backing = self.get_transport()
1047
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1048
branch = self.make_branch('.', format='knit')
1049
repo = branch.repository
1050
repo_token = repo.lock_write()
1051
repo.leave_lock_in_place()
1053
response = request.execute('')
1055
SmartServerResponse(('LockContention',)), response)
1057
repo.lock_write(repo_token)
1058
repo.dont_leave_lock_in_place()
1061
def test_lock_write_on_readonly_transport(self):
1062
backing = self.get_readonly_transport()
1063
request = smart.branch.SmartServerBranchRequestLockWrite(backing)
1064
branch = self.make_branch('.')
1065
root = self.get_transport().clone('/')
1066
path = urlutils.relative_url(root.base, self.get_transport().base)
1067
response = request.execute(path)
1068
error_name, lock_str, why_str = response.args
1069
self.assertFalse(response.is_successful())
1070
self.assertEqual('LockFailed', error_name)
1073
class TestSmartServerBranchRequestUnlock(tests.TestCaseWithMemoryTransport):
1076
tests.TestCaseWithMemoryTransport.setUp(self)
1078
def test_unlock_on_locked_branch_and_repo(self):
1079
backing = self.get_transport()
1080
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1081
branch = self.make_branch('.', format='knit')
1083
branch_token = branch.lock_write()
1084
repo_token = branch.repository.lock_write()
1085
branch.repository.unlock()
1086
# Unlock the branch (and repo) object, leaving the physical locks
1088
branch.leave_lock_in_place()
1089
branch.repository.leave_lock_in_place()
1091
response = request.execute('',
1092
branch_token, repo_token)
1094
SmartServerResponse(('ok',)), response)
1095
# The branch is now unlocked. Verify that with a new branch
1097
new_branch = branch.bzrdir.open_branch()
1098
new_branch.lock_write()
1101
def test_unlock_on_unlocked_branch_unlocked_repo(self):
1102
backing = self.get_transport()
1103
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1104
branch = self.make_branch('.', format='knit')
1105
response = request.execute(
1106
'', 'branch token', 'repo token')
1108
SmartServerResponse(('TokenMismatch',)), response)
1110
def test_unlock_on_unlocked_branch_locked_repo(self):
1111
backing = self.get_transport()
1112
request = smart.branch.SmartServerBranchRequestUnlock(backing)
1113
branch = self.make_branch('.', format='knit')
1114
# Lock the repository.
1115
repo_token = branch.repository.lock_write()
1116
branch.repository.leave_lock_in_place()
1117
branch.repository.unlock()
1118
# Issue branch lock_write request on the unlocked branch (with locked
1120
response = request.execute(
1121
'', 'branch token', repo_token)
1123
SmartServerResponse(('TokenMismatch',)), response)
1125
branch.repository.lock_write(repo_token)
1126
branch.repository.dont_leave_lock_in_place()
1127
branch.repository.unlock()
1130
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
1132
def test_no_repository(self):
1133
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
1134
# we test this using a shared repository above the named path,
1135
# thus checking the right search logic is used - that is, that
1136
# its the exact path being looked at and the server is not
1138
backing = self.get_transport()
1139
request = smart.repository.SmartServerRepositoryRequest(backing)
1140
self.make_repository('.', shared=True)
1141
self.make_bzrdir('subdir')
1142
self.assertRaises(errors.NoRepositoryPresent,
1143
request.execute, 'subdir')
1146
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
1148
def test_trivial_bzipped(self):
1149
# This tests that the wire encoding is actually bzipped
1150
backing = self.get_transport()
1151
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1152
tree = self.make_branch_and_memory_tree('.')
1154
self.assertEqual(None,
1155
request.execute('', 'missing-id'))
1156
# Note that it returns a body that is bzipped.
1158
SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
1159
request.do_body('\n\n0\n'))
1161
def test_trivial_include_missing(self):
1162
backing = self.get_transport()
1163
request = smart.repository.SmartServerRepositoryGetParentMap(backing)
1164
tree = self.make_branch_and_memory_tree('.')
1166
self.assertEqual(None,
1167
request.execute('', 'missing-id', 'include-missing:'))
1169
SuccessfulSmartServerResponse(('ok', ),
1170
bz2.compress('missing:missing-id')),
1171
request.do_body('\n\n0\n'))
1174
class TestSmartServerRepositoryGetRevisionGraph(tests.TestCaseWithMemoryTransport):
1176
def test_none_argument(self):
1177
backing = self.get_transport()
1178
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1179
tree = self.make_branch_and_memory_tree('.')
1182
r1 = tree.commit('1st commit')
1183
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1186
# the lines of revision_id->revision_parent_list has no guaranteed
1187
# order coming out of a dict, so sort both our test and response
1188
lines = sorted([' '.join([r2, r1]), r1])
1189
response = request.execute('', '')
1190
response.body = '\n'.join(sorted(response.body.split('\n')))
1193
SmartServerResponse(('ok', ), '\n'.join(lines)), response)
1195
def test_specific_revision_argument(self):
1196
backing = self.get_transport()
1197
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1198
tree = self.make_branch_and_memory_tree('.')
1201
rev_id_utf8 = u'\xc9'.encode('utf-8')
1202
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
1203
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1206
self.assertEqual(SmartServerResponse(('ok', ), rev_id_utf8),
1207
request.execute('', rev_id_utf8))
1209
def test_no_such_revision(self):
1210
backing = self.get_transport()
1211
request = smart.repository.SmartServerRepositoryGetRevisionGraph(backing)
1212
tree = self.make_branch_and_memory_tree('.')
1215
r1 = tree.commit('1st commit')
1218
# Note that it still returns body (of zero bytes).
1220
SmartServerResponse(('nosuchrevision', 'missingrevision', ), ''),
1221
request.execute('', 'missingrevision'))
1224
class TestSmartServerRepositoryGetRevIdForRevno(tests.TestCaseWithMemoryTransport):
1226
def test_revno_found(self):
1227
backing = self.get_transport()
1228
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1229
tree = self.make_branch_and_memory_tree('.')
1232
rev1_id_utf8 = u'\xc8'.encode('utf-8')
1233
rev2_id_utf8 = u'\xc9'.encode('utf-8')
1234
tree.commit('1st commit', rev_id=rev1_id_utf8)
1235
tree.commit('2nd commit', rev_id=rev2_id_utf8)
1238
self.assertEqual(SmartServerResponse(('ok', rev1_id_utf8)),
1239
request.execute('', 1, (2, rev2_id_utf8)))
1241
def test_known_revid_missing(self):
1242
backing = self.get_transport()
1243
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1244
repo = self.make_repository('.')
1246
FailedSmartServerResponse(('nosuchrevision', 'ghost')),
1247
request.execute('', 1, (2, 'ghost')))
1249
def test_history_incomplete(self):
1250
backing = self.get_transport()
1251
request = smart.repository.SmartServerRepositoryGetRevIdForRevno(backing)
1252
parent = self.make_branch_and_memory_tree('parent', format='1.9')
1254
parent.add([''], ['TREE_ROOT'])
1255
r1 = parent.commit(message='first commit')
1256
r2 = parent.commit(message='second commit')
1258
local = self.make_branch_and_memory_tree('local', format='1.9')
1259
local.branch.pull(parent.branch)
1260
local.set_parent_ids([r2])
1261
r3 = local.commit(message='local commit')
1262
local.branch.create_clone_on_transport(
1263
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
1265
SmartServerResponse(('history-incomplete', 2, r2)),
1266
request.execute('stacked', 1, (3, r3)))
1269
class TestSmartServerRepositoryGetStream(tests.TestCaseWithMemoryTransport):
1271
def make_two_commit_repo(self):
1272
tree = self.make_branch_and_memory_tree('.')
1275
r1 = tree.commit('1st commit')
1276
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1278
repo = tree.branch.repository
1281
def test_ancestry_of(self):
1282
"""The search argument may be a 'ancestry-of' some heads'."""
1283
backing = self.get_transport()
1284
request = smart.repository.SmartServerRepositoryGetStream(backing)
1285
repo, r1, r2 = self.make_two_commit_repo()
1286
fetch_spec = ['ancestry-of', r2]
1287
lines = '\n'.join(fetch_spec)
1288
request.execute('', repo._format.network_name())
1289
response = request.do_body(lines)
1290
self.assertEqual(('ok',), response.args)
1291
stream_bytes = ''.join(response.body_stream)
1292
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1294
def test_search(self):
1295
"""The search argument may be a 'search' of some explicit keys."""
1296
backing = self.get_transport()
1297
request = smart.repository.SmartServerRepositoryGetStream(backing)
1298
repo, r1, r2 = self.make_two_commit_repo()
1299
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1300
lines = '\n'.join(fetch_spec)
1301
request.execute('', repo._format.network_name())
1302
response = request.do_body(lines)
1303
self.assertEqual(('ok',), response.args)
1304
stream_bytes = ''.join(response.body_stream)
1305
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1308
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1310
def test_missing_revision(self):
1311
"""For a missing revision, ('no', ) is returned."""
1312
backing = self.get_transport()
1313
request = smart.repository.SmartServerRequestHasRevision(backing)
1314
self.make_repository('.')
1315
self.assertEqual(SmartServerResponse(('no', )),
1316
request.execute('', 'revid'))
1318
def test_present_revision(self):
1319
"""For a present revision, ('yes', ) is returned."""
1320
backing = self.get_transport()
1321
request = smart.repository.SmartServerRequestHasRevision(backing)
1322
tree = self.make_branch_and_memory_tree('.')
1325
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1326
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1328
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1329
self.assertEqual(SmartServerResponse(('yes', )),
1330
request.execute('', rev_id_utf8))
1333
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1335
def test_empty_revid(self):
1336
"""With an empty revid, we get only size an number and revisions"""
1337
backing = self.get_transport()
1338
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1339
repository = self.make_repository('.')
1340
stats = repository.gather_stats()
1341
expected_body = 'revisions: 0\n'
1342
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1343
request.execute('', '', 'no'))
1345
def test_revid_with_committers(self):
1346
"""For a revid we get more infos."""
1347
backing = self.get_transport()
1348
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1349
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1350
tree = self.make_branch_and_memory_tree('.')
1353
# Let's build a predictable result
1354
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1355
tree.commit('a commit', timestamp=654321.4, timezone=0,
1359
stats = tree.branch.repository.gather_stats()
1360
expected_body = ('firstrev: 123456.200 3600\n'
1361
'latestrev: 654321.400 0\n'
1363
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1367
def test_not_empty_repository_with_committers(self):
1368
"""For a revid and requesting committers we get the whole thing."""
1369
backing = self.get_transport()
1370
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1371
request = smart.repository.SmartServerRepositoryGatherStats(backing)
1372
tree = self.make_branch_and_memory_tree('.')
1375
# Let's build a predictable result
1376
tree.commit('a commit', timestamp=123456.2, timezone=3600,
1378
tree.commit('a commit', timestamp=654321.4, timezone=0,
1379
committer='bar', rev_id=rev_id_utf8)
1381
stats = tree.branch.repository.gather_stats()
1383
expected_body = ('committers: 2\n'
1384
'firstrev: 123456.200 3600\n'
1385
'latestrev: 654321.400 0\n'
1387
self.assertEqual(SmartServerResponse(('ok', ), expected_body),
1389
rev_id_utf8, 'yes'))
1392
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
1394
def test_is_shared(self):
1395
"""For a shared repository, ('yes', ) is returned."""
1396
backing = self.get_transport()
1397
request = smart.repository.SmartServerRepositoryIsShared(backing)
1398
self.make_repository('.', shared=True)
1399
self.assertEqual(SmartServerResponse(('yes', )),
1400
request.execute('', ))
1402
def test_is_not_shared(self):
1403
"""For a shared repository, ('no', ) is returned."""
1404
backing = self.get_transport()
1405
request = smart.repository.SmartServerRepositoryIsShared(backing)
1406
self.make_repository('.', shared=False)
1407
self.assertEqual(SmartServerResponse(('no', )),
1408
request.execute('', ))
1411
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
1413
def test_lock_write_on_unlocked_repo(self):
1414
backing = self.get_transport()
1415
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1416
repository = self.make_repository('.', format='knit')
1417
response = request.execute('')
1418
nonce = repository.control_files._lock.peek().get('nonce')
1419
self.assertEqual(SmartServerResponse(('ok', nonce)), response)
1420
# The repository is now locked. Verify that with a new repository
1422
new_repo = repository.bzrdir.open_repository()
1423
self.assertRaises(errors.LockContention, new_repo.lock_write)
1425
request = smart.repository.SmartServerRepositoryUnlock(backing)
1426
response = request.execute('', nonce)
1428
def test_lock_write_on_locked_repo(self):
1429
backing = self.get_transport()
1430
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1431
repository = self.make_repository('.', format='knit')
1432
repo_token = repository.lock_write()
1433
repository.leave_lock_in_place()
1435
response = request.execute('')
1437
SmartServerResponse(('LockContention',)), response)
1439
repository.lock_write(repo_token)
1440
repository.dont_leave_lock_in_place()
1443
def test_lock_write_on_readonly_transport(self):
1444
backing = self.get_readonly_transport()
1445
request = smart.repository.SmartServerRepositoryLockWrite(backing)
1446
repository = self.make_repository('.', format='knit')
1447
response = request.execute('')
1448
self.assertFalse(response.is_successful())
1449
self.assertEqual('LockFailed', response.args[0])
1452
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
1454
def make_empty_byte_stream(self, repo):
1455
byte_stream = smart.repository._stream_to_byte_stream([], repo._format)
1456
return ''.join(byte_stream)
1459
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
1461
def test_insert_stream_empty(self):
1462
backing = self.get_transport()
1463
request = smart.repository.SmartServerRepositoryInsertStream(backing)
1464
repository = self.make_repository('.')
1465
response = request.execute('', '')
1466
self.assertEqual(None, response)
1467
response = request.do_chunk(self.make_empty_byte_stream(repository))
1468
self.assertEqual(None, response)
1469
response = request.do_end()
1470
self.assertEqual(SmartServerResponse(('ok', )), response)
1473
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
1475
def test_insert_stream_empty(self):
1476
backing = self.get_transport()
1477
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1479
repository = self.make_repository('.', format='knit')
1480
lock_token = repository.lock_write()
1481
response = request.execute('', '', lock_token)
1482
self.assertEqual(None, response)
1483
response = request.do_chunk(self.make_empty_byte_stream(repository))
1484
self.assertEqual(None, response)
1485
response = request.do_end()
1486
self.assertEqual(SmartServerResponse(('ok', )), response)
1489
def test_insert_stream_with_wrong_lock_token(self):
1490
backing = self.get_transport()
1491
request = smart.repository.SmartServerRepositoryInsertStreamLocked(
1493
repository = self.make_repository('.', format='knit')
1494
lock_token = repository.lock_write()
1496
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
1500
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
1503
tests.TestCaseWithMemoryTransport.setUp(self)
1505
def test_unlock_on_locked_repo(self):
1506
backing = self.get_transport()
1507
request = smart.repository.SmartServerRepositoryUnlock(backing)
1508
repository = self.make_repository('.', format='knit')
1509
token = repository.lock_write()
1510
repository.leave_lock_in_place()
1512
response = request.execute('', token)
1514
SmartServerResponse(('ok',)), response)
1515
# The repository is now unlocked. Verify that with a new repository
1517
new_repo = repository.bzrdir.open_repository()
1518
new_repo.lock_write()
1521
def test_unlock_on_unlocked_repo(self):
1522
backing = self.get_transport()
1523
request = smart.repository.SmartServerRepositoryUnlock(backing)
1524
repository = self.make_repository('.', format='knit')
1525
response = request.execute('', 'some token')
1527
SmartServerResponse(('TokenMismatch',)), response)
1530
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1532
def test_is_readonly_no(self):
1533
backing = self.get_transport()
1534
request = smart.request.SmartServerIsReadonly(backing)
1535
response = request.execute()
1537
SmartServerResponse(('no',)), response)
1539
def test_is_readonly_yes(self):
1540
backing = self.get_readonly_transport()
1541
request = smart.request.SmartServerIsReadonly(backing)
1542
response = request.execute()
1544
SmartServerResponse(('yes',)), response)
1547
class TestSmartServerRepositorySetMakeWorkingTrees(tests.TestCaseWithMemoryTransport):
1549
def test_set_false(self):
1550
backing = self.get_transport()
1551
repo = self.make_repository('.', shared=True)
1552
repo.set_make_working_trees(True)
1553
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1554
request = request_class(backing)
1555
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1556
request.execute('', 'False'))
1557
repo = repo.bzrdir.open_repository()
1558
self.assertFalse(repo.make_working_trees())
1560
def test_set_true(self):
1561
backing = self.get_transport()
1562
repo = self.make_repository('.', shared=True)
1563
repo.set_make_working_trees(False)
1564
request_class = smart.repository.SmartServerRepositorySetMakeWorkingTrees
1565
request = request_class(backing)
1566
self.assertEqual(SuccessfulSmartServerResponse(('ok',)),
1567
request.execute('', 'True'))
1568
repo = repo.bzrdir.open_repository()
1569
self.assertTrue(repo.make_working_trees())
1572
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
1574
def make_repo_needing_autopacking(self, path='.'):
1575
# Make a repo in need of autopacking.
1576
tree = self.make_branch_and_tree('.', format='pack-0.92')
1577
repo = tree.branch.repository
1578
# monkey-patch the pack collection to disable autopacking
1579
repo._pack_collection._max_pack_count = lambda count: count
1581
tree.commit('commit %s' % x)
1582
self.assertEqual(10, len(repo._pack_collection.names()))
1583
del repo._pack_collection._max_pack_count
1586
def test_autopack_needed(self):
1587
repo = self.make_repo_needing_autopacking()
1589
self.addCleanup(repo.unlock)
1590
backing = self.get_transport()
1591
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1593
response = request.execute('')
1594
self.assertEqual(SmartServerResponse(('ok',)), response)
1595
repo._pack_collection.reload_pack_names()
1596
self.assertEqual(1, len(repo._pack_collection.names()))
1598
def test_autopack_not_needed(self):
1599
tree = self.make_branch_and_tree('.', format='pack-0.92')
1600
repo = tree.branch.repository
1602
self.addCleanup(repo.unlock)
1604
tree.commit('commit %s' % x)
1605
backing = self.get_transport()
1606
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1608
response = request.execute('')
1609
self.assertEqual(SmartServerResponse(('ok',)), response)
1610
repo._pack_collection.reload_pack_names()
1611
self.assertEqual(9, len(repo._pack_collection.names()))
1613
def test_autopack_on_nonpack_format(self):
1614
"""A request to autopack a non-pack repo is a no-op."""
1615
repo = self.make_repository('.', format='knit')
1616
backing = self.get_transport()
1617
request = smart.packrepository.SmartServerPackRepositoryAutopack(
1619
response = request.execute('')
1620
self.assertEqual(SmartServerResponse(('ok',)), response)
1623
class TestHandlers(tests.TestCase):
1624
"""Tests for the request.request_handlers object."""
1626
def test_all_registrations_exist(self):
1627
"""All registered request_handlers can be found."""
1628
# If there's a typo in a register_lazy call, this loop will fail with
1629
# an AttributeError.
1630
for key, item in smart.request.request_handlers.iteritems():
1633
def assertHandlerEqual(self, verb, handler):
1634
self.assertEqual(smart.request.request_handlers.get(verb), handler)
1636
def test_registered_methods(self):
1637
"""Test that known methods are registered to the correct object."""
1638
self.assertHandlerEqual('Branch.get_config_file',
1639
smart.branch.SmartServerBranchGetConfigFile)
1640
self.assertHandlerEqual('Branch.get_parent',
1641
smart.branch.SmartServerBranchGetParent)
1642
self.assertHandlerEqual('Branch.get_tags_bytes',
1643
smart.branch.SmartServerBranchGetTagsBytes)
1644
self.assertHandlerEqual('Branch.lock_write',
1645
smart.branch.SmartServerBranchRequestLockWrite)
1646
self.assertHandlerEqual('Branch.last_revision_info',
1647
smart.branch.SmartServerBranchRequestLastRevisionInfo)
1648
self.assertHandlerEqual('Branch.revision_history',
1649
smart.branch.SmartServerRequestRevisionHistory)
1650
self.assertHandlerEqual('Branch.set_config_option',
1651
smart.branch.SmartServerBranchRequestSetConfigOption)
1652
self.assertHandlerEqual('Branch.set_last_revision',
1653
smart.branch.SmartServerBranchRequestSetLastRevision)
1654
self.assertHandlerEqual('Branch.set_last_revision_info',
1655
smart.branch.SmartServerBranchRequestSetLastRevisionInfo)
1656
self.assertHandlerEqual('Branch.set_last_revision_ex',
1657
smart.branch.SmartServerBranchRequestSetLastRevisionEx)
1658
self.assertHandlerEqual('Branch.set_parent_location',
1659
smart.branch.SmartServerBranchRequestSetParentLocation)
1660
self.assertHandlerEqual('Branch.unlock',
1661
smart.branch.SmartServerBranchRequestUnlock)
1662
self.assertHandlerEqual('BzrDir.find_repository',
1663
smart.bzrdir.SmartServerRequestFindRepositoryV1)
1664
self.assertHandlerEqual('BzrDir.find_repositoryV2',
1665
smart.bzrdir.SmartServerRequestFindRepositoryV2)
1666
self.assertHandlerEqual('BzrDirFormat.initialize',
1667
smart.bzrdir.SmartServerRequestInitializeBzrDir)
1668
self.assertHandlerEqual('BzrDirFormat.initialize_ex_1.16',
1669
smart.bzrdir.SmartServerRequestBzrDirInitializeEx)
1670
self.assertHandlerEqual('BzrDir.cloning_metadir',
1671
smart.bzrdir.SmartServerBzrDirRequestCloningMetaDir)
1672
self.assertHandlerEqual('BzrDir.get_config_file',
1673
smart.bzrdir.SmartServerBzrDirRequestConfigFile)
1674
self.assertHandlerEqual('BzrDir.open_branch',
1675
smart.bzrdir.SmartServerRequestOpenBranch)
1676
self.assertHandlerEqual('BzrDir.open_branchV2',
1677
smart.bzrdir.SmartServerRequestOpenBranchV2)
1678
self.assertHandlerEqual('PackRepository.autopack',
1679
smart.packrepository.SmartServerPackRepositoryAutopack)
1680
self.assertHandlerEqual('Repository.gather_stats',
1681
smart.repository.SmartServerRepositoryGatherStats)
1682
self.assertHandlerEqual('Repository.get_parent_map',
1683
smart.repository.SmartServerRepositoryGetParentMap)
1684
self.assertHandlerEqual('Repository.get_rev_id_for_revno',
1685
smart.repository.SmartServerRepositoryGetRevIdForRevno)
1686
self.assertHandlerEqual('Repository.get_revision_graph',
1687
smart.repository.SmartServerRepositoryGetRevisionGraph)
1688
self.assertHandlerEqual('Repository.get_stream',
1689
smart.repository.SmartServerRepositoryGetStream)
1690
self.assertHandlerEqual('Repository.has_revision',
1691
smart.repository.SmartServerRequestHasRevision)
1692
self.assertHandlerEqual('Repository.insert_stream',
1693
smart.repository.SmartServerRepositoryInsertStream)
1694
self.assertHandlerEqual('Repository.insert_stream_locked',
1695
smart.repository.SmartServerRepositoryInsertStreamLocked)
1696
self.assertHandlerEqual('Repository.is_shared',
1697
smart.repository.SmartServerRepositoryIsShared)
1698
self.assertHandlerEqual('Repository.lock_write',
1699
smart.repository.SmartServerRepositoryLockWrite)
1700
self.assertHandlerEqual('Repository.tarball',
1701
smart.repository.SmartServerRepositoryTarball)
1702
self.assertHandlerEqual('Repository.unlock',
1703
smart.repository.SmartServerRepositoryUnlock)
1704
self.assertHandlerEqual('Transport.is_readonly',
1705
smart.request.SmartServerIsReadonly)
added
removed
bzrlib/tests/test_smart.py
