~abentley/bzrtools/bzrtools.dev

# Copyright (C) 2005, 2006 by Aaron Bentley

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

from bzrlib.bzrdir import BzrDir

import bzrlib.bzrdir as bzrdir

from bzrlib.errors import (BzrError,

                           NotBranchError,

                           NoWorkingTree,

                           BzrCommandError,

                           NoSuchRevision,

                           NoRepositoryPresent,

                          )

from bzrlib.branch import Branch

from bzrlib.commit import Commit, NullCommitReporter

from bzrlib.commands import Command

from bzrlib.option import _global_option, Option

from bzrlib.merge import merge_inner

from bzrlib.revision import NULL_REVISION

import bzrlib.ui

import bzrlib.ui.text

from bzrlib.workingtree import WorkingTree

from errors import NoPyBaz

try:

    import pybaz

    import pybaz.errors

    from pybaz import NameParser as NameParser

    from pybaz.backends.baz import null_cmd

except ImportError:

    raise NoPyBaz

from fai import iter_new_merges, direct_merges

import tempfile

import os

import os.path

import shutil

import bzrlib

import bzrlib.trace

import bzrlib.merge

import bzrlib.inventory

import bzrlib.osutils

import sys

import email.Utils

from progress import *

BAZ_IMPORT_ROOT = 'TREE_ROOT'

class ImportCommitReporter(NullCommitReporter):

    def escaped(self, escape_count, message):

        bzrlib.trace.warning("replaced %d control characters in message" %

                             escape_count)

def add_id(files, id=None):

    """Adds an explicit id to a list of files.

    :param files: the name of the file to add an id to

    :type files: list of str

    :param id: tag one file using the specified id, instead of generating id

    :type id: str

    """

    args = ["add-id"]

    if id is not None:

        args.extend(["--id", id])

    args.extend(files)

    return null_cmd(args)

saved_dir = None

def make_archive(name, location):

    pb_location = pybaz.ArchiveLocation(location)

    pb_location.create_master(pybaz.Archive(name),

                              pybaz.ArchiveLocationParams())

def test_environ():

    """

    >>> q = test_environ()

    >>> os.path.exists(q)

    True

    >>> os.path.exists(os.path.join(q, "home", ".arch-params"))

    True

    >>> teardown_environ(q)

    >>> os.path.exists(q)

    False

    """

    global saved_dir

    saved_dir = os.getcwdu()

    tdir = tempfile.mkdtemp(prefix="testdir-")

    os.environ["HOME"] = os.path.join(tdir, "home")

    os.mkdir(os.environ["HOME"])

    arch_dir = os.path.join(tdir, "archive_dir")

    make_archive("test@example.com", arch_dir)

    work_dir = os.path.join(tdir, "work_dir")

    os.mkdir(work_dir)

    os.chdir(work_dir)

    pybaz.init_tree(work_dir, "test@example.com/test--test--0")

    lib_dir = os.path.join(tdir, "lib_dir")

    os.mkdir(lib_dir)

    pybaz.register_revision_library(lib_dir)

    pybaz.set_my_id("Test User<test@example.org>")

    return tdir

def add_file(path, text, id):

    """

    >>> q = test_environ()

    >>> add_file("path with space", "text", "lalala")

    >>> tree = pybaz.tree_root(".")

    >>> inv = list(tree.iter_inventory_ids(source=True, both=True))

    >>> ("x_lalala", "path with space") in inv

    True

    >>> teardown_environ(q)

    """

    file(path, "wb").write(text)

    add_id([path], id)

def add_dir(path, id):

    """

    >>> q = test_environ()

    >>> add_dir("path with\(sp) space", "lalala")

    >>> tree = pybaz.tree_root(".")

    >>> inv = list(tree.iter_inventory_ids(source=True, both=True))

    >>> ("x_lalala", "path with\(sp) space") in inv

    True

    >>> teardown_environ(q)

    """

    os.mkdir(path)

    add_id([path], id)

