~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2005-2012 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

"""builtin bzr commands"""

from __future__ import absolute_import

import os

import bzrlib.bzrdir

from bzrlib import lazy_import

lazy_import.lazy_import(globals(), """

import cStringIO

import errno

import sys

import time

import bzrlib

from bzrlib import (

    bugtracker,

    bundle,

    btree_index,

    controldir,

    directory_service,

    delta,

    config as _mod_config,

    errors,

    globbing,

    hooks,

    log,

    merge as _mod_merge,

    merge_directive,

    osutils,

    reconfigure,

    rename_map,

    revision as _mod_revision,

    static_tuple,

    timestamp,

    transport,

    ui,

    urlutils,

    views,

    gpg,

    )

from bzrlib.branch import Branch

from bzrlib.conflicts import ConflictList

from bzrlib.transport import memory

from bzrlib.revisionspec import RevisionSpec, RevisionInfo

from bzrlib.smtp_connection import SMTPConnection

from bzrlib.workingtree import WorkingTree

from bzrlib.i18n import gettext, ngettext

""")

from bzrlib.commands import (

    Command,

    builtin_command_registry,

    display_command,

    )

from bzrlib.option import (

    ListOption,

    Option,

    RegistryOption,

    custom_help,

    _parse_revision_str,

    )

from bzrlib.trace import mutter, note, warning, is_quiet, get_verbosity_level

from bzrlib import (

    symbol_versioning,

    )

def _get_branch_location(control_dir, possible_transports=None):

    """Return location of branch for this control dir."""

    try:

        target = control_dir.get_branch_reference()

    except errors.NotBranchError:

        return control_dir.root_transport.base

    if target is not None:

        return target

    this_branch = control_dir.open_branch(

        possible_transports=possible_transports)

    # This may be a heavy checkout, where we want the master branch

    master_location = this_branch.get_bound_location()

    if master_location is not None:

        return master_location

    # If not, use a local sibling

    return this_branch.base

def _is_colocated(control_dir, possible_transports=None):

    """Check if the branch in control_dir is colocated.

    :param control_dir: Control directory

    :return: Tuple with boolean indicating whether the branch is colocated

        and the full URL to the actual branch

    """

    # This path is meant to be relative to the existing branch

    this_url = _get_branch_location(control_dir,

        possible_transports=possible_transports)

    # Perhaps the target control dir supports colocated branches?

    try:

        root = controldir.ControlDir.open(this_url,

            possible_transports=possible_transports)

    except errors.NotBranchError:

        return (False, this_url)

    else:

        try:

            wt = control_dir.open_workingtree()

        except (errors.NoWorkingTree, errors.NotLocalUrl):

            return (False, this_url)

        else:

            return (

                root._format.colocated_branches and

                control_dir.control_url == root.control_url,

                this_url)

def lookup_new_sibling_branch(control_dir, location, possible_transports=None):

    """Lookup the location for a new sibling branch.

    :param control_dir: Control directory to find sibling branches from

    :param location: Name of the new branch

    :return: Full location to the new branch

    """

    location = directory_service.directories.dereference(location)

    if '/' not in location and '\\' not in location:

        (colocated, this_url) = _is_colocated(control_dir, possible_transports)

        if colocated:

            return urlutils.join_segment_parameters(this_url,

                {"branch": urlutils.escape(location)})

        else:

            return urlutils.join(this_url, '..', urlutils.escape(location))

    return location

def open_sibling_branch(control_dir, location, possible_transports=None):

    """Open a branch, possibly a sibling of another.

    :param control_dir: Control directory relative to which to lookup the

        location.

    :param location: Location to look up

    :return: branch to open

    """

    try:

        # Perhaps it's a colocated branch?

        return control_dir.open_branch(location, 

            possible_transports=possible_transports)

    except (errors.NotBranchError, errors.NoColocatedBranchSupport):

        this_url = _get_branch_location(control_dir)

        return Branch.open(

            urlutils.join(

                this_url, '..', urlutils.escape(location)))

