~abentley/bzrtools/bzrtools.dev

#!/usr/bin/env python

# Copyright (C) 2005 Aaron Bentley

# <aaron.bentley@utoronto.ca>

#

#    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

try:

    import pybaz

except ImportError:

    print "This command requires PyBaz.  Please ensure that it is installed."

    import sys

    sys.exit(1)

from pybaz.backends.baz import null_cmd

import tempfile

import os

import os.path

import shutil

import bzrlib

import bzrlib.trace

import sys

import email.Utils

from progress import *

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)

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

    """

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

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

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

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

    pybaz.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("/")

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

    >>> teardown_environ(q)

    """

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

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

    ancestors.reverse()

    return ancestors

def import_version(output_dir, version, fancy=True):

    """

    >>> q = test_environ()

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

    >>> commit_test_revisions()

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

    >>> import_version(result_path, version, fancy=False)

    not fancy

    ....

    Import complete.

    >>> teardown_environ(q)

    """

    progress_bar = ProgressBar()

    tempdir = tempfile.mkdtemp(prefix="baz2bzr")

    try:

        if not fancy:

            print "not fancy"

        for result in iter_import_version(output_dir, version, tempdir):

            if fancy:

                progress_bar(result)

            else:

                sys.stdout.write('.')

    finally:

        if fancy:

            clear_progress_bar()

        else:

            sys.stdout.write('\n')

        shutil.rmtree(tempdir)

    print "Import complete."

class UserError(Exception):

    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

        """

        Exception.__init__(self, message)

def revision_id(arch_revision):

    """

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

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

    """

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

def iter_import_version(output_dir, version, tempdir):

    revdir = None

    ancestors = version_ancestry(version)

    for i in range(len(ancestors)):

        revision = ancestors[i]

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

        if revdir is None:

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

            baz_inv, log = get_revision(revdir, revision)

            branch = bzrlib.Branch(revdir, init=True)

        else:

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

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

            os.rename(old, new)

            baz_inv, log = apply_revision(revdir, revision)

            os.rename(new, old)

            branch = bzrlib.Branch(revdir)

        branch.set_inventory(baz_inv)

        timestamp = email.Utils.mktime_tz(log.date + (0,))

        rev_id = revision_id(revision)

        bzrlib.trace.silent = True

        try:

            branch.commit(log.summary, verbose=False, committer=log.creator,

                          timestamp=timestamp, timezone=0, rev_id=rev_id)

        finally:

            bzrlib.trace.silent = False   

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

    unlink_unversioned(branch, revdir)

    os.rename(revdir, output_dir)   

def unlink_unversioned(branch, revdir):

    for unversioned in branch.working_tree().extras():

        path = os.path.join(revdir, unversioned)

        if os.path.isdir(path):

            shutil.rmtree(path)

        else:

            os.unlink(path)

def get_log(tree, revision):

    log = tree.iter_logs(version=revision.version, reverse=True).next()

    assert log.revision == revision

    return log

def get_revision(revdir, revision):

    revision.get(revdir)

    tree = pybaz.tree_root(revdir)

    log = get_log(tree, revision)

    try:

        return bzr_inventory_data(tree), log 

    except BadFileKind, e:

        raise UserError("Cannot convert %s because %s is a %s" % (revision,e.path, e.kind) )

def apply_revision(revdir, revision):

    tree = pybaz.tree_root(revdir)

    revision.apply(tree)

    log = get_log(tree, revision)

    try:

        return bzr_inventory_data(tree), log

    except BadFileKind, e:

        raise UserError("Cannot convert %s because %s is a %s" % (revision,e.path, e.kind) )

class BadFileKind(Exception):

    """The file kind is not permitted in bzr inventories"""

    def __init__(self, tree_root, path, kind):

        self.tree_root = tree_root

        self.path = path

        self.kind = kind

        Exception.__init__(self, "File %s is of forbidden type %s" %

                           (os.path.join(tree_root, path), kind))

def bzr_inventory_data(tree):

    inv_iter = tree.iter_inventory_ids(source=True, both=True)

    inv_map = {}

    for file_id, path in inv_iter:

        inv_map[path] = file_id 

    bzr_inv = []

    for path, file_id in inv_map.iteritems():

        full_path = os.path.join(tree, path)

        kind = bzrlib.osutils.file_kind(full_path)

        if kind not in ("file", "directory"):

            raise BadFileKind(tree, path, kind)

        parent_dir = os.path.dirname(path)

        if parent_dir != "":

            parent_id = inv_map[parent_dir]

        else:

            parent_id = bzrlib.inventory.ROOT_ID

        bzr_inv.append((path, file_id, parent_id, kind))

    bzr_inv.sort()

    return bzr_inv

if len(sys.argv) == 2 and sys.argv[1] == "test":

    print "Running tests"

    import doctest

    doctest.testmod()

elif len(sys.argv) == 3:

    try:

        output_dir = sys.argv[2]

        if os.path.exists(output_dir):

            raise UserError("Directory \"%s\" already exists" % output_dir)

        import_version(output_dir, pybaz.Version(sys.argv[1]))

    except UserError, e:

        print e

    except KeyboardInterrupt:

        print "Aborted."

else:

    print "usage: %s VERSION OUTDIR" % os.path.basename(sys.argv[0])