/usr/lib/python2.7/dist-packages/bzrlib/upgrade.py is in python-bzrlib 2.7.0-2ubuntu3.1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 | # Copyright (C) 2005, 2006, 2008-2011 Canonical Ltd
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
"""bzr upgrade logic."""
from __future__ import absolute_import
from bzrlib import (
errors,
trace,
ui,
urlutils,
)
from bzrlib.controldir import (
ControlDir,
format_registry,
)
from bzrlib.i18n import gettext
from bzrlib.remote import RemoteBzrDir
class Convert(object):
def __init__(self, url=None, format=None, control_dir=None):
"""Convert a Bazaar control directory to a given format.
Either the url or control_dir parameter must be given.
:param url: the URL of the control directory or None if the
control_dir is explicitly given instead
:param format: the format to convert to or None for the default
:param control_dir: the control directory or None if it is
specified via the URL parameter instead
"""
self.format = format
# XXX: Change to cleanup
warning_id = 'cross_format_fetch'
saved_warning = warning_id in ui.ui_factory.suppressed_warnings
if url is None and control_dir is None:
raise AssertionError(
"either the url or control_dir parameter must be set.")
if control_dir is not None:
self.bzrdir = control_dir
else:
self.bzrdir = ControlDir.open_unsupported(url)
if isinstance(self.bzrdir, RemoteBzrDir):
self.bzrdir._ensure_real()
self.bzrdir = self.bzrdir._real_bzrdir
if self.bzrdir.root_transport.is_readonly():
raise errors.UpgradeReadonly
self.transport = self.bzrdir.root_transport
ui.ui_factory.suppressed_warnings.add(warning_id)
try:
self.convert()
finally:
if not saved_warning:
ui.ui_factory.suppressed_warnings.remove(warning_id)
def convert(self):
try:
branch = self.bzrdir.open_branch()
if branch.user_url != self.bzrdir.user_url:
ui.ui_factory.note(gettext(
'This is a checkout. The branch (%s) needs to be upgraded'
' separately.') % (urlutils.unescape_for_display(
branch.user_url, 'utf-8')))
del branch
except (errors.NotBranchError, errors.IncompatibleRepositories):
# might not be a format we can open without upgrading; see e.g.
# https://bugs.launchpad.net/bzr/+bug/253891
pass
if self.format is None:
try:
rich_root = self.bzrdir.find_repository()._format.rich_root_data
except errors.NoRepositoryPresent:
rich_root = False # assume no rich roots
if rich_root:
format_name = "default-rich-root"
else:
format_name = "default"
format = format_registry.make_bzrdir(format_name)
else:
format = self.format
if not self.bzrdir.needs_format_conversion(format):
raise errors.UpToDateFormat(self.bzrdir._format)
if not self.bzrdir.can_convert_format():
raise errors.BzrError(gettext("cannot upgrade from bzrdir format %s") %
self.bzrdir._format)
self.bzrdir.check_conversion_target(format)
ui.ui_factory.note(gettext('starting upgrade of %s') %
urlutils.unescape_for_display(self.transport.base, 'utf-8'))
self.backup_oldpath, self.backup_newpath = self.bzrdir.backup_bzrdir()
while self.bzrdir.needs_format_conversion(format):
converter = self.bzrdir._format.get_converter(format)
self.bzrdir = converter.convert(self.bzrdir, None)
ui.ui_factory.note(gettext('finished'))
def clean_up(self):
"""Clean-up after a conversion.
This removes the backup.bzr directory.
"""
transport = self.transport
backup_relpath = transport.relpath(self.backup_newpath)
child_pb = ui.ui_factory.nested_progress_bar()
child_pb.update(gettext('Deleting backup.bzr'))
try:
transport.delete_tree(backup_relpath)
finally:
child_pb.finished()
def upgrade(url, format=None, clean_up=False, dry_run=False):
"""Upgrade locations to format.
This routine wraps the smart_upgrade() routine with a nicer UI.
In particular, it ensures all URLs can be opened before starting
and reports a summary at the end if more than one upgrade was attempted.
This routine is useful for command line tools. Other bzrlib clients
probably ought to use smart_upgrade() instead.
:param url: a URL of the locations to upgrade.
:param format: the format to convert to or None for the best default
:param clean-up: if True, the backup.bzr directory is removed if the
upgrade succeeded for a given repo/branch/tree
:param dry_run: show what would happen but don't actually do any upgrades
:return: the list of exceptions encountered
"""
control_dirs = [ControlDir.open_unsupported(url)]
attempted, succeeded, exceptions = smart_upgrade(control_dirs,
format, clean_up=clean_up, dry_run=dry_run)
if len(attempted) > 1:
attempted_count = len(attempted)
succeeded_count = len(succeeded)
failed_count = attempted_count - succeeded_count
ui.ui_factory.note(
gettext('\nSUMMARY: {0} upgrades attempted, {1} succeeded,'\
' {2} failed').format(
attempted_count, succeeded_count, failed_count))
return exceptions
def smart_upgrade(control_dirs, format, clean_up=False,
dry_run=False):
"""Convert control directories to a new format intelligently.
If the control directory is a shared repository, dependent branches
are also converted provided the repository converted successfully.
If the conversion of a branch fails, remaining branches are still tried.
:param control_dirs: the BzrDirs to upgrade
:param format: the format to convert to or None for the best default
:param clean_up: if True, the backup.bzr directory is removed if the
upgrade succeeded for a given repo/branch/tree
:param dry_run: show what would happen but don't actually do any upgrades
:return: attempted-control-dirs, succeeded-control-dirs, exceptions
"""
all_attempted = []
all_succeeded = []
all_exceptions = []
for control_dir in control_dirs:
attempted, succeeded, exceptions = _smart_upgrade_one(control_dir,
format, clean_up=clean_up, dry_run=dry_run)
all_attempted.extend(attempted)
all_succeeded.extend(succeeded)
all_exceptions.extend(exceptions)
return all_attempted, all_succeeded, all_exceptions
def _smart_upgrade_one(control_dir, format, clean_up=False,
dry_run=False):
"""Convert a control directory to a new format intelligently.
See smart_upgrade for parameter details.
"""
# If the URL is a shared repository, find the dependent branches
dependents = None
try:
repo = control_dir.open_repository()
except errors.NoRepositoryPresent:
# A branch or checkout using a shared repository higher up
pass
else:
# The URL is a repository. If it successfully upgrades,
# then upgrade the dependent branches as well.
if repo.is_shared():
dependents = repo.find_branches(using=True)
# Do the conversions
attempted = [control_dir]
succeeded, exceptions = _convert_items([control_dir], format, clean_up,
dry_run)
if succeeded and dependents:
ui.ui_factory.note(gettext('Found %d dependent branches - upgrading ...')
% (len(dependents),))
# Convert dependent branches
branch_cdirs = [b.bzrdir for b in dependents]
successes, problems = _convert_items(branch_cdirs, format, clean_up,
dry_run, label="branch")
attempted.extend(branch_cdirs)
succeeded.extend(successes)
exceptions.extend(problems)
# Return the result
return attempted, succeeded, exceptions
# FIXME: There are several problems below:
# - RemoteRepository doesn't support _unsupported (really ?)
# - raising AssertionError is rude and may not be necessary
# - no tests
# - the only caller uses only the label
def _get_object_and_label(control_dir):
"""Return the primary object and type label for a control directory.
:return: object, label where:
* object is a Branch, Repository or WorkingTree and
* label is one of:
* branch - a branch
* repository - a repository
* tree - a lightweight checkout
"""
try:
try:
br = control_dir.open_branch(unsupported=True,
ignore_fallbacks=True)
except NotImplementedError:
# RemoteRepository doesn't support the unsupported parameter
br = control_dir.open_branch(ignore_fallbacks=True)
except errors.NotBranchError:
pass
else:
return br, "branch"
try:
repo = control_dir.open_repository()
except errors.NoRepositoryPresent:
pass
else:
return repo, "repository"
try:
wt = control_dir.open_workingtree()
except (errors.NoWorkingTree, errors.NotLocalUrl):
pass
else:
return wt, "tree"
raise AssertionError("unknown type of control directory %s", control_dir)
def _convert_items(items, format, clean_up, dry_run, label=None):
"""Convert a sequence of control directories to the given format.
:param items: the control directories to upgrade
:param format: the format to convert to or None for the best default
:param clean-up: if True, the backup.bzr directory is removed if the
upgrade succeeded for a given repo/branch/tree
:param dry_run: show what would happen but don't actually do any upgrades
:param label: the label for these items or None to calculate one
:return: items successfully upgraded, exceptions
"""
succeeded = []
exceptions = []
child_pb = ui.ui_factory.nested_progress_bar()
child_pb.update(gettext('Upgrading bzrdirs'), 0, len(items))
for i, control_dir in enumerate(items):
# Do the conversion
location = control_dir.root_transport.base
bzr_object, bzr_label = _get_object_and_label(control_dir)
type_label = label or bzr_label
child_pb.update(gettext("Upgrading %s") % (type_label), i+1, len(items))
ui.ui_factory.note(gettext('Upgrading {0} {1} ...').format(type_label,
urlutils.unescape_for_display(location, 'utf-8'),))
try:
if not dry_run:
cv = Convert(control_dir=control_dir, format=format)
except errors.UpToDateFormat, ex:
ui.ui_factory.note(str(ex))
succeeded.append(control_dir)
continue
except Exception, ex:
trace.warning('conversion error: %s' % ex)
exceptions.append(ex)
continue
# Do any required post processing
succeeded.append(control_dir)
if clean_up:
try:
ui.ui_factory.note(gettext('Removing backup ...'))
if not dry_run:
cv.clean_up()
except Exception, ex:
trace.warning(gettext('failed to clean-up {0}: {1}') % (location, ex))
exceptions.append(ex)
child_pb.finished()
# Return the result
return succeeded, exceptions
|