397
397
request.execute(''))
400
class TestSmartServerBranchRequestSetLastRevision(tests.TestCaseWithMemoryTransport):
402
def test_empty(self):
403
backing = self.get_transport()
404
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
405
b = self.make_branch('.')
406
branch_token = b.lock_write()
407
repo_token = b.repository.lock_write()
408
b.repository.unlock()
410
self.assertEqual(SmartServerResponse(('ok',)),
412
'', branch_token, repo_token,
417
def test_not_present_revision_id(self):
418
backing = self.get_transport()
419
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
420
b = self.make_branch('.')
421
branch_token = b.lock_write()
422
repo_token = b.repository.lock_write()
423
b.repository.unlock()
425
revision_id = 'non-existent revision'
427
SmartServerResponse(('NoSuchRevision', revision_id)),
429
'', branch_token, repo_token,
434
def test_revision_id_present(self):
435
backing = self.get_transport()
436
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
437
tree = self.make_branch_and_memory_tree('.')
440
rev_id_utf8 = u'\xc8'.encode('utf-8')
441
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
442
r2 = tree.commit('2nd commit')
444
branch_token = tree.branch.lock_write()
445
repo_token = tree.branch.repository.lock_write()
446
tree.branch.repository.unlock()
449
SmartServerResponse(('ok',)),
451
'', branch_token, repo_token,
453
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
457
def test_revision_id_present2(self):
458
backing = self.get_transport()
459
request = smart.branch.SmartServerBranchRequestSetLastRevision(backing)
460
tree = self.make_branch_and_memory_tree('.')
463
rev_id_utf8 = u'\xc8'.encode('utf-8')
464
r1 = tree.commit('1st commit', rev_id=rev_id_utf8)
465
r2 = tree.commit('2nd commit')
467
tree.branch.set_revision_history([])
468
branch_token = tree.branch.lock_write()
469
repo_token = tree.branch.repository.lock_write()
470
tree.branch.repository.unlock()
473
SmartServerResponse(('ok',)),
475
'', branch_token, repo_token,
477
self.assertEqual([rev_id_utf8], tree.branch.revision_history())
482
class TestSmartServerBranchRequestSetLastRevisionInfo(tests.TestCaseWithTransport):
484
def lock_branch(self, branch):
485
branch_token = branch.lock_write()
486
repo_token = branch.repository.lock_write()
487
branch.repository.unlock()
488
self.addCleanup(branch.unlock)
400
class SetLastRevisionTestBase(tests.TestCaseWithMemoryTransport):
401
"""Base test case for verbs that implement set_last_revision."""
404
tests.TestCaseWithMemoryTransport.setUp(self)
405
backing_transport = self.get_transport()
406
self.request = self.request_class(backing_transport)
407
self.tree = self.make_branch_and_memory_tree('.')
409
def lock_branch(self):
411
branch_token = b.lock_write()
412
repo_token = b.repository.lock_write()
413
b.repository.unlock()
489
414
return branch_token, repo_token
491
def make_locked_branch(self, format=None):
492
branch = self.make_branch('.', format=format)
493
branch_token, repo_token = self.lock_branch(branch)
494
return branch, branch_token, repo_token
496
def test_empty(self):
416
def unlock_branch(self):
417
self.tree.branch.unlock()
419
def set_last_revision(self, revision_id, revno):
420
branch_token, repo_token = self.lock_branch()
421
response = self._set_last_revision(
422
revision_id, revno, branch_token, repo_token)
426
def assertRequestSucceeds(self, revision_id, revno):
427
response = self.set_last_revision(revision_id, revno)
428
self.assertEqual(SuccessfulSmartServerResponse(('ok',)), response)
431
class TestSetLastRevisionVerbMixin(object):
432
"""Mixin test case for verbs that implement set_last_revision."""
434
def test_set_null_to_null(self):
497
435
"""An empty branch can have its last revision set to 'null:'."""
498
b, branch_token, repo_token = self.make_locked_branch()
499
backing = self.get_transport()
500
request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(
502
response = request.execute('', branch_token, repo_token, '0', 'null:')
503
self.assertEqual(SmartServerResponse(('ok',)), response)
505
def assertBranchLastRevisionInfo(self, expected_info, branch_relpath):
506
branch = bzrdir.BzrDir.open(branch_relpath).open_branch()
507
self.assertEqual(expected_info, branch.last_revision_info())
509
def test_branch_revision_info_is_updated(self):
510
"""This method really does update the branch last revision info."""
511
tree = self.make_branch_and_memory_tree('.')
514
tree.commit('First commit', rev_id='revision-1')
515
tree.commit('Second commit', rev_id='revision-2')
519
branch_token, repo_token = self.lock_branch(branch)
520
backing = self.get_transport()
521
request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(
523
self.assertBranchLastRevisionInfo((2, 'revision-2'), '.')
524
response = request.execute(
525
'', branch_token, repo_token, '1', 'revision-1')
526
self.assertEqual(SmartServerResponse(('ok',)), response)
527
self.assertBranchLastRevisionInfo((1, 'revision-1'), '.')
529
def test_not_present_revid(self):
530
"""Some branch formats will check that the revision is present in the
531
repository. When that check fails, a NoSuchRevision error is returned
534
# Make a knit format branch, because that format checks the values
535
# given to set_last_revision_info.
536
b, branch_token, repo_token = self.make_locked_branch(format='knit')
537
backing = self.get_transport()
538
request = smart.branch.SmartServerBranchRequestSetLastRevisionInfo(
540
response = request.execute(
541
'', branch_token, repo_token, '1', 'not-present')
543
SmartServerResponse(('NoSuchRevision', 'not-present')), response)
436
self.assertRequestSucceeds('null:', 0)
438
def test_NoSuchRevision(self):
439
"""If the revision_id is not present, the verb returns NoSuchRevision.
441
revision_id = 'non-existent revision'
443
FailedSmartServerResponse(('NoSuchRevision', revision_id)),
444
self.set_last_revision(revision_id, 1))
446
def make_tree_with_two_commits(self):
447
self.tree.lock_write()
449
rev_id_utf8 = u'\xc8'.encode('utf-8')
450
r1 = self.tree.commit('1st commit', rev_id=rev_id_utf8)
451
r2 = self.tree.commit('2nd commit', rev_id='rev-2')
454
def test_branch_last_revision_info_is_updated(self):
455
"""A branch's tip can be set to a revision that is present in its
458
# Make a branch with an empty revision history, but two revisions in
460
self.make_tree_with_two_commits()
461
rev_id_utf8 = u'\xc8'.encode('utf-8')
462
self.tree.branch.set_revision_history([])
464
(0, 'null:'), self.tree.branch.last_revision_info())
465
# We can update the branch to a revision that is present in the
467
self.assertRequestSucceeds(rev_id_utf8, 1)
469
(1, rev_id_utf8), self.tree.branch.last_revision_info())
471
def test_branch_last_revision_info_rewind(self):
472
"""A branch's tip can be set to a revision that is an ancestor of the
475
self.make_tree_with_two_commits()
476
rev_id_utf8 = u'\xc8'.encode('utf-8')
478
(2, 'rev-2'), self.tree.branch.last_revision_info())
479
self.assertRequestSucceeds(rev_id_utf8, 1)
481
(1, rev_id_utf8), self.tree.branch.last_revision_info())
484
class TestSmartServerBranchRequestSetLastRevision(
485
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
486
"""Tests for Branch.set_last_revision verb."""
488
request_class = smart.branch.SmartServerBranchRequestSetLastRevision
490
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
491
return self.request.execute(
492
'', branch_token, repo_token, revision_id)
495
class TestSmartServerBranchRequestSetLastRevisionInfo(
496
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
497
"""Tests for Branch.set_last_revision_info verb."""
499
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionInfo
501
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
502
return self.request.execute(
503
'', branch_token, repo_token, revno, revision_id)
505
def test_NoSuchRevision(self):
506
"""Branch.set_last_revision_info does not have to return
507
NoSuchRevision if the revision_id is absent.
509
raise tests.TestNotApplicable()
512
class TestSmartServerBranchRequestSetLastRevisionEx(
513
SetLastRevisionTestBase, TestSetLastRevisionVerbMixin):
514
"""Tests for Branch.set_last_revision_ex verb."""
516
request_class = smart.branch.SmartServerBranchRequestSetLastRevisionEx
518
def _set_last_revision(self, revision_id, revno, branch_token, repo_token):
519
return self.request.execute(
520
'', branch_token, repo_token, revision_id, 0, 0)
522
def assertRequestSucceeds(self, revision_id, revno):
523
response = self.set_last_revision(revision_id, revno)
525
SuccessfulSmartServerResponse(('ok', revno, revision_id)),
528
def test_branch_last_revision_info_rewind(self):
529
"""A branch's tip can be set to a revision that is an ancestor of the
530
current tip, but only if allow_overwrite_descendant is passed.
532
self.make_tree_with_two_commits()
533
rev_id_utf8 = u'\xc8'.encode('utf-8')
535
(2, 'rev-2'), self.tree.branch.last_revision_info())
536
# If allow_overwrite_descendant flag is 0, then trying to set the tip
537
# to an older revision ID has no effect.
538
branch_token, repo_token = self.lock_branch()
539
response = self.request.execute(
540
'', branch_token, repo_token, rev_id_utf8, 0, 0)
542
SuccessfulSmartServerResponse(('ok', 2, 'rev-2')),
545
(2, 'rev-2'), self.tree.branch.last_revision_info())
547
# If allow_overwrite_descendant flag is 1, then setting the tip to an
549
response = self.request.execute(
550
'', branch_token, repo_token, rev_id_utf8, 0, 1)
552
SuccessfulSmartServerResponse(('ok', 1, rev_id_utf8)),
556
(1, rev_id_utf8), self.tree.branch.last_revision_info())
558
def make_branch_with_divergent_history(self):
559
"""Make a branch with divergent history in its repo.
561
The branch's tip will be 'child-2', and the repo will also contain
562
'child-1', which diverges from a common base revision.
564
self.tree.lock_write()
566
r1 = self.tree.commit('1st commit')
567
revno_1, revid_1 = self.tree.branch.last_revision_info()
568
r2 = self.tree.commit('2nd commit', rev_id='child-1')
569
# Undo the second commit
570
self.tree.branch.set_last_revision_info(revno_1, revid_1)
571
self.tree.set_parent_ids([revid_1])
572
# Make a new second commit, child-2. child-2 has diverged from
574
new_r2 = self.tree.commit('2nd commit', rev_id='child-2')
577
def test_not_allow_diverged(self):
578
"""If allow_diverged is not passed, then setting a divergent history
579
returns a Diverged error.
581
self.make_branch_with_divergent_history()
583
FailedSmartServerResponse(('Diverged',)),
584
self.set_last_revision('child-1', 2))
585
# The branch tip was not changed.
586
self.assertEqual('child-2', self.tree.branch.last_revision())
588
def test_allow_diverged(self):
589
"""If allow_diverged is passed, then setting a divergent history
592
self.make_branch_with_divergent_history()
593
branch_token, repo_token = self.lock_branch()
594
response = self.request.execute(
595
'', branch_token, repo_token, 'child-1', 1, 0)
597
SuccessfulSmartServerResponse(('ok', 2, 'child-1')),
600
# The branch tip was changed.
601
self.assertEqual('child-1', self.tree.branch.last_revision())
546
604
class TestSmartServerBranchRequestLockWrite(tests.TestCaseWithMemoryTransport):
938
991
SmartServerResponse(('TokenMismatch',)), response)
941
class TestSmartServerRepositoryTarball(tests.TestCaseWithTransport):
943
def test_repository_tarball(self):
944
backing = self.get_transport()
945
request = smart.repository.SmartServerRepositoryTarball(backing)
946
repository = self.make_repository('.')
947
# make some extraneous junk in the repository directory which should
949
self.build_tree(['.bzr/repository/extra-junk'])
950
response = request.execute('', 'bz2')
951
self.assertEqual(('ok',), response.args)
952
# body should be a tbz2
953
body_file = StringIO(response.body)
954
body_tar = tarfile.open('body_tar.tbz2', fileobj=body_file,
956
# let's make sure there are some key repository components inside it.
957
# the tarfile returns directories with trailing slashes...
958
names = set([n.rstrip('/') for n in body_tar.getnames()])
959
self.assertTrue('.bzr/repository/lock' in names)
960
self.assertTrue('.bzr/repository/format' in names)
961
self.assertTrue('.bzr/repository/extra-junk' not in names,
962
"extraneous file present in tar file")
965
class TestSmartServerRepositoryStreamKnitData(tests.TestCaseWithMemoryTransport):
967
def test_fetch_revisions(self):
968
backing = self.get_transport()
969
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
970
tree = self.make_branch_and_memory_tree('.')
973
rev_id1_utf8 = u'\xc8'.encode('utf-8')
974
rev_id2_utf8 = u'\xc9'.encode('utf-8')
975
r1 = tree.commit('1st commit', rev_id=rev_id1_utf8)
976
r1 = tree.commit('2nd commit', rev_id=rev_id2_utf8)
979
response = request.execute('', rev_id2_utf8)
980
self.assertEqual(('ok',), response.args)
981
unpacker = pack.ContainerReader(StringIO(response.body))
983
for [name], read_bytes in unpacker.iter_records():
985
bytes = read_bytes(None)
986
# The bytes should be a valid bencoded string.
987
bencode.bdecode(bytes)
988
# XXX: assert that the bencoded knit records have the right
991
def test_no_such_revision_error(self):
992
backing = self.get_transport()
993
request = smart.repository.SmartServerRepositoryStreamKnitDataForRevisions(backing)
994
repo = self.make_repository('.')
995
rev_id1_utf8 = u'\xc8'.encode('utf-8')
996
response = request.execute('', rev_id1_utf8)
998
SmartServerResponse(('NoSuchRevision', rev_id1_utf8)),
1002
class TestSmartServerRepositoryStreamRevisionsChunked(tests.TestCaseWithMemoryTransport):
1004
def test_fetch_revisions(self):
1005
backing = self.get_transport()
1006
request = smart.repository.SmartServerRepositoryStreamRevisionsChunked(
1008
tree = self.make_branch_and_memory_tree('.')
1011
rev_id1_utf8 = u'\xc8'.encode('utf-8')
1012
rev_id2_utf8 = u'\xc9'.encode('utf-8')
1013
tree.commit('1st commit', rev_id=rev_id1_utf8)
1014
tree.commit('2nd commit', rev_id=rev_id2_utf8)
1017
response = request.execute('')
1018
self.assertEqual(None, response)
1019
response = request.do_body("%s\n%s\n1" % (rev_id2_utf8, rev_id1_utf8))
1020
self.assertEqual(('ok',), response.args)
1021
parser = pack.ContainerPushParser()
1023
for stream_bytes in response.body_stream:
1024
parser.accept_bytes(stream_bytes)
1025
for [name], record_bytes in parser.read_pending_records():
1027
# The bytes should be a valid bencoded string.
1028
bencode.bdecode(record_bytes)
1029
# XXX: assert that the bencoded knit records have the right
1032
def test_no_such_revision_error(self):
1033
backing = self.get_transport()
1034
request = smart.repository.SmartServerRepositoryStreamRevisionsChunked(
1036
repo = self.make_repository('.')
1037
rev_id1_utf8 = u'\xc8'.encode('utf-8')
1038
response = request.execute('')
1039
self.assertEqual(None, response)
1040
response = request.do_body("%s\n\n1" % (rev_id1_utf8,))
1042
FailedSmartServerResponse(('NoSuchRevision', )),
1046
994
class TestSmartServerIsReadonly(tests.TestCaseWithMemoryTransport):
1048
996
def test_is_readonly_no(self):
added
removed
bzrlib/tests/test_smart.py
