~abentley/bzrtools/bzrtools.dev

# Copyright (C) 2005, 2006, 2007 Aaron Bentley <aaron@aaronbentley.com>

# Copyright (C) 2007 John Arbash Meinel

#

#    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 codecs

import errno

import os

import re

import tempfile

import shutil

from subprocess import Popen, PIPE

import sys

import bzrlib

from bzrlib import revision as _mod_revision, trace, urlutils

import bzrlib.errors

from bzrlib.errors import (

    BzrCommandError,

    BzrError,

    ConnectionError,

    NotBranchError,

    NoSuchFile,

    NoWorkingTree,

    PermissionDenied,

    UnsupportedFormatError,

    TransportError,

    )

from bzrlib.bzrdir import BzrDir, BzrDirFormat

from bzrlib.transport import get_transport

def temp_tree():

    dirname = tempfile.mkdtemp("temp-branch")

    return BzrDir.create_standalone_workingtree(dirname)

def rm_tree(tree):

    shutil.rmtree(tree.basedir)

def is_clean(cur_tree):

    """

    Return true if no files are modifed or unknown

    """

    old_tree = cur_tree.basis_tree()

    new_tree = cur_tree

    non_source = []

    new_tree.lock_read()

    try:

        for path, file_class, kind, file_id, entry in new_tree.list_files():

            if file_class in ('?', 'I'):

                non_source.append(path)

        delta = new_tree.changes_from(old_tree, want_unchanged=False)

    finally:

        new_tree.unlock()

    return not delta.has_changed(), non_source

def set_push_data(tree, location):

    tree.branch.control_files.put_utf8("x-push-data", "%s\n" % location)

def get_push_data(tree):

    """

    >>> tree = temp_tree()

    >>> get_push_data(tree) is None

    True

    >>> set_push_data(tree, 'http://somewhere')

    >>> get_push_data(tree)

    u'http://somewhere'

    >>> rm_tree(tree)

    """

    try:

        location = tree.branch.control_files.get_utf8('x-push-data').read()

    except NoSuchFile:

        return None

    return location.rstrip('\n')

"""

>>> shell_escape('hello')

'\h\e\l\l\o'

"""

def shell_escape(arg):

    return "".join(['\\'+c for c in arg])

def safe_system(args):

    """

    >>> real_system = os.system

    >>> os.system = sys.stdout.write

    >>> safe_system(['a', 'b', 'cd'])

    \\a \\b \\c\\d

    >>> os.system = real_system

    """

    arg_str = " ".join([shell_escape(a) for a in args])

    return os.system(arg_str)

class RsyncUnknownStatus(Exception):

    def __init__(self, status):

        Exception.__init__(self, "Unknown status: %d" % status)

class NoRsync(Exception):

    def __init__(self, rsync_name):

        Exception.__init__(self, "%s not found." % rsync_name)

def rsync(source, target, ssh=False, excludes=(), silent=False,

          rsync_name="rsync"):

    cmd = [rsync_name, "-av", "--delete"]

    if ssh:

        cmd.extend(('-e', 'ssh'))

    if len(excludes) > 0:

        cmd.extend(('--exclude-from', '-'))

    cmd.extend((source, target))

    if silent:

        stderr = PIPE

        stdout = PIPE

    else:

        stderr = None

        stdout = None

    try:

        proc = Popen(cmd, stdin=PIPE, stderr=stderr, stdout=stdout)

    except OSError, e:

        if e.errno == errno.ENOENT:

            raise NoRsync(rsync_name)

    proc.stdin.write('\n'.join(excludes)+'\n')

    proc.stdin.close()

    if silent:

        proc.stderr.read()

        proc.stderr.close()

        proc.stdout.read()

        proc.stdout.close()

    proc.wait()

    if proc.returncode == 12:

        raise RsyncStreamIO()

    elif proc.returncode == 23:

        raise RsyncNoFile(source)

    elif proc.returncode != 0:

        raise RsyncUnknownStatus(proc.returncode)

    return cmd

