~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2004, 2005, 2006 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

"""builtin bzr commands"""

import os

from bzrlib.lazy_import import lazy_import

lazy_import(globals(), """

import codecs

import errno

import sys

import bzrlib

from bzrlib import (

    branch,

    bundle,

    bzrdir,

    config,

    errors,

    ignores,

    log,

    merge as _mod_merge,

    osutils,

    repository,

    transport,

    tree as _mod_tree,

    ui,

    urlutils,

    )

from bzrlib.branch import Branch

from bzrlib.bundle.apply_bundle import install_bundle, merge_bundle

from bzrlib.conflicts import ConflictList

from bzrlib.revision import common_ancestor

from bzrlib.revisionspec import RevisionSpec

from bzrlib.workingtree import WorkingTree

""")

from bzrlib.commands import Command, display_command

from bzrlib.option import Option

from bzrlib.progress import DummyProgress, ProgressPhase

from bzrlib.trace import mutter, note, log_error, warning, is_quiet, info

def tree_files(file_list, default_branch=u'.'):

    try:

        return internal_tree_files(file_list, default_branch)

    except errors.FileInWrongBranch, e:

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

                                     (e.path, file_list[0]))

# 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'.'):

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

    :return: workingtree, [relative_paths]

    """

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

        return WorkingTree.open_containing(default_branch)[0], file_list

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

    new_list = []

    for filename in file_list:

        try:

            new_list.append(tree.relpath(filename))

        except errors.PathNotChild:

            raise errors.FileInWrongBranch(tree.branch, filename)

    return tree, new_list

def get_format_type(typestring):

    """Parse and return a format specifier."""

    if typestring == "weave":

        return bzrdir.BzrDirFormat6()

    if typestring == "default":

        return bzrdir.BzrDirMetaFormat1()

    if typestring == "metaweave":

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormat7()

        return format

    if typestring == "knit":

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit1()

        return format

    if typestring == "experimental-knit2":

        format = bzrdir.BzrDirMetaFormat1()

        format.repository_format = repository.RepositoryFormatKnit2()

        return format

    msg = "Unknown bzr format %s. Current formats are: default, knit,\n" \

          "metaweave and weave" % typestring

    raise errors.BzrCommandError(msg)

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

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

    unknown

        Not versioned and not matching an ignore pattern.

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

    changes to file texts, use 'bzr diff'.

    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.

    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']

    aliases = ['st', 'stat']

    encoding_type = 'replace'

    @display_command

    def run(self, show_ids=False, file_list=None, revision=None):

        from bzrlib.status import show_tree_status

        tree, file_list = tree_files(file_list)

        show_tree_status(tree, show_ids=show_ids,

                         specific_files=file_list, revision=revision,

                         to_file=self.outf)

class cmd_cat_revision(Command):

    """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 = ['revision']

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

    encoding = 'strict'

    @display_command

    def run(self, revision_id=None, revision=None):

        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(u'.')[0].branch

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

        if revision_id is not None:

            self.outf.write(b.repository.get_revision_xml(revision_id).decode('utf-8'))

        elif revision is not None:

            for rev in revision:

                if rev is None:

                    raise errors.BzrCommandError('You cannot specify a NULL'

                                                 ' revision.')

                revno, rev_id = rev.in_history(b)

                self.outf.write(b.repository.get_revision_xml(rev_id).decode('utf-8'))

class cmd_revno(Command):

    """Show current revision number.

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

    """

    takes_args = ['location?']

    @display_command

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

        self.outf.write(str(Branch.open_containing(location)[0].revno()))

        self.outf.write('\n')

