/brz/remove-bazaar

# Copyright (C) 2006 Canonical Ltd
# Authors: Robert Collins <robert.collins@canonical.com>
#
# 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


"""A GIT branch and repository format implementation for bzr."""


from StringIO import StringIO

import stgit
import stgit.git as git

from bzrlib import config, iterablefile, graph, osutils, urlutils
from bzrlib.decorators import *
import bzrlib.branch
import bzrlib.bzrdir
import bzrlib.errors as errors
import bzrlib.repository
from bzrlib.revision import Revision


class GitBranchConfig(config.BranchConfig):
    """BranchConfig that uses locations.conf in place of branch.conf""" 

    def __init__(self, branch):
        config.BranchConfig.__init__(self, branch)
        # do not provide a BranchDataConfig
        self.option_sources = self.option_sources[0], self.option_sources[2]

    def set_user_option(self, name, value, local=False):
        """Force local to True"""
        config.BranchConfig.set_user_option(self, name, value, local=True)


def gitrevid_from_bzr(revision_id):
    if revision_id is None:
        return None
    return revision_id[4:]


def bzrrevid_from_git(revision_id):
    return "git:" + revision_id


class GitLock(object):
    """A lock that thunks through to Git."""

    def lock_write(self):
        pass

    def lock_read(self):
        pass

    def unlock(self):
        pass


class GitLockableFiles(bzrlib.lockable_files.LockableFiles):
    """Git specific lockable files abstraction."""

    def __init__(self, lock):
        self._lock = lock
        self._transaction = None
        self._lock_mode = None
        self._lock_count = 0


class GitDir(bzrlib.bzrdir.BzrDir):
    """An adapter to the '.git' dir used by git."""

    def __init__(self, transport, lockfiles, format):
        self._format = format
        self.root_transport = transport
        self.transport = transport.clone('.git')
        self._lockfiles = lockfiles

    def get_branch_transport(self, branch_format):
        if branch_format is None:
            return self.transport
        if isinstance(branch_format, GitBzrDirFormat):
            return self.transport
        raise errors.IncompatibleFormat(branch_format, self._format)

    get_repository_transport = get_branch_transport
    get_workingtree_transport = get_branch_transport

    def is_supported(self):
        return True

    def open_branch(self, ignored=None):
        """'crate' a branch for this dir."""
        return GitBranch(self, self._lockfiles)

    def open_repository(self, shared=False):
        """'open' a repository for this dir."""
        return GitRepository(self._gitrepo, self, self._lockfiles)

    def open_workingtree(self):
        loc = urlutils.unescape_for_display(self.root_transport.base, 'ascii')
        raise errors.NoWorkingTree(loc)


class GitBzrDirFormat(bzrlib.bzrdir.BzrDirFormat):
    """The .git directory control format."""

    @classmethod
    def _known_formats(self):
        return set([GitBzrDirFormat()])

    def open(self, transport, _create=False, _found=None):
        """Open this directory.
        
        :param _create: create the git dir on the fly. private to GitDirFormat.
        """
        # we dont grok readonly - git isn't integrated with transport.
        url = transport.base
        if url.startswith('readonly+'):
            url = url[len('readonly+'):]
        if url.startswith('file://'):
            url = url[len('file://'):]
        url = url.encode('utf8')
        if not transport.has('.git'):
            raise errors.NotBranchError(path=transport.base)
        lockfiles = GitLockableFiles(GitLock())
        return GitDir(transport, lockfiles, self)

    @classmethod
    def probe_transport(klass, transport):
        """Our format is present if the transport ends in '.not/'."""
        # little ugly, but works
        format = klass() 
        # try a manual probe first, its a little faster perhaps ?
        if transport.has('.git'):
            return format
        # delegate to the main opening code. This pays a double rtt cost at the
        # moment, so perhaps we want probe_transport to return the opened thing
        # rather than an openener ? or we could return a curried thing with the
        # dir to open already instantiated ? Needs more thought.
        try:
            format.open(transport)
            return format
        except Exception, e:
            raise errors.NotBranchError(path=transport.base)
        raise errors.NotBranchError(path=transport.base)


bzrlib.bzrdir.BzrDirFormat.register_control_format(GitBzrDirFormat)