def rsync_ls(source, ssh=False, silent=True):

    cmd = ["rsync"]

    if ssh:

        cmd.extend(('-e', 'ssh'))

    cmd.append(source)

    if silent:

        stderr = PIPE

    else:

        stderr = None

    proc = Popen(cmd, stderr=stderr, stdout=PIPE)

    result = proc.stdout.read()

    proc.stdout.close()

    if silent:

        proc.stderr.read()

        proc.stderr.close()

    proc.wait()

    if proc.returncode == 12:

        raise RsyncStreamIO()

    elif proc.returncode == 23:

        raise RsyncNoFile(source)

    elif proc.returncode != 0:

        raise RsyncUnknownStatus(proc.returncode)

    return [l.split(' ')[-1].rstrip('\n') for l in result.splitlines(True)]

exclusions = ('.bzr/x-push-data', '.bzr/branch/x-push/data', '.bzr/parent',

              '.bzr/branch/parent', '.bzr/x-pull-data', '.bzr/x-pull',

              '.bzr/pull', '.bzr/stat-cache', '.bzr/x-rsync-data',

              '.bzr/basis-inventory', '.bzr/inventory.backup.weave')

def read_revision_history(fname):

    return [l.rstrip('\r\n') for l in

            codecs.open(fname, 'rb', 'utf-8').readlines()]

def read_revision_info(path):

    """Parse a last_revision file to determine revision_info"""

    line = open(path, 'rb').readlines()[0].strip('\n')

    revno, revision_id = line.split(' ', 1)

    revno = int(revno)

    return revno, revision_id

class RsyncNoFile(Exception):

    def __init__(self, path):

        Exception.__init__(self, "No such file %s" % path)

class RsyncStreamIO(Exception):

    def __init__(self):

        Exception.__init__(self, "Error in rsync protocol data stream.")

class NotStandalone(BzrError):

    _format = '%(location) is not a standalone tree.'

    _internal = False

    def __init__(self, location):

        BzrError.__init__(self, location=location)

def get_revision_history(location, _rsync):

    tempdir = tempfile.mkdtemp('push')

    my_rsync = _rsync

    if my_rsync is None:

        my_rsync = rsync

    try:

        history_fname = os.path.join(tempdir, 'revision-history')

        try:

            cmd = my_rsync(location+'.bzr/revision-history', history_fname,

                        silent=True)

        except RsyncNoFile:

            cmd = rsync(location+'.bzr/branch/revision-history', history_fname,

                        silent=True)

        history = read_revision_history(history_fname)

    finally:

        shutil.rmtree(tempdir)

    return history

def get_revision_info(location, _rsync):

    """Get the revsision_info for an rsync-able branch"""

    tempdir = tempfile.mkdtemp('push')

    my_rsync = _rsync

    if my_rsync is None:

        my_rsync = rsync

    try:

        info_fname = os.path.join(tempdir, 'last-revision')

        cmd = rsync(location+'.bzr/branch/last-revision', info_fname,

                    silent=True)

        return read_revision_info(info_fname)

    finally:

        shutil.rmtree(tempdir)

def history_subset(location, branch, _rsync=None):

    local_history = branch.revision_history()

    try:

        remote_history = get_revision_history(location, _rsync)

    except RsyncNoFile:

        revno, revision_id = get_revision_info(location, _rsync)

        if revision_id == _mod_revision.NULL_REVISION:

            return True

        return bool(revision_id.decode('utf-8') in local_history)

    else:

        if len(remote_history) > len(local_history):

            return False

        for local, remote in zip(remote_history, local_history):

            if local != remote:

                return False

        return True

def empty_or_absent(location):

    try:

        files = rsync_ls(location)

        return files == ['.']

    except RsyncNoFile:

        return True

def rspush(tree, location=None, overwrite=False, working_tree=True,

    _rsync=None):

    tree.lock_write()

    try:

        my_rsync = _rsync

        if my_rsync is None:

            my_rsync = rsync

        if (tree.bzrdir.root_transport.base !=

            tree.branch.bzrdir.root_transport.base):

            raise NotStandalone(tree.bzrdir.root_transport.base)

        if (tree.branch.get_bound_location() is not None):

            raise NotStandalone(tree.bzrdir.root_transport.base)

        if (tree.branch.repository.is_shared()):

            raise NotStandalone(tree.bzrdir.root_transport.base)

        push_location = get_push_data(tree)

        if location is not None:

            if not location.endswith('/'):

                location += '/'

            push_location = location

        if push_location is None:

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

        if (push_location.find('::') != -1):

            usessh=False

        else:

            usessh=True

        if (push_location.find('://') != -1 or

            push_location.find(':') == -1):

            raise BzrCommandError("Invalid rsync path %r." % push_location)

        if working_tree:

            clean, non_source = is_clean(tree)

            if not clean:

                raise bzrlib.errors.BzrCommandError(

                    'This tree has uncommitted changes or unknown'

                    ' (?) files.  Use "bzr status" to list them.')

                sys.exit(1)

            final_exclusions = non_source[:]

        else:

            wt = tree

            final_exclusions = []

            for path, status, kind, file_id, entry in wt.list_files():

                final_exclusions.append(path)

        final_exclusions.extend(exclusions)

        if not overwrite:

            try:

                if not history_subset(push_location, tree.branch,

                                      _rsync=my_rsync):

                    raise bzrlib.errors.BzrCommandError(

                        "Local branch is not a newer version of remote"

                        " branch.")

            except RsyncNoFile:

                if not empty_or_absent(push_location):

                    raise bzrlib.errors.BzrCommandError(

                        "Remote location is not a bzr branch (or empty"

                        " directory)")

            except RsyncStreamIO:

                raise bzrlib.errors.BzrCommandError("Rsync could not use the"

                    " specified location.  Please ensure that"

                    ' "%s" is of the form "machine:/path".' % push_location)

        trace.note("Pushing to %s", push_location)

        my_rsync(tree.basedir+'/', push_location, ssh=usessh,

                 excludes=final_exclusions)

        set_push_data(tree, push_location)

    finally:

        tree.unlock()

def short_committer(committer):

    new_committer = re.sub('<.*>', '', committer).strip(' ')

    if len(new_committer) < 2:

        return committer

    return new_committer

def apache_ls(t):

    """Screen-scrape Apache listings"""

    apache_dir = '<img border="0" src="/icons/folder.gif" alt="[dir]">'\

        ' <a href="'

    t = t.clone()

    t._remote_path = lambda x: t.base

    try:

        lines = t.get('')

    except bzrlib.errors.NoSuchFile:

        return

    expr = re.compile('<a[^>]*href="([^>]*)\/"[^>]*>', flags=re.I)

    for line in lines:

        match = expr.search(line)

        if match is None:

            continue

        url = match.group(1)

        if url.startswith('http://') or url.startswith('/') or '../' in url:

            continue

        if '?' in url:

            continue

        yield url.rstrip('/')

def list_branches(t):

    def is_inside(branch):

        return bool(branch.base.startswith(t.base))

    if t.base.startswith('http://'):

        def evaluate(bzrdir):

            try:

                branch = bzrdir.open_branch()

                if is_inside(branch):

                    return True, branch

                else:

                    return True, None

            except NotBranchError:

                return True, None

        return [b for b in BzrDir.find_bzrdirs(t, list_current=apache_ls,

                evaluate=evaluate) if b is not None]

    elif not t.listable():

        raise BzrCommandError("Can't list this type of location.")

    return [b for b in BzrDir.find_branches(t) if is_inside(b)]

def evaluate_branch_tree(bzrdir):

    try:

        tree, branch = bzrdir._get_tree_branch()

    except NotBranchError:

        return True, None

    else:

        return True, (branch, tree)

def iter_branch_tree(t, lister=None):

    return (x for x in BzrDir.find_bzrdirs(t, evaluate=evaluate_branch_tree,

            list_current=lister) if x is not None)

def open_from_url(location):

    location = urlutils.normalize_url(location)

    dirname, basename = urlutils.split(location)

    return get_transport(dirname).get(basename)

def run_tests():

    import doctest

    result = doctest.testmod()

    if result[1] > 0:

        if result[0] == 0:

            print "All tests passed"

    else:

        print "No tests to run"

if __name__ == "__main__":

    run_tests()