class cmd_revision_info(Command):

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

    """

    hidden = True

    takes_args = ['revision_info*']

    takes_options = ['revision']

    @display_command

    def run(self, revision=None, revision_info_list=[]):

        revs = []

        if revision is not None:

            revs.extend(revision)

        if revision_info_list is not None:

            for rev in revision_info_list:

                revs.append(RevisionSpec.from_string(rev))

        if len(revs) == 0:

            raise errors.BzrCommandError('You must supply a revision identifier')

        b = WorkingTree.open_containing(u'.')[0].branch

        for rev in revs:

            revinfo = rev.in_history(b)

            if revinfo.revno is None:

                print '     %s' % revinfo.rev_id

            else:

                print '%4d %s' % (revinfo.revno, revinfo.rev_id)

class cmd_add(Command):

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

    """

    takes_args = ['file*']

    takes_options = ['no-recurse', 'dry-run', 'verbose',

                     Option('file-ids-from', type=unicode,

                            help='Lookup file ids from here')]

    encoding_type = 'replace'

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

            file_ids_from=None):

        import bzrlib.add

        if file_ids_from is not None:

            try:

                base_tree, base_path = WorkingTree.open_containing(

                                            file_ids_from)

            except errors.NoWorkingTree:

                base_branch, base_path = Branch.open_containing(

                                            file_ids_from)

                base_tree = base_branch.basis_tree()

            action = bzrlib.add.AddFromBaseAction(base_tree, base_path,

                          to_file=self.outf, should_print=(not is_quiet()))

        else:

            action = bzrlib.add.AddAction(to_file=self.outf,

                should_print=(not is_quiet()))

        added, ignored = bzrlib.add.smart_add(file_list, not no_recurse,

                                              action=action, save=not dry_run)

        if len(ignored) > 0:

            if verbose:

                for glob in sorted(ignored.keys()):

                    for path in ignored[glob]:

                        self.outf.write("ignored %s matching \"%s\"\n" 

                                        % (path, glob))

            else:

                match_len = 0

                for glob, paths in ignored.items():

                    match_len += len(paths)

                self.outf.write("ignored %d file(s).\n" % match_len)

            self.outf.write("If you wish to add some of these files,"

                            " please add them by name.\n")

class cmd_mkdir(Command):

    """Create a new versioned directory.

    This is equivalent to creating the directory and then adding it.

    """

    takes_args = ['dir+']

    encoding_type = 'replace'

    def run(self, dir_list):

        for d in dir_list:

            os.mkdir(d)

            wt, dd = WorkingTree.open_containing(d)

            wt.add([dd])

            self.outf.write('added %s\n' % d)

class cmd_relpath(Command):

    """Show path of a file relative to root"""

    takes_args = ['filename']

    hidden = True

    @display_command

    def run(self, filename):

        # TODO: jam 20050106 Can relpath return a munged path if

        #       sys.stdout encoding cannot represent it?

        tree, relpath = WorkingTree.open_containing(filename)

        self.outf.write(relpath)

        self.outf.write('\n')

class cmd_inventory(Command):

    """Show inventory of the current working copy or a revision.

    It is possible to limit the output to a particular entry

    type using the --kind option.  For example: --kind file.

    It is also possible to restrict the list of files to a specific

    set. For example: bzr inventory --show-ids this/file

    """

    takes_options = ['revision', 'show-ids', 'kind']

    takes_args = ['file*']

    @display_command

    def run(self, revision=None, show_ids=False, kind=None, file_list=None):

        if kind and kind not in ['file', 'directory', 'symlink']:

            raise errors.BzrCommandError('invalid kind specified')

        work_tree, file_list = tree_files(file_list)

        if revision is not None:

            if len(revision) > 1:

                raise errors.BzrCommandError('bzr inventory --revision takes'

                                             ' exactly one revision identifier')

            revision_id = revision[0].in_history(work_tree.branch).rev_id

            tree = work_tree.branch.repository.revision_tree(revision_id)

            # We include work_tree as well as 'tree' here

            # So that doing '-r 10 path/foo' will lookup whatever file

            # exists now at 'path/foo' even if it has been renamed, as

            # well as whatever files existed in revision 10 at path/foo

            trees = [tree, work_tree]

        else:

            tree = work_tree

            trees = [tree]

        if file_list is not None:

            file_ids = _mod_tree.find_ids_across_trees(file_list, trees,

                                                      require_versioned=True)

            # find_ids_across_trees may include some paths that don't

            # exist in 'tree'.

            entries = sorted((tree.id2path(file_id), tree.inventory[file_id])

                             for file_id in file_ids if file_id in tree)

        else:

            entries = tree.inventory.entries()

        for path, entry in entries:

            if kind and kind != entry.kind:

                continue

            if show_ids:

                self.outf.write('%-50s %s\n' % (path, entry.file_id))

            else:

                self.outf.write(path)

                self.outf.write('\n')