class GitBranch(bzrlib.branch.Branch):
    """An adapter to git repositories for bzr Branch objects."""

    def __init__(self, gitdir, lockfiles):
        self.bzrdir = gitdir
        self.control_files = lockfiles
        self.repository = GitRepository(gitdir, lockfiles)
        self.base = gitdir.root_transport.base
        if '.git' not in gitdir.root_transport.list_dir('.'):
            raise errors.NotBranchError(self.base)

    def lock_write(self):
        self.control_files.lock_write()

    @needs_read_lock
    def last_revision(self):
        # perhaps should escape this ?
        return bzrrevid_from_git(self.repository.git.get_head())

    @needs_read_lock
    def revision_history(self):
        node = self.last_revision()
        ancestors = self.repository.get_revision_graph(node)
        history = []
        while node is not None:
            history.append(node)
            if len(ancestors[node]) > 0:
                node = ancestors[node][0]
            else:
                node = None
        return list(reversed(history))

    def get_config(self):
        return GitBranchConfig(self)

    def lock_read(self):
        self.control_files.lock_read()

    def unlock(self):
        self.control_files.unlock()

    def get_push_location(self):
        """See Branch.get_push_location."""
        push_loc = self.get_config().get_user_option('push_location')
        return push_loc

    def set_push_location(self, location):
        """See Branch.set_push_location."""
        self.get_config().set_user_option('push_location', location, 
                                          local=True)


class GitRepository(bzrlib.repository.Repository):
    """An adapter to git repositories for bzr."""

    def __init__(self, gitdir, lockfiles):
        self.bzrdir = gitdir
        self.control_files = lockfiles
        gitdirectory = urlutils.local_path_from_url(gitdir.transport.base)
        self.git = GitModel(gitdirectory)
        self._revision_cache = {}

    def _ancestor_revisions(self, revision_ids):
        if revision_ids is not None:
            git_revisions = [gitrevid_from_bzr(r) for r in revision_ids]
        else:
            git_revisions = None
        for lines in self.git.ancestor_lines(git_revisions):
            yield self.parse_rev(lines)

    def is_shared(self):
        return True

    def get_revision_graph(self, revision_id=None):
        if revision_id is None:
            revisions = None
        else:
            revisions = [revision_id]
        return self.get_revision_graph_with_ghosts(revisions).get_ancestors()

    def get_revision_graph_with_ghosts(self, revision_ids=None):
        result = graph.Graph()
        for revision in self._ancestor_revisions(revision_ids):
            result.add_node(revision.revision_id, revision.parent_ids)
            self._revision_cache[revision.revision_id] = revision
        return result

    def get_revision(self, revision_id):
        if revision_id in self._revision_cache:
            return self._revision_cache[revision_id]
        raw = self.git.rev_list([gitrevid_from_bzr(revision_id)], max_count=1,
                                header=True)
        return self.parse_rev(raw)

    def has_revision(self, revision_id):
        try:
            self.get_revision(revision_id)
        except NoSuchRevision:
            return False
        else:
            return True

    def get_revisions(self, revisions):
        return [self.get_revision(r) for r in revisions]

    def parse_rev(self, raw):
        # first field is the rev itself.
        # then its 'field value'
        # until the EOF??
        parents = []
        log = []
        in_log = False
        committer = None
        revision_id = bzrrevid_from_git(raw[0][:-1])
        for field in raw[1:]:
            #if field.startswith('author '):
            #    committer = field[7:]
            if field.startswith('parent '):
                parents.append(bzrrevid_from_git(field.split()[1]))
            elif field.startswith('committer '):
                commit_fields = field.split()
                if committer is None:
                    committer = ' '.join(commit_fields[1:-3])
                timestamp = commit_fields[-2]
                timezone = commit_fields[-1]
            elif field.startswith('tree '):
                tree_id = field.split()[1]
            elif in_log:
                log.append(field[4:])
            elif field == '\n':
                in_log = True

        log = ''.join(log)
        result = Revision(revision_id)
        result.parent_ids = parents
        result.message = log
        result.inventory_sha1 = ""
        result.timezone = timezone and int(timezone)
        result.timestamp = float(timestamp)
        result.committer = committer 
        result.properties['git-tree-id'] = tree_id
        return result

    def revision_tree(self, revision_id):
        return GitRevisionTree(self, revision_id)

    def get_inventory(self, revision_id):
        revision = self.get_revision(revision_id)
        inventory = GitInventory(revision_id)
        tree_id = revision.properties['git-tree-id']
        type_map = {'blob': 'file', 'tree': 'directory' }
        def get_inventory(tree_id, prefix):
            for perms, type, obj_id, name in self.git.get_inventory(tree_id):
                full_path = prefix + name
                if type == 'blob':
                    text_sha1 = obj_id
                else:
                    text_sha1 = None
                executable = (perms[-3] in ('1', '3', '5', '7'))
                entry = GitEntry(full_path, type_map[type], revision_id,
                                 text_sha1, executable)
                inventory.entries[full_path] = entry
                if type == 'tree':
                    get_inventory(obj_id, full_path+'/')
        get_inventory(tree_id, '')
        return inventory


class GitRevisionTree(object):

    def __init__(self, repository, revision_id):
        self.repository = repository
        self.revision_id = revision_id
        self.inventory = repository.get_inventory(revision_id)

    def get_file(self, file_id):
        obj_id = self.inventory[file_id].text_sha1
        lines = self.repository.git.cat_file('blob', obj_id)
        return iterablefile.IterableFile(lines)

    def is_executable(self, file_id):
        return self.inventory[file_id].executable


class GitInventory(object):

    def __init__(self, revision_id):
        self.entries = {}
        self.root = GitEntry('', 'directory', revision_id)
        self.entries[''] = self.root

    def __getitem__(self, key):
        return self.entries[key]

    def iter_entries(self):
        return iter(sorted(self.entries.items()))

    def iter_entries_by_dir(self):
        return self.iter_entries()

    def __len__(self):
        return len(self.entries)


class GitEntry(object):

    def __init__(self, path, kind, revision, text_sha1=None, executable=False,
                 text_size=None):
        self.path = path
        self.file_id = path
        self.kind = kind
        self.executable = executable
        self.name = osutils.basename(path)
        if path == '':
            self.parent_id = None
        else:
            self.parent_id = osutils.dirname(path)
        self.revision = revision
        self.symlink_target = None
        self.text_sha1 = text_sha1
        self.text_size = None

    def __repr__(self):
        return "GitEntry(%r, %r, %r, %r)" % (self.path, self.kind, 
                                             self.revision, self.parent_id)


class GitModel(object):
    """API that follows GIT model closely"""

    def __init__(self, git_dir):
        self.git_dir = git_dir

    def git_command(self, command, args):
        args = ' '.join("'%s'" % arg for arg in args)
        return 'git --git-dir=%s %s %s' % (self.git_dir, command, args) 

    def git_lines(self, command, args):
        return stgit.git._output_lines(self.git_command(command, args))

    def git_line(self, command, args):
        return stgit.git._output_one_line(self.git_command(command, args))

    def cat_file(self, type, object_id, pretty=False):
        args = []
        if pretty:
            args.append('-p')
        else:
            args.append(type)
        args.append(object_id)
        return self.git_lines('cat-file', args)

    def rev_list(self, heads, max_count=None, header=False):
        args = []
        if max_count is not None:
            args.append('--max-count=%d' % max_count)
        if header is not False:
            args.append('--header')
        if heads is None:
            args.append('--all')
        else:
            args.extend(heads)
        return self.git_lines('rev-list', args)

    def rev_parse(self, git_id):
        args = ['--verify', git_id]
        return self.git_line('rev-parse', args)

    def get_head(self):
        return self.rev_parse('HEAD')

    def ancestor_lines(self, revisions):
        revision_lines = []
        for line in self.rev_list(revisions, header=True):
            if line.startswith('\x00'):
                yield revision_lines
                revision_lines = [line[1:].decode('latin-1')]
            else:
                revision_lines.append(line.decode('latin-1'))
        assert revision_lines == ['']

    def get_inventory(self, tree_id):
        for line in self.cat_file('tree', tree_id, True):
            sections = line.split(' ', 2)
            obj_id, name = sections[2].split('\t', 1)
            name = name.rstrip('\n')
            if name.startswith('"'):
                name = name[1:-1].decode('string_escape').decode('utf-8')
            yield (sections[0], sections[1], obj_id, name)

def test_suite():
    from unittest import TestSuite, TestLoader
    import tests

    suite = TestSuite()

    suite.addTest(tests.test_suite())

    return suite