~bzr-pqm/bzr/bzr.dev

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

import sys, os

import bzrlib

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

from bzrlib.errors import BzrError, BzrCheckError, BzrCommandError

from bzrlib.branch import find_branch

from bzrlib import BZRDIR

plugin_cmds = {}

def register_command(cmd):

    "Utility function to help register a command"

    global plugin_cmds

    k = cmd.__name__

    if k.startswith("cmd_"):

        k_unsquished = _unsquish_command_name(k)

    else:

        k_unsquished = k

    if not plugin_cmds.has_key(k_unsquished):

        plugin_cmds[k_unsquished] = cmd

    else:

        log_error('Two plugins defined the same command: %r' % k)

        log_error('Not loading the one in %r' % sys.modules[cmd.__module__])

def _squish_command_name(cmd):

    return 'cmd_' + cmd.replace('-', '_')

def _unsquish_command_name(cmd):

    assert cmd.startswith("cmd_")

    return cmd[4:].replace('_','-')

def _parse_revision_str(revstr):

    """This handles a revision string -> revno. 

    It supports integers directly, but everything else it

    defers for passing to Branch.get_revision_info()

    >>> _parse_revision_str('234')

    [234]

    >>> _parse_revision_str('234..567')

    [234, 567]

    >>> _parse_revision_str('..')

    [None, None]

    >>> _parse_revision_str('..234')

    [None, 234]

    >>> _parse_revision_str('234..')

    [234, None]

    >>> _parse_revision_str('234..456..789') # Maybe this should be an error

    [234, 456, 789]

    >>> _parse_revision_str('234....789') # Error?

    [234, None, 789]

    >>> _parse_revision_str('revid:test@other.com-234234')

    ['revid:test@other.com-234234']

    >>> _parse_revision_str('revid:test@other.com-234234..revid:test@other.com-234235')

    ['revid:test@other.com-234234', 'revid:test@other.com-234235']

    >>> _parse_revision_str('revid:test@other.com-234234..23')

    ['revid:test@other.com-234234', 23]

    >>> _parse_revision_str('date:2005-04-12')

    ['date:2005-04-12']

    >>> _parse_revision_str('date:2005-04-12 12:24:33')

    ['date:2005-04-12 12:24:33']

    >>> _parse_revision_str('date:2005-04-12T12:24:33')

    ['date:2005-04-12T12:24:33']

    >>> _parse_revision_str('date:2005-04-12,12:24:33')

    ['date:2005-04-12,12:24:33']

    >>> _parse_revision_str('-5..23')

    [-5, 23]

    >>> _parse_revision_str('-5')

    [-5]

    >>> _parse_revision_str('123a')

    ['123a']

    >>> _parse_revision_str('abc')

    ['abc']

    """

    import re

    old_format_re = re.compile('\d*:\d*')

    m = old_format_re.match(revstr)

    if m:

        warning('Colon separator for revision numbers is deprecated.'

                ' Use .. instead')

        revs = []

        for rev in revstr.split(':'):

            if rev:

                revs.append(int(rev))

            else:

                revs.append(None)

        return revs

    revs = []

    for x in revstr.split('..'):

        if not x:

            revs.append(None)

        else:

            try:

                revs.append(int(x))

            except ValueError:

                revs.append(x)

    return revs

def _get_cmd_dict(plugins_override=True):

    d = {}

    for k, v in globals().iteritems():

        if k.startswith("cmd_"):

            d[_unsquish_command_name(k)] = v

    # If we didn't load plugins, the plugin_cmds dict will be empty

    if plugins_override:

        d.update(plugin_cmds)

    else:

        d2 = plugin_cmds.copy()

        d2.update(d)

        d = d2

    return d

def get_all_cmds(plugins_override=True):

    """Return canonical name and class for all registered commands."""

    for k, v in _get_cmd_dict(plugins_override=plugins_override).iteritems():

        yield k,v

def get_cmd_class(cmd, plugins_override=True):

    """Return the canonical name and command class for a command.

    """

    cmd = str(cmd)                      # not unicode

    # first look up this command under the specified name

    cmds = _get_cmd_dict(plugins_override=plugins_override)

    try:

        return cmd, cmds[cmd]

    except KeyError:

        pass

    # look for any command which claims this as an alias

    for cmdname, cmdclass in cmds.iteritems():

        if cmd in cmdclass.aliases:

            return cmdname, cmdclass

    cmdclass = ExternalCommand.find_command(cmd)

    if cmdclass:

        return cmd, cmdclass

    raise BzrCommandError("unknown command %r" % cmd)

