1
# Copyright (C) 2006-2012, 2016 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.
32
branch as _mod_branch,
40
from breezy.bzr import (
41
branch as _mod_bzrbranch,
45
from breezy.bzr.smart import (
46
branch as smart_branch,
48
repository as smart_repo,
49
packrepository as smart_packrepo,
54
from breezy.testament import Testament
55
from breezy.tests import test_server
56
from breezy.transport import (
62
def load_tests(loader, standard_tests, pattern):
63
"""Multiply tests version and protocol consistency."""
64
# FindRepository tests.
67
"_request_class": smart_dir.SmartServerRequestFindRepositoryV1}),
68
("find_repositoryV2", {
69
"_request_class": smart_dir.SmartServerRequestFindRepositoryV2}),
70
("find_repositoryV3", {
71
"_request_class": smart_dir.SmartServerRequestFindRepositoryV3}),
73
to_adapt, result = tests.split_suite_by_re(standard_tests,
74
"TestSmartServerRequestFindRepository")
75
v2_only, v1_and_2 = tests.split_suite_by_re(to_adapt,
77
tests.multiply_tests(v1_and_2, scenarios, result)
78
# The first scenario is only applicable to v1 protocols, it is deleted
80
tests.multiply_tests(v2_only, scenarios[1:], result)
84
class TestCaseWithChrootedTransport(tests.TestCaseWithTransport):
87
self.vfs_transport_factory = memory.MemoryServer
88
super(TestCaseWithChrootedTransport, self).setUp()
89
self._chroot_server = None
91
def get_transport(self, relpath=None):
92
if self._chroot_server is None:
93
backing_transport = tests.TestCaseWithTransport.get_transport(self)
94
self._chroot_server = chroot.ChrootServer(backing_transport)
95
self.start_server(self._chroot_server)
96
t = transport.get_transport_from_url(self._chroot_server.get_url())
97
if relpath is not None:
102
class TestCaseWithSmartMedium(tests.TestCaseWithMemoryTransport):
105
super(TestCaseWithSmartMedium, self).setUp()
106
# We're allowed to set the transport class here, so that we don't use
107
# the default or a parameterized class, but rather use the
108
# TestCaseWithTransport infrastructure to set up a smart server and
110
self.overrideAttr(self, "transport_server", self.make_transport_server)
112
def make_transport_server(self):
113
return test_server.SmartTCPServer_for_testing('-' + self.id())
115
def get_smart_medium(self):
116
"""Get a smart medium to use in tests."""
117
return self.get_transport().get_smart_medium()
120
class TestByteStreamToStream(tests.TestCase):
122
def test_repeated_substreams_same_kind_are_one_stream(self):
123
# Make a stream - an iterable of bytestrings.
124
stream = [('text', [versionedfile.FulltextContentFactory(('k1',), None,
125
None, 'foo')]),('text', [
126
versionedfile.FulltextContentFactory(('k2',), None, None, 'bar')])]
127
fmt = controldir.format_registry.get('pack-0.92')().repository_format
128
bytes = smart_repo._stream_to_byte_stream(stream, fmt)
130
# Iterate the resulting iterable; checking that we get only one stream
132
fmt, stream = smart_repo._byte_stream_to_stream(bytes)
133
for kind, substream in stream:
134
streams.append((kind, list(substream)))
135
self.assertLength(1, streams)
136
self.assertLength(2, streams[0][1])
139
class TestSmartServerResponse(tests.TestCase):
141
def test__eq__(self):
142
self.assertEqual(smart_req.SmartServerResponse(('ok', )),
143
smart_req.SmartServerResponse(('ok', )))
144
self.assertEqual(smart_req.SmartServerResponse(('ok', ), 'body'),
145
smart_req.SmartServerResponse(('ok', ), 'body'))
146
self.assertNotEqual(smart_req.SmartServerResponse(('ok', )),
147
smart_req.SmartServerResponse(('notok', )))
148
self.assertNotEqual(smart_req.SmartServerResponse(('ok', ), 'body'),
149
smart_req.SmartServerResponse(('ok', )))
150
self.assertNotEqual(None,
151
smart_req.SmartServerResponse(('ok', )))
153
def test__str__(self):
154
"""SmartServerResponses can be stringified."""
156
"<SuccessfulSmartServerResponse args=('args',) body='body'>",
157
str(smart_req.SuccessfulSmartServerResponse(('args',), 'body')))
159
"<FailedSmartServerResponse args=('args',) body='body'>",
160
str(smart_req.FailedSmartServerResponse(('args',), 'body')))
163
class TestSmartServerRequest(tests.TestCaseWithMemoryTransport):
165
def test_translate_client_path(self):
166
transport = self.get_transport()
167
request = smart_req.SmartServerRequest(transport, 'foo/')
168
self.assertEqual('./', request.translate_client_path('foo/'))
170
errors.InvalidURLJoin, request.translate_client_path, 'foo/..')
172
errors.PathNotChild, request.translate_client_path, '/')
174
errors.PathNotChild, request.translate_client_path, 'bar/')
175
self.assertEqual('./baz', request.translate_client_path('foo/baz'))
176
e_acute = u'\N{LATIN SMALL LETTER E WITH ACUTE}'.encode('utf-8')
177
self.assertEqual('./' + urlutils.escape(e_acute),
178
request.translate_client_path('foo/' + e_acute))
180
def test_translate_client_path_vfs(self):
181
"""VfsRequests receive escaped paths rather than raw UTF-8."""
182
transport = self.get_transport()
183
request = vfs.VfsRequest(transport, 'foo/')
184
e_acute = u'\N{LATIN SMALL LETTER E WITH ACUTE}'.encode('utf-8')
185
escaped = urlutils.escape('foo/' + e_acute)
186
self.assertEqual('./' + urlutils.escape(e_acute),
187
request.translate_client_path(escaped))
189
def test_transport_from_client_path(self):
190
transport = self.get_transport()
191
request = smart_req.SmartServerRequest(transport, 'foo/')
194
request.transport_from_client_path('foo/').base)
197
class TestSmartServerBzrDirRequestCloningMetaDir(
198
tests.TestCaseWithMemoryTransport):
199
"""Tests for BzrDir.cloning_metadir."""
201
def test_cloning_metadir(self):
202
"""When there is a bzrdir present, the call succeeds."""
203
backing = self.get_transport()
204
dir = self.make_controldir('.')
205
local_result = dir.cloning_metadir()
206
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
207
request = request_class(backing)
208
expected = smart_req.SuccessfulSmartServerResponse(
209
(local_result.network_name(),
210
local_result.repository_format.network_name(),
211
('branch', local_result.get_branch_format().network_name())))
212
self.assertEqual(expected, request.execute('', 'False'))
214
def test_cloning_metadir_reference(self):
215
"""The request fails when bzrdir contains a branch reference."""
216
backing = self.get_transport()
217
referenced_branch = self.make_branch('referenced')
218
dir = self.make_controldir('.')
219
local_result = dir.cloning_metadir()
220
reference = _mod_bzrbranch.BranchReferenceFormat().initialize(
221
dir, target_branch=referenced_branch)
222
reference_url = _mod_bzrbranch.BranchReferenceFormat().get_reference(dir)
223
# The server shouldn't try to follow the branch reference, so it's fine
224
# if the referenced branch isn't reachable.
225
backing.rename('referenced', 'moved')
226
request_class = smart_dir.SmartServerBzrDirRequestCloningMetaDir
227
request = request_class(backing)
228
expected = smart_req.FailedSmartServerResponse(('BranchReference',))
229
self.assertEqual(expected, request.execute('', 'False'))
232
class TestSmartServerBzrDirRequestCloningMetaDir(
233
tests.TestCaseWithMemoryTransport):
234
"""Tests for BzrDir.checkout_metadir."""
236
def test_checkout_metadir(self):
237
backing = self.get_transport()
238
request = smart_dir.SmartServerBzrDirRequestCheckoutMetaDir(
240
branch = self.make_branch('.', format='2a')
241
response = request.execute('')
243
smart_req.SmartServerResponse(
244
('Bazaar-NG meta directory, format 1\n',
245
'Bazaar repository format 2a (needs bzr 1.16 or later)\n',
246
'Bazaar Branch Format 7 (needs bzr 1.6)\n')),
250
class TestSmartServerBzrDirRequestDestroyBranch(
251
tests.TestCaseWithMemoryTransport):
252
"""Tests for BzrDir.destroy_branch."""
254
def test_destroy_branch_default(self):
255
"""The default branch can be removed."""
256
backing = self.get_transport()
257
dir = self.make_branch('.').controldir
258
request_class = smart_dir.SmartServerBzrDirRequestDestroyBranch
259
request = request_class(backing)
260
expected = smart_req.SuccessfulSmartServerResponse(('ok',))
261
self.assertEqual(expected, request.execute('', None))
263
def test_destroy_branch_named(self):
264
"""A named branch can be removed."""
265
backing = self.get_transport()
266
dir = self.make_repository('.', format="development-colo").controldir
267
dir.create_branch(name="branchname")
268
request_class = smart_dir.SmartServerBzrDirRequestDestroyBranch
269
request = request_class(backing)
270
expected = smart_req.SuccessfulSmartServerResponse(('ok',))
271
self.assertEqual(expected, request.execute('', "branchname"))
273
def test_destroy_branch_missing(self):
274
"""An error is raised if the branch didn't exist."""
275
backing = self.get_transport()
276
dir = self.make_controldir('.', format="development-colo")
277
request_class = smart_dir.SmartServerBzrDirRequestDestroyBranch
278
request = request_class(backing)
279
expected = smart_req.FailedSmartServerResponse(('nobranch',), None)
280
self.assertEqual(expected, request.execute('', "branchname"))
283
class TestSmartServerBzrDirRequestHasWorkingTree(
284
tests.TestCaseWithTransport):
285
"""Tests for BzrDir.has_workingtree."""
287
def test_has_workingtree_yes(self):
288
"""A working tree is present."""
289
backing = self.get_transport()
290
dir = self.make_branch_and_tree('.').controldir
291
request_class = smart_dir.SmartServerBzrDirRequestHasWorkingTree
292
request = request_class(backing)
293
expected = smart_req.SuccessfulSmartServerResponse(('yes',))
294
self.assertEqual(expected, request.execute(''))
296
def test_has_workingtree_no(self):
297
"""A working tree is missing."""
298
backing = self.get_transport()
299
dir = self.make_controldir('.')
300
request_class = smart_dir.SmartServerBzrDirRequestHasWorkingTree
301
request = request_class(backing)
302
expected = smart_req.SuccessfulSmartServerResponse(('no',))
303
self.assertEqual(expected, request.execute(''))
306
class TestSmartServerBzrDirRequestDestroyRepository(
307
tests.TestCaseWithMemoryTransport):
308
"""Tests for BzrDir.destroy_repository."""
310
def test_destroy_repository_default(self):
311
"""The repository can be removed."""
312
backing = self.get_transport()
313
dir = self.make_repository('.').controldir
314
request_class = smart_dir.SmartServerBzrDirRequestDestroyRepository
315
request = request_class(backing)
316
expected = smart_req.SuccessfulSmartServerResponse(('ok',))
317
self.assertEqual(expected, request.execute(''))
319
def test_destroy_repository_missing(self):
320
"""An error is raised if the repository didn't exist."""
321
backing = self.get_transport()
322
dir = self.make_controldir('.')
323
request_class = smart_dir.SmartServerBzrDirRequestDestroyRepository
324
request = request_class(backing)
325
expected = smart_req.FailedSmartServerResponse(
326
('norepository',), None)
327
self.assertEqual(expected, request.execute(''))
330
class TestSmartServerRequestCreateRepository(tests.TestCaseWithMemoryTransport):
331
"""Tests for BzrDir.create_repository."""
333
def test_makes_repository(self):
334
"""When there is a bzrdir present, the call succeeds."""
335
backing = self.get_transport()
336
self.make_controldir('.')
337
request_class = smart_dir.SmartServerRequestCreateRepository
338
request = request_class(backing)
339
reference_bzrdir_format = controldir.format_registry.get('pack-0.92')()
340
reference_format = reference_bzrdir_format.repository_format
341
network_name = reference_format.network_name()
342
expected = smart_req.SuccessfulSmartServerResponse(
343
('ok', 'no', 'no', 'no', network_name))
344
self.assertEqual(expected, request.execute('', network_name, 'True'))
347
class TestSmartServerRequestFindRepository(tests.TestCaseWithMemoryTransport):
348
"""Tests for BzrDir.find_repository."""
350
def test_no_repository(self):
351
"""When there is no repository to be found, ('norepository', ) is returned."""
352
backing = self.get_transport()
353
request = self._request_class(backing)
354
self.make_controldir('.')
355
self.assertEqual(smart_req.SmartServerResponse(('norepository', )),
358
def test_nonshared_repository(self):
359
# nonshared repositorys only allow 'find' to return a handle when the
360
# path the repository is being searched on is the same as that that
361
# the repository is at.
362
backing = self.get_transport()
363
request = self._request_class(backing)
364
result = self._make_repository_and_result()
365
self.assertEqual(result, request.execute(''))
366
self.make_controldir('subdir')
367
self.assertEqual(smart_req.SmartServerResponse(('norepository', )),
368
request.execute('subdir'))
370
def _make_repository_and_result(self, shared=False, format=None):
371
"""Convenience function to setup a repository.
373
:result: The SmartServerResponse to expect when opening it.
375
repo = self.make_repository('.', shared=shared, format=format)
376
if repo.supports_rich_root():
380
if repo._format.supports_tree_reference:
384
if repo._format.supports_external_lookups:
388
if (smart_dir.SmartServerRequestFindRepositoryV3 ==
389
self._request_class):
390
return smart_req.SuccessfulSmartServerResponse(
391
('ok', '', rich_root, subtrees, external,
392
repo._format.network_name()))
393
elif (smart_dir.SmartServerRequestFindRepositoryV2 ==
394
self._request_class):
395
# All tests so far are on formats, and for non-external
397
return smart_req.SuccessfulSmartServerResponse(
398
('ok', '', rich_root, subtrees, external))
400
return smart_req.SuccessfulSmartServerResponse(
401
('ok', '', rich_root, subtrees))
403
def test_shared_repository(self):
404
"""When there is a shared repository, we get 'ok', 'relpath-to-repo'."""
405
backing = self.get_transport()
406
request = self._request_class(backing)
407
result = self._make_repository_and_result(shared=True)
408
self.assertEqual(result, request.execute(''))
409
self.make_controldir('subdir')
410
result2 = smart_req.SmartServerResponse(
411
result.args[0:1] + ('..', ) + result.args[2:])
412
self.assertEqual(result2,
413
request.execute('subdir'))
414
self.make_controldir('subdir/deeper')
415
result3 = smart_req.SmartServerResponse(
416
result.args[0:1] + ('../..', ) + result.args[2:])
417
self.assertEqual(result3,
418
request.execute('subdir/deeper'))
420
def test_rich_root_and_subtree_encoding(self):
421
"""Test for the format attributes for rich root and subtree support."""
422
backing = self.get_transport()
423
request = self._request_class(backing)
424
result = self._make_repository_and_result(
425
format='development-subtree')
426
# check the test will be valid
427
self.assertEqual('yes', result.args[2])
428
self.assertEqual('yes', result.args[3])
429
self.assertEqual(result, request.execute(''))
431
def test_supports_external_lookups_no_v2(self):
432
"""Test for the supports_external_lookups attribute."""
433
backing = self.get_transport()
434
request = self._request_class(backing)
435
result = self._make_repository_and_result(
436
format='development-subtree')
437
# check the test will be valid
438
self.assertEqual('yes', result.args[4])
439
self.assertEqual(result, request.execute(''))
442
class TestSmartServerBzrDirRequestGetConfigFile(
443
tests.TestCaseWithMemoryTransport):
444
"""Tests for BzrDir.get_config_file."""
446
def test_present(self):
447
backing = self.get_transport()
448
dir = self.make_controldir('.')
449
dir.get_config().set_default_stack_on("/")
450
local_result = dir._get_config()._get_config_file().read()
451
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
452
request = request_class(backing)
453
expected = smart_req.SuccessfulSmartServerResponse((), local_result)
454
self.assertEqual(expected, request.execute(''))
456
def test_missing(self):
457
backing = self.get_transport()
458
dir = self.make_controldir('.')
459
request_class = smart_dir.SmartServerBzrDirRequestConfigFile
460
request = request_class(backing)
461
expected = smart_req.SuccessfulSmartServerResponse((), '')
462
self.assertEqual(expected, request.execute(''))
465
class TestSmartServerBzrDirRequestGetBranches(
466
tests.TestCaseWithMemoryTransport):
467
"""Tests for BzrDir.get_branches."""
469
def test_simple(self):
470
backing = self.get_transport()
471
branch = self.make_branch('.')
472
request_class = smart_dir.SmartServerBzrDirRequestGetBranches
473
request = request_class(backing)
474
local_result = bencode.bencode(
475
{"": ("branch", branch._format.network_name())})
476
expected = smart_req.SuccessfulSmartServerResponse(
477
("success", ), local_result)
478
self.assertEqual(expected, request.execute(''))
480
def test_empty(self):
481
backing = self.get_transport()
482
dir = self.make_controldir('.')
483
request_class = smart_dir.SmartServerBzrDirRequestGetBranches
484
request = request_class(backing)
485
local_result = bencode.bencode({})
486
expected = smart_req.SuccessfulSmartServerResponse(
487
('success',), local_result)
488
self.assertEqual(expected, request.execute(''))
491
class TestSmartServerRequestInitializeBzrDir(tests.TestCaseWithMemoryTransport):
493
def test_empty_dir(self):
494
"""Initializing an empty dir should succeed and do it."""
495
backing = self.get_transport()
496
request = smart_dir.SmartServerRequestInitializeBzrDir(backing)
497
self.assertEqual(smart_req.SmartServerResponse(('ok', )),
499
made_dir = controldir.ControlDir.open_from_transport(backing)
500
# no branch, tree or repository is expected with the current
502
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
503
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
504
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
506
def test_missing_dir(self):
507
"""Initializing a missing directory should fail like the bzrdir api."""
508
backing = self.get_transport()
509
request = smart_dir.SmartServerRequestInitializeBzrDir(backing)
510
self.assertRaises(errors.NoSuchFile,
511
request.execute, 'subdir')
513
def test_initialized_dir(self):
514
"""Initializing an extant bzrdir should fail like the bzrdir api."""
515
backing = self.get_transport()
516
request = smart_dir.SmartServerRequestInitializeBzrDir(backing)
517
self.make_controldir('subdir')
518
self.assertRaises(errors.AlreadyControlDirError,
519
request.execute, 'subdir')
522
class TestSmartServerRequestBzrDirInitializeEx(
523
tests.TestCaseWithMemoryTransport):
524
"""Basic tests for BzrDir.initialize_ex_1.16 in the smart server.
526
The main unit tests in test_bzrdir exercise the API comprehensively.
529
def test_empty_dir(self):
530
"""Initializing an empty dir should succeed and do it."""
531
backing = self.get_transport()
532
name = self.make_controldir('reference')._format.network_name()
533
request = smart_dir.SmartServerRequestBzrDirInitializeEx(backing)
535
smart_req.SmartServerResponse(('', '', '', '', '', '', name,
536
'False', '', '', '')),
537
request.execute(name, '', 'True', 'False', 'False', '', '', '', '',
539
made_dir = controldir.ControlDir.open_from_transport(backing)
540
# no branch, tree or repository is expected with the current
542
self.assertRaises(errors.NoWorkingTree, made_dir.open_workingtree)
543
self.assertRaises(errors.NotBranchError, made_dir.open_branch)
544
self.assertRaises(errors.NoRepositoryPresent, made_dir.open_repository)
546
def test_missing_dir(self):
547
"""Initializing a missing directory should fail like the bzrdir api."""
548
backing = self.get_transport()
549
name = self.make_controldir('reference')._format.network_name()
550
request = smart_dir.SmartServerRequestBzrDirInitializeEx(backing)
551
self.assertRaises(errors.NoSuchFile, request.execute, name,
552
'subdir/dir', 'False', 'False', 'False', '', '', '', '', 'False')
554
def test_initialized_dir(self):
555
"""Initializing an extant directory should fail like the bzrdir api."""
556
backing = self.get_transport()
557
name = self.make_controldir('reference')._format.network_name()
558
request = smart_dir.SmartServerRequestBzrDirInitializeEx(backing)
559
self.make_controldir('subdir')
560
self.assertRaises(errors.FileExists, request.execute, name, 'subdir',
561
'False', 'False', 'False', '', '', '', '', 'False')
564
class TestSmartServerRequestOpenBzrDir(tests.TestCaseWithMemoryTransport):
566
def test_no_directory(self):
567
backing = self.get_transport()
568
request = smart_dir.SmartServerRequestOpenBzrDir(backing)
569
self.assertEqual(smart_req.SmartServerResponse(('no', )),
570
request.execute('does-not-exist'))
572
def test_empty_directory(self):
573
backing = self.get_transport()
574
backing.mkdir('empty')
575
request = smart_dir.SmartServerRequestOpenBzrDir(backing)
576
self.assertEqual(smart_req.SmartServerResponse(('no', )),
577
request.execute('empty'))
579
def test_outside_root_client_path(self):
580
backing = self.get_transport()
581
request = smart_dir.SmartServerRequestOpenBzrDir(backing,
582
root_client_path='root')
583
self.assertEqual(smart_req.SmartServerResponse(('no', )),
584
request.execute('not-root'))
587
class TestSmartServerRequestOpenBzrDir_2_1(tests.TestCaseWithMemoryTransport):
589
def test_no_directory(self):
590
backing = self.get_transport()
591
request = smart_dir.SmartServerRequestOpenBzrDir_2_1(backing)
592
self.assertEqual(smart_req.SmartServerResponse(('no', )),
593
request.execute('does-not-exist'))
595
def test_empty_directory(self):
596
backing = self.get_transport()
597
backing.mkdir('empty')
598
request = smart_dir.SmartServerRequestOpenBzrDir_2_1(backing)
599
self.assertEqual(smart_req.SmartServerResponse(('no', )),
600
request.execute('empty'))
602
def test_present_without_workingtree(self):
603
backing = self.get_transport()
604
request = smart_dir.SmartServerRequestOpenBzrDir_2_1(backing)
605
self.make_controldir('.')
606
self.assertEqual(smart_req.SmartServerResponse(('yes', 'no')),
609
def test_outside_root_client_path(self):
610
backing = self.get_transport()
611
request = smart_dir.SmartServerRequestOpenBzrDir_2_1(backing,
612
root_client_path='root')
613
self.assertEqual(smart_req.SmartServerResponse(('no',)),
614
request.execute('not-root'))
617
class TestSmartServerRequestOpenBzrDir_2_1_disk(TestCaseWithChrootedTransport):
619
def test_present_with_workingtree(self):
620
self.vfs_transport_factory = test_server.LocalURLServer
621
backing = self.get_transport()
622
request = smart_dir.SmartServerRequestOpenBzrDir_2_1(backing)
623
bd = self.make_controldir('.')
624
bd.create_repository()
626
bd.create_workingtree()
627
self.assertEqual(smart_req.SmartServerResponse(('yes', 'yes')),
631
class TestSmartServerRequestOpenBranch(TestCaseWithChrootedTransport):
633
def test_no_branch(self):
634
"""When there is no branch, ('nobranch', ) is returned."""
635
backing = self.get_transport()
636
request = smart_dir.SmartServerRequestOpenBranch(backing)
637
self.make_controldir('.')
638
self.assertEqual(smart_req.SmartServerResponse(('nobranch', )),
641
def test_branch(self):
642
"""When there is a branch, 'ok' is returned."""
643
backing = self.get_transport()
644
request = smart_dir.SmartServerRequestOpenBranch(backing)
645
self.make_branch('.')
646
self.assertEqual(smart_req.SmartServerResponse(('ok', '')),
649
def test_branch_reference(self):
650
"""When there is a branch reference, the reference URL is returned."""
651
self.vfs_transport_factory = test_server.LocalURLServer
652
backing = self.get_transport()
653
request = smart_dir.SmartServerRequestOpenBranch(backing)
654
branch = self.make_branch('branch')
655
checkout = branch.create_checkout('reference',lightweight=True)
656
reference_url = _mod_bzrbranch.BranchReferenceFormat().get_reference(
658
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
659
self.assertEqual(smart_req.SmartServerResponse(('ok', reference_url)),
660
request.execute('reference'))
662
def test_notification_on_branch_from_repository(self):
663
"""When there is a repository, the error should return details."""
664
backing = self.get_transport()
665
request = smart_dir.SmartServerRequestOpenBranch(backing)
666
repo = self.make_repository('.')
667
self.assertEqual(smart_req.SmartServerResponse(('nobranch',)),
671
class TestSmartServerRequestOpenBranchV2(TestCaseWithChrootedTransport):
673
def test_no_branch(self):
674
"""When there is no branch, ('nobranch', ) is returned."""
675
backing = self.get_transport()
676
self.make_controldir('.')
677
request = smart_dir.SmartServerRequestOpenBranchV2(backing)
678
self.assertEqual(smart_req.SmartServerResponse(('nobranch', )),
681
def test_branch(self):
682
"""When there is a branch, 'ok' is returned."""
683
backing = self.get_transport()
684
expected = self.make_branch('.')._format.network_name()
685
request = smart_dir.SmartServerRequestOpenBranchV2(backing)
686
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
687
('branch', expected)),
690
def test_branch_reference(self):
691
"""When there is a branch reference, the reference URL is returned."""
692
self.vfs_transport_factory = test_server.LocalURLServer
693
backing = self.get_transport()
694
request = smart_dir.SmartServerRequestOpenBranchV2(backing)
695
branch = self.make_branch('branch')
696
checkout = branch.create_checkout('reference',lightweight=True)
697
reference_url = _mod_bzrbranch.BranchReferenceFormat().get_reference(
699
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
700
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
701
('ref', reference_url)),
702
request.execute('reference'))
704
def test_stacked_branch(self):
705
"""Opening a stacked branch does not open the stacked-on branch."""
706
trunk = self.make_branch('trunk')
707
feature = self.make_branch('feature')
708
feature.set_stacked_on_url(trunk.base)
710
_mod_branch.Branch.hooks.install_named_hook(
711
'open', opened_branches.append, None)
712
backing = self.get_transport()
713
request = smart_dir.SmartServerRequestOpenBranchV2(backing)
716
response = request.execute('feature')
718
request.teardown_jail()
719
expected_format = feature._format.network_name()
720
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
721
('branch', expected_format)),
723
self.assertLength(1, opened_branches)
725
def test_notification_on_branch_from_repository(self):
726
"""When there is a repository, the error should return details."""
727
backing = self.get_transport()
728
request = smart_dir.SmartServerRequestOpenBranchV2(backing)
729
repo = self.make_repository('.')
730
self.assertEqual(smart_req.SmartServerResponse(('nobranch',)),
734
class TestSmartServerRequestOpenBranchV3(TestCaseWithChrootedTransport):
736
def test_no_branch(self):
737
"""When there is no branch, ('nobranch', ) is returned."""
738
backing = self.get_transport()
739
self.make_controldir('.')
740
request = smart_dir.SmartServerRequestOpenBranchV3(backing)
741
self.assertEqual(smart_req.SmartServerResponse(('nobranch',)),
744
def test_branch(self):
745
"""When there is a branch, 'ok' is returned."""
746
backing = self.get_transport()
747
expected = self.make_branch('.')._format.network_name()
748
request = smart_dir.SmartServerRequestOpenBranchV3(backing)
749
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
750
('branch', expected)),
753
def test_branch_reference(self):
754
"""When there is a branch reference, the reference URL is returned."""
755
self.vfs_transport_factory = test_server.LocalURLServer
756
backing = self.get_transport()
757
request = smart_dir.SmartServerRequestOpenBranchV3(backing)
758
branch = self.make_branch('branch')
759
checkout = branch.create_checkout('reference',lightweight=True)
760
reference_url = _mod_bzrbranch.BranchReferenceFormat().get_reference(
762
self.assertFileEqual(reference_url, 'reference/.bzr/branch/location')
763
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
764
('ref', reference_url)),
765
request.execute('reference'))
767
def test_stacked_branch(self):
768
"""Opening a stacked branch does not open the stacked-on branch."""
769
trunk = self.make_branch('trunk')
770
feature = self.make_branch('feature')
771
feature.set_stacked_on_url(trunk.base)
773
_mod_branch.Branch.hooks.install_named_hook(
774
'open', opened_branches.append, None)
775
backing = self.get_transport()
776
request = smart_dir.SmartServerRequestOpenBranchV3(backing)
779
response = request.execute('feature')
781
request.teardown_jail()
782
expected_format = feature._format.network_name()
783
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
784
('branch', expected_format)),
786
self.assertLength(1, opened_branches)
788
def test_notification_on_branch_from_repository(self):
789
"""When there is a repository, the error should return details."""
790
backing = self.get_transport()
791
request = smart_dir.SmartServerRequestOpenBranchV3(backing)
792
repo = self.make_repository('.')
793
self.assertEqual(smart_req.SmartServerResponse(
794
('nobranch', 'location is a repository')),
798
class TestSmartServerRequestRevisionHistory(tests.TestCaseWithMemoryTransport):
800
def test_empty(self):
801
"""For an empty branch, the body is empty."""
802
backing = self.get_transport()
803
request = smart_branch.SmartServerRequestRevisionHistory(backing)
804
self.make_branch('.')
805
self.assertEqual(smart_req.SmartServerResponse(('ok', ), ''),
808
def test_not_empty(self):
809
"""For a non-empty branch, the body is empty."""
810
backing = self.get_transport()
811
request = smart_branch.SmartServerRequestRevisionHistory(backing)
812
tree = self.make_branch_and_memory_tree('.')
815
r1 = tree.commit('1st commit')
816
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
819
smart_req.SmartServerResponse(('ok', ), ('\x00'.join([r1, r2]))),
823
class TestSmartServerBranchRequest(tests.TestCaseWithMemoryTransport):
825
def test_no_branch(self):
826
"""When there is a bzrdir and no branch, NotBranchError is raised."""
827
backing = self.get_transport()
828
request = smart_branch.SmartServerBranchRequest(backing)
829
self.make_controldir('.')
830
self.assertRaises(errors.NotBranchError,
833
def test_branch_reference(self):
834
"""When there is a branch reference, NotBranchError is raised."""
835
backing = self.get_transport()
836
request = smart_branch.SmartServerBranchRequest(backing)
837
branch = self.make_branch('branch')
838
checkout = branch.create_checkout('reference',lightweight=True)
839
self.assertRaises(errors.NotBranchError,
840
request.execute, 'checkout')
843
class TestSmartServerBranchRequestLastRevisionInfo(
844
tests.TestCaseWithMemoryTransport):
846
def test_empty(self):
847
"""For an empty branch, the result is ('ok', '0', 'null:')."""
848
backing = self.get_transport()
849
request = smart_branch.SmartServerBranchRequestLastRevisionInfo(backing)
850
self.make_branch('.')
851
self.assertEqual(smart_req.SmartServerResponse(('ok', '0', 'null:')),
854
def test_not_empty(self):
855
"""For a non-empty branch, the result is ('ok', 'revno', 'revid')."""
856
backing = self.get_transport()
857
request = smart_branch.SmartServerBranchRequestLastRevisionInfo(backing)
858
tree = self.make_branch_and_memory_tree('.')
861
rev_id_utf8 = u'\xc8'.encode('utf-8')
862
r1 = tree.commit('1st commit')
863
r2 = tree.commit('2nd commit', rev_id=rev_id_utf8)
866
smart_req.SmartServerResponse(('ok', '2', rev_id_utf8)),
870
class TestSmartServerBranchRequestRevisionIdToRevno(
871
tests.TestCaseWithMemoryTransport):
874
backing = self.get_transport()
875
request = smart_branch.SmartServerBranchRequestRevisionIdToRevno(
877
self.make_branch('.')
878
self.assertEqual(smart_req.SmartServerResponse(('ok', '0')),
879
request.execute('', 'null:'))
881
def test_simple(self):
882
backing = self.get_transport()
883
request = smart_branch.SmartServerBranchRequestRevisionIdToRevno(
885
tree = self.make_branch_and_memory_tree('.')
888
r1 = tree.commit('1st commit')
891
smart_req.SmartServerResponse(('ok', '1')),
892
request.execute('', r1))
894
def test_not_found(self):
895
backing = self.get_transport()
896
request = smart_branch.SmartServerBranchRequestRevisionIdToRevno(
898
branch = self.make_branch('.')
900
smart_req.FailedSmartServerResponse(
901
('NoSuchRevision', 'idontexist')),
902
request.execute('', 'idontexist'))
905
class TestSmartServerBranchRequestGetConfigFile(
906
tests.TestCaseWithMemoryTransport):
908
def test_default(self):
909
"""With no file, we get empty content."""
910
backing = self.get_transport()
911
request = smart_branch.SmartServerBranchGetConfigFile(backing)
912
branch = self.make_branch('.')
913
# there should be no file by default
915
self.assertEqual(smart_req.SmartServerResponse(('ok', ), content),
918
def test_with_content(self):
919
# SmartServerBranchGetConfigFile should return the content from
920
# branch.control_files.get('branch.conf') for now - in the future it may
921
# perform more complex processing.
922
backing = self.get_transport()
923
request = smart_branch.SmartServerBranchGetConfigFile(backing)
924
branch = self.make_branch('.')
925
branch._transport.put_bytes('branch.conf', 'foo bar baz')
926
self.assertEqual(smart_req.SmartServerResponse(('ok', ), 'foo bar baz'),
930
class TestLockedBranch(tests.TestCaseWithMemoryTransport):
932
def get_lock_tokens(self, branch):
933
branch_token = branch.lock_write().branch_token
934
repo_token = branch.repository.lock_write().repository_token
935
branch.repository.unlock()
936
return branch_token, repo_token
939
class TestSmartServerBranchRequestPutConfigFile(TestLockedBranch):
941
def test_with_content(self):
942
backing = self.get_transport()
943
request = smart_branch.SmartServerBranchPutConfigFile(backing)
944
branch = self.make_branch('.')
945
branch_token, repo_token = self.get_lock_tokens(branch)
946
self.assertIs(None, request.execute('', branch_token, repo_token))
948
smart_req.SmartServerResponse(('ok', )),
949
request.do_body('foo bar baz'))
951
branch.control_transport.get_bytes('branch.conf'),
956
class TestSmartServerBranchRequestSetConfigOption(TestLockedBranch):
958
def test_value_name(self):
959
branch = self.make_branch('.')
960
request = smart_branch.SmartServerBranchRequestSetConfigOption(
961
branch.controldir.root_transport)
962
branch_token, repo_token = self.get_lock_tokens(branch)
963
config = branch._get_config()
964
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
966
self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), result)
967
self.assertEqual('bar', config.get_option('foo'))
971
def test_value_name_section(self):
972
branch = self.make_branch('.')
973
request = smart_branch.SmartServerBranchRequestSetConfigOption(
974
branch.controldir.root_transport)
975
branch_token, repo_token = self.get_lock_tokens(branch)
976
config = branch._get_config()
977
result = request.execute('', branch_token, repo_token, 'bar', 'foo',
979
self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), result)
980
self.assertEqual('bar', config.get_option('foo', 'gam'))
985
class TestSmartServerBranchRequestSetConfigOptionDict(TestLockedBranch):
988
TestLockedBranch.setUp(self)
989
# A dict with non-ascii keys and values to exercise unicode
991
self.encoded_value_dict = (
992
'd5:ascii1:a11:unicode \xe2\x8c\x9a3:\xe2\x80\xbde')
994
'ascii': 'a', u'unicode \N{WATCH}': u'\N{INTERROBANG}'}
996
def test_value_name(self):
997
branch = self.make_branch('.')
998
request = smart_branch.SmartServerBranchRequestSetConfigOptionDict(
999
branch.controldir.root_transport)
1000
branch_token, repo_token = self.get_lock_tokens(branch)
1001
config = branch._get_config()
1002
result = request.execute('', branch_token, repo_token,
1003
self.encoded_value_dict, 'foo', '')
1004
self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), result)
1005
self.assertEqual(self.value_dict, config.get_option('foo'))
1009
def test_value_name_section(self):
1010
branch = self.make_branch('.')
1011
request = smart_branch.SmartServerBranchRequestSetConfigOptionDict(
1012
branch.controldir.root_transport)
1013
branch_token, repo_token = self.get_lock_tokens(branch)
1014
config = branch._get_config()
1015
result = request.execute('', branch_token, repo_token,
1016
self.encoded_value_dict, 'foo', 'gam')
1017
self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), result)
1018
self.assertEqual(self.value_dict, config.get_option('foo', 'gam'))
1023
class TestSmartServerBranchRequestSetTagsBytes(TestLockedBranch):
1024
# Only called when the branch format and tags match [yay factory
1025
# methods] so only need to test straight forward cases.
1027
def test_set_bytes(self):
1028
base_branch = self.make_branch('base')
1029
tag_bytes = base_branch._get_tags_bytes()
1030
# get_lock_tokens takes out a lock.
1031
branch_token, repo_token = self.get_lock_tokens(base_branch)
1032
request = smart_branch.SmartServerBranchSetTagsBytes(
1033
self.get_transport())
1034
response = request.execute('base', branch_token, repo_token)
1035
self.assertEqual(None, response)
1036
response = request.do_chunk(tag_bytes)
1037
self.assertEqual(None, response)
1038
response = request.do_end()
1040
smart_req.SuccessfulSmartServerResponse(()), response)
1041
base_branch.unlock()
1043
def test_lock_failed(self):
1044
base_branch = self.make_branch('base')
1045
base_branch.lock_write()
1046
tag_bytes = base_branch._get_tags_bytes()
1047
request = smart_branch.SmartServerBranchSetTagsBytes(
1048
self.get_transport())
1049
self.assertRaises(errors.TokenMismatch, request.execute,
1050
'base', 'wrong token', 'wrong token')
1051
# The request handler will keep processing the message parts, so even
1052
# if the request fails immediately do_chunk and do_end are still
1054
request.do_chunk(tag_bytes)
1056
base_branch.unlock()
1060
class SetLastRevisionTestBase(TestLockedBranch):
1061
"""Base test case for verbs that implement set_last_revision."""
1064
super(SetLastRevisionTestBase, self).setUp()
1065
backing_transport = self.get_transport()
1066
self.request = self.request_class(backing_transport)
1067
self.tree = self.make_branch_and_memory_tree('.')
1069
def lock_branch(self):
1070
return self.get_lock_tokens(self.tree.branch)
1072
def unlock_branch(self):
1073
self.tree.branch.unlock()
1075
def set_last_revision(self, revision_id, revno):
1076
branch_token, repo_token = self.lock_branch()
1077
response = self._set_last_revision(
1078
revision_id, revno, branch_token, repo_token)
1079
self.unlock_branch()
1082
def assertRequestSucceeds(self, revision_id, revno):
1083
response = self.set_last_revision(revision_id, revno)
1084
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok',)),
1088
class TestSetLastRevisionVerbMixin(object):
1089
"""Mixin test case for verbs that implement set_last_revision."""
1091
def test_set_null_to_null(self):
1092
"""An empty branch can have its last revision set to 'null:'."""
1093
self.assertRequestSucceeds('null:', 0)
1095
def test_NoSuchRevision(self):
1096
"""If the revision_id is not present, the verb returns NoSuchRevision.
1098
revision_id = 'non-existent revision'
1099
self.assertEqual(smart_req.FailedSmartServerResponse(('NoSuchRevision',
1101
self.set_last_revision(revision_id, 1))
1103
def make_tree_with_two_commits(self):
1104
self.tree.lock_write()
1106
rev_id_utf8 = u'\xc8'.encode('utf-8')
1107
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
1108
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
1111
def test_branch_last_revision_info_is_updated(self):
1112
"""A branch's tip can be set to a revision that is present in its
1115
# Make a branch with an empty revision history, but two revisions in
1117
self.make_tree_with_two_commits()
1118
rev_id_utf8 = u'\xc8'.encode('utf-8')
1119
self.tree.branch.set_last_revision_info(0, 'null:')
1121
(0, 'null:'), self.tree.branch.last_revision_info())
1122
# We can update the branch to a revision that is present in the
1124
self.assertRequestSucceeds(rev_id_utf8, 1)
1126
(1, rev_id_utf8), self.tree.branch.last_revision_info())
1128
def test_branch_last_revision_info_rewind(self):
1129
"""A branch's tip can be set to a revision that is an ancestor of the
1132
self.make_tree_with_two_commits()
1133
rev_id_utf8 = u'\xc8'.encode('utf-8')
1135
(2, 'rev-2'), self.tree.branch.last_revision_info())
1136
self.assertRequestSucceeds(rev_id_utf8, 1)
1138
(1, rev_id_utf8), self.tree.branch.last_revision_info())
1140
def test_TipChangeRejected(self):
1141
"""If a pre_change_branch_tip hook raises TipChangeRejected, the verb
1142
returns TipChangeRejected.
1144
rejection_message = u'rejection message\N{INTERROBANG}'
1145
def hook_that_rejects(params):
1146
raise errors.TipChangeRejected(rejection_message)
1147
_mod_branch.Branch.hooks.install_named_hook(
1148
'pre_change_branch_tip', hook_that_rejects, None)
1150
smart_req.FailedSmartServerResponse(
1151
('TipChangeRejected', rejection_message.encode('utf-8'))),
1152
self.set_last_revision('null:', 0))
1155
class TestSmartServerBranchRequestSetLastRevision(
1156
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
1157
"""Tests for Branch.set_last_revision verb."""
1159
request_class = smart_branch.SmartServerBranchRequestSetLastRevision
1161
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
1162
return self.request.execute(
1163
'', branch_token, repo_token, revision_id)
1166
class TestSmartServerBranchRequestSetLastRevisionInfo(
1167
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
1168
"""Tests for Branch.set_last_revision_info verb."""
1170
request_class = smart_branch.SmartServerBranchRequestSetLastRevisionInfo
1172
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
1173
return self.request.execute(
1174
'', branch_token, repo_token, revno, revision_id)
1176
def test_NoSuchRevision(self):
1177
"""Branch.set_last_revision_info does not have to return
1178
NoSuchRevision if the revision_id is absent.
1180
raise tests.TestNotApplicable()
1183
class TestSmartServerBranchRequestSetLastRevisionEx(
1184
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
1185
"""Tests for Branch.set_last_revision_ex verb."""
1187
request_class = smart_branch.SmartServerBranchRequestSetLastRevisionEx
1189
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
1190
return self.request.execute(
1191
'', branch_token, repo_token, revision_id, 0, 0)
1193
def assertRequestSucceeds(self, revision_id, revno):
1194
response = self.set_last_revision(revision_id, revno)
1196
smart_req.SuccessfulSmartServerResponse(('ok', revno, revision_id)),
1199
def test_branch_last_revision_info_rewind(self):
1200
"""A branch's tip can be set to a revision that is an ancestor of the
1201
current tip, but only if allow_overwrite_descendant is passed.
1203
self.make_tree_with_two_commits()
1204
rev_id_utf8 = u'\xc8'.encode('utf-8')
1206
(2, 'rev-2'), self.tree.branch.last_revision_info())
1207
# If allow_overwrite_descendant flag is 0, then trying to set the tip
1208
# to an older revision ID has no effect.
1209
branch_token, repo_token = self.lock_branch()
1210
response = self.request.execute(
1211
'', branch_token, repo_token, rev_id_utf8, 0, 0)
1213
smart_req.SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
1216
(2, 'rev-2'), self.tree.branch.last_revision_info())
1218
# If allow_overwrite_descendant flag is 1, then setting the tip to an
1220
response = self.request.execute(
1221
'', branch_token, repo_token, rev_id_utf8, 0, 1)
1223
smart_req.SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
1225
self.unlock_branch()
1227
(1, rev_id_utf8), self.tree.branch.last_revision_info())
1229
def make_branch_with_divergent_history(self):
1230
"""Make a branch with divergent history in its repo.
1232
The branch's tip will be 'child-2', and the repo will also contain
1233
'child-1', which diverges from a common base revision.
1235
self.tree.lock_write()
1237
r1 = self.tree.commit('1st commit')
1238
revno_1, revid_1 = self.tree.branch.last_revision_info()
1239
r2 = self.tree.commit('2nd commit', rev_id='child-1')
1240
# Undo the second commit
1241
self.tree.branch.set_last_revision_info(revno_1, revid_1)
1242
self.tree.set_parent_ids([revid_1])
1243
# Make a new second commit, child-2. child-2 has diverged from
1245
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
1248
def test_not_allow_diverged(self):
1249
"""If allow_diverged is not passed, then setting a divergent history
1250
returns a Diverged error.
1252
self.make_branch_with_divergent_history()
1254
smart_req.FailedSmartServerResponse(('Diverged',)),
1255
self.set_last_revision('child-1', 2))
1256
# The branch tip was not changed.
1257
self.assertEqual('child-2', self.tree.branch.last_revision())
1259
def test_allow_diverged(self):
1260
"""If allow_diverged is passed, then setting a divergent history
1263
self.make_branch_with_divergent_history()
1264
branch_token, repo_token = self.lock_branch()
1265
response = self.request.execute(
1266
'', branch_token, repo_token, 'child-1', 1, 0)
1268
smart_req.SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
1270
self.unlock_branch()
1271
# The branch tip was changed.
1272
self.assertEqual('child-1', self.tree.branch.last_revision())
1275
class TestSmartServerBranchBreakLock(tests.TestCaseWithMemoryTransport):
1277
def test_lock_to_break(self):
1278
base_branch = self.make_branch('base')
1279
request = smart_branch.SmartServerBranchBreakLock(
1280
self.get_transport())
1281
base_branch.lock_write()
1283
smart_req.SuccessfulSmartServerResponse(('ok', ), None),
1284
request.execute('base'))
1286
def test_nothing_to_break(self):
1287
base_branch = self.make_branch('base')
1288
request = smart_branch.SmartServerBranchBreakLock(
1289
self.get_transport())
1291
smart_req.SuccessfulSmartServerResponse(('ok', ), None),
1292
request.execute('base'))
1295
class TestSmartServerBranchRequestGetParent(tests.TestCaseWithMemoryTransport):
1297
def test_get_parent_none(self):
1298
base_branch = self.make_branch('base')
1299
request = smart_branch.SmartServerBranchGetParent(self.get_transport())
1300
response = request.execute('base')
1302
smart_req.SuccessfulSmartServerResponse(('',)), response)
1304
def test_get_parent_something(self):
1305
base_branch = self.make_branch('base')
1306
base_branch.set_parent(self.get_url('foo'))
1307
request = smart_branch.SmartServerBranchGetParent(self.get_transport())
1308
response = request.execute('base')
1310
smart_req.SuccessfulSmartServerResponse(("../foo",)),
1314
class TestSmartServerBranchRequestSetParent(TestLockedBranch):
1316
def test_set_parent_none(self):
1317
branch = self.make_branch('base', format="1.9")
1319
branch._set_parent_location('foo')
1321
request = smart_branch.SmartServerBranchRequestSetParentLocation(
1322
self.get_transport())
1323
branch_token, repo_token = self.get_lock_tokens(branch)
1325
response = request.execute('base', branch_token, repo_token, '')
1328
self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), response)
1329
# Refresh branch as SetParentLocation modified it
1330
branch = branch.controldir.open_branch()
1331
self.assertEqual(None, branch.get_parent())
1333
def test_set_parent_something(self):
1334
branch = self.make_branch('base', format="1.9")
1335
request = smart_branch.SmartServerBranchRequestSetParentLocation(
1336
self.get_transport())
1337
branch_token, repo_token = self.get_lock_tokens(branch)
1339
response = request.execute('base', branch_token, repo_token,
1343
self.assertEqual(smart_req.SuccessfulSmartServerResponse(()), response)
1344
refreshed = _mod_branch.Branch.open(branch.base)
1345
self.assertEqual('http://bar/', refreshed.get_parent())
1348
class TestSmartServerBranchRequestGetTagsBytes(
1349
tests.TestCaseWithMemoryTransport):
1350
# Only called when the branch format and tags match [yay factory
1351
# methods] so only need to test straight forward cases.
1353
def test_get_bytes(self):
1354
base_branch = self.make_branch('base')
1355
request = smart_branch.SmartServerBranchGetTagsBytes(
1356
self.get_transport())
1357
response = request.execute('base')
1359
smart_req.SuccessfulSmartServerResponse(('',)), response)
1362
class TestSmartServerBranchRequestGetStackedOnURL(tests.TestCaseWithMemoryTransport):
1364
def test_get_stacked_on_url(self):
1365
base_branch = self.make_branch('base', format='1.6')
1366
stacked_branch = self.make_branch('stacked', format='1.6')
1367
# typically should be relative
1368
stacked_branch.set_stacked_on_url('../base')
1369
request = smart_branch.SmartServerBranchRequestGetStackedOnURL(
1370
self.get_transport())
1371
response = request.execute('stacked')
1373
smart_req.SmartServerResponse(('ok', '../base')),
1377
class TestSmartServerBranchRequestLockWrite(TestLockedBranch):
1379
def test_lock_write_on_unlocked_branch(self):
1380
backing = self.get_transport()
1381
request = smart_branch.SmartServerBranchRequestLockWrite(backing)
1382
branch = self.make_branch('.', format='knit')
1383
repository = branch.repository
1384
response = request.execute('')
1385
branch_nonce = branch.control_files._lock.peek().get('nonce')
1386
repository_nonce = repository.control_files._lock.peek().get('nonce')
1387
self.assertEqual(smart_req.SmartServerResponse(
1388
('ok', branch_nonce, repository_nonce)),
1390
# The branch (and associated repository) is now locked. Verify that
1391
# with a new branch object.
1392
new_branch = repository.controldir.open_branch()
1393
self.assertRaises(errors.LockContention, new_branch.lock_write)
1395
request = smart_branch.SmartServerBranchRequestUnlock(backing)
1396
response = request.execute('', branch_nonce, repository_nonce)
1398
def test_lock_write_on_locked_branch(self):
1399
backing = self.get_transport()
1400
request = smart_branch.SmartServerBranchRequestLockWrite(backing)
1401
branch = self.make_branch('.')
1402
branch_token = branch.lock_write().branch_token
1403
branch.leave_lock_in_place()
1405
response = request.execute('')
1407
smart_req.SmartServerResponse(('LockContention',)), response)
1409
branch.lock_write(branch_token)
1410
branch.dont_leave_lock_in_place()
1413
def test_lock_write_with_tokens_on_locked_branch(self):
1414
backing = self.get_transport()
1415
request = smart_branch.SmartServerBranchRequestLockWrite(backing)
1416
branch = self.make_branch('.', format='knit')
1417
branch_token, repo_token = self.get_lock_tokens(branch)
1418
branch.leave_lock_in_place()
1419
branch.repository.leave_lock_in_place()
1421
response = request.execute('',
1422
branch_token, repo_token)
1424
smart_req.SmartServerResponse(('ok', branch_token, repo_token)),
1427
branch.repository.lock_write(repo_token)
1428
branch.repository.dont_leave_lock_in_place()
1429
branch.repository.unlock()
1430
branch.lock_write(branch_token)
1431
branch.dont_leave_lock_in_place()
1434
def test_lock_write_with_mismatched_tokens_on_locked_branch(self):
1435
backing = self.get_transport()
1436
request = smart_branch.SmartServerBranchRequestLockWrite(backing)
1437
branch = self.make_branch('.', format='knit')
1438
branch_token, repo_token = self.get_lock_tokens(branch)
1439
branch.leave_lock_in_place()
1440
branch.repository.leave_lock_in_place()
1442
response = request.execute('',
1443
branch_token+'xxx', repo_token)
1445
smart_req.SmartServerResponse(('TokenMismatch',)), response)
1447
branch.repository.lock_write(repo_token)
1448
branch.repository.dont_leave_lock_in_place()
1449
branch.repository.unlock()
1450
branch.lock_write(branch_token)
1451
branch.dont_leave_lock_in_place()
1454
def test_lock_write_on_locked_repo(self):
1455
backing = self.get_transport()
1456
request = smart_branch.SmartServerBranchRequestLockWrite(backing)
1457
branch = self.make_branch('.', format='knit')
1458
repo = branch.repository
1459
repo_token = repo.lock_write().repository_token
1460
repo.leave_lock_in_place()
1462
response = request.execute('')
1464
smart_req.SmartServerResponse(('LockContention',)), response)
1466
repo.lock_write(repo_token)
1467
repo.dont_leave_lock_in_place()
1470
def test_lock_write_on_readonly_transport(self):
1471
backing = self.get_readonly_transport()
1472
request = smart_branch.SmartServerBranchRequestLockWrite(backing)
1473
branch = self.make_branch('.')
1474
root = self.get_transport().clone('/')
1475
path = urlutils.relative_url(root.base, self.get_transport().base)
1476
response = request.execute(path)
1477
error_name, lock_str, why_str = response.args
1478
self.assertFalse(response.is_successful())
1479
self.assertEqual('LockFailed', error_name)
1482
class TestSmartServerBranchRequestGetPhysicalLockStatus(TestLockedBranch):
1484
def test_true(self):
1485
backing = self.get_transport()
1486
request = smart_branch.SmartServerBranchRequestGetPhysicalLockStatus(
1488
branch = self.make_branch('.')
1489
branch_token, repo_token = self.get_lock_tokens(branch)
1490
self.assertEqual(True, branch.get_physical_lock_status())
1491
response = request.execute('')
1493
smart_req.SmartServerResponse(('yes',)), response)
1496
def test_false(self):
1497
backing = self.get_transport()
1498
request = smart_branch.SmartServerBranchRequestGetPhysicalLockStatus(
1500
branch = self.make_branch('.')
1501
self.assertEqual(False, branch.get_physical_lock_status())
1502
response = request.execute('')
1504
smart_req.SmartServerResponse(('no',)), response)
1507
class TestSmartServerBranchRequestUnlock(TestLockedBranch):
1509
def test_unlock_on_locked_branch_and_repo(self):
1510
backing = self.get_transport()
1511
request = smart_branch.SmartServerBranchRequestUnlock(backing)
1512
branch = self.make_branch('.', format='knit')
1514
branch_token, repo_token = self.get_lock_tokens(branch)
1515
# Unlock the branch (and repo) object, leaving the physical locks
1517
branch.leave_lock_in_place()
1518
branch.repository.leave_lock_in_place()
1520
response = request.execute('',
1521
branch_token, repo_token)
1523
smart_req.SmartServerResponse(('ok',)), response)
1524
# The branch is now unlocked. Verify that with a new branch
1526
new_branch = branch.controldir.open_branch()
1527
new_branch.lock_write()
1530
def test_unlock_on_unlocked_branch_unlocked_repo(self):
1531
backing = self.get_transport()
1532
request = smart_branch.SmartServerBranchRequestUnlock(backing)
1533
branch = self.make_branch('.', format='knit')
1534
response = request.execute(
1535
'', 'branch token', 'repo token')
1537
smart_req.SmartServerResponse(('TokenMismatch',)), response)
1539
def test_unlock_on_unlocked_branch_locked_repo(self):
1540
backing = self.get_transport()
1541
request = smart_branch.SmartServerBranchRequestUnlock(backing)
1542
branch = self.make_branch('.', format='knit')
1543
# Lock the repository.
1544
repo_token = branch.repository.lock_write().repository_token
1545
branch.repository.leave_lock_in_place()
1546
branch.repository.unlock()
1547
# Issue branch lock_write request on the unlocked branch (with locked
1549
response = request.execute('', 'branch token', repo_token)
1551
smart_req.SmartServerResponse(('TokenMismatch',)), response)
1553
branch.repository.lock_write(repo_token)
1554
branch.repository.dont_leave_lock_in_place()
1555
branch.repository.unlock()
1558
class TestSmartServerRepositoryRequest(tests.TestCaseWithMemoryTransport):
1560
def test_no_repository(self):
1561
"""Raise NoRepositoryPresent when there is a bzrdir and no repo."""
1562
# we test this using a shared repository above the named path,
1563
# thus checking the right search logic is used - that is, that
1564
# its the exact path being looked at and the server is not
1566
backing = self.get_transport()
1567
request = smart_repo.SmartServerRepositoryRequest(backing)
1568
self.make_repository('.', shared=True)
1569
self.make_controldir('subdir')
1570
self.assertRaises(errors.NoRepositoryPresent,
1571
request.execute, 'subdir')
1574
class TestSmartServerRepositoryAddSignatureText(tests.TestCaseWithMemoryTransport):
1576
def test_add_text(self):
1577
backing = self.get_transport()
1578
request = smart_repo.SmartServerRepositoryAddSignatureText(backing)
1579
tree = self.make_branch_and_memory_tree('.')
1580
write_token = tree.lock_write()
1581
self.addCleanup(tree.unlock)
1583
tree.commit("Message", rev_id='rev1')
1584
tree.branch.repository.start_write_group()
1585
write_group_tokens = tree.branch.repository.suspend_write_group()
1586
self.assertEqual(None, request.execute('', write_token,
1587
'rev1', *write_group_tokens))
1588
response = request.do_body('somesignature')
1589
self.assertTrue(response.is_successful())
1590
self.assertEqual(response.args[0], 'ok')
1591
write_group_tokens = response.args[1:]
1592
tree.branch.repository.resume_write_group(write_group_tokens)
1593
tree.branch.repository.commit_write_group()
1595
self.assertEqual("somesignature",
1596
tree.branch.repository.get_signature_text("rev1"))
1599
class TestSmartServerRepositoryAllRevisionIds(
1600
tests.TestCaseWithMemoryTransport):
1602
def test_empty(self):
1603
"""An empty body should be returned for an empty repository."""
1604
backing = self.get_transport()
1605
request = smart_repo.SmartServerRepositoryAllRevisionIds(backing)
1606
self.make_repository('.')
1608
smart_req.SuccessfulSmartServerResponse(("ok", ), ""),
1609
request.execute(''))
1611
def test_some_revisions(self):
1612
"""An empty body should be returned for an empty repository."""
1613
backing = self.get_transport()
1614
request = smart_repo.SmartServerRepositoryAllRevisionIds(backing)
1615
tree = self.make_branch_and_memory_tree('.')
1618
tree.commit(rev_id='origineel', message="message")
1619
tree.commit(rev_id='nog-een-revisie', message="message")
1622
smart_req.SuccessfulSmartServerResponse(("ok", ),
1623
"origineel\nnog-een-revisie"),
1624
request.execute(''))
1627
class TestSmartServerRepositoryBreakLock(tests.TestCaseWithMemoryTransport):
1629
def test_lock_to_break(self):
1630
backing = self.get_transport()
1631
request = smart_repo.SmartServerRepositoryBreakLock(backing)
1632
tree = self.make_branch_and_memory_tree('.')
1633
tree.branch.repository.lock_write()
1635
smart_req.SuccessfulSmartServerResponse(('ok', ), None),
1636
request.execute(''))
1638
def test_nothing_to_break(self):
1639
backing = self.get_transport()
1640
request = smart_repo.SmartServerRepositoryBreakLock(backing)
1641
tree = self.make_branch_and_memory_tree('.')
1643
smart_req.SuccessfulSmartServerResponse(('ok', ), None),
1644
request.execute(''))
1647
class TestSmartServerRepositoryGetParentMap(tests.TestCaseWithMemoryTransport):
1649
def test_trivial_bzipped(self):
1650
# This tests that the wire encoding is actually bzipped
1651
backing = self.get_transport()
1652
request = smart_repo.SmartServerRepositoryGetParentMap(backing)
1653
tree = self.make_branch_and_memory_tree('.')
1655
self.assertEqual(None,
1656
request.execute('', 'missing-id'))
1657
# Note that it returns a body that is bzipped.
1659
smart_req.SuccessfulSmartServerResponse(('ok', ), bz2.compress('')),
1660
request.do_body('\n\n0\n'))
1662
def test_trivial_include_missing(self):
1663
backing = self.get_transport()
1664
request = smart_repo.SmartServerRepositoryGetParentMap(backing)
1665
tree = self.make_branch_and_memory_tree('.')
1667
self.assertEqual(None,
1668
request.execute('', 'missing-id', 'include-missing:'))
1670
smart_req.SuccessfulSmartServerResponse(('ok', ),
1671
bz2.compress('missing:missing-id')),
1672
request.do_body('\n\n0\n'))
1675
class TestSmartServerRepositoryGetRevisionGraph(
1676
tests.TestCaseWithMemoryTransport):
1678
def test_none_argument(self):
1679
backing = self.get_transport()
1680
request = smart_repo.SmartServerRepositoryGetRevisionGraph(backing)
1681
tree = self.make_branch_and_memory_tree('.')
1684
r1 = tree.commit('1st commit')
1685
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1688
# the lines of revision_id->revision_parent_list has no guaranteed
1689
# order coming out of a dict, so sort both our test and response
1690
lines = sorted([' '.join([r2, r1]), r1])
1691
response = request.execute('', '')
1692
response.body = '\n'.join(sorted(response.body.split('\n')))
1695
smart_req.SmartServerResponse(('ok', ), '\n'.join(lines)), response)
1697
def test_specific_revision_argument(self):
1698
backing = self.get_transport()
1699
request = smart_repo.SmartServerRepositoryGetRevisionGraph(backing)
1700
tree = self.make_branch_and_memory_tree('.')
1703
rev_id_utf8 = u'\xc9'.encode('utf-8')
1704
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
1705
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1708
self.assertEqual(smart_req.SmartServerResponse(('ok', ), rev_id_utf8),
1709
request.execute('', rev_id_utf8))
1711
def test_no_such_revision(self):
1712
backing = self.get_transport()
1713
request = smart_repo.SmartServerRepositoryGetRevisionGraph(backing)
1714
tree = self.make_branch_and_memory_tree('.')
1717
r1 = tree.commit('1st commit')
1720
# Note that it still returns body (of zero bytes).
1721
self.assertEqual(smart_req.SmartServerResponse(
1722
('nosuchrevision', 'missingrevision', ), ''),
1723
request.execute('', 'missingrevision'))
1726
class TestSmartServerRepositoryGetRevIdForRevno(
1727
tests.TestCaseWithMemoryTransport):
1729
def test_revno_found(self):
1730
backing = self.get_transport()
1731
request = smart_repo.SmartServerRepositoryGetRevIdForRevno(backing)
1732
tree = self.make_branch_and_memory_tree('.')
1735
rev1_id_utf8 = u'\xc8'.encode('utf-8')
1736
rev2_id_utf8 = u'\xc9'.encode('utf-8')
1737
tree.commit('1st commit', rev_id=rev1_id_utf8)
1738
tree.commit('2nd commit', rev_id=rev2_id_utf8)
1741
self.assertEqual(smart_req.SmartServerResponse(('ok', rev1_id_utf8)),
1742
request.execute('', 1, (2, rev2_id_utf8)))
1744
def test_known_revid_missing(self):
1745
backing = self.get_transport()
1746
request = smart_repo.SmartServerRepositoryGetRevIdForRevno(backing)
1747
repo = self.make_repository('.')
1749
smart_req.FailedSmartServerResponse(('nosuchrevision', 'ghost')),
1750
request.execute('', 1, (2, 'ghost')))
1752
def test_history_incomplete(self):
1753
backing = self.get_transport()
1754
request = smart_repo.SmartServerRepositoryGetRevIdForRevno(backing)
1755
parent = self.make_branch_and_memory_tree('parent', format='1.9')
1757
parent.add([''], ['TREE_ROOT'])
1758
r1 = parent.commit(message='first commit')
1759
r2 = parent.commit(message='second commit')
1761
local = self.make_branch_and_memory_tree('local', format='1.9')
1762
local.branch.pull(parent.branch)
1763
local.set_parent_ids([r2])
1764
r3 = local.commit(message='local commit')
1765
local.branch.create_clone_on_transport(
1766
self.get_transport('stacked'), stacked_on=self.get_url('parent'))
1768
smart_req.SmartServerResponse(('history-incomplete', 2, r2)),
1769
request.execute('stacked', 1, (3, r3)))
1772
class TestSmartServerRepositoryIterRevisions(
1773
tests.TestCaseWithMemoryTransport):
1775
def test_basic(self):
1776
backing = self.get_transport()
1777
request = smart_repo.SmartServerRepositoryIterRevisions(backing)
1778
tree = self.make_branch_and_memory_tree('.', format='2a')
1781
tree.commit('1st commit', rev_id="rev1")
1782
tree.commit('2nd commit', rev_id="rev2")
1785
self.assertIs(None, request.execute(''))
1786
response = request.do_body("rev1\nrev2")
1787
self.assertTrue(response.is_successful())
1788
# Format 2a uses serializer format 10
1789
self.assertEqual(response.args, ("ok", "10"))
1791
self.addCleanup(tree.branch.lock_read().unlock)
1792
entries = [zlib.compress(record.get_bytes_as("fulltext")) for record in
1793
tree.branch.repository.revisions.get_record_stream(
1794
[("rev1", ), ("rev2", )], "unordered", True)]
1796
contents = "".join(response.body_stream)
1797
self.assertTrue(contents in (
1798
"".join([entries[0], entries[1]]),
1799
"".join([entries[1], entries[0]])))
1801
def test_missing(self):
1802
backing = self.get_transport()
1803
request = smart_repo.SmartServerRepositoryIterRevisions(backing)
1804
tree = self.make_branch_and_memory_tree('.', format='2a')
1806
self.assertIs(None, request.execute(''))
1807
response = request.do_body("rev1\nrev2")
1808
self.assertTrue(response.is_successful())
1809
# Format 2a uses serializer format 10
1810
self.assertEqual(response.args, ("ok", "10"))
1812
contents = "".join(response.body_stream)
1813
self.assertEqual(contents, "")
1816
class GetStreamTestBase(tests.TestCaseWithMemoryTransport):
1818
def make_two_commit_repo(self):
1819
tree = self.make_branch_and_memory_tree('.')
1822
r1 = tree.commit('1st commit')
1823
r2 = tree.commit('2nd commit', rev_id=u'\xc8'.encode('utf-8'))
1825
repo = tree.branch.repository
1829
class TestSmartServerRepositoryGetStream(GetStreamTestBase):
1831
def test_ancestry_of(self):
1832
"""The search argument may be a 'ancestry-of' some heads'."""
1833
backing = self.get_transport()
1834
request = smart_repo.SmartServerRepositoryGetStream(backing)
1835
repo, r1, r2 = self.make_two_commit_repo()
1836
fetch_spec = ['ancestry-of', r2]
1837
lines = '\n'.join(fetch_spec)
1838
request.execute('', repo._format.network_name())
1839
response = request.do_body(lines)
1840
self.assertEqual(('ok',), response.args)
1841
stream_bytes = ''.join(response.body_stream)
1842
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1844
def test_search(self):
1845
"""The search argument may be a 'search' of some explicit keys."""
1846
backing = self.get_transport()
1847
request = smart_repo.SmartServerRepositoryGetStream(backing)
1848
repo, r1, r2 = self.make_two_commit_repo()
1849
fetch_spec = ['search', '%s %s' % (r1, r2), 'null:', '2']
1850
lines = '\n'.join(fetch_spec)
1851
request.execute('', repo._format.network_name())
1852
response = request.do_body(lines)
1853
self.assertEqual(('ok',), response.args)
1854
stream_bytes = ''.join(response.body_stream)
1855
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1857
def test_search_everything(self):
1858
"""A search of 'everything' returns a stream."""
1859
backing = self.get_transport()
1860
request = smart_repo.SmartServerRepositoryGetStream_1_19(backing)
1861
repo, r1, r2 = self.make_two_commit_repo()
1862
serialised_fetch_spec = 'everything'
1863
request.execute('', repo._format.network_name())
1864
response = request.do_body(serialised_fetch_spec)
1865
self.assertEqual(('ok',), response.args)
1866
stream_bytes = ''.join(response.body_stream)
1867
self.assertStartsWith(stream_bytes, 'Bazaar pack format 1')
1870
class TestSmartServerRequestHasRevision(tests.TestCaseWithMemoryTransport):
1872
def test_missing_revision(self):
1873
"""For a missing revision, ('no', ) is returned."""
1874
backing = self.get_transport()
1875
request = smart_repo.SmartServerRequestHasRevision(backing)
1876
self.make_repository('.')
1877
self.assertEqual(smart_req.SmartServerResponse(('no', )),
1878
request.execute('', 'revid'))
1880
def test_present_revision(self):
1881
"""For a present revision, ('yes', ) is returned."""
1882
backing = self.get_transport()
1883
request = smart_repo.SmartServerRequestHasRevision(backing)
1884
tree = self.make_branch_and_memory_tree('.')
1887
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1888
r1 = tree.commit('a commit', rev_id=rev_id_utf8)
1890
self.assertTrue(tree.branch.repository.has_revision(rev_id_utf8))
1891
self.assertEqual(smart_req.SmartServerResponse(('yes', )),
1892
request.execute('', rev_id_utf8))
1895
class TestSmartServerRepositoryIterFilesBytes(tests.TestCaseWithTransport):
1897
def test_single(self):
1898
backing = self.get_transport()
1899
request = smart_repo.SmartServerRepositoryIterFilesBytes(backing)
1900
t = self.make_branch_and_tree('.')
1901
self.addCleanup(t.lock_write().unlock)
1902
self.build_tree_contents([("file", "somecontents")])
1903
t.add(["file"], ["thefileid"])
1904
t.commit(rev_id='somerev', message="add file")
1905
self.assertIs(None, request.execute(''))
1906
response = request.do_body("thefileid\0somerev\n")
1907
self.assertTrue(response.is_successful())
1908
self.assertEqual(response.args, ("ok", ))
1909
self.assertEqual("".join(response.body_stream),
1910
"ok\x000\n" + zlib.compress("somecontents"))
1912
def test_missing(self):
1913
backing = self.get_transport()
1914
request = smart_repo.SmartServerRepositoryIterFilesBytes(backing)
1915
t = self.make_branch_and_tree('.')
1916
self.addCleanup(t.lock_write().unlock)
1917
self.assertIs(None, request.execute(''))
1918
response = request.do_body("thefileid\0revision\n")
1919
self.assertTrue(response.is_successful())
1920
self.assertEqual(response.args, ("ok", ))
1921
self.assertEqual("".join(response.body_stream),
1922
"absent\x00thefileid\x00revision\x000\n")
1925
class TestSmartServerRequestHasSignatureForRevisionId(
1926
tests.TestCaseWithMemoryTransport):
1928
def test_missing_revision(self):
1929
"""For a missing revision, NoSuchRevision is returned."""
1930
backing = self.get_transport()
1931
request = smart_repo.SmartServerRequestHasSignatureForRevisionId(
1933
self.make_repository('.')
1935
smart_req.FailedSmartServerResponse(
1936
('nosuchrevision', 'revid'), None),
1937
request.execute('', 'revid'))
1939
def test_missing_signature(self):
1940
"""For a missing signature, ('no', ) is returned."""
1941
backing = self.get_transport()
1942
request = smart_repo.SmartServerRequestHasSignatureForRevisionId(
1944
tree = self.make_branch_and_memory_tree('.')
1947
r1 = tree.commit('a commit', rev_id='A')
1949
self.assertTrue(tree.branch.repository.has_revision('A'))
1950
self.assertEqual(smart_req.SmartServerResponse(('no', )),
1951
request.execute('', 'A'))
1953
def test_present_signature(self):
1954
"""For a present signature, ('yes', ) is returned."""
1955
backing = self.get_transport()
1956
request = smart_repo.SmartServerRequestHasSignatureForRevisionId(
1958
strategy = gpg.LoopbackGPGStrategy(None)
1959
tree = self.make_branch_and_memory_tree('.')
1962
r1 = tree.commit('a commit', rev_id='A')
1963
tree.branch.repository.start_write_group()
1964
tree.branch.repository.sign_revision('A', strategy)
1965
tree.branch.repository.commit_write_group()
1967
self.assertTrue(tree.branch.repository.has_revision('A'))
1968
self.assertEqual(smart_req.SmartServerResponse(('yes', )),
1969
request.execute('', 'A'))
1972
class TestSmartServerRepositoryGatherStats(tests.TestCaseWithMemoryTransport):
1974
def test_empty_revid(self):
1975
"""With an empty revid, we get only size an number and revisions"""
1976
backing = self.get_transport()
1977
request = smart_repo.SmartServerRepositoryGatherStats(backing)
1978
repository = self.make_repository('.')
1979
stats = repository.gather_stats()
1980
expected_body = 'revisions: 0\n'
1981
self.assertEqual(smart_req.SmartServerResponse(('ok', ), expected_body),
1982
request.execute('', '', 'no'))
1984
def test_revid_with_committers(self):
1985
"""For a revid we get more infos."""
1986
backing = self.get_transport()
1987
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
1988
request = smart_repo.SmartServerRepositoryGatherStats(backing)
1989
tree = self.make_branch_and_memory_tree('.')
1992
# Let's build a predictable result
1993
tree.commit('a commit', timestamp=123456.2, timezone=3600)
1994
tree.commit('a commit', timestamp=654321.4, timezone=0,
1998
stats = tree.branch.repository.gather_stats()
1999
expected_body = ('firstrev: 123456.200 3600\n'
2000
'latestrev: 654321.400 0\n'
2002
self.assertEqual(smart_req.SmartServerResponse(('ok', ), expected_body),
2006
def test_not_empty_repository_with_committers(self):
2007
"""For a revid and requesting committers we get the whole thing."""
2008
backing = self.get_transport()
2009
rev_id_utf8 = u'\xc8abc'.encode('utf-8')
2010
request = smart_repo.SmartServerRepositoryGatherStats(backing)
2011
tree = self.make_branch_and_memory_tree('.')
2014
# Let's build a predictable result
2015
tree.commit('a commit', timestamp=123456.2, timezone=3600,
2017
tree.commit('a commit', timestamp=654321.4, timezone=0,
2018
committer='bar', rev_id=rev_id_utf8)
2020
stats = tree.branch.repository.gather_stats()
2022
expected_body = ('committers: 2\n'
2023
'firstrev: 123456.200 3600\n'
2024
'latestrev: 654321.400 0\n'
2026
self.assertEqual(smart_req.SmartServerResponse(('ok', ), expected_body),
2028
rev_id_utf8, 'yes'))
2030
def test_unknown_revid(self):
2031
"""An unknown revision id causes a 'nosuchrevision' error."""
2032
backing = self.get_transport()
2033
request = smart_repo.SmartServerRepositoryGatherStats(backing)
2034
repository = self.make_repository('.')
2035
expected_body = 'revisions: 0\n'
2037
smart_req.FailedSmartServerResponse(
2038
('nosuchrevision', 'mia'), None),
2039
request.execute('', 'mia', 'yes'))
2042
class TestSmartServerRepositoryIsShared(tests.TestCaseWithMemoryTransport):
2044
def test_is_shared(self):
2045
"""For a shared repository, ('yes', ) is returned."""
2046
backing = self.get_transport()
2047
request = smart_repo.SmartServerRepositoryIsShared(backing)
2048
self.make_repository('.', shared=True)
2049
self.assertEqual(smart_req.SmartServerResponse(('yes', )),
2050
request.execute('', ))
2052
def test_is_not_shared(self):
2053
"""For a shared repository, ('no', ) is returned."""
2054
backing = self.get_transport()
2055
request = smart_repo.SmartServerRepositoryIsShared(backing)
2056
self.make_repository('.', shared=False)
2057
self.assertEqual(smart_req.SmartServerResponse(('no', )),
2058
request.execute('', ))
2061
class TestSmartServerRepositoryGetRevisionSignatureText(
2062
tests.TestCaseWithMemoryTransport):
2064
def test_get_signature(self):
2065
backing = self.get_transport()
2066
request = smart_repo.SmartServerRepositoryGetRevisionSignatureText(
2068
bb = self.make_branch_builder('.')
2069
bb.build_commit(rev_id='A')
2070
repo = bb.get_branch().repository
2071
strategy = gpg.LoopbackGPGStrategy(None)
2072
self.addCleanup(repo.lock_write().unlock)
2073
repo.start_write_group()
2074
repo.sign_revision('A', strategy)
2075
repo.commit_write_group()
2077
'-----BEGIN PSEUDO-SIGNED CONTENT-----\n' +
2078
Testament.from_revision(repo, 'A').as_short_text() +
2079
'-----END PSEUDO-SIGNED CONTENT-----\n')
2081
smart_req.SmartServerResponse(('ok', ), expected_body),
2082
request.execute('', 'A'))
2085
class TestSmartServerRepositoryMakeWorkingTrees(
2086
tests.TestCaseWithMemoryTransport):
2088
def test_make_working_trees(self):
2089
"""For a repository with working trees, ('yes', ) is returned."""
2090
backing = self.get_transport()
2091
request = smart_repo.SmartServerRepositoryMakeWorkingTrees(backing)
2092
r = self.make_repository('.')
2093
r.set_make_working_trees(True)
2094
self.assertEqual(smart_req.SmartServerResponse(('yes', )),
2095
request.execute('', ))
2097
def test_is_not_shared(self):
2098
"""For a repository with working trees, ('no', ) is returned."""
2099
backing = self.get_transport()
2100
request = smart_repo.SmartServerRepositoryMakeWorkingTrees(backing)
2101
r = self.make_repository('.')
2102
r.set_make_working_trees(False)
2103
self.assertEqual(smart_req.SmartServerResponse(('no', )),
2104
request.execute('', ))
2107
class TestSmartServerRepositoryLockWrite(tests.TestCaseWithMemoryTransport):
2109
def test_lock_write_on_unlocked_repo(self):
2110
backing = self.get_transport()
2111
request = smart_repo.SmartServerRepositoryLockWrite(backing)
2112
repository = self.make_repository('.', format='knit')
2113
response = request.execute('')
2114
nonce = repository.control_files._lock.peek().get('nonce')
2115
self.assertEqual(smart_req.SmartServerResponse(('ok', nonce)), response)
2116
# The repository is now locked. Verify that with a new repository
2118
new_repo = repository.controldir.open_repository()
2119
self.assertRaises(errors.LockContention, new_repo.lock_write)
2121
request = smart_repo.SmartServerRepositoryUnlock(backing)
2122
response = request.execute('', nonce)
2124
def test_lock_write_on_locked_repo(self):
2125
backing = self.get_transport()
2126
request = smart_repo.SmartServerRepositoryLockWrite(backing)
2127
repository = self.make_repository('.', format='knit')
2128
repo_token = repository.lock_write().repository_token
2129
repository.leave_lock_in_place()
2131
response = request.execute('')
2133
smart_req.SmartServerResponse(('LockContention',)), response)
2135
repository.lock_write(repo_token)
2136
repository.dont_leave_lock_in_place()
2139
def test_lock_write_on_readonly_transport(self):
2140
backing = self.get_readonly_transport()
2141
request = smart_repo.SmartServerRepositoryLockWrite(backing)
2142
repository = self.make_repository('.', format='knit')
2143
response = request.execute('')
2144
self.assertFalse(response.is_successful())
2145
self.assertEqual('LockFailed', response.args[0])
2148
class TestInsertStreamBase(tests.TestCaseWithMemoryTransport):
2150
def make_empty_byte_stream(self, repo):
2151
byte_stream = smart_repo._stream_to_byte_stream([], repo._format)
2152
return ''.join(byte_stream)
2155
class TestSmartServerRepositoryInsertStream(TestInsertStreamBase):
2157
def test_insert_stream_empty(self):
2158
backing = self.get_transport()
2159
request = smart_repo.SmartServerRepositoryInsertStream(backing)
2160
repository = self.make_repository('.')
2161
response = request.execute('', '')
2162
self.assertEqual(None, response)
2163
response = request.do_chunk(self.make_empty_byte_stream(repository))
2164
self.assertEqual(None, response)
2165
response = request.do_end()
2166
self.assertEqual(smart_req.SmartServerResponse(('ok', )), response)
2169
class TestSmartServerRepositoryInsertStreamLocked(TestInsertStreamBase):
2171
def test_insert_stream_empty(self):
2172
backing = self.get_transport()
2173
request = smart_repo.SmartServerRepositoryInsertStreamLocked(
2175
repository = self.make_repository('.', format='knit')
2176
lock_token = repository.lock_write().repository_token
2177
response = request.execute('', '', lock_token)
2178
self.assertEqual(None, response)
2179
response = request.do_chunk(self.make_empty_byte_stream(repository))
2180
self.assertEqual(None, response)
2181
response = request.do_end()
2182
self.assertEqual(smart_req.SmartServerResponse(('ok', )), response)
2185
def test_insert_stream_with_wrong_lock_token(self):
2186
backing = self.get_transport()
2187
request = smart_repo.SmartServerRepositoryInsertStreamLocked(
2189
repository = self.make_repository('.', format='knit')
2190
lock_token = repository.lock_write().repository_token
2192
errors.TokenMismatch, request.execute, '', '', 'wrong-token')
2196
class TestSmartServerRepositoryUnlock(tests.TestCaseWithMemoryTransport):
2198
def test_unlock_on_locked_repo(self):
2199
backing = self.get_transport()
2200
request = smart_repo.SmartServerRepositoryUnlock(backing)
2201
repository = self.make_repository('.', format='knit')
2202
token = repository.lock_write().repository_token
2203
repository.leave_lock_in_place()
2205
response = request.execute('', token)
2207
smart_req.SmartServerResponse(('ok',)), response)
2208
# The repository is now unlocked. Verify that with a new repository
2210
new_repo = repository.controldir.open_repository()
2211
new_repo.lock_write()
2214
def test_unlock_on_unlocked_repo(self):
2215
backing = self.get_transport()
2216
request = smart_repo.SmartServerRepositoryUnlock(backing)
2217
repository = self.make_repository('.', format='knit')
2218
response = request.execute('', 'some token')
2220
smart_req.SmartServerResponse(('TokenMismatch',)), response)
2223
class TestSmartServerRepositoryGetPhysicalLockStatus(
2224
tests.TestCaseWithTransport):
2226
def test_with_write_lock(self):
2227
backing = self.get_transport()
2228
repo = self.make_repository('.')
2229
self.addCleanup(repo.lock_write().unlock)
2230
# lock_write() doesn't necessarily actually take a physical
2232
if repo.get_physical_lock_status():
2236
request_class = smart_repo.SmartServerRepositoryGetPhysicalLockStatus
2237
request = request_class(backing)
2238
self.assertEqual(smart_req.SuccessfulSmartServerResponse((expected,)),
2239
request.execute('', ))
2241
def test_without_write_lock(self):
2242
backing = self.get_transport()
2243
repo = self.make_repository('.')
2244
self.assertEqual(False, repo.get_physical_lock_status())
2245
request_class = smart_repo.SmartServerRepositoryGetPhysicalLockStatus
2246
request = request_class(backing)
2247
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('no',)),
2248
request.execute('', ))
2251
class TestSmartServerRepositoryReconcile(tests.TestCaseWithTransport):
2253
def test_reconcile(self):
2254
backing = self.get_transport()
2255
repo = self.make_repository('.')
2256
token = repo.lock_write().repository_token
2257
self.addCleanup(repo.unlock)
2258
request_class = smart_repo.SmartServerRepositoryReconcile
2259
request = request_class(backing)
2260
self.assertEqual(smart_req.SuccessfulSmartServerResponse(
2262
'garbage_inventories: 0\n'
2263
'inconsistent_parents: 0\n'),
2264
request.execute('', token))
2267
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
2269
def test_is_readonly_no(self):
2270
backing = self.get_transport()
2271
request = smart_req.SmartServerIsReadonly(backing)
2272
response = request.execute()
2274
smart_req.SmartServerResponse(('no',)), response)
2276
def test_is_readonly_yes(self):
2277
backing = self.get_readonly_transport()
2278
request = smart_req.SmartServerIsReadonly(backing)
2279
response = request.execute()
2281
smart_req.SmartServerResponse(('yes',)), response)
2284
class TestSmartServerRepositorySetMakeWorkingTrees(
2285
tests.TestCaseWithMemoryTransport):
2287
def test_set_false(self):
2288
backing = self.get_transport()
2289
repo = self.make_repository('.', shared=True)
2290
repo.set_make_working_trees(True)
2291
request_class = smart_repo.SmartServerRepositorySetMakeWorkingTrees
2292
request = request_class(backing)
2293
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok',)),
2294
request.execute('', 'False'))
2295
repo = repo.controldir.open_repository()
2296
self.assertFalse(repo.make_working_trees())
2298
def test_set_true(self):
2299
backing = self.get_transport()
2300
repo = self.make_repository('.', shared=True)
2301
repo.set_make_working_trees(False)
2302
request_class = smart_repo.SmartServerRepositorySetMakeWorkingTrees
2303
request = request_class(backing)
2304
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok',)),
2305
request.execute('', 'True'))
2306
repo = repo.controldir.open_repository()
2307
self.assertTrue(repo.make_working_trees())
2310
class TestSmartServerRepositoryGetSerializerFormat(
2311
tests.TestCaseWithMemoryTransport):
2313
def test_get_serializer_format(self):
2314
backing = self.get_transport()
2315
repo = self.make_repository('.', format='2a')
2316
request_class = smart_repo.SmartServerRepositoryGetSerializerFormat
2317
request = request_class(backing)
2319
smart_req.SuccessfulSmartServerResponse(('ok', '10')),
2320
request.execute(''))
2323
class TestSmartServerRepositoryWriteGroup(
2324
tests.TestCaseWithMemoryTransport):
2326
def test_start_write_group(self):
2327
backing = self.get_transport()
2328
repo = self.make_repository('.')
2329
lock_token = repo.lock_write().repository_token
2330
self.addCleanup(repo.unlock)
2331
request_class = smart_repo.SmartServerRepositoryStartWriteGroup
2332
request = request_class(backing)
2333
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok', [])),
2334
request.execute('', lock_token))
2336
def test_start_write_group_unsuspendable(self):
2337
backing = self.get_transport()
2338
repo = self.make_repository('.', format='knit')
2339
lock_token = repo.lock_write().repository_token
2340
self.addCleanup(repo.unlock)
2341
request_class = smart_repo.SmartServerRepositoryStartWriteGroup
2342
request = request_class(backing)
2344
smart_req.FailedSmartServerResponse(('UnsuspendableWriteGroup',)),
2345
request.execute('', lock_token))
2347
def test_commit_write_group(self):
2348
backing = self.get_transport()
2349
repo = self.make_repository('.')
2350
lock_token = repo.lock_write().repository_token
2351
self.addCleanup(repo.unlock)
2352
repo.start_write_group()
2353
tokens = repo.suspend_write_group()
2354
request_class = smart_repo.SmartServerRepositoryCommitWriteGroup
2355
request = request_class(backing)
2356
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok',)),
2357
request.execute('', lock_token, tokens))
2359
def test_abort_write_group(self):
2360
backing = self.get_transport()
2361
repo = self.make_repository('.')
2362
lock_token = repo.lock_write().repository_token
2363
repo.start_write_group()
2364
tokens = repo.suspend_write_group()
2365
self.addCleanup(repo.unlock)
2366
request_class = smart_repo.SmartServerRepositoryAbortWriteGroup
2367
request = request_class(backing)
2368
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok',)),
2369
request.execute('', lock_token, tokens))
2371
def test_check_write_group(self):
2372
backing = self.get_transport()
2373
repo = self.make_repository('.')
2374
lock_token = repo.lock_write().repository_token
2375
repo.start_write_group()
2376
tokens = repo.suspend_write_group()
2377
self.addCleanup(repo.unlock)
2378
request_class = smart_repo.SmartServerRepositoryCheckWriteGroup
2379
request = request_class(backing)
2380
self.assertEqual(smart_req.SuccessfulSmartServerResponse(('ok',)),
2381
request.execute('', lock_token, tokens))
2383
def test_check_write_group_invalid(self):
2384
backing = self.get_transport()
2385
repo = self.make_repository('.')
2386
lock_token = repo.lock_write().repository_token
2387
self.addCleanup(repo.unlock)
2388
request_class = smart_repo.SmartServerRepositoryCheckWriteGroup
2389
request = request_class(backing)
2390
self.assertEqual(smart_req.FailedSmartServerResponse(
2391
('UnresumableWriteGroup', ['random'],
2392
'Malformed write group token')),
2393
request.execute('', lock_token, ["random"]))
2396
class TestSmartServerPackRepositoryAutopack(tests.TestCaseWithTransport):
2398
def make_repo_needing_autopacking(self, path='.'):
2399
# Make a repo in need of autopacking.
2400
tree = self.make_branch_and_tree('.', format='pack-0.92')
2401
repo = tree.branch.repository
2402
# monkey-patch the pack collection to disable autopacking
2403
repo._pack_collection._max_pack_count = lambda count: count
2405
tree.commit('commit %s' % x)
2406
self.assertEqual(10, len(repo._pack_collection.names()))
2407
del repo._pack_collection._max_pack_count
2410
def test_autopack_needed(self):
2411
repo = self.make_repo_needing_autopacking()
2413
self.addCleanup(repo.unlock)
2414
backing = self.get_transport()
2415
request = smart_packrepo.SmartServerPackRepositoryAutopack(
2417
response = request.execute('')
2418
self.assertEqual(smart_req.SmartServerResponse(('ok',)), response)
2419
repo._pack_collection.reload_pack_names()
2420
self.assertEqual(1, len(repo._pack_collection.names()))
2422
def test_autopack_not_needed(self):
2423
tree = self.make_branch_and_tree('.', format='pack-0.92')
2424
repo = tree.branch.repository
2426
self.addCleanup(repo.unlock)
2428
tree.commit('commit %s' % x)
2429
backing = self.get_transport()
2430
request = smart_packrepo.SmartServerPackRepositoryAutopack(
2432
response = request.execute('')
2433
self.assertEqual(smart_req.SmartServerResponse(('ok',)), response)
2434
repo._pack_collection.reload_pack_names()
2435
self.assertEqual(9, len(repo._pack_collection.names()))
2437
def test_autopack_on_nonpack_format(self):
2438
"""A request to autopack a non-pack repo is a no-op."""
2439
repo = self.make_repository('.', format='knit')
2440
backing = self.get_transport()
2441
request = smart_packrepo.SmartServerPackRepositoryAutopack(
2443
response = request.execute('')
2444
self.assertEqual(smart_req.SmartServerResponse(('ok',)), response)
2447
class TestSmartServerVfsGet(tests.TestCaseWithMemoryTransport):
2449
def test_unicode_path(self):
2450
"""VFS requests expect unicode paths to be escaped."""
2451
filename = u'foo\N{INTERROBANG}'
2452
filename_escaped = urlutils.escape(filename)
2453
backing = self.get_transport()
2454
request = vfs.GetRequest(backing)
2455
backing.put_bytes_non_atomic(filename_escaped, 'contents')
2456
self.assertEqual(smart_req.SmartServerResponse(('ok', ), 'contents'),
2457
request.execute(filename_escaped))
2460
class TestHandlers(tests.TestCase):
2461
"""Tests for the request.request_handlers object."""
2463
def test_all_registrations_exist(self):
2464
"""All registered request_handlers can be found."""
2465
# If there's a typo in a register_lazy call, this loop will fail with
2466
# an AttributeError.
2467
for key in smart_req.request_handlers.keys():
2469
item = smart_req.request_handlers.get(key)
2470
except AttributeError as e:
2471
raise AttributeError('failed to get %s: %s' % (key, e))
2473
def assertHandlerEqual(self, verb, handler):
2474
self.assertEqual(smart_req.request_handlers.get(verb), handler)
2476
def test_registered_methods(self):
2477
"""Test that known methods are registered to the correct object."""
2478
self.assertHandlerEqual('Branch.break_lock',
2479
smart_branch.SmartServerBranchBreakLock)
2480
self.assertHandlerEqual('Branch.get_config_file',
2481
smart_branch.SmartServerBranchGetConfigFile)
2482
self.assertHandlerEqual('Branch.put_config_file',
2483
smart_branch.SmartServerBranchPutConfigFile)
2484
self.assertHandlerEqual('Branch.get_parent',
2485
smart_branch.SmartServerBranchGetParent)
2486
self.assertHandlerEqual('Branch.get_physical_lock_status',
2487
smart_branch.SmartServerBranchRequestGetPhysicalLockStatus)
2488
self.assertHandlerEqual('Branch.get_tags_bytes',
2489
smart_branch.SmartServerBranchGetTagsBytes)
2490
self.assertHandlerEqual('Branch.lock_write',
2491
smart_branch.SmartServerBranchRequestLockWrite)
2492
self.assertHandlerEqual('Branch.last_revision_info',
2493
smart_branch.SmartServerBranchRequestLastRevisionInfo)
2494
self.assertHandlerEqual('Branch.revision_history',
2495
smart_branch.SmartServerRequestRevisionHistory)
2496
self.assertHandlerEqual('Branch.revision_id_to_revno',
2497
smart_branch.SmartServerBranchRequestRevisionIdToRevno)
2498
self.assertHandlerEqual('Branch.set_config_option',
2499
smart_branch.SmartServerBranchRequestSetConfigOption)
2500
self.assertHandlerEqual('Branch.set_last_revision',
2501
smart_branch.SmartServerBranchRequestSetLastRevision)
2502
self.assertHandlerEqual('Branch.set_last_revision_info',
2503
smart_branch.SmartServerBranchRequestSetLastRevisionInfo)
2504
self.assertHandlerEqual('Branch.set_last_revision_ex',
2505
smart_branch.SmartServerBranchRequestSetLastRevisionEx)
2506
self.assertHandlerEqual('Branch.set_parent_location',
2507
smart_branch.SmartServerBranchRequestSetParentLocation)
2508
self.assertHandlerEqual('Branch.unlock',
2509
smart_branch.SmartServerBranchRequestUnlock)
2510
self.assertHandlerEqual('BzrDir.destroy_branch',
2511
smart_dir.SmartServerBzrDirRequestDestroyBranch)
2512
self.assertHandlerEqual('BzrDir.find_repository',
2513
smart_dir.SmartServerRequestFindRepositoryV1)
2514
self.assertHandlerEqual('BzrDir.find_repositoryV2',
2515
smart_dir.SmartServerRequestFindRepositoryV2)
2516
self.assertHandlerEqual('BzrDirFormat.initialize',
2517
smart_dir.SmartServerRequestInitializeBzrDir)
2518
self.assertHandlerEqual('BzrDirFormat.initialize_ex_1.16',
2519
smart_dir.SmartServerRequestBzrDirInitializeEx)
2520
self.assertHandlerEqual('BzrDir.checkout_metadir',
2521
smart_dir.SmartServerBzrDirRequestCheckoutMetaDir)
2522
self.assertHandlerEqual('BzrDir.cloning_metadir',
2523
smart_dir.SmartServerBzrDirRequestCloningMetaDir)
2524
self.assertHandlerEqual('BzrDir.get_branches',
2525
smart_dir.SmartServerBzrDirRequestGetBranches)
2526
self.assertHandlerEqual('BzrDir.get_config_file',
2527
smart_dir.SmartServerBzrDirRequestConfigFile)
2528
self.assertHandlerEqual('BzrDir.open_branch',
2529
smart_dir.SmartServerRequestOpenBranch)
2530
self.assertHandlerEqual('BzrDir.open_branchV2',
2531
smart_dir.SmartServerRequestOpenBranchV2)
2532
self.assertHandlerEqual('BzrDir.open_branchV3',
2533
smart_dir.SmartServerRequestOpenBranchV3)
2534
self.assertHandlerEqual('PackRepository.autopack',
2535
smart_packrepo.SmartServerPackRepositoryAutopack)
2536
self.assertHandlerEqual('Repository.add_signature_text',
2537
smart_repo.SmartServerRepositoryAddSignatureText)
2538
self.assertHandlerEqual('Repository.all_revision_ids',
2539
smart_repo.SmartServerRepositoryAllRevisionIds)
2540
self.assertHandlerEqual('Repository.break_lock',
2541
smart_repo.SmartServerRepositoryBreakLock)
2542
self.assertHandlerEqual('Repository.gather_stats',
2543
smart_repo.SmartServerRepositoryGatherStats)
2544
self.assertHandlerEqual('Repository.get_parent_map',
2545
smart_repo.SmartServerRepositoryGetParentMap)
2546
self.assertHandlerEqual('Repository.get_physical_lock_status',
2547
smart_repo.SmartServerRepositoryGetPhysicalLockStatus)
2548
self.assertHandlerEqual('Repository.get_rev_id_for_revno',
2549
smart_repo.SmartServerRepositoryGetRevIdForRevno)
2550
self.assertHandlerEqual('Repository.get_revision_graph',
2551
smart_repo.SmartServerRepositoryGetRevisionGraph)
2552
self.assertHandlerEqual('Repository.get_revision_signature_text',
2553
smart_repo.SmartServerRepositoryGetRevisionSignatureText)
2554
self.assertHandlerEqual('Repository.get_stream',
2555
smart_repo.SmartServerRepositoryGetStream)
2556
self.assertHandlerEqual('Repository.get_stream_1.19',
2557
smart_repo.SmartServerRepositoryGetStream_1_19)
2558
self.assertHandlerEqual('Repository.iter_revisions',
2559
smart_repo.SmartServerRepositoryIterRevisions)
2560
self.assertHandlerEqual('Repository.has_revision',
2561
smart_repo.SmartServerRequestHasRevision)
2562
self.assertHandlerEqual('Repository.insert_stream',
2563
smart_repo.SmartServerRepositoryInsertStream)
2564
self.assertHandlerEqual('Repository.insert_stream_locked',
2565
smart_repo.SmartServerRepositoryInsertStreamLocked)
2566
self.assertHandlerEqual('Repository.is_shared',
2567
smart_repo.SmartServerRepositoryIsShared)
2568
self.assertHandlerEqual('Repository.iter_files_bytes',
2569
smart_repo.SmartServerRepositoryIterFilesBytes)
2570
self.assertHandlerEqual('Repository.lock_write',
2571
smart_repo.SmartServerRepositoryLockWrite)
2572
self.assertHandlerEqual('Repository.make_working_trees',
2573
smart_repo.SmartServerRepositoryMakeWorkingTrees)
2574
self.assertHandlerEqual('Repository.pack',
2575
smart_repo.SmartServerRepositoryPack)
2576
self.assertHandlerEqual('Repository.reconcile',
2577
smart_repo.SmartServerRepositoryReconcile)
2578
self.assertHandlerEqual('Repository.tarball',
2579
smart_repo.SmartServerRepositoryTarball)
2580
self.assertHandlerEqual('Repository.unlock',
2581
smart_repo.SmartServerRepositoryUnlock)
2582
self.assertHandlerEqual('Repository.start_write_group',
2583
smart_repo.SmartServerRepositoryStartWriteGroup)
2584
self.assertHandlerEqual('Repository.check_write_group',
2585
smart_repo.SmartServerRepositoryCheckWriteGroup)
2586
self.assertHandlerEqual('Repository.commit_write_group',
2587
smart_repo.SmartServerRepositoryCommitWriteGroup)
2588
self.assertHandlerEqual('Repository.abort_write_group',
2589
smart_repo.SmartServerRepositoryAbortWriteGroup)
2590
self.assertHandlerEqual('VersionedFileRepository.get_serializer_format',
2591
smart_repo.SmartServerRepositoryGetSerializerFormat)
2592
self.assertHandlerEqual('VersionedFileRepository.get_inventories',
2593
smart_repo.SmartServerRepositoryGetInventories)
2594
self.assertHandlerEqual('Transport.is_readonly',
2595
smart_req.SmartServerIsReadonly)
2598
class SmartTCPServerHookTests(tests.TestCaseWithMemoryTransport):
2599
"""Tests for SmartTCPServer hooks."""
2602
super(SmartTCPServerHookTests, self).setUp()
2603
self.server = server.SmartTCPServer(self.get_transport())
2605
def test_run_server_started_hooks(self):
2606
"""Test the server started hooks get fired properly."""
2608
server.SmartTCPServer.hooks.install_named_hook('server_started',
2609
lambda backing_urls, url: started_calls.append((backing_urls, url)),
2611
started_ex_calls = []
2612
server.SmartTCPServer.hooks.install_named_hook('server_started_ex',
2613
lambda backing_urls, url: started_ex_calls.append((backing_urls, url)),
2615
self.server._sockname = ('example.com', 42)
2616
self.server.run_server_started_hooks()
2617
self.assertEqual(started_calls,
2618
[([self.get_transport().base], 'bzr://example.com:42/')])
2619
self.assertEqual(started_ex_calls,
2620
[([self.get_transport().base], self.server)])
2622
def test_run_server_started_hooks_ipv6(self):
2623
"""Test that socknames can contain 4-tuples."""
2624
self.server._sockname = ('::', 42, 0, 0)
2626
server.SmartTCPServer.hooks.install_named_hook('server_started',
2627
lambda backing_urls, url: started_calls.append((backing_urls, url)),
2629
self.server.run_server_started_hooks()
2630
self.assertEqual(started_calls,
2631
[([self.get_transport().base], 'bzr://:::42/')])
2633
def test_run_server_stopped_hooks(self):
2634
"""Test the server stopped hooks."""
2635
self.server._sockname = ('example.com', 42)
2637
server.SmartTCPServer.hooks.install_named_hook('server_stopped',
2638
lambda backing_urls, url: stopped_calls.append((backing_urls, url)),
2640
self.server.run_server_stopped_hooks()
2641
self.assertEqual(stopped_calls,
2642
[([self.get_transport().base], 'bzr://example.com:42/')])
2645
class TestSmartServerRepositoryPack(tests.TestCaseWithMemoryTransport):
2647
def test_pack(self):
2648
backing = self.get_transport()
2649
request = smart_repo.SmartServerRepositoryPack(backing)
2650
tree = self.make_branch_and_memory_tree('.')
2651
repo_token = tree.branch.repository.lock_write().repository_token
2653
self.assertIs(None, request.execute('', repo_token, False))
2656
smart_req.SuccessfulSmartServerResponse(('ok', ), ),
2657
request.do_body(''))
2660
class TestSmartServerRepositoryGetInventories(tests.TestCaseWithTransport):
2662
def _get_serialized_inventory_delta(self, repository, base_revid, revid):
2663
base_inv = repository.revision_tree(base_revid).root_inventory
2664
inv = repository.revision_tree(revid).root_inventory
2665
inv_delta = inv._make_delta(base_inv)
2666
serializer = inventory_delta.InventoryDeltaSerializer(True, False)
2667
return "".join(serializer.delta_to_lines(base_revid, revid, inv_delta))
2669
def test_single(self):
2670
backing = self.get_transport()
2671
request = smart_repo.SmartServerRepositoryGetInventories(backing)
2672
t = self.make_branch_and_tree('.', format='2a')
2673
self.addCleanup(t.lock_write().unlock)
2674
self.build_tree_contents([("file", "somecontents")])
2675
t.add(["file"], ["thefileid"])
2676
t.commit(rev_id='somerev', message="add file")
2677
self.assertIs(None, request.execute('', 'unordered'))
2678
response = request.do_body("somerev\n")
2679
self.assertTrue(response.is_successful())
2680
self.assertEqual(response.args, ("ok", ))
2681
stream = [('inventory-deltas', [
2682
versionedfile.FulltextContentFactory('somerev', None, None,
2683
self._get_serialized_inventory_delta(
2684
t.branch.repository, 'null:', 'somerev'))])]
2685
fmt = controldir.format_registry.get('2a')().repository_format
2687
"".join(response.body_stream),
2688
"".join(smart_repo._stream_to_byte_stream(stream, fmt)))
2690
def test_empty(self):
2691
backing = self.get_transport()
2692
request = smart_repo.SmartServerRepositoryGetInventories(backing)
2693
t = self.make_branch_and_tree('.', format='2a')
2694
self.addCleanup(t.lock_write().unlock)
2695
self.build_tree_contents([("file", "somecontents")])
2696
t.add(["file"], ["thefileid"])
2697
t.commit(rev_id='somerev', message="add file")
2698
self.assertIs(None, request.execute('', 'unordered'))
2699
response = request.do_body("")
2700
self.assertTrue(response.is_successful())
2701
self.assertEqual(response.args, ("ok", ))
2702
self.assertEqual("".join(response.body_stream),
2703
"Bazaar pack format 1 (introduced in 0.18)\nB54\n\nBazaar repository format 2a (needs bzr 1.16 or later)\nE")
added
removed
breezy/tests/test_smart.py
