interfaces/email/interactive/libbe/darcs.py

   1 # Copyright (C) 2009 W. Trevor King <wking@drexel.edu>
   2 #
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License as published by
   5 # the Free Software Foundation; either version 2 of the License, or
   6 # (at your option) any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License along
  14 # with this program; if not, write to the Free Software Foundation, Inc.,
  15 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16
  17 """
  18 Darcs backend.
  19 """
  20
  21 import codecs
  22 import os
  23 import re
  24 import sys
  25 try: # import core module, Python >= 2.5
  26     from xml.etree import ElementTree
  27 except ImportError: # look for non-core module
  28     from elementtree import ElementTree
  29 from xml.sax.saxutils import unescape
  30 import doctest
  31 import unittest
  32
  33 import vcs
  34
  35
  36 def new():
  37     return Darcs()
  38
  39 class Darcs(vcs.VCS):
  40     name="darcs"
  41     client="darcs"
  42     versioned=True
  43     def _vcs_help(self):
  44         status,output,error = self._u_invoke_client("--help")
  45         return output
  46     def _vcs_detect(self, path):
  47         if self._u_search_parent_directories(path, "_darcs") != None :
  48             return True
  49         return False
  50     def _vcs_root(self, path):
  51         """Find the root of the deepest repository containing path."""
  52         # Assume that nothing funny is going on; in particular, that we aren't
  53         # dealing with a bare repo.
  54         if os.path.isdir(path) != True:
  55             path = os.path.dirname(path)
  56         darcs_dir = self._u_search_parent_directories(path, "_darcs")
  57         if darcs_dir == None:
  58             return None
  59         return os.path.dirname(darcs_dir)
  60     def _vcs_init(self, path):
  61         self._u_invoke_client("init", directory=path)
  62     def _vcs_get_user_id(self):
  63         # following http://darcs.net/manual/node4.html#SECTION00410030000000000000
  64         # as of June 29th, 2009
  65         if self.rootdir == None:
  66             return None
  67         darcs_dir = os.path.join(self.rootdir, "_darcs")
  68         if darcs_dir != None:
  69             for pref_file in ["author", "email"]:
  70                 pref_path = os.path.join(darcs_dir, "prefs", pref_file)
  71                 if os.path.exists(pref_path):
  72                     return self.get_file_contents(pref_path)
  73         for env_variable in ["DARCS_EMAIL", "EMAIL"]:
  74             if env_variable in os.environ:
  75                 return os.environ[env_variable]
  76         return None
  77     def _vcs_set_user_id(self, value):
  78         if self.rootdir == None:
  79             self.root(".")
  80             if self.rootdir == None:
  81                 raise vcs.SettingIDnotSupported
  82         author_path = os.path.join(self.rootdir, "_darcs", "prefs", "author")
  83         f = codecs.open(author_path, "w", self.encoding)
  84         f.write(value)
  85         f.close()
  86     def _vcs_add(self, path):
  87         if os.path.isdir(path):
  88             return
  89         self._u_invoke_client("add", path)
  90     def _vcs_remove(self, path):
  91         if not os.path.isdir(self._u_abspath(path)):
  92             os.remove(os.path.join(self.rootdir, path)) # darcs notices removal
  93     def _vcs_update(self, path):
  94         pass # darcs notices changes
  95     def _vcs_get_file_contents(self, path, revision=None, binary=False):
  96         if revision == None:
  97             return vcs.VCS._vcs_get_file_contents(self, path, revision,
  98                                               binary=binary)
  99         else:
 100             try:
 101                 return self._u_invoke_client("show", "contents", "--patch", revision, path)
 102             except vcs.CommandError:
 103                 # Darcs versions < 2.0.0pre2 lack the "show contents" command
 104
 105                 status,output,error = self._u_invoke_client("diff", "--unified",
 106                                                             "--from-patch",
 107                                                             revision, path)
 108                 major_patch = output
 109                 status,output,error = self._u_invoke_client("diff", "--unified",
 110                                                             "--patch",
 111                                                             revision, path)
 112                 target_patch = output
 113
 114                 # "--output -" to be supported in GNU patch > 2.5.9
 115                 # but that hasn't been released as of June 30th, 2009.
 116
 117                 # Rewrite path to status before the patch we want
 118                 args=["patch", "--reverse", path]
 119                 status,output,error = self._u_invoke(args, stdin=major_patch)
 120                 # Now apply the patch we want
 121                 args=["patch", path]
 122                 status,output,error = self._u_invoke(args, stdin=target_patch)
 123
 124                 if os.path.exists(os.path.join(self.rootdir, path)) == True:
 125                     contents = vcs.VCS._vcs_get_file_contents(self, path,
 126                                                           binary=binary)
 127                 else:
 128                     contents = ""
 129
 130                 # Now restore path to it's current incarnation
 131                 args=["patch", "--reverse", path]
 132                 status,output,error = self._u_invoke(args, stdin=target_patch)
 133                 args=["patch", path]
 134                 status,output,error = self._u_invoke(args, stdin=major_patch)
 135                 current_contents = vcs.VCS._vcs_get_file_contents(self, path,
 136                                                               binary=binary)
 137                 return contents
 138     def _vcs_duplicate_repo(self, directory, revision=None):
 139         if revision==None:
 140             vcs.VCS._vcs_duplicate_repo(self, directory, revision)
 141         else:
 142             self._u_invoke_client("put", "--to-patch", revision, directory)
 143     def _vcs_commit(self, commitfile, allow_empty=False):
 144         id = self.get_user_id()
 145         if '@' not in id:
 146             id = "%s <%s@invalid.com>" % (id, id)
 147         args = ['record', '--all', '--author', id, '--logfile', commitfile]
 148         status,output,error = self._u_invoke_client(*args)
 149         empty_strings = ["No changes!"]
 150         if self._u_any_in_string(empty_strings, output) == True:
 151             if allow_empty == False:
 152                 raise vcs.EmptyCommit()
 153             # note that darcs does _not_ make an empty revision.
 154             # this returns the last non-empty revision id...
 155             revision = self._vcs_revision_id(-1)
 156         else:
 157             revline = re.compile("Finished recording patch '(.*)'")
 158             match = revline.search(output)
 159             assert match != None, output+error
 160             assert len(match.groups()) == 1
 161             revision = match.groups()[0]
 162         return revision
 163     def _vcs_revision_id(self, index):
 164         status,output,error = self._u_invoke_client("changes", "--xml")
 165         revisions = []
 166         xml_str = output.encode("unicode_escape").replace(r"\n", "\n")
 167         element = ElementTree.XML(xml_str)
 168         assert element.tag == "changelog", element.tag
 169         for patch in element.getchildren():
 170             assert patch.tag == "patch", patch.tag
 171             for child in patch.getchildren():
 172                 if child.tag == "name":
 173                     text = unescape(unicode(child.text).decode("unicode_escape").strip())
 174                     revisions.append(text)
 175         revisions.reverse()
 176         try:
 177             return revisions[index]
 178         except IndexError:
 179             return None
 180 \f
 181 vcs.make_vcs_testcase_subclasses(Darcs, sys.modules[__name__])
 182
 183 unitsuite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
 184 suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])