# Copyright (C) 2009 Gianluca Montecchi # W. Trevor King # # 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()])