class cmd_mv(Command):

    """Move or rename a file.

    usage:

        bzr mv OLDNAME NEWNAME

        bzr mv SOURCE... DESTINATION

    If the last argument is a versioned directory, all the other names

    are moved into it.  Otherwise, there must be exactly two arguments

    and the file is changed to a new name, which must not already exist.

    Files cannot be moved between branches.

    """

    takes_args = ['names*']

    aliases = ['move', 'rename']

    encoding_type = 'replace'

    def run(self, names_list):

        if names_list is None:

            names_list = []

        if len(names_list) < 2:

            raise errors.BzrCommandError("missing file argument")

        tree, rel_names = tree_files(names_list)

        if os.path.isdir(names_list[-1]):

            # move into existing directory

            for pair in tree.move(rel_names[:-1], rel_names[-1]):

                self.outf.write("%s => %s\n" % pair)

        else:

            if len(names_list) != 2:

                raise errors.BzrCommandError('to mv multiple files the destination '

                                             'must be a versioned directory')

            tree.rename_one(rel_names[0], rel_names[1])

            self.outf.write("%s => %s\n" % (rel_names[0], rel_names[1]))

class cmd_pull(Command):

    """Turn this branch into a mirror of another branch.

    This command only works on branches that have not diverged.  Branches are

    considered diverged if the destination branch's most recent commit is one

    that has not been merged (directly or indirectly) into the parent.

    If branches have diverged, you can use 'bzr merge' to integrate the changes

    from one into the other.  Once one branch has merged, the other should

    be able to pull it again.

    If you want to forget your local changes and just update your branch to

    match the remote one, use pull --overwrite.

    If there is no default location set, the first pull will set it.  After

    that, you can omit the location to use the default.  To change the

    default, use --remember. The value will only be saved if the remote

    location can be accessed.

    """

    takes_options = ['remember', 'overwrite', 'revision', 'verbose']

    takes_args = ['location?']

    encoding_type = 'replace'

    def run(self, location=None, remember=False, overwrite=False, revision=None, verbose=False):

        # FIXME: too much stuff is in the command class

        try:

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

            branch_to = tree_to.branch

        except errors.NoWorkingTree:

            tree_to = None

            branch_to = Branch.open_containing(u'.')[0]

        reader = None

        if location is not None:

            try:

                reader = bundle.read_bundle_from_url(location)

            except errors.NotABundle:

                pass # Continue on considering this url a Branch

        stored_loc = branch_to.get_parent()

        if location is None:

            if stored_loc is None:

                raise errors.BzrCommandError("No pull location known or"

                                             " specified.")

            else:

                display_url = urlutils.unescape_for_display(stored_loc,

                        self.outf.encoding)

                self.outf.write("Using saved location: %s\n" % display_url)

                location = stored_loc

        if reader is not None:

            install_bundle(branch_to.repository, reader)

            branch_from = branch_to

        else:

            branch_from = Branch.open(location)

            if branch_to.get_parent() is None or remember:

                branch_to.set_parent(branch_from.base)

        rev_id = None

        if revision is None:

            if reader is not None:

                rev_id = reader.target

        elif len(revision) == 1:

            rev_id = revision[0].in_history(branch_from).rev_id

        else:

            raise errors.BzrCommandError('bzr pull --revision takes one value.')

        old_rh = branch_to.revision_history()

        if tree_to is not None:

            count = tree_to.pull(branch_from, overwrite, rev_id)

        else:

            count = branch_to.pull(branch_from, overwrite, rev_id)

        note('%d revision(s) pulled.' % (count,))

        if verbose:

            new_rh = branch_to.revision_history()

            if old_rh != new_rh:

                # Something changed

                from bzrlib.log import show_changed_revisions

                show_changed_revisions(branch_to, old_rh, new_rh,

                                       to_file=self.outf)