def open_nearby_branch(near=None, location=None, possible_transports=None):

    """Open a nearby branch.

    :param near: Optional location of container from which to open branch

    :param location: Location of the branch

    :return: Branch instance

    """

    if near is None:

        if location is None:

            location = "."

        try:

            return Branch.open(location,

                possible_transports=possible_transports)

        except errors.NotBranchError:

            near = "."

    cdir = controldir.ControlDir.open(near,

        possible_transports=possible_transports)

    return open_sibling_branch(cdir, location,

        possible_transports=possible_transports)

def iter_sibling_branches(control_dir, possible_transports=None):

    """Iterate over the siblings of a branch.

    :param control_dir: Control directory for which to look up the siblings

    :return: Iterator over tuples with branch name and branch object

    """

    seen_urls = set()

    try:

        reference = control_dir.get_branch_reference()

    except errors.NotBranchError:

        # There is no active branch, just return the colocated branches.

        for name, branch in control_dir.get_branches().iteritems():

            yield name, branch

        return

    if reference is not None:

        ref_branch = Branch.open(reference,

            possible_transports=possible_transports)

    else:

        ref_branch = None

    if ref_branch is None or ref_branch.name:

        if ref_branch is not None:

            control_dir = ref_branch.bzrdir

        for name, branch in control_dir.get_branches().iteritems():

            yield name, branch

    else:

        repo = ref_branch.bzrdir.find_repository()

        for branch in repo.find_branches(using=True):

            name = urlutils.relative_url(repo.user_url,

                branch.user_url).rstrip("/")

            yield name, branch

def tree_files_for_add(file_list):

    """

    Return a tree and list of absolute paths from a file list.

    Similar to tree_files, but add handles files a bit differently, so it a

    custom implementation.  In particular, MutableTreeTree.smart_add expects

    absolute paths, which it immediately converts to relative paths.

    """

    # FIXME Would be nice to just return the relative paths like

    # internal_tree_files does, but there are a large number of unit tests

    # that assume the current interface to mutabletree.smart_add

    if file_list:

        tree, relpath = WorkingTree.open_containing(file_list[0])

        if tree.supports_views():

            view_files = tree.views.lookup_view()

            if view_files:

                for filename in file_list:

                    if not osutils.is_inside_any(view_files, filename):

                        raise errors.FileOutsideView(filename, view_files)

        file_list = file_list[:]

        file_list[0] = tree.abspath(relpath)

    else:

        tree = WorkingTree.open_containing(u'.')[0]

        if tree.supports_views():

            view_files = tree.views.lookup_view()

            if view_files:

                file_list = view_files

                view_str = views.view_display_str(view_files)

                note(gettext("Ignoring files outside view. View is %s") % view_str)

    return tree, file_list

def _get_one_revision(command_name, revisions):

    if revisions is None:

        return None

    if len(revisions) != 1:

        raise errors.BzrCommandError(gettext(

            'bzr %s --revision takes exactly one revision identifier') % (

                command_name,))

    return revisions[0]

def _get_one_revision_tree(command_name, revisions, branch=None, tree=None):

    """Get a revision tree. Not suitable for commands that change the tree.

    Specifically, the basis tree in dirstate trees is coupled to the dirstate

    and doing a commit/uncommit/pull will at best fail due to changing the

    basis revision data.

    If tree is passed in, it should be already locked, for lifetime management

    of the trees internal cached state.

    """

    if branch is None:

        branch = tree.branch

    if revisions is None:

        if tree is not None:

            rev_tree = tree.basis_tree()

        else:

            rev_tree = branch.basis_tree()

    else:

        revision = _get_one_revision(command_name, revisions)

        rev_tree = revision.as_tree(branch)

    return rev_tree

def _get_view_info_for_change_reporter(tree):

    """Get the view information from a tree for change reporting."""

    view_info = None

    try:

        current_view = tree.views.get_view_info()[0]

        if current_view is not None:

            view_info = (current_view, tree.views.lookup_view())

    except errors.ViewsNotSupported:

        pass

    return view_info

def _open_directory_or_containing_tree_or_branch(filename, directory):

    """Open the tree or branch containing the specified file, unless

    the --directory option is used to specify a different branch."""

    if directory is not None:

        return (None, Branch.open(directory), filename)

    return controldir.ControlDir.open_containing_tree_or_branch(filename)