def teardown_environ(tdir):

    os.chdir(saved_dir)

    shutil.rmtree(tdir)

def timport(tree, summary):

    msg = tree.log_message()

    msg["summary"] = summary

    tree.import_(msg)

def commit(tree, summary):

    """

    >>> q = test_environ()

    >>> tree = pybaz.tree_root(".")

    >>> timport(tree, "import")

    >>> commit(tree, "commit")

    >>> logs = [str(l.revision) for l in tree.iter_logs()]

    >>> len(logs)

    2

    >>> logs[0]

    'test@example.com/test--test--0--base-0'

    >>> logs[1]

    'test@example.com/test--test--0--patch-1'

    >>> teardown_environ(q)

    """

    msg = tree.log_message()

    msg["summary"] = summary

    tree.commit(msg)

def commit_test_revisions():

    """

    >>> q = test_environ()

    >>> commit_test_revisions()

    >>> a = pybaz.Archive("test@example.com")

    >>> revisions = list(a.iter_revisions("test--test--0"))

    >>> len(revisions)

    3

    >>> str(revisions[2])

    'test@example.com/test--test--0--base-0'

    >>> str(revisions[1])

    'test@example.com/test--test--0--patch-1'

    >>> str(revisions[0])

    'test@example.com/test--test--0--patch-2'

    >>> teardown_environ(q)

    """

    tree = pybaz.tree_root(".")

    add_file("mainfile", "void main(void){}", "mainfile by aaron")

    timport(tree, "Created mainfile")

    file("mainfile", "wb").write("or something like that")

    commit(tree, "altered mainfile")

    add_file("ofile", "this is another file", "ofile by aaron")

    commit(tree, "altered mainfile")

def commit_more_test_revisions():

    """

    >>> q = test_environ()

    >>> commit_test_revisions()

    >>> commit_more_test_revisions()

    >>> a = pybaz.Archive("test@example.com")

    >>> revisions = list(a.iter_revisions("test--test--0"))

    >>> len(revisions)

    4

    >>> str(revisions[0])

    'test@example.com/test--test--0--patch-3'

    >>> teardown_environ(q)

    """

    tree = pybaz.tree_root(".")

    add_file("trainfile", "void train(void){}", "trainfile by aaron")

    commit(tree, "altered trainfile")

class NoSuchVersion(Exception):

    def __init__(self, version):

        Exception.__init__(self, "The version %s does not exist." % version)

        self.version = version

def version_ancestry(version):

    """

    >>> q = test_environ()

    >>> commit_test_revisions()

    >>> version = pybaz.Version("test@example.com/test--test--0")

    >>> ancestors = version_ancestry(version)

    >>> str(ancestors[0])

    'test@example.com/test--test--0--base-0'

    >>> str(ancestors[1])

    'test@example.com/test--test--0--patch-1'

    >>> version = pybaz.Version("test@example.com/test--test--0.5")

    >>> ancestors = version_ancestry(version)

    Traceback (most recent call last):

    NoSuchVersion: The version test@example.com/test--test--0.5 does not exist.

    >>> teardown_environ(q)

    """

    try:

        revision = version.iter_revisions(reverse=True).next()

    except StopIteration:

        return ()

    except:

        print version

        if not version.exists():

            raise NoSuchVersion(version)

        else:

            raise

    ancestors = list(revision.iter_ancestors(metoo=True))

    ancestors.reverse()

    return ancestors

def get_last_revision(branch):

    last_patch = branch.last_revision()

    try:

        return arch_revision(last_patch)

    except NotArchRevision:

        raise UserError(

            "Directory \"%s\" already exists, and the last revision is not"

            " an Arch revision (%s)" % (branch.base, last_patch))