class cmd_push(Command):

    """Update a mirror of this branch.

    The target branch will not have its working tree populated because this

    is both expensive, and is not supported on remote file systems.

    Some smart servers or protocols *may* put the working tree in place in

    the future.

    This command only works on branches that have not diverged.  Branches are

    considered diverged if the destination branch's most recent commit is one

    that has not been merged (directly or indirectly) by the source branch.

    If branches have diverged, you can use 'bzr push --overwrite' to replace

    the other branch completely, discarding its unmerged changes.

    If you want to ensure you have the different changes in the other branch,

    do a merge (see bzr help merge) from the other branch, and commit that.

    After that you will be able to do a push without '--overwrite'.

    If there is no default push location set, the first push will set it.

    After that, you can omit the location to use the default.  To change the

    default, use --remember. The value will only be saved if the remote

    location can be accessed.

    """

    takes_options = ['remember', 'overwrite', 'verbose',

                     Option('create-prefix', 

                            help='Create the path leading up to the branch '

                                 'if it does not already exist')]

    takes_args = ['location?']

    encoding_type = 'replace'

    def run(self, location=None, remember=False, overwrite=False,

            create_prefix=False, verbose=False):

        # FIXME: Way too big!  Put this into a function called from the

        # command.

        br_from = Branch.open_containing('.')[0]

        stored_loc = br_from.get_push_location()

        if location is None:

            if stored_loc is None:

                raise errors.BzrCommandError("No push location known or specified.")

            else:

                display_url = urlutils.unescape_for_display(stored_loc,

                        self.outf.encoding)

                self.outf.write("Using saved location: %s\n" % display_url)

                location = stored_loc

        to_transport = transport.get_transport(location)

        location_url = to_transport.base

        old_rh = []

        try:

            dir_to = bzrdir.BzrDir.open(location_url)

            br_to = dir_to.open_branch()

        except errors.NotBranchError:

            # create a branch.

            to_transport = to_transport.clone('..')

            if not create_prefix:

                try:

                    relurl = to_transport.relpath(location_url)

                    mutter('creating directory %s => %s', location_url, relurl)

                    to_transport.mkdir(relurl)

                except errors.NoSuchFile:

                    raise errors.BzrCommandError("Parent directory of %s "

                                                 "does not exist." % location)

            else:

                current = to_transport.base

                needed = [(to_transport, to_transport.relpath(location_url))]

                while needed:

                    try:

                        to_transport, relpath = needed[-1]

                        to_transport.mkdir(relpath)

                        needed.pop()

                    except errors.NoSuchFile:

                        new_transport = to_transport.clone('..')

                        needed.append((new_transport,

                                       new_transport.relpath(to_transport.base)))

                        if new_transport.base == to_transport.base:

                            raise errors.BzrCommandError("Could not create "

                                                         "path prefix.")

            dir_to = br_from.bzrdir.clone(location_url,

                revision_id=br_from.last_revision())

            br_to = dir_to.open_branch()

            count = len(br_to.revision_history())

            # We successfully created the target, remember it

            if br_from.get_push_location() is None or remember:

                br_from.set_push_location(br_to.base)

        else:

            # We were able to connect to the remote location, so remember it

            # we don't need to successfully push because of possible divergence.

            if br_from.get_push_location() is None or remember:

                br_from.set_push_location(br_to.base)

            old_rh = br_to.revision_history()

            try:

                try:

                    tree_to = dir_to.open_workingtree()

                except errors.NotLocalUrl:

                    warning('This transport does not update the working '

                            'tree of: %s' % (br_to.base,))

                    count = br_to.pull(br_from, overwrite)

                except errors.NoWorkingTree:

                    count = br_to.pull(br_from, overwrite)

                else:

                    count = tree_to.pull(br_from, overwrite)

            except errors.DivergedBranches:

                raise errors.BzrCommandError('These branches have diverged.'