class Command(object):

    """Base class for commands.

    The docstring for an actual command should give a single-line

    summary, then a complete description of the command.  A grammar

    description will be inserted.

    takes_args

        List of argument forms, marked with whether they are optional,

        repeated, etc.

    takes_options

        List of options that may be given for this command.

    hidden

        If true, this command isn't advertised.

    """

    aliases = []

    takes_args = []

    takes_options = []

    hidden = False

    def __init__(self, options, arguments):

        """Construct and run the command.

        Sets self.status to the return value of run()."""

        assert isinstance(options, dict)

        assert isinstance(arguments, dict)

        cmdargs = options.copy()

        cmdargs.update(arguments)

        assert self.__doc__ != Command.__doc__, \

               ("No help message set for %r" % self)

        self.status = self.run(**cmdargs)

    def run(self):

        """Override this in sub-classes.

        This is invoked with the options and arguments bound to

        keyword parameters.

        Return 0 or None if the command was successful, or a shell

        error code if not.

        """

        return 0

class ExternalCommand(Command):

    """Class to wrap external commands.

    We cheat a little here, when get_cmd_class() calls us we actually give it back

    an object we construct that has the appropriate path, help, options etc for the

    specified command.

    When run_bzr() tries to instantiate that 'class' it gets caught by the __call__

    method, which we override to call the Command.__init__ method. That then calls

    our run method which is pretty straight forward.

    The only wrinkle is that we have to map bzr's dictionary of options and arguments

    back into command line options and arguments for the script.

    """

    def find_command(cls, cmd):

        import os.path

        bzrpath = os.environ.get('BZRPATH', '')

        for dir in bzrpath.split(os.pathsep):

            path = os.path.join(dir, cmd)

            if os.path.isfile(path):

                return ExternalCommand(path)

        return None

    find_command = classmethod(find_command)

    def __init__(self, path):

        self.path = path

        pipe = os.popen('%s --bzr-usage' % path, 'r')

        self.takes_options = pipe.readline().split()

        for opt in self.takes_options:

            if not opt in OPTIONS:

                raise BzrError("Unknown option '%s' returned by external command %s"

                               % (opt, path))

        # TODO: Is there any way to check takes_args is valid here?

        self.takes_args = pipe.readline().split()

        if pipe.close() is not None:

            raise BzrError("Failed funning '%s --bzr-usage'" % path)

        pipe = os.popen('%s --bzr-help' % path, 'r')

        self.__doc__ = pipe.read()

        if pipe.close() is not None:

            raise BzrError("Failed funning '%s --bzr-help'" % path)

    def __call__(self, options, arguments):

        Command.__init__(self, options, arguments)

        return self

    def run(self, **kargs):

        opts = []

        args = []

        keys = kargs.keys()

        keys.sort()

        for name in keys:

            optname = name.replace('_','-')

            value = kargs[name]

            if OPTIONS.has_key(optname):

                # it's an option

                opts.append('--%s' % optname)

                if value is not None and value is not True:

                    opts.append(str(value))

            else:

                # it's an arg, or arg list

                if type(value) is not list:

                    value = [value]

                for v in value:

                    if v is not None:

                        args.append(str(v))

        self.status = os.spawnv(os.P_WAIT, self.path, [self.path] + opts + args)

        return self.status

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.

    unchanged

        Nothing about this file has changed since the previous revision.

        Only shown with --all.

    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.

    """

    takes_args = ['file*']

    takes_options = ['all', 'show-ids']

    aliases = ['st', 'stat']

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

        if file_list:

            b = find_branch(file_list[0])

            file_list = [b.relpath(x) for x in file_list]

            # special case: only one path was given and it's the root

            # of the branch

            if file_list == ['']:

                file_list = None

        else:

            b = find_branch('.')

        import status

        status.show_status(b, show_unchanged=all, show_ids=show_ids,

                           specific_files=file_list)

class cmd_cat_revision(Command):

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

    hidden = True

    takes_args = ['revision_id']

    def run(self, revision_id):

        from bzrlib.xml import pack_xml

        pack_xml(find_branch('.').get_revision(revision_id), sys.stdout)

class cmd_revno(Command):

    """Show current revision number.

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

    def run(self):

        print find_branch('.').revno()

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

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

        from bzrlib.branch import find_branch

        revs = []

        if revision is not None:

            revs.extend(revision)

        if revision_info_list is not None:

            revs.extend(revision_info_list)

        if len(revs) == 0:

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

        b = find_branch('.')

        for rev in revs:

            print '%4d %s' % b.get_revision_info(rev)

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.

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

    are currently unknown.

    TODO: Perhaps adding a file whose directly is not versioned should

    recursively add that parent, rather than giving an error?

    """

    takes_args = ['file+']

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

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

        from bzrlib.add import smart_add

        smart_add(file_list, verbose, not no_recurse)

class cmd_mkdir(Command):

    """Create a new versioned directory.

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

    """

    takes_args = ['dir+']

    def run(self, dir_list):

        b = None

        for d in dir_list:

            os.mkdir(d)

            if not b:

                b = find_branch(d)

            b.add([d], verbose=True)

class cmd_relpath(Command):

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

    takes_args = ['filename']

    hidden = True

    def run(self, filename):

        print find_branch(filename).relpath(filename)

class cmd_inventory(Command):

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

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

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

        b = find_branch('.')

        if revision == None:

            inv = b.read_working_inventory()

        else:

            if len(revision) > 1:

                raise BzrCommandError('bzr inventory --revision takes'

                    ' exactly one revision identifier')

            inv = b.get_revision_inventory(b.lookup_revision(revision[0]))

        for path, entry in inv.entries():

            if show_ids:

                print '%-50s %s' % (path, entry.file_id)

            else:

                print path

class cmd_move(Command):

    """Move files to a different directory.

    examples:

        bzr move *.txt doc

    The destination must be a versioned directory in the same branch.

    """

    takes_args = ['source$', 'dest']

    def run(self, source_list, dest):

        b = find_branch('.')

        b.move([b.relpath(s) for s in source_list], b.relpath(dest))

class cmd_rename(Command):

    """Change the name of an entry.

    examples:

      bzr rename frob.c frobber.c

      bzr rename src/frob.c lib/frob.c

    It is an error if the destination name exists.

    See also the 'move' command, which moves files into a different

    directory without changing their name.

    TODO: Some way to rename multiple files without invoking bzr for each

    one?"""

    takes_args = ['from_name', 'to_name']

    def run(self, from_name, to_name):

        b = find_branch('.')

        b.rename_one(b.relpath(from_name), b.relpath(to_name))

class cmd_pull(Command):

    """Pull any changes from another branch into the current one.

    If the location is omitted, the last-used location will be used.

    Both the revision history and the working directory will be

    updated.

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

    considered diverged if both branches have had commits without first

    pulling from the other.

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

    from one into the other.

    """

    takes_args = ['location?']

    def run(self, location=None):

        from bzrlib.merge import merge

        import tempfile

        from shutil import rmtree

        import errno

        br_to = find_branch('.')

        stored_loc = None

        try:

            stored_loc = br_to.controlfile("x-pull", "rb").read().rstrip('\n')

        except IOError, e:

            if e.errno != errno.ENOENT:

                raise

        if location is None:

            if stored_loc is None:

                raise BzrCommandError("No pull location known or specified.")

            else:

                print "Using last location: %s" % stored_loc

                location = stored_loc

        cache_root = tempfile.mkdtemp()

        from bzrlib.branch import DivergedBranches

        br_from = find_branch(location)

        location = pull_loc(br_from)

        old_revno = br_to.revno()

        try:

            from branch import find_cached_branch, DivergedBranches

            br_from = find_cached_branch(location, cache_root)

            location = pull_loc(br_from)

            old_revno = br_to.revno()

            try:

                br_to.update_revisions(br_from)

            except DivergedBranches:

                raise BzrCommandError("These branches have diverged."

                    "  Try merge.")

            merge(('.', -1), ('.', old_revno), check_clean=False)

            if location != stored_loc:

                br_to.controlfile("x-pull", "wb").write(location + "\n")

        finally:

            rmtree(cache_root)

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

    """

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

    takes_options = ['revision']

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

        import errno

        from bzrlib.merge import merge

        from bzrlib.branch import DivergedBranches, NoSuchRevision, \

             find_cached_branch, Branch

        from shutil import rmtree

        from meta_store import CachedStore

        import tempfile

        cache_root = tempfile.mkdtemp()

        if revision is not None:

            if len(revision) > 1:

                raise BzrCommandError('bzr branch --revision takes exactly 1 revision value')

        try:

            try:

                br_from = find_cached_branch(from_location, cache_root)

            except OSError, e:

                if e.errno == errno.ENOENT:

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

                                          ' exist.' % to_location)

                else:

                    raise

            if to_location is None:

                to_location = os.path.basename(from_location.rstrip("/\\"))

            try:

                os.mkdir(to_location)

            except OSError, e:

                if e.errno == errno.EEXIST:

                    raise BzrCommandError('Target directory "%s" already'

                                          ' exists.' % to_location)

                if e.errno == errno.ENOENT:

                    raise BzrCommandError('Parent of "%s" does not exist.' %

                                          to_location)

                else:

                    raise

            br_to = Branch(to_location, init=True)

            br_to.set_root_id(br_from.get_root_id())

            if revision:

                revno = br_to.lookup_revision(revision[0])

                try:

                    br_to.update_revisions(br_from, stop_revision=revno)

                except NoSuchRevision:

                    rmtree(to_location)

                    msg = "The branch %s has no revision %d." % (from_location,

                                                                 revno)

                    raise BzrCommandError(msg)

            merge((to_location, -1), (to_location, 0), this_dir=to_location,

                  check_clean=False, ignore_zero=True)

            from_location = pull_loc(br_from)

            br_to.controlfile("x-pull", "wb").write(from_location + "\n")

        finally:

            rmtree(cache_root)

def pull_loc(branch):

    # TODO: Should perhaps just make attribute be 'base' in

    # RemoteBranch and Branch?

    if hasattr(branch, "baseurl"):

        return branch.baseurl

    else:

        return branch.base

class cmd_renames(Command):

    """Show list of renamed files.

    TODO: Option to show renames between two historical versions.

    TODO: Only show renames under dir, rather than in the whole branch.

    """

    takes_args = ['dir?']

    def run(self, dir='.'):

        b = find_branch(dir)

        old_inv = b.basis_tree().inventory

        new_inv = b.read_working_inventory()

        renames = list(bzrlib.tree.find_renames(old_inv, new_inv))

        renames.sort()

        for old_name, new_name in renames:

            print "%s => %s" % (old_name, new_name)        

class cmd_info(Command):

    """Show statistical information about a branch."""

    takes_args = ['branch?']

    def run(self, branch=None):

        import info

        b = find_branch(branch)

        info.show_info(b)

class cmd_remove(Command):

    """Make a file unversioned.

    This makes bzr stop tracking changes to a versioned file.  It does

    not delete the working copy.

    """

    takes_args = ['file+']

    takes_options = ['verbose']

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

        b = find_branch(file_list[0])

        b.remove([b.relpath(f) for f in file_list], verbose=verbose)

class cmd_file_id(Command):

    """Print file_id of a particular file or directory.

    The file_id is assigned when the file is first added and remains the

    same through all revisions where the file exists, even when it is

    moved or renamed.

    """

    hidden = True

    takes_args = ['filename']

    def run(self, filename):

        b = find_branch(filename)

        i = b.inventory.path2id(b.relpath(filename))

        if i == None:

            raise BzrError("%r is not a versioned file" % filename)

        else:

            print i

class cmd_file_path(Command):

    """Print path of file_ids to a file or directory.

    This prints one line for each directory down to the target,

    starting at the branch root."""

    hidden = True

    takes_args = ['filename']

    def run(self, filename):

        b = find_branch(filename)

        inv = b.inventory

        fid = inv.path2id(b.relpath(filename))

        if fid == None:

            raise BzrError("%r is not a versioned file" % filename)

        for fip in inv.get_idpath(fid):

            print fip

class cmd_revision_history(Command):

    """Display list of revision ids on this branch."""

    hidden = True

    def run(self):

        for patchid in find_branch('.').revision_history():

            print patchid

class cmd_directories(Command):

    """Display list of versioned directories in this branch."""

    def run(self):

        for name, ie in find_branch('.').read_working_inventory().directories():