# TODO: Make sure no commands unconditionally use the working directory as a

# branch.  If a filename argument is used, the first of them should be used to

# specify the branch.  (Perhaps this can be factored out into some kind of

# Argument class, representing a file in a branch, where the first occurrence

# opens the branch?)

class cmd_status(Command):

    __doc__ = """Display status summary.

    This reports on versioned and unknown files, reporting them

    grouped by state.  Possible states are:

    added

        Versioned in the working copy but not in the previous revision.

    removed

        Versioned in the previous revision but removed or deleted

        in the working copy.

    renamed

        Path of this file changed from the previous revision;

        the text may also have changed.  This includes files whose

        parent directory was renamed.

    modified

        Text has changed since the previous revision.

    kind changed

        File kind has been changed (e.g. from file to directory).

    unknown

        Not versioned and not matching an ignore pattern.

    Additionally for directories, symlinks and files with a changed

    executable bit, Bazaar indicates their type using a trailing

    character: '/', '@' or '*' respectively. These decorations can be

    disabled using the '--no-classify' option.

    To see ignored files use 'bzr ignored'.  For details on the

    changes to file texts, use 'bzr diff'.

    Note that --short or -S gives status flags for each item, similar

    to Subversion's status command. To get output similar to svn -q,

    use bzr status -SV.

    If no arguments are specified, the status of the entire working

    directory is shown.  Otherwise, only the status of the specified

    files or directories is reported.  If a directory is given, status

    is reported for everything inside that directory.

    Before merges are committed, the pending merge tip revisions are

    shown. To see all pending merge revisions, use the -v option.

    To skip the display of pending merge information altogether, use

    the no-pending option or specify a file/directory.

    To compare the working directory to a specific revision, pass a

    single revision to the revision argument.

    To see which files have changed in a specific revision, or between

    two revisions, pass a revision range to the revision argument.

    This will produce the same results as calling 'bzr diff --summarize'.

    """

    # TODO: --no-recurse/-N, --recurse options

    takes_args = ['file*']

    takes_options = ['show-ids', 'revision', 'change', 'verbose',

                     Option('short', help='Use short status indicators.',

                            short_name='S'),

                     Option('versioned', help='Only show versioned files.',

                            short_name='V'),

                     Option('no-pending', help='Don\'t show pending merges.',

                           ),

                     Option('no-classify',

                            help='Do not mark object type using indicator.',

                           ),

                     ]

    aliases = ['st', 'stat']

    encoding_type = 'replace'

    _see_also = ['diff', 'revert', 'status-flags']

    @display_command

    def run(self, show_ids=False, file_list=None, revision=None, short=False,

            versioned=False, no_pending=False, verbose=False,

            no_classify=False):

        from bzrlib.status import show_tree_status

        if revision and len(revision) > 2:

            raise errors.BzrCommandError(gettext('bzr status --revision takes exactly'

                                         ' one or two revision specifiers'))

        tree, relfile_list = WorkingTree.open_containing_paths(file_list)

        # Avoid asking for specific files when that is not needed.

        if relfile_list == ['']:

            relfile_list = None

            # Don't disable pending merges for full trees other than '.'.

            if file_list == ['.']:

                no_pending = True

        # A specific path within a tree was given.

        elif relfile_list is not None:

            no_pending = True

        show_tree_status(tree, show_ids=show_ids,

                         specific_files=relfile_list, revision=revision,

                         to_file=self.outf, short=short, versioned=versioned,

                         show_pending=(not no_pending), verbose=verbose,

                         classify=not no_classify)

class cmd_cat_revision(Command):

    __doc__ = """Write out metadata for a revision.

    The revision to print can either be specified by a specific

    revision identifier, or you can use --revision.

    """

    hidden = True

    takes_args = ['revision_id?']

    takes_options = ['directory', 'revision']

    # cat-revision is more for frontends so should be exact

    encoding = 'strict'

    def print_revision(self, revisions, revid):

        stream = revisions.get_record_stream([(revid,)], 'unordered', True)

        record = stream.next()

        if record.storage_kind == 'absent':

            raise errors.NoSuchRevision(revisions, revid)

        revtext = record.get_bytes_as('fulltext')

        self.outf.write(revtext.decode('utf-8'))

    @display_command

    def run(self, revision_id=None, revision=None, directory=u'.'):

        if revision_id is not None and revision is not None:

            raise errors.BzrCommandError(gettext('You can only supply one of'

                                         ' revision_id or --revision'))

        if revision_id is None and revision is None:

            raise errors.BzrCommandError(gettext('You must supply either'

                                         ' --revision or a revision_id'))

        b = controldir.ControlDir.open_containing_tree_or_branch(directory)[1]

        revisions = b.repository.revisions

        if revisions is None:

            raise errors.BzrCommandError(gettext('Repository %r does not support '

                'access to raw revision texts'))

        b.repository.lock_read()

        try:

            # TODO: jam 20060112 should cat-revision always output utf-8?

            if revision_id is not None:

                revision_id = osutils.safe_revision_id(revision_id, warn=False)

                try:

                    self.print_revision(revisions, revision_id)

                except errors.NoSuchRevision:

                    msg = gettext("The repository {0} contains no revision {1}.").format(

                        b.repository.base, revision_id)

                    raise errors.BzrCommandError(msg)

            elif revision is not None:

                for rev in revision:

                    if rev is None:

                        raise errors.BzrCommandError(

                            gettext('You cannot specify a NULL revision.'))

                    rev_id = rev.as_revision_id(b)

                    self.print_revision(revisions, rev_id)

        finally:

            b.repository.unlock()

class cmd_dump_btree(Command):

    __doc__ = """Dump the contents of a btree index file to stdout.

    PATH is a btree index file, it can be any URL. This includes things like

    .bzr/repository/pack-names, or .bzr/repository/indices/a34b3a...ca4a4.iix

    By default, the tuples stored in the index file will be displayed. With

    --raw, we will uncompress the pages, but otherwise display the raw bytes

    stored in the index.

    """

    # TODO: Do we want to dump the internal nodes as well?

    # TODO: It would be nice to be able to dump the un-parsed information,

    #       rather than only going through iter_all_entries. However, this is

    #       good enough for a start

    hidden = True

    encoding_type = 'exact'

    takes_args = ['path']

    takes_options = [Option('raw', help='Write the uncompressed bytes out,'

                                        ' rather than the parsed tuples.'),

                    ]

    def run(self, path, raw=False):

        dirname, basename = osutils.split(path)

        t = transport.get_transport(dirname)

        if raw:

            self._dump_raw_bytes(t, basename)

        else:

            self._dump_entries(t, basename)

    def _get_index_and_bytes(self, trans, basename):

        """Create a BTreeGraphIndex and raw bytes."""

        bt = btree_index.BTreeGraphIndex(trans, basename, None)

        bytes = trans.get_bytes(basename)

        bt._file = cStringIO.StringIO(bytes)

        bt._size = len(bytes)

        return bt, bytes

    def _dump_raw_bytes(self, trans, basename):

        import zlib

        # We need to parse at least the root node.

        # This is because the first page of every row starts with an

        # uncompressed header.

        bt, bytes = self._get_index_and_bytes(trans, basename)

        for page_idx, page_start in enumerate(xrange(0, len(bytes),

                                                     btree_index._PAGE_SIZE)):

            page_end = min(page_start + btree_index._PAGE_SIZE, len(bytes))

            page_bytes = bytes[page_start:page_end]

            if page_idx == 0:

                self.outf.write('Root node:\n')

                header_end, data = bt._parse_header_from_bytes(page_bytes)

                self.outf.write(page_bytes[:header_end])

                page_bytes = data

            self.outf.write('\nPage %d\n' % (page_idx,))

            if len(page_bytes) == 0:

                self.outf.write('(empty)\n');

            else:

                decomp_bytes = zlib.decompress(page_bytes)

                self.outf.write(decomp_bytes)

                self.outf.write('\n')

    def _dump_entries(self, trans, basename):

        try:

            st = trans.stat(basename)

        except errors.TransportNotPossible:

            # We can't stat, so we'll fake it because we have to do the 'get()'

            # anyway.

            bt, _ = self._get_index_and_bytes(trans, basename)

        else:

            bt = btree_index.BTreeGraphIndex(trans, basename, st.st_size)

        for node in bt.iter_all_entries():

            # Node is made up of:

            # (index, key, value, [references])

            try:

                refs = node[3]

            except IndexError:

                refs_as_tuples = None

            else:

                refs_as_tuples = static_tuple.as_tuples(refs)

            as_tuple = (tuple(node[1]), node[2], refs_as_tuples)

            self.outf.write('%s\n' % (as_tuple,))

class cmd_remove_tree(Command):

    __doc__ = """Remove the working tree from a given branch/checkout.

    Since a lightweight checkout is little more than a working tree

    this will refuse to run against one.

    To re-create the working tree, use "bzr checkout".

    """

    _see_also = ['checkout', 'working-trees']

    takes_args = ['location*']

    takes_options = [

        Option('force',

               help='Remove the working tree even if it has '

                    'uncommitted or shelved changes.'),

        ]

    def run(self, location_list, force=False):

        if not location_list:

            location_list=['.']

        for location in location_list:

            d = controldir.ControlDir.open(location)

            try:

                working = d.open_workingtree()

            except errors.NoWorkingTree:

                raise errors.BzrCommandError(gettext("No working tree to remove"))

            except errors.NotLocalUrl:

                raise errors.BzrCommandError(gettext("You cannot remove the working tree"

                                             " of a remote path"))

            if not force:

                if (working.has_changes()):

                    raise errors.UncommittedChanges(working)

                if working.get_shelf_manager().last_shelf() is not None:

                    raise errors.ShelvedChanges(working)

            if working.user_url != working.branch.user_url:

                raise errors.BzrCommandError(gettext("You cannot remove the working tree"

                                             " from a lightweight checkout"))

            d.destroy_workingtree()

class cmd_repair_workingtree(Command):

    __doc__ = """Reset the working tree state file.

    This is not meant to be used normally, but more as a way to recover from

    filesystem corruption, etc. This rebuilds the working inventory back to a

    'known good' state. Any new modifications (adding a file, renaming, etc)

    will be lost, though modified files will still be detected as such.

    Most users will want something more like "bzr revert" or "bzr update"

    unless the state file has become corrupted.

    By default this attempts to recover the current state by looking at the

    headers of the state file. If the state file is too corrupted to even do

    that, you can supply --revision to force the state of the tree.

    """

    takes_options = ['revision', 'directory',

        Option('force',

               help='Reset the tree even if it doesn\'t appear to be'

                    ' corrupted.'),

    ]

    hidden = True

    def run(self, revision=None, directory='.', force=False):

        tree, _ = WorkingTree.open_containing(directory)

        self.add_cleanup(tree.lock_tree_write().unlock)

        if not force:

            try:

                tree.check_state()

            except errors.BzrError:

                pass # There seems to be a real error here, so we'll reset

            else:

                # Refuse

                raise errors.BzrCommandError(gettext(

                    'The tree does not appear to be corrupt. You probably'

                    ' want "bzr revert" instead. Use "--force" if you are'

                    ' sure you want to reset the working tree.'))

        if revision is None:

            revision_ids = None

        else:

            revision_ids = [r.as_revision_id(tree.branch) for r in revision]

        try:

            tree.reset_state(revision_ids)

        except errors.BzrError, e:

            if revision_ids is None:

                extra = (gettext(', the header appears corrupt, try passing -r -1'

                         ' to set the state to the last commit'))

            else:

                extra = ''

            raise errors.BzrCommandError(gettext('failed to reset the tree state{0}').format(extra))

class cmd_revno(Command):

    __doc__ = """Show current revision number.

    This is equal to the number of revisions on this branch.

    """

    _see_also = ['info']

    takes_args = ['location?']

    takes_options = [

        Option('tree', help='Show revno of working tree.'),

        'revision',

        ]

    @display_command

    def run(self, tree=False, location=u'.', revision=None):

        if revision is not None and tree:

            raise errors.BzrCommandError(gettext("--tree and --revision can "

                "not be used together"))

        if tree:

            try:

                wt = WorkingTree.open_containing(location)[0]