20
20
from bzrlib.lazy_import import lazy_import
21
21
lazy_import(globals(), """
22
from copy import deepcopy
23
from unittest import TestSuite
24
22
from warnings import warn
174
172
return self.get_config().get_nickname()
176
174
def _set_nick(self, nick):
177
self.get_config().set_user_option('nickname', nick)
175
self.get_config().set_user_option('nickname', nick, warn_masked=True)
179
177
nick = property(_get_nick, _set_nick)
508
506
def revision_id_to_revno(self, revision_id):
509
507
"""Given a revision id, return its revno"""
510
if revision_id is None:
508
if _mod_revision.is_null(revision_id):
512
510
revision_id = osutils.safe_revision_id(revision_id)
513
511
history = self.revision_history()
587
585
elif make_relative:
588
586
url = urlutils.relative_url(self.base, url)
589
config.set_user_option(name, url)
587
config.set_user_option(name, url, warn_masked=True)
591
589
def _get_config_location(self, name, config=None):
592
590
if config is None:
612
610
pattern is that the user can override it by specifying a
615
self.get_config().set_user_option('submit_branch', location)
613
self.get_config().set_user_option('submit_branch', location,
617
616
def get_public_branch(self):
618
617
"""Return the public location of the branch.
702
701
:param revision_id: The revision-id to truncate history at. May
703
702
be None to copy complete history.
704
if revision_id == _mod_revision.NULL_REVISION:
705
706
new_history = self.revision_history()
706
if revision_id is not None:
707
if revision_id is not None and new_history != []:
707
708
revision_id = osutils.safe_revision_id(revision_id)
709
710
new_history = new_history[:new_history.index(revision_id) + 1]
1399
1406
other_branch=None):
1400
1407
# stop_revision must be a descendant of last_revision
1401
1408
stop_graph = self.repository.get_revision_graph(revision_id)
1402
if last_rev is not None and last_rev not in stop_graph:
1409
if (last_rev is not None and last_rev != _mod_revision.NULL_REVISION
1410
and last_rev not in stop_graph):
1403
1411
# our previous tip is not merged into stop_revision
1404
1412
raise errors.DivergedBranches(self, other_branch)
1405
1413
# make a new revision history from the graph
1444
1452
stop_revision = osutils.safe_revision_id(stop_revision)
1445
1453
# whats the current last revision, before we fetch [and change it
1447
last_rev = self.last_revision()
1455
last_rev = _mod_revision.ensure_null(self.last_revision())
1448
1456
# we fetch here regardless of whether we need to so that we pickup
1449
1457
# filled in ghosts.
1450
1458
self.fetch(other, stop_revision)
1451
my_ancestry = self.repository.get_ancestry(last_rev)
1459
my_ancestry = self.repository.get_ancestry(last_rev,
1452
1461
if stop_revision in my_ancestry:
1453
1462
# last_revision is a descendant of stop_revision
1773
1782
# last_rev is not in the other_last_rev history, AND
1774
1783
# other_last_rev is not in our history, and do it without pulling
1775
1784
# history around
1776
last_rev = self.last_revision()
1777
if last_rev is not None:
1785
last_rev = _mod_revision.ensure_null(self.last_revision())
1786
if last_rev != _mod_revision.NULL_REVISION:
1778
1787
other.lock_read()
1780
1789
other_last_rev = other.last_revision()
1781
if other_last_rev is not None:
1790
if not _mod_revision.is_null(other_last_rev):
1782
1791
# neither branch is new, we have to do some work to
1783
1792
# ascertain diversion.
1784
1793
remote_graph = other.repository.get_revision_graph(
1807
1816
if master is not None:
1808
1817
old_tip = self.last_revision()
1809
1818
self.pull(master, overwrite=True)
1810
if old_tip in self.repository.get_ancestry(self.last_revision()):
1819
if old_tip in self.repository.get_ancestry(
1820
_mod_revision.ensure_null(self.last_revision()),
1954
1965
self._clear_cached_state()
1956
1967
def _check_history_violation(self, revision_id):
1957
last_revision = self.last_revision()
1958
if last_revision is None:
1968
last_revision = _mod_revision.ensure_null(self.last_revision())
1969
if _mod_revision.is_null(last_revision):
1960
1971
if last_revision not in self._lefthand_history(revision_id):
1961
1972
raise errors.AppendRevisionsOnlyViolation(self.base)
2025
2036
if config.get_user_option('bound') != 'True':
2028
config.set_user_option('bound', 'False')
2039
config.set_user_option('bound', 'False', warn_masked=True)
2031
2042
self._set_config_location('bound_location', location,
2033
config.set_user_option('bound', 'True')
2044
config.set_user_option('bound', 'True', warn_masked=True)
2036
2047
def _get_bound_location(self, bound):
2058
2069
value = 'False'
2059
self.get_config().set_user_option('append_revisions_only', value)
2070
self.get_config().set_user_option('append_revisions_only', value,
2061
2073
def _get_append_revisions_only(self):
2062
2074
value = self.get_config().get_user_option('append_revisions_only')
2077
2089
if revision_id is None:
2078
2090
revno, revision_id = self.last_revision_info()
2080
revno = self.revision_id_to_revno(revision_id)
2092
# To figure out the revno for a random revision, we need to build
2093
# the revision history, and count its length.
2094
# We don't care about the order, just how long it is.
2095
# Alternatively, we could start at the current location, and count
2096
# backwards. But there is no guarantee that we will find it since
2097
# it may be a merged revision.
2098
revno = len(list(self.repository.iter_reverse_revision_history(
2081
2100
destination.set_last_revision_info(revno, revision_id)
2083
2102
def _make_tags(self):
2084
2103
return BasicTags(self)
2087
class BranchTestProviderAdapter(object):
2088
"""A tool to generate a suite testing multiple branch formats at once.
2090
This is done by copying the test once for each transport and injecting
2091
the transport_server, transport_readonly_server, and branch_format
2092
classes into each copy. Each copy is also given a new id() to make it
2096
def __init__(self, transport_server, transport_readonly_server, formats,
2097
vfs_transport_factory=None):
2098
self._transport_server = transport_server
2099
self._transport_readonly_server = transport_readonly_server
2100
self._formats = formats
2102
def adapt(self, test):
2103
result = TestSuite()
2104
for branch_format, bzrdir_format in self._formats:
2105
new_test = deepcopy(test)
2106
new_test.transport_server = self._transport_server
2107
new_test.transport_readonly_server = self._transport_readonly_server
2108
new_test.bzrdir_format = bzrdir_format
2109
new_test.branch_format = branch_format
2110
def make_new_test_id():
2111
# the format can be either a class or an instance
2112
name = getattr(branch_format, '__name__',
2113
branch_format.__class__.__name__)
2114
new_id = "%s(%s)" % (new_test.id(), name)
2115
return lambda: new_id
2116
new_test.id = make_new_test_id()
2117
result.addTest(new_test)
2121
2106
######################################################################
2122
2107
# results of operations