Initial directory restructuring to clarify dependencies

1 files changed, 192 insertions, 0 deletions

diff --git a/libbe/storage/vcs/darcs.py b/libbe/storage/vcs/darcs.py
new file mode 100644
index 0000000..d94eaef
--- /dev/null
+++ b/libbe/storage/vcs/darcs.py
@@ -0,0 +1,192 @@
+# 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 sys
+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 vcs
+if libbe.TESTING == True:
+    import doctest
+    import unittest
+
+
+def new():
+    return Darcs()
+
+class Darcs(vcs.VCS):
+    name="darcs"
+    client="darcs"
+    versioned=True
+    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_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_get_user_id(self):
+        # following http://darcs.net/manual/node4.html#SECTION00410030000000000000
+        # as of June 29th, 2009
+        if self.rootdir == None:
+            return None
+        darcs_dir = os.path.join(self.rootdir, "_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_set_user_id(self, value):
+        if self.rootdir == None:
+            self.root(".")
+            if self.rootdir == None:
+                raise vcs.SettingIDnotSupported
+        author_path = os.path.join(self.rootdir, "_darcs", "prefs", "author")
+        f = codecs.open(author_path, "w", self.encoding)
+        f.write(value)
+        f.close()
+    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.rootdir, path)) # darcs notices removal
+    def _vcs_update(self, path):
+        pass # darcs notices changes
+    def _vcs_get_file_contents(self, path, revision=None, binary=False):
+        if revision == None:
+            return vcs.VCS._vcs_get_file_contents(self, path, revision,
+                                                  binary=binary)
+        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.rootdir, path)) == True:
+                    contents = vcs.VCS._vcs_get_file_contents(self, path,
+                                                          binary=binary)
+                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 = vcs.VCS._vcs_get_file_contents(self, path,
+                                                              binary=binary)
+                return contents
+    def _vcs_duplicate_repo(self, directory, revision=None):
+        if revision==None:
+            vcs.VCS._vcs_duplicate_repo(self, directory, revision)
+        else:
+            self._u_invoke_client("put", "--to-patch", revision, directory)
+    def _vcs_commit(self, commitfile, allow_empty=False):
+        id = self.get_user_id()
+        if '@' 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!"]
+        if self._u_any_in_string(empty_strings, output) == True:
+            if allow_empty == False:
+                raise vcs.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:
+            return revisions[index]
+        except IndexError:
+            return None
+    
+if libbe.TESTING == True:
+    vcs.make_vcs_testcase_subclasses(Darcs, sys.modules[__name__])
+
+    unitsuite =unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
+    suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])