Don't assume there is always a stage 2 in git.get_conflicts

[stgit] / stgit / git.py

"""Python GIT interface
"""

__copyright__ = """
Copyright (C) 2005, Catalin Marinas <catalin.marinas@gmail.com>

This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License version 2 as
published by the Free Software Foundation.

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, re, gitmergeonefile
from shutil import copyfile

from stgit.exception import *
from stgit import basedir
from stgit.utils import *
from stgit.out import *
from stgit.run import *
from stgit.config import config

# git exception class
class GitException(StgException):
    pass

# When a subprocess has a problem, we want the exception to be a
# subclass of GitException.
class GitRunException(GitException):
    pass
class GRun(Run):
    exc = GitRunException
    def __init__(self, *cmd):
        """Initialise the Run object and insert the 'git' command name.
        """
        Run.__init__(self, 'git', *cmd)

#
# Classes
#

class Person:
    """An author, committer, etc."""
    def __init__(self, name = None, email = None, date = '',
                 desc = None):
        self.name = self.email = self.date = None
        if name or email or date:
            assert not desc
            self.name = name
            self.email = email
            self.date = date
        elif desc:
            assert not (name or email or date)
            def parse_desc(s):
                m = re.match(r'^(.+)<(.+)>(.*)$', s)
                assert m
                return [x.strip() or None for x in m.groups()]
            self.name, self.email, self.date = parse_desc(desc)
    def set_name(self, val):
        if val:
            self.name = val
    def set_email(self, val):
        if val:
            self.email = val
    def set_date(self, val):
        if val:
            self.date = val
    def __str__(self):
        if self.name and self.email:
            return '%s <%s>' % (self.name, self.email)
        else:
            raise GitException, 'not enough identity data'

class Commit:
    """Handle the commit objects
    """
    def __init__(self, id_hash):
        self.__id_hash = id_hash

        lines = GRun('cat-file', 'commit', id_hash).output_lines()
        for i in range(len(lines)):
            line = lines[i]
            if not line:
                break # we've seen all the header fields
            key, val = line.split(' ', 1)
            if key == 'tree':
                self.__tree = val
            elif key == 'author':
                self.__author = val
            elif key == 'committer':
                self.__committer = val
            else:
                pass # ignore other headers
        self.__log = '\n'.join(lines[i+1:])

    def get_id_hash(self):
        return self.__id_hash

    def get_tree(self):
        return self.__tree

    def get_parent(self):
        parents = self.get_parents()
        if parents:
            return parents[0]
        else:
            return None

    def get_parents(self):
        return GRun('rev-list', '--parents', '--max-count=1', self.__id_hash
                    ).output_one_line().split()[1:]

    def get_author(self):
        return self.__author

    def get_committer(self):
        return self.__committer

    def get_log(self):
        return self.__log

    def __str__(self):
        return self.get_id_hash()

# dictionary of Commit objects, used to avoid multiple calls to git
__commits = dict()

#
# Functions
#

def get_commit(id_hash):
    """Commit objects factory. Save/look-up them in the __commits
    dictionary
    """
    global __commits

    if id_hash in __commits:
        return __commits[id_hash]
    else:
        commit = Commit(id_hash)
        __commits[id_hash] = commit
        return commit

def get_conflicts():
    """Return the list of file conflicts
    """
    names = set()
    for line in GRun('ls-files', '-z', '--unmerged'
                     ).raw_output().split('\0')[:-1]:
        stat, path = line.split('\t', 1)
        names.add(path)
    return list(names)

def exclude_files():
    files = [os.path.join(basedir.get(), 'info', 'exclude')]
    user_exclude = config.get('core.excludesfile')
    if user_exclude:
        files.append(user_exclude)
    return files

def ls_files(files, tree = None, full_name = True):
    """Return the files known to GIT or raise an error otherwise. It also
    converts the file to the full path relative the the .git directory.
    """
    if not files:
        return []

    args = []
    if tree:
        args.append('--with-tree=%s' % tree)
    if full_name:
        args.append('--full-name')
    args.append('--')
    args.extend(files)
    try:
        return GRun('ls-files', '--error-unmatch', *args).output_lines()
    except GitRunException:
        # just hide the details of the 'git ls-files' command we use
        raise GitException, \
            'Some of the given paths are either missing or not known to GIT'

def tree_status(files = None, tree_id = 'HEAD', unknown = False,
                  noexclude = True, verbose = False, diff_flags = []):
    """Get the status of all changed files, or of a selected set of
    files. Returns a list of pairs - (status, filename).

    If 'not files', it will check all files, and optionally all
    unknown files.  If 'files' is a list, it will only check the files
    in the list.
    """
    assert not files or not unknown

    if verbose:
        out.start('Checking for changes in the working directory')

    refresh_index()

    cache_files = []

    # unknown files
    if unknown:
        cmd = ['ls-files', '-z', '--others', '--directory',
               '--no-empty-directory']
        if not noexclude:
            cmd += ['--exclude=%s' % s for s in
                    ['*.[ao]', '*.pyc', '.*', '*~', '#*', 'TAGS', 'tags']]
            cmd += ['--exclude-per-directory=.gitignore']
            cmd += ['--exclude-from=%s' % fn
                    for fn in exclude_files()
                    if os.path.exists(fn)]

        lines = GRun(*cmd).raw_output().split('\0')
        cache_files += [('?', line) for line in lines if line]

    # conflicted files
    conflicts = get_conflicts()
    cache_files += [('C', filename) for filename in conflicts
                    if not files or filename in files]
    reported_files = set(conflicts)

    # files in the index
    args = diff_flags + [tree_id]
    if files:
        args += ['--'] + files
    for line in GRun('diff-index', *args).output_lines():
        fs = tuple(line.rstrip().split(' ',4)[-1].split('\t',1))
        if fs[1] not in reported_files:
            cache_files.append(fs)
            reported_files.add(fs[1])

    # files in the index but changed on (or removed from) disk
    args = list(diff_flags)
    if files:
        args += ['--'] + files
    for line in GRun('diff-files', *args).output_lines():
        fs = tuple(line.rstrip().split(' ',4)[-1].split('\t',1))
        if fs[1] not in reported_files:
            cache_files.append(fs)
            reported_files.add(fs[1])

    if verbose:
        out.done()

    return cache_files

def local_changes(verbose = True):
    """Return true if there are local changes in the tree
    """
    return len(tree_status(verbose = verbose)) != 0

def get_heads():
    heads = []
    hr = re.compile(r'^[0-9a-f]{40} refs/heads/(.+)$')
    for line in GRun('show-ref', '--heads').output_lines():
        m = hr.match(line)
        heads.append(m.group(1))
    return heads

# HEAD value cached
__head = None

def get_head():
    """Verifies the HEAD and returns the SHA1 id that represents it
    """
    global __head

    if not __head:
        __head = rev_parse('HEAD')
    return __head

class DetachedHeadException(GitException):
    def __init__(self):
        GitException.__init__(self, 'Not on any branch')

def get_head_file():
    """Return the name of the file pointed to by the HEAD symref.
    Throw an exception if HEAD is detached."""
    try:
        return strip_prefix(
            'refs/heads/', GRun('symbolic-ref', '-q', 'HEAD'
                                ).output_one_line())
    except GitRunException:
        raise DetachedHeadException()

def set_head_file(ref):
    """Resets HEAD to point to a new ref
    """
    # head cache flushing is needed since we might have a different value
    # in the new head
    __clear_head_cache()
    try:
        GRun('symbolic-ref', 'HEAD', 'refs/heads/%s' % ref).run()
    except GitRunException:
        raise GitException, 'Could not set head to "%s"' % ref

def set_ref(ref, val):
    """Point ref at a new commit object."""
    try:
        GRun('update-ref', ref, val).run()
    except GitRunException:
        raise GitException, 'Could not update %s to "%s".' % (ref, val)

def set_branch(branch, val):
    set_ref('refs/heads/%s' % branch, val)

def __set_head(val):
    """Sets the HEAD value
    """
    global __head

    if not __head or __head != val:
        set_ref('HEAD', val)
        __head = val

    # only allow SHA1 hashes
    assert(len(__head) == 40)

def __clear_head_cache():
    """Sets the __head to None so that a re-read is forced
    """
    global __head

    __head = None

def refresh_index():
    """Refresh index with stat() information from the working directory.
    """
    GRun('update-index', '-q', '--unmerged', '--refresh').run()

def rev_parse(git_id):
    """Parse the string and return a verified SHA1 id
    """
    try:
        return GRun('rev-parse', '--verify', git_id
                    ).discard_stderr().output_one_line()
    except GitRunException:
        raise GitException, 'Unknown revision: %s' % git_id

def ref_exists(ref):
    try:
        rev_parse(ref)
        return True
    except GitException:
        return False

def branch_exists(branch):
    return ref_exists('refs/heads/%s' % branch)

def create_branch(new_branch, tree_id = None):
    """Create a new branch in the git repository
    """