~bzr-pqm/bzr/bzr.dev

# Copyright (C) 2004, 2005, 2006 by 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 codecs

import errno

import os

import os.path

import sys

import bzrlib

from bzrlib import (

    branch,

    bundle,

    bzrdir,

    config,

    errors,

    ignores,

    log,

    osutils,

    repository,

    transport,

    ui,

    urlutils,

    )

from bzrlib.branch import Branch, BranchReferenceFormat

from bzrlib.bundle import read_bundle_from_url

from bzrlib.bundle.apply_bundle import install_bundle, merge_bundle

from bzrlib.conflicts import ConflictList

from bzrlib.commands import Command, display_command

from bzrlib.errors import (BzrError, BzrCheckError, BzrCommandError, 

                           NotBranchError, DivergedBranches, NotConflicted,

                           NoSuchFile, NoWorkingTree, FileInWrongBranch,

                           NotVersionedError, NotABundle)

from bzrlib.merge import Merge3Merger

from bzrlib.option import Option

from bzrlib.progress import DummyProgress, ProgressPhase

from bzrlib.revision import common_ancestor

from bzrlib.revisionspec import RevisionSpec

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

from bzrlib.transport.local import LocalTransport

from bzrlib.workingtree import WorkingTree

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

    try:

        return internal_tree_files(file_list, default_branch)

    except FileInWrongBranch, e:

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

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

          "metaweave and weave" % typestring

    raise 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 BzrCommandError('You can only supply one of revision_id or --revision')

        if revision_id is None and revision is None:

            raise 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 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(rev))

        if len(revs) == 0:

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

    """

    takes_args = ['file*']

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

    encoding_type = 'replace'

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

        import bzrlib.add

        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.

    """

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

    @display_command

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

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

            raise BzrCommandError('invalid kind specified')

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

        if revision is None:

            inv = tree.read_working_inventory()

        else:

            if len(revision) > 1:

                raise BzrCommandError('bzr inventory --revision takes'

                    ' exactly one revision identifier')

            inv = tree.branch.repository.get_revision_inventory(

                revision[0].in_history(tree.branch).rev_id)

        for path, entry in inv.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 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 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 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 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 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 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 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 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 NoSuchFile:

                    raise 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 NoSuchFile:

                        new_transport = to_transport.clone('..')

                        needed.append((new_transport,

                                       new_transport.relpath(to_transport.base)))

                        if new_transport.base == to_transport.base:

                            raise 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 NoWorkingTree:

                    count = br_to.pull(br_from, overwrite)

                else:

                    count = tree_to.pull(br_from, overwrite)

            except DivergedBranches:

                raise BzrCommandError("These branches have diverged."

                                      "  Try a merge then push with overwrite.")

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

        if verbose:

            new_rh = br_to.revision_history()

            if old_rh != new_rh:

                # Something changed

                from bzrlib.log import show_changed_revisions

                show_changed_revisions(br_to, old_rh, new_rh,

                                       to_file=self.outf)

class cmd_branch(Command):

    """Create a new copy of a branch.

    If the TO_LOCATION is omitted, the last component of the FROM_LOCATION will

    be used.  In other words, "branch ../foo/bar" will attempt to create ./bar.

    To retrieve the branch as of a particular revision, supply the --revision

    parameter, as in "branch foo/bar -r 5".

    --basis is to speed up branching from remote branches.  When specified, it

    copies all the file-contents, inventory and revision data from the basis

    branch before copying anything from the remote branch.

    """

    takes_args = ['from_location', 'to_location?']

    takes_options = ['revision', 'basis']

    aliases = ['get', 'clone']

    def run(self, from_location, to_location=None, revision=None, basis=None):

        if revision is None:

            revision = [None]

        elif len(revision) > 1:

            raise BzrCommandError(

                'bzr branch --revision takes exactly 1 revision value')

        try:

            br_from = Branch.open(from_location)

        except OSError, e:

            if e.errno == errno.ENOENT:

                raise BzrCommandError('Source location "%s" does not'

                                      ' exist.' % to_location)

            else:

                raise

        br_from.lock_read()

        try:

            if basis is not None:

                basis_dir = bzrdir.BzrDir.open_containing(basis)[0]

            else:

                basis_dir = None

            if len(revision) == 1 and revision[0] is not None:

                revision_id = revision[0].in_history(br_from)[1]