def do_branch(br_from, to_location, revision_id):

    """Derived from branch in builtins."""

    br_from.lock_read()

    try:

        try:

            os.mkdir(to_location)

        except OSError, e:

            if e.errno == errno.EEXIST:

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

                                      ' exists.' % to_location)

            if e.errno == errno.ENOENT:

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

                                      to_location)

            else:

                raise

        try:

            br_from.bzrdir.clone(to_location, revision_id)

        except NoSuchRevision:

            rmtree(to_location)

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

                                                         revision_id)

            raise UserError(msg)

    finally:

        br_from.unlock()

def get_remaining_revisions(output_dir, version, encoding,

                            reuse_history_from=[]):

    last_patch = None

    old_revno = None

    output_exists = os.path.exists(output_dir)

    if output_exists:

        # We are starting from an existing directory, figure out what

        # the current version is

        branch = Branch.open(output_dir)

        last_patch, last_encoding = get_last_revision(branch)

        assert encoding == last_encoding

        if last_patch is None:

            if branch.last_revision() != None:

                raise NotPreviousImport(branch.base)

        elif version is None:

            version = last_patch.version

    elif version is None:

        raise UserError("No version specified, and directory does not exist.")

    try:

        ancestors = version_ancestry(version)

        if not output_exists and reuse_history_from != []:

            for ancestor in reversed(ancestors):

                if last_patch is not None:

                    # found something to copy

                    break

                # try to grab a copy of ancestor

                # note that is not optimised: we could look for namespace

                # transitions and only look for the past after the

                # transition.

                for history_root in reuse_history_from:

                    possible_source = os.path.join(history_root,

                        map_namespace(ancestor.version))

                    try:

                        source = Branch.open(possible_source)

                        rev_id = revision_id(ancestor, encoding)

                        if rev_id in source.revision_history():

                            do_branch(source, output_dir, rev_id)

                            last_patch = ancestor

                            break

                    except NotBranchError:

                        pass

    except NoSuchVersion, e:

        raise UserError(str(e))

    if last_patch:

        for i in range(len(ancestors)):

            if ancestors[i] == last_patch:

                break

        else:

            raise UserError("Directory \"%s\" already exists, and the last "

                "revision (%s) is not in the ancestry of %s" %

                (output_dir, last_patch, version))

        # Strip off all of the ancestors which are already present

        # And get a directory starting with the latest ancestor

        latest_ancestor = ancestors[i]

        old_revno = Branch.open(output_dir).revno()

        ancestors = ancestors[i+1:]

    return ancestors, old_revno

###class Importer(object):

###    """An importer.

###

###    Currently this is used as a parameter object, though more behaviour is

###    possible later.

###    """

###

###    def __init__(self, output_dir, version, fast=False,

###                 verbose=False, dry_run=False, max_count=None,

###                   reuse_history_from=[]):

###        self.output_dir = output_dir

###        self.version = version

###        self.

def import_version(output_dir, version, encoding, fast=False,

                   verbose=False, dry_run=False, max_count=None,

                   reuse_history_from=[], standalone=True):

    """

    >>> q = test_environ()

    Progress bars output to stderr, but doctest does not capture that.

    >>> old_stderr = sys.stderr

    >>> sys.stderr = sys.stdout

    >>> result_path = os.path.join(q, "result")

    >>> commit_test_revisions()

    >>> version = pybaz.Version("test@example.com/test--test--0.1")

    >>> old_ui = bzrlib.ui.ui_factory

    >>> bzrlib.ui.ui_factory = bzrlib.ui.text.TextUIFactory(

    ...     bar_type=bzrlib.progress.DotsProgressBar)

    >>> import_version('/', version, None, dry_run=True)

    Traceback (most recent call last):

    NotPreviousImport: / is not the location of a previous import.

    >>> import_version(result_path, version, None, dry_run=True)

    Traceback (most recent call last):

    UserError: The version test@example.com/test--test--0.1 does not exist.

    >>> version = pybaz.Version("test@example.com/test--test--0")

    >>> import_version(result_path, version, None, dry_run=True) #doctest: +ELLIPSIS

    importing test@example.com/test--test--0 into ...

    ...

    revisions: ..........................................

    Dry run, not modifying output_dir

    Cleaning up

    >>> import_version(result_path, version, None) #doctest: +ELLIPSIS

    importing test@example.com/test--test--0 into ...

    ...

    revisions: .....................................................................

    Cleaning up

    Import complete.

    >>> import_version(result_path, version, None) #doctest: +ELLIPSIS

    Tree is up-to-date with test@example.com/test--test--0--patch-2

    >>> commit_more_test_revisions()

    >>> import_version(result_path, version, None) #doctest: +ELLIPSIS

    importing test@example.com/test--test--0 into ...

    revisions: ....................................................

    Cleaning up

    Import complete.

    >>> bzrlib.ui.ui_factory = old_ui

    >>> sys.stderr = old_stderr

    >>> teardown_environ(q)

    """

    progress_bar = bzrlib.ui.ui_factory.nested_progress_bar()

    try:

        try:

            ancestors, old_revno = get_remaining_revisions(output_dir, version,

                                                           encoding,

                                                           reuse_history_from)

        except NotBranchError, e:

            raise NotPreviousImport(e.path)

        if old_revno is None and len(ancestors) == 0:

            progress_bar.note('Version %s has no revisions.' % version)

            return

        if len(ancestors) == 0:

            last_revision, last_encoding = \

                get_last_revision(Branch.open(output_dir))

            progress_bar.note('Tree is up-to-date with %s' % last_revision)

            return

        progress_bar.note("importing %s into %s" % (version, output_dir))

        tempdir = tempfile.mkdtemp(prefix="baz2bzr-",

                                   dir=os.path.dirname(output_dir))

        try:

            wt = WorkingTree.open(output_dir)

        except (NotBranchError, NoWorkingTree):

            wt = None

        try:

            for result in iter_import_version(output_dir, ancestors, tempdir,

                    pb=progress_bar, encoding=encoding, fast=fast,

                    verbose=verbose, dry_run=dry_run, max_count=max_count,

                    standalone=standalone):

                show_progress(progress_bar, result)

            if dry_run:

                progress_bar.note('Dry run, not modifying output_dir')

                return

            # Update the working tree of the branch

            try:

                wt = WorkingTree.open(output_dir)

            except NoWorkingTree:

                wt = None

            if wt is not None:

                wt.set_last_revision(wt.branch.last_revision())

                wt.set_root_id(BAZ_IMPORT_ROOT)

                wt.revert([])

        finally:

            progress_bar.note('Cleaning up')

            shutil.rmtree(tempdir)

        progress_bar.note("Import complete.")

    finally:

        progress_bar.finished()

class UserError(BzrCommandError):

    def __init__(self, message):

        """Exception to throw when a user makes an impossible request

        :param message: The message to emit when printing this exception

        :type message: string

        """

        BzrCommandError.__init__(self, message)

class NotPreviousImport(UserError):

    def __init__(self, path):

        UserError.__init__(self, "%s is not the location of a previous import."

                           % path)

def revision_id(arch_revision, encoding):

    """

    Generate a Bzr revision id from an Arch revision id.  'x' in the id

    designates a revision imported with an experimental algorithm.  A number

    would indicate a particular standardized version.

    :param arch_revision: The Arch revision to generate an ID for.

    >>> revision_id(pybaz.Revision("you@example.com/cat--br--0--base-0"), None)

    'Arch-1:you@example.com%cat--br--0--base-0'

    >>> revision_id(pybaz.Revision("you@example.com/cat--br--0--base-0"), 'utf-8')

    'Arch-1-utf-8:you@example.com%cat--br--0--base-0'

    """

    if encoding is None:

        encoding = ''

    else:

        encoding = '-' + encoding

    return "Arch-1%s:%s" % (encoding, str(arch_revision).replace('/', '%'))

class NotArchRevision(Exception):

    def __init__(self, revision_id):

        msg = "The revision id %s does not look like it came from Arch."\

            % revision_id

        Exception.__init__(self, msg)

def arch_revision(revision_id):

    """

    >>> str(arch_revision("Arch-1:jrandom@example.com%test--test--0"))

    Traceback (most recent call last):

    NotArchRevision: The revision id Arch-1:jrandom@example.com%test--test--0 does not look like it came from Arch.

    >>> str(arch_revision("Arch-1:jrandom@example.com%test--test--0--base-5"))

    Traceback (most recent call last):

    NotArchRevision: The revision id Arch-1:jrandom@example.com%test--test--0--base-5 does not look like it came from Arch.

    >>> str(arch_revision("Arch-1:jrandom@example.com%test--test--0--patch-5")[0])

    'jrandom@example.com/test--test--0--patch-5'

    >>> str(arch_revision("Arch-1:jrandom@example.com%test--test--0--patch-5")[0])

    'jrandom@example.com/test--test--0--patch-5'

    >>> str(arch_revision("Arch-1:jrandom@example.com%test--test--0--patch-5")[1])

    'None'

    >>> str(arch_revision("Arch-1-utf-8:jrandom@example.com%test--test--0--patch-5")[1])

    'utf-8'

    """

    if revision_id is None:

        return None, None

    if revision_id[:7] not in ('Arch-1:', 'Arch-1-'):

        raise NotArchRevision(revision_id)

    else:

        try:

            encoding, arch_name = revision_id[6:].split(':', 1)

            arch_name = arch_name.replace('%', '/')

            if encoding == '':

                encoding = None

            else:

                encoding = encoding[1:]

            return pybaz.Revision(arch_name), encoding

        except pybaz.errors.NamespaceError, e:

            raise NotArchRevision(revision_id)

def create_shared_repository(output_dir):

    bd = bzrdir.BzrDirMetaFormat1().initialize(output_dir)

    bd.create_repository(shared=True)

def create_branch(output_dir):

    os.mkdir(output_dir)

    bd = bzrdir.BzrDirMetaFormat1().initialize(output_dir)

    return bd.create_branch()

def create_checkout(source, to_location, revision_id=None):

    checkout = bzrdir.BzrDirMetaFormat1().initialize(to_location)

    bzrlib.branch.BranchReferenceFormat().initialize(checkout, source)

    return checkout.create_workingtree(revision_id)

def create_checkout_metadata(source, to_location, revision_id=None):

    if revision_id is None:

        revision_id = source.last_revision()

    wt = create_checkout(source, to_location, NULL_REVISION)

    wt.lock_write()

    try:

        wt.set_last_revision(revision_id)

        wt.flush()

        if revision_id not in (NULL_REVISION, None):

            basis = wt.basis_tree()

            basis.lock_read()

            try:

                wt._write_inventory(basis.inventory)

            finally:

                basis.unlock()

    finally:

        wt.unlock()

    return wt

def iter_import_version(output_dir, ancestors, tempdir, pb, encoding,

                        fast=False, verbose=False, dry_run=False,

                        max_count=None, standalone=False):

    revdir = None

    log_encoding = 'ascii'

    if encoding is not None:

        log_encoding = encoding

    # Uncomment this for testing, it basically just has baz2bzr only update

    # 5 patches at a time

    if max_count:

        ancestors = ancestors[:max_count]

    # Not sure if I want this output. basically it tells you ahead of time

    # what it is going to do, but then later it tells you as it is doing it.

    # what probably would be best would be to collapse it into ranges, so that

    # this gives the simple view, and then later it gives the blow by blow.

    #if verbose:

    #    print 'Adding the following revisions:'

    #    for a in ancestors:

    #        print '\t%s' % a

    previous_version=None

    missing_ancestor = None

    if dry_run:

        dry_output_dir = os.path.join(tempdir, 'od')

        if os.path.exists(output_dir):

            shutil.copytree(output_dir, dry_output_dir)

        output_dir = dry_output_dir

    if os.path.exists(output_dir):

        target_branch = Branch.open(output_dir)

    else:

        if standalone:

            wt = BzrDir.create_standalone_workingtree(output_dir)

            target_branch = wt.branch

        else:

            target_branch = create_branch(output_dir)

    for i in range(len(ancestors)):

        revision = ancestors[i]

        rev_id = revision_id(revision, encoding)

        direct_merges = []

        if verbose:

            version = str(revision.version)

            if version != previous_version:

                pb.note('On version: %s' % version)

            yield Progress(str(revision.patchlevel), i, len(ancestors))

            previous_version = version

        else:

            yield Progress("revisions", i, len(ancestors))

        if target_branch.repository.has_revision(rev_id):

            target_branch.append_revision(rev_id)

            continue

        if revdir is None:

            revdir = os.path.join(tempdir, "rd")

            try:

                tree, baz_inv, log = get_revision(revdir, revision)

            except pybaz.errors.ExecProblem, e:

                if ("%s" % e.args).find('could not connect') == -1:

                    raise

                missing_ancestor = revision

                revdir = None

                pb.note("unable to access ancestor %s, making into a merge."

                       % missing_ancestor)

                continue

            target_tree = create_checkout_metadata(target_branch, revdir)

            branch = target_tree.branch

        else:

            old = os.path.join(revdir, ".bzr")

            new = os.path.join(tempdir, ".bzr")

            os.rename(old, new)

            baz_inv, log = apply_revision(tree, revision)

            os.rename(new, old)

            target_tree = WorkingTree.open(revdir)

            branch = target_tree.branch

        # cached so we can delete the log

        log_date = log.date

        log_summary = log.summary

        log_description = log.description

        is_continuation = log.continuation_of is not None

        log_creator = log.creator

        direct_merges = get_direct_merges(revdir, revision, log)

        timestamp = email.Utils.mktime_tz(log_date + (0,))

        if log_summary is None:

            log_summary = ""

        # log_descriptions of None and "" are ignored.

        if not is_continuation and log_description:

            log_message = "\n".join((log_summary, log_description))

        else:

            log_message = log_summary

        target_tree.lock_write()

        branch.lock_write()

        try:

            if missing_ancestor:

                # if we want it to be in revision-history, do that here.

                target_tree.set_parent_ids(

                    [revision_id(missing_ancestor, encoding)],

                    allow_leftmost_as_ghost=True)

                missing_ancestor = None

            for merged_rev in direct_merges:

                target_tree.add_pending_merge(revision_id(merged_rev,

                                                          encoding))

            target_tree.set_root_id(BAZ_IMPORT_ROOT)

            target_tree.flush()

            target_tree.set_inventory(baz_inv)

            commitobj = Commit(reporter=ImportCommitReporter())

            commitobj.commit(working_tree=target_tree,

                message=log_message.decode(log_encoding, 'replace'),

                verbose=False, committer=log_creator, timestamp=timestamp,

                timezone=0, rev_id=rev_id, revprops={})

        finally:

            target_tree.unlock()

            branch.unlock()

    yield Progress("revisions", len(ancestors), len(ancestors))

def get_direct_merges(revdir, revision, log):

    continuation = log.continuation_of

    previous_version = revision.version

    if pybaz.WorkingTree(revdir).tree_version != previous_version:

        pybaz.WorkingTree(revdir).set_tree_version(previous_version)

    log_path = "%s/{arch}/%s/%s/%s/%s/patch-log/%s" % (revdir,

        revision.category.nonarch, revision.branch.nonarch,

        revision.version.nonarch, revision.archive, revision.patchlevel)

    temp_path = tempfile.mktemp(dir=os.path.dirname(revdir))