path: root/libbe/storage/vcs/darcs.py

# Copyright (C) 2009 Gianluca Montecchi <gian@grys.it>
#                    W. Trevor King <wking@drexel.edu>
#
# 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.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""
Darcs backend.
"""

import codecs
import os
import re
import shutil
import sys
import time # work around http://mercurial.selenic.com/bts/issue618
try: # import core module, Python >= 2.5
    from xml.etree import ElementTree
except ImportError: # look for non-core module
    from elementtree import ElementTree
from xml.sax.saxutils import unescape

import libbe
import base

if libbe.TESTING == True:
    import doctest
    import unittest


def new():
    return Darcs()

class Darcs(base.VCS):
    name='darcs'
    client='darcs'

    def __init__(self, *args, **kwargs):
        base.VCS.__init__(self, *args, **kwargs)
        self.versioned = True
        self.__updated = [] # work around http://mercurial.selenic.com/bts/issue618

    def _vcs_version(self):
        status,output,error = self._u_invoke_client('--version')
        num_part = output.split(' ')[0]
        self.parsed_version = [int(i) for i in num_part.split('.')]
        return output

    def _vcs_get_user_id(self):
        # following http://darcs.net/manual/node4.html#SECTION00410030000000000000
        # as of June 29th, 2009
        if self.repo == None:
            return None
        darcs_dir = os.path.join(self.repo, '_darcs')
        if darcs_dir != None:
            for pref_file in ['author', 'email']:
                pref_path = os.path.join(darcs_dir, 'prefs', pref_file)
                if os.path.exists(pref_path):
                    return self.get_file_contents(pref_path)
        for env_variable in ['DARCS_EMAIL', 'EMAIL']:
            if env_variable in os.environ:
                return os.environ[env_variable]
        return None

    def _vcs_detect(self, path):
        if self._u_search_parent_directories(path, "_darcs") != None :
            return True
        return False

    def _vcs_root(self, path):
        """Find the root of the deepest repository containing path."""
        # Assume that nothing funny is going on; in particular, that we aren't
        # dealing with a bare repo.
        if os.path.isdir(path) != True:
            path = os.path.dirname(path)
        darcs_dir = self._u_search_parent_directories(path, '_darcs')
        if darcs_dir == None:
            return None
        return os.path.dirname(darcs_dir)

    def _vcs_init(self, path):
        self._u_invoke_client('init', cwd=path)

    def _vcs_destroy(self):
        vcs_dir = os.path.join(self.repo, '_darcs')
        if os.path.exists(vcs_dir):
            shutil.rmtree(vcs_dir)

    def _vcs_add(self, path):
        if os.path.isdir(path):
            return
        self._u_invoke_client('add', path)

    def _vcs_remove(self, path):
        if not os.path.isdir(self._u_abspath(path)):
            os.remove(os.path.join(self.repo, path)) # darcs notices removal

    def _vcs_update(self, path):
        self.__updated.append(path) # work around http://mercurial.selenic.com/bts/issue618
        pass # darcs notices changes

    def _vcs_get_file_contents(self, path, revision=None):
        if revision == None:
            return base.VCS._vcs_get_file_contents(self, path, revision)
        else:
            if self.parsed_version[0] >= 2:
                status,output,error = self._u_invoke_client( \
                    'show', 'contents', '--patch', revision, path)
                return output
            else:
                # Darcs versions < 2.0.0pre2 lack the 'show contents' command

                status,output,error = self._u_invoke_client( \
                    'diff', '--unified', '--from-patch', revision, path,
                    unicode_output=False)
                major_patch = output
                status,output,error = self._u_invoke_client( \
                    'diff', '--unified', '--patch', revision, path,
                    unicode_output=False)
                target_patch = output

                # '--output -' to be supported in GNU patch > 2.5.9
                # but that hasn't been released as of June 30th, 2009.

                # Rewrite path to status before the patch we want
                args=['patch', '--reverse', path]
                status,output,error = self._u_invoke(args, stdin=major_patch)
                # Now apply the patch we want
                args=['patch', path]
                status,output,error = self._u_invoke(args, stdin=target_patch)

                if os.path.exists(os.path.join(self.repo, path)) == True:
                    contents = base.VCS._vcs_get_file_contents(self, path)
                else:
                    contents = ''

                # Now restore path to it's current incarnation
                args=['patch', '--reverse', path]
                status,output,error = self._u_invoke(args, stdin=target_patch)
                args=['patch', path]
                status,output,error = self._u_invoke(args, stdin=major_patch)
                current_contents = base.VCS._vcs_get_file_contents(self, path)
                return contents

    def _vcs_commit(self, commitfile, allow_empty=False):
        id = self.get_user_id()
        if id == None or '@' not in id:
            id = '%s <%s@invalid.com>' % (id, id)
        args = ['record', '--all', '--author', id, '--logfile', commitfile]
        status,output,error = self._u_invoke_client(*args)
        empty_strings = ['No changes!']
        # work around http://mercurial.selenic.com/bts/issue618
        if self._u_any_in_string(empty_strings, output) == True \
                and len(self.__updated) > 0:
            time.sleep(1)
            for path in self.__updated:
                os.utime(os.path.join(self.repo, path), None)
            status,output,error = self._u_invoke_client(*args)
        self.__updated = []
        # end work around
        if self._u_any_in_string(empty_strings, output) == True:
            if allow_empty == False:
                raise base.EmptyCommit()
            # note that darcs does _not_ make an empty revision.
            # this returns the last non-empty revision id...
            revision = self._vcs_revision_id(-1)
        else:
            revline = re.compile("Finished recording patch '(.*)'")
            match = revline.search(output)
            assert match != None, output+error
            assert len(match.groups()) == 1
            revision = match.groups()[0]
        return revision

    def _vcs_revision_id(self, index):
        status,output,error = self._u_invoke_client('changes', '--xml')
        revisions = []
        xml_str = output.encode('unicode_escape').replace(r'\n', '\n')
        element = ElementTree.XML(xml_str)
        assert element.tag == 'changelog', element.tag
        for patch in element.getchildren():
            assert patch.tag == 'patch', patch.tag
            for child in patch.getchildren():
                if child.tag == 'name':
                    text = unescape(unicode(child.text).decode('unicode_escape').strip())
                    revisions.append(text)
        revisions.reverse()
        try:
            if index > 0:
                return revisions[index-1]
            elif index < 0:
                return revisions[index]
            else:
                return None
        except IndexError:
            return None


if libbe.TESTING == True:
    base.make_vcs_testcase_subclasses(Darcs, sys.modules[__name__])

    unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
    suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])