~bzr-pqm/bzr/bzr.dev

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

import os

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

import codecs

import cStringIO

import sys

import time

import bzrlib

from bzrlib import (

    bugtracker,

    bundle,

    btree_index,

    bzrdir,

    directory_service,

    delta,

    config,

    errors,

    globbing,

    hooks,

    log,

    merge as _mod_merge,

    merge_directive,

    osutils,

    reconfigure,

    rename_map,

    revision as _mod_revision,

    static_tuple,

    symbol_versioning,

    timestamp,

    transport,

    ui,

    urlutils,

    views,

    )

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.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

def tree_files(file_list, default_branch=u'.', canonicalize=True,

    apply_view=True):

    try:

        return internal_tree_files(file_list, default_branch, canonicalize,

            apply_view)

    except errors.FileInWrongBranch, e:

        raise errors.BzrCommandError("%s is not in the same branch as %s" %

                                     (e.path, file_list[0]))

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("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(

            '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

# XXX: Bad function name; should possibly also be a class method of

# WorkingTree rather than a function.

def internal_tree_files(file_list, default_branch=u'.', canonicalize=True,

    apply_view=True):

    """Convert command-line paths to a WorkingTree and relative paths.

    This is typically used for command-line processors that take one or

    more filenames, and infer the workingtree that contains them.

    The filenames given are not required to exist.

    :param file_list: Filenames to convert.

    :param default_branch: Fallback tree path to use if file_list is empty or

        None.

    :param apply_view: if True and a view is set, apply it or check that

        specified files are within it

    :return: workingtree, [relative_paths]

    """

    if file_list is None or len(file_list) == 0:

        tree = WorkingTree.open_containing(default_branch)[0]

        if tree.supports_views() and apply_view:

            view_files = tree.views.lookup_view()

            if view_files:

                file_list = view_files

                view_str = views.view_display_str(view_files)

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

        return tree, file_list

    tree = WorkingTree.open_containing(osutils.realpath(file_list[0]))[0]

    return tree, safe_relpath_files(tree, file_list, canonicalize,

        apply_view=apply_view)

def safe_relpath_files(tree, file_list, canonicalize=True, apply_view=True):

    """Convert file_list into a list of relpaths in tree.

    :param tree: A tree to operate on.

    :param file_list: A list of user provided paths or None.

    :param apply_view: if True and a view is set, apply it or check that

        specified files are within it

    :return: A list of relative paths.

    :raises errors.PathNotChild: When a provided path is in a different tree

        than tree.

    """

    if file_list is None:

        return None

    if tree.supports_views() and apply_view:

        view_files = tree.views.lookup_view()

    else:

        view_files = []

    new_list = []

    # tree.relpath exists as a "thunk" to osutils, but canonical_relpath

    # doesn't - fix that up here before we enter the loop.

    if canonicalize:

        fixer = lambda p: osutils.canonical_relpath(tree.basedir, p)

    else:

        fixer = tree.relpath

    for filename in file_list:

        try:

            relpath = fixer(osutils.dereference_path(filename))

            if  view_files and not osutils.is_inside_any(view_files, relpath):

                raise errors.FileOutsideView(filename, view_files)

            new_list.append(relpath)

        except errors.PathNotChild:

            raise errors.FileInWrongBranch(tree.branch, filename)

    return new_list

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 bzrdir.BzrDir.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 an executable

    bit, Bazaar indicates their type using a trailing character: '/', '@'

    or '*' respectively.

    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.

    If a revision argument is given, the status is calculated against

    that revision, or between two revisions if two are provided.

    """

    # TODO: --no-recurse, --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.',

                           ),

                     ]

    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):

        from bzrlib.status import show_tree_status

        if revision and len(revision) > 2:

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

                                         ' one or two revision specifiers')

        tree, relfile_list = tree_files(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)

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('You can only supply one of'

                                         ' revision_id or --revision')

        if revision_id is None and revision is None:

            raise errors.BzrCommandError('You must supply either'

                                         ' --revision or a revision_id')

        b = WorkingTree.open_containing(directory)[0].branch

        revisions = b.repository.revisions

        if revisions is None:

            raise errors.BzrCommandError('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 = "The repository %s contains no revision %s." % (

                        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(

                            '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,))

            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 = bzrdir.BzrDir.open(location)

            try:

                working = d.open_workingtree()

            except errors.NoWorkingTree:

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

            except errors.NotLocalUrl:

                raise errors.BzrCommandError("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("You cannot remove the working tree"

                                             " from a lightweight checkout")

            d.destroy_workingtree()

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'),

        ]

    @display_command

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

        if tree:

            try:

                wt = WorkingTree.open_containing(location)[0]

                self.add_cleanup(wt.lock_read().unlock)

            except (errors.NoWorkingTree, errors.NotLocalUrl):

                raise errors.NoWorkingTree(location)

            revid = wt.last_revision()

            try:

                revno_t = wt.branch.revision_id_to_dotted_revno(revid)

            except errors.NoSuchRevision:

                revno_t = ('???',)

            revno = ".".join(str(n) for n in revno_t)

        else:

            b = Branch.open_containing(location)[0]

            self.add_cleanup(b.lock_read().unlock)

            revno = b.revno()

        self.cleanup_now()

        self.outf.write(str(revno) + '\n')

class cmd_revision_info(Command):

    __doc__ = """Show revision number and revision id for a given revision identifier.

    """

    hidden = True

    takes_args = ['revision_info*']

    takes_options = [

        'revision',

        custom_help('directory',

            help='Branch to examine, '

                 'rather than the one containing the working directory.'),

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

        ]

    @display_command

    def run(self, revision=None, directory=u'.', tree=False,

            revision_info_list=[]):

        try:

            wt = WorkingTree.open_containing(directory)[0]

            b = wt.branch

            self.add_cleanup(wt.lock_read().unlock)

        except (errors.NoWorkingTree, errors.NotLocalUrl):

            wt = None

            b = Branch.open_containing(directory)[0]

            self.add_cleanup(b.lock_read().unlock)

        revision_ids = []

        if revision is not None:

            revision_ids.extend(rev.as_revision_id(b) for rev in revision)

        if revision_info_list is not None:

            for rev_str in revision_info_list:

                rev_spec = RevisionSpec.from_string(rev_str)

                revision_ids.append(rev_spec.as_revision_id(b))

        # No arguments supplied, default to the last revision

        if len(revision_ids) == 0:

            if tree:

                if wt is None:

                    raise errors.NoWorkingTree(directory)

                revision_ids.append(wt.last_revision())

            else:

                revision_ids.append(b.last_revision())

        revinfos = []

        maxlen = 0

        for revision_id in revision_ids:

            try:

                dotted_revno = b.revision_id_to_dotted_revno(revision_id)

                revno = '.'.join(str(i) for i in dotted_revno)

            except errors.NoSuchRevision:

                revno = '???'

            maxlen = max(maxlen, len(revno))

            revinfos.append([revno, revision_id])

        self.cleanup_now()

        for ri in revinfos:

            self.outf.write('%*s %s\n' % (maxlen, ri[0], ri[1]))

class cmd_add(Command):

    __doc__ = """Add specified files or directories.

    In non-recursive mode, all the named items are added, regardless

    of whether they were previously ignored.  A warning is given if

    any of the named files are already versioned.

    In recursive mode (the default), files are treated the same way

    but the behaviour for directories is different.  Directories that

    are already versioned do not give a warning.  All directories,

    whether already versioned or not, are searched for files or

    subdirectories that are neither versioned or ignored, and these

    are added.  This search proceeds recursively into versioned

    directories.  If no names are given '.' is assumed.

    Therefore simply saying 'bzr add' will version all files that

    are currently unknown.

    Adding a file whose parent directory is not versioned will

    implicitly add the parent, and so on up to the root. This means

    you should never need to explicitly add a directory, they'll just

    get added when you add a file in the directory.

    --dry-run will show which files would be added, but not actually

    add them.

    --file-ids-from will try to use the file ids from the supplied path.

    It looks up ids trying to find a matching parent directory with the

    same filename, and then by pure path. This option is rarely needed

    but can be useful when adding the same logical file into two

    branches that will be merged later (without showing the two different

    adds as a conflict). It is also useful when merging another project

    into a subdirectory of this one.

    Any files matching patterns in the ignore list will not be added

    unless they are explicitly mentioned.

    """

    takes_args = ['file*']

    takes_options = [

        Option('no-recurse',

               help="Don't recursively add the contents of directories."),

        Option('dry-run',

               help="Show what would be done, but don't actually do anything."),

        'verbose',

        Option('file-ids-from',

               type=unicode,

               help='Lookup file ids from this tree.'),

        ]

    encoding_type = 'replace'

    _see_also = ['remove', 'ignore']

    def run(self, file_list, no_recurse=False, dry_run=False, verbose=False,