Merge from W. Trevor King's tree.

1 files changed, 608 insertions, 45 deletions

diff --git a/libbe/rcs.py b/libbe/rcs.py
index 4487fba..3519c3d 100644
--- a/libbe/rcs.py
+++ b/libbe/rcs.py
@@ -15,42 +15,45 @@
 #    along with this program; if not, write to the Free Software
 #    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 from subprocess import Popen, PIPE
+import os
+import os.path
+from socket import gethostname
+import re
 import sys
+import tempfile
+import shutil
+import unittest
+import doctest
 
-def rcs_by_name(rcs_name):
-    """Return the module for the RCS with the given name"""
-    if rcs_name == "Arch":
-        import arch
-        return arch
-    elif rcs_name == "bzr":
-        import bzr
-        return bzr
-    elif rcs_name == "hg":
-        import hg
-        return hg
-    elif rcs_name == "git":
-        import git
-        return git
-    elif rcs_name == "None":
-        import no_rcs
-        return no_rcs
-
-def detect(dir):
-    """Return the module for the rcs being used in this directory"""
+from utility import Dir, search_parent_directories
+
+
+def _get_matching_rcs(matchfn):
+    """Return the first module for which matchfn(RCS_instance) is true"""
     import arch
     import bzr
     import hg
     import git
-    if arch.detect(dir):
-        return arch
-    elif bzr.detect(dir):
-        return bzr
-    elif hg.detect(dir):
-        return hg
-    elif git.detect(dir):
-        return git
-    import no_rcs
-    return no_rcs
+    for module in [arch, bzr, hg, git]:
+        rcs = module.new()
+        if matchfn(rcs) == True:
+            return rcs
+        else:
+            del(rcs)
+    return RCS()
+    
+def rcs_by_name(rcs_name):
+    """Return the module for the RCS with the given name"""
+    return _get_matching_rcs(lambda rcs: rcs.name == rcs_name)
+
+def detect_rcs(dir):
+    """Return an RCS instance for the rcs being used in this directory"""
+    return _get_matching_rcs(lambda rcs: rcs.detect(dir))
+
+def installed_rcs():
+    """Return an instance of an installed RCS"""
+    return _get_matching_rcs(lambda rcs: rcs.installed())
+
 
 class CommandError(Exception):
     def __init__(self, err_str, status):
@@ -58,19 +61,579 @@ class CommandError(Exception):
         self.err_str = err_str
         self.status = status
 
-def invoke(args, expect=(0,), cwd=None):
-    try :
-        if sys.platform != "win32":
-            q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, cwd=cwd)
+class SettingIDnotSupported(NotImplementedError):
+    pass
+
+class RCSnotRooted(Exception):
+    def __init__(self):
+        msg = "RCS not rooted"
+        Exception.__init__(self, msg)
+
+class PathNotInRoot(Exception):
+    def __init__(self, path, root):
+        msg = "Path '%s' not in root '%s'" % (path, root)
+        Exception.__init__(self, msg)
+        self.path = path
+        self.root = root
+
+class NoSuchFile(Exception):
+    def __init__(self, pathname):
+        Exception.__init__(self, "No such file: %s" % pathname)
+
+
+def new():
+    return RCS()
+
+class RCS(object):
+    """
+    This class implements a 'no-rcs' interface.
+
+    Support for other RCSs can be added by subclassing this class, and
+    overriding methods _rcs_*() with code appropriate for your RCS.
+    
+    The methods _u_*() are utility methods available to the _rcs_*()
+    methods.
+    """
+    name = "None"
+    client = "" # command-line tool for _u_invoke_client
+    versioned = False
+    def __init__(self, paranoid=False):
+        self.paranoid = paranoid
+        self.verboseInvoke = False
+        self.rootdir = None
+        self._duplicateBasedir = None
+        self._duplicateDirname = None
+    def __del__(self):
+        self.cleanup()
+
+    def _rcs_help(self):
+        """
+        Return the command help string.
+        (Allows a simple test to see if the client is installed.)
+        """
+        pass
+    def _rcs_detect(self, path=None):
+        """
+        Detect whether a directory is revision controlled with this RCS.
+        """
+        return True
+    def _rcs_root(self, path):
+        """
+        Get the RCS root.  This is the default working directory for
+        future invocations.  You would normally set this to the root
+        directory for your RCS.
+        """
+        if os.path.isdir(path)==False:
+            path = os.path.dirname(path)
+            if path == "":
+                path = os.path.abspath(".")
+        return path
+    def _rcs_init(self, path):
+        """
+        Begin versioning the tree based at path.
+        """
+        pass
+    def _rcs_cleanup(self):
+        """
+        Remove any cruft that _rcs_init() created outside of the
+        versioned tree.
+        """
+        pass
+    def _rcs_get_user_id(self):
+        """
+        Get the RCS's suggested user id (e.g. "John Doe <jdoe@example.com>").
+        If the RCS has not been configured with a username, return None.
+        """
+        return None
+    def _rcs_set_user_id(self, value):
+        """
+        Set the RCS's suggested user id (e.g "John Doe <jdoe@example.com>").
+        This is run if the RCS has not been configured with a usename, so
+        that commits will have a reasonable FROM value.
+        """
+        raise SettingIDnotSupported
+    def _rcs_add(self, path):
+        """
+        Add the already created file at path to version control.
+        """
+        pass
+    def _rcs_remove(self, path):
+        """
+        Remove the file at path from version control.  Optionally
+        remove the file from the filesystem as well.
+        """
+        pass
+    def _rcs_update(self, path):
+        """
+        Notify the versioning system of changes to the versioned file
+        at path.
+        """
+        pass
+    def _rcs_get_file_contents(self, path, revision=None):
+        """
+        Get the file contents as they were in a given revision.  Don't
+        worry about decoding the contents, the RCS.get_file_contents()
+        method will handle that.
+        
+        Revision==None specifies the current revision.
+        """
+        assert revision == None, \
+            "The %s RCS does not support revision specifiers" % self.name
+        return file(os.path.join(self.rootdir, path), "rb").read()
+    def _rcs_duplicate_repo(self, directory, revision=None):
+        """
+        Get the repository as it was in a given revision.
+        revision==None specifies the current revision.
+        dir specifies a directory to create the duplicate in.
+        """
+        shutil.copytree(self.rootdir, directory, True)
+    def _rcs_commit(self, commitfile):
+        """
+        Commit the current working directory, using the contents of
+        commitfile as the comment.  Return the name of the old
+        revision.
+        """
+        return None
+    def installed(self):
+        try:
+            self._rcs_help()
+            return True
+        except OSError, e:
+            if e.errno == errno.ENOENT:
+                return False
+            raise e
+    def detect(self, path="."):
+        """
+        Detect whether a directory is revision controlled with this RCS.
+        """
+        return self._rcs_detect(path)
+    def root(self, path):
+        """
+        Set the root directory to the path's RCS root.  This is the
+        default working directory for future invocations.
+        """
+        self.rootdir = self._rcs_root(path)
+    def init(self, path):
+        """
+        Begin versioning the tree based at path.
+        Also roots the rcs at path.
+        """
+        if os.path.isdir(path)==False:
+            path = os.path.dirname(path)
+        self._rcs_init(path)
+        self.root(path)
+    def cleanup(self):
+        self._rcs_cleanup()
+    def get_user_id(self):
+        """
+        Get the RCS's suggested user id (e.g. "John Doe <jdoe@example.com>").
+        If the RCS has not been configured with a username, return the user's
+        id.  You can override the automatic lookup procedure by setting the
+        RCS.user_id attribute to a string of your choice.
+        """
+        if hasattr(self, "user_id"):
+            if self.user_id != None:
+                return self.user_id
+        id = self._rcs_get_user_id()
+        if id == None:
+            name = self._u_get_fallback_username()
+            email = self._u_get_fallback_email()
+            id = self._u_create_id(name, email)
+            print >> sys.stderr, "Guessing id '%s'" % id
+            try:
+                self.set_user_id(id)
+            except SettingIDnotSupported:
+                pass
+        return id
+    def set_user_id(self, value):
+        """
+        Set the RCS's suggested user id (e.g "John Doe <jdoe@example.com>").
+        This is run if the RCS has not been configured with a usename, so
+        that commits will have a reasonable FROM value.
+        """
+        self._rcs_set_user_id(value)
+    def add(self, path):
+        """
+        Add the already created file at path to version control.
+        """
+        self._rcs_add(self._u_rel_path(path))
+    def remove(self, path):
+        """
+        Remove a file from both version control and the filesystem.
+        """
+        self._rcs_remove(self._u_rel_path(path))
+        if os.path.exists(path):
+            os.remove(path)
+    def recursive_remove(self, dirname):
+        """
+        Remove a file/directory and all its decendents from both
+        version control and the filesystem.
+        """
+        if not os.path.exists(dirname):
+            raise NoSuchFile(dirname)
+        for dirpath,dirnames,filenames in os.walk(dirname, topdown=False):
+            filenames.extend(dirnames)
+            for path in filenames:
+                fullpath = os.path.join(dirpath, path)
+                if os.path.exists(fullpath) == False:
+                    continue
+                self._rcs_remove(self._u_rel_path(fullpath))
+        if os.path.exists(dirname):
+            shutil.rmtree(dirname)
+    def update(self, path):
+        """
+        Notify the versioning system of changes to the versioned file
+        at path.
+        """
+        self._rcs_update(self._u_rel_path(path))
+    def get_file_contents(self, path, revision=None, allow_no_rcs=False):
+        """
+        Get the file as it was in a given revision.
+        Revision==None specifies the current revision.
+        """
+        if not os.path.exists(path):
+            raise NoSuchFile(path)
+        if self._use_rcs(path, allow_no_rcs):
+            relpath = self._u_rel_path(path)
+            contents = self._rcs_get_file_contents(relpath,revision)
+        else:
+            contents = file(path, "rb").read()
+        return contents.decode("utf-8")
+    def set_file_contents(self, path, contents, allow_no_rcs=False):
+        """
+        Set the file contents under version control.
+        """
+        add = not os.path.exists(path)
+        file(path, "wb").write(contents.encode("utf-8"))
+        
+        if self._use_rcs(path, allow_no_rcs):
+            if add:
+                self.add(path)
+            else:
+                self.update(path)
+    def mkdir(self, path, allow_no_rcs=False):
+        """
+        Create (if neccessary) a directory at path under version
+        control.
+        """
+        if not os.path.exists(path):
+            os.mkdir(path)
+            if self._use_rcs(path, allow_no_rcs):
+                self.add(path)
         else:
-            # win32 don't have os.execvp() so have to run command in a shell
-            q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, shell=True, 
-                      cwd=cwd)
-    except OSError, e :
-        strerror = "%s\nwhile executing %s" % (e.args[1], args)
-        raise CommandError(strerror, e.args[0])
-    output, error = q.communicate()
-    status = q.wait()
-    if status not in expect:
-        raise CommandError(error, status)
-    return status, output, error
+            assert os.path.isdir(path)
+            if self._use_rcs(path, allow_no_rcs):
+                self.update(path)
+    def duplicate_repo(self, revision=None):
+        """
+        Get the repository as it was in a given revision.
+        revision==None specifies the current revision.
+        Return the path to the arbitrary directory at the base of the new repo.
+        """
+        # Dirname in Baseir to protect against simlink attacks.
+        if self._duplicateBasedir == None:
+            self._duplicateBasedir = tempfile.mkdtemp(prefix='BErcs')
+            self._duplicateDirname = \
+                os.path.join(self._duplicateBasedir, "duplicate")
+            self._rcs_duplicate_repo(directory=self._duplicateDirname,
+                                     revision=revision)
+        return self._duplicateDirname
+    def remove_duplicate_repo(self):
+        """
+        Clean up a duplicate repo created with duplicate_repo().
+        """
+        if self._duplicateBasedir != None:
+            shutil.rmtree(self._duplicateBasedir)
+            self._duplicateBasedir = None
+            self._duplicateDirname = None
+    def commit(self, summary, body=None):
+        """
+        Commit the current working directory, with a commit message
+        string summary and body.  Return the name of the old revision
+        (or None if versioning is not supported).
+        """
+        if body is not None:
+            summary += '\n' + body
+        descriptor, filename = tempfile.mkstemp()
+        revision = None
+        try:
+            temp_file = os.fdopen(descriptor, 'wb')
+            temp_file.write(summary)
+            temp_file.flush()
+            revision = self._rcs_commit(filename)
+            temp_file.close()
+        finally:
+            os.remove(filename)
+        return revision
+    def precommit(self, directory):
+        pass
+    def postcommit(self, directory):
+        pass
+    def _u_invoke(self, args, expect=(0,), cwd=None):
+        if cwd == None:
+            cwd = self.rootdir
+        if self.verboseInvoke == True:
+            print >> sys.stderr, "%s$ %s" % (cwd, " ".join(args))
+        try :
+            if sys.platform != "win32":
+                q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, cwd=cwd)
+            else:
+                # win32 don't have os.execvp() so have to run command in a shell
+                q = Popen(args, stdin=PIPE, stdout=PIPE, stderr=PIPE, 
+                          shell=True, cwd=cwd)
+        except OSError, e :
+            strerror = "%s\nwhile executing %s" % (e.args[1], args)
+            raise CommandError(strerror, e.args[0])
+        output, error = q.communicate()
+        status = q.wait()
+        if self.verboseInvoke == True:
+            print >> sys.stderr, "%d\n%s%s" % (status, output, error)
+        if status not in expect:
+            strerror = "%s\nwhile executing %s\n%s" % (args[1], args, error)
+            raise CommandError(strerror, status)
+        return status, output, error
+    def _u_invoke_client(self, *args, **kwargs):
+        directory = kwargs.get('directory',None)
+        expect = kwargs.get('expect', (0,))
+        cl_args = [self.client]
+        cl_args.extend(args)
+        return self._u_invoke(cl_args, expect, cwd=directory)
+    def _u_search_parent_directories(self, path, filename):
+        """
+        Find the file (or directory) named filename in path or in any
+        of path's parents.
+        
+        e.g.
+          search_parent_directories("/a/b/c", ".be")
+        will return the path to the first existing file from
+          /a/b/c/.be
+          /a/b/.be
+          /a/.be
+          /.be
+        or None if none of those files exist.
+        """
+        return search_parent_directories(path, filename)
+    def _use_rcs(self, path, allow_no_rcs):
+        """
+        Try and decide if _rcs_add/update/mkdir/etc calls will
+        succeed.  Returns True is we think the rcs_call would
+        succeeed, and False otherwise.
+        """
+        use_rcs = True
+        exception = None
+        if self.rootdir != None:
+            if self.path_in_root(path) == False:
+                use_rcs = False
+                exception = PathNotInRoot(path, self.rootdir)
+        else:
+            use_rcs = False
+            exception = RCSnotRooted
+        if use_rcs == False and allow_no_rcs==False:
+            raise exception
+        return use_rcs
+    def path_in_root(self, path, root=None):
+        """
+        Return the relative path to path from root.
+        >>> rcs = new()
+        >>> rcs.path_in_root("/a.b/c/.be", "/a.b/c")
+        True
+        >>> rcs.path_in_root("/a.b/.be", "/a.b/c")
+        False
+        """
+        if root == None:
+            if self.rootdir == None:
+                raise RCSnotRooted
+            root = self.rootdir
+        path = os.path.abspath(path)
+        absRoot = os.path.abspath(root)
+        absRootSlashedDir = os.path.join(absRoot,"")
+        if not path.startswith(absRootSlashedDir):
+            return False
+        return True
+    def _u_rel_path(self, path, root=None):
+        """
+        Return the relative path to path from root.
+        >>> rcs = new()
+        >>> rcs._u_rel_path("/a.b/c/.be", "/a.b/c")
+        '.be'
+        """
+        if root == None:
+            if self.rootdir == None:
+                raise RCSnotRooted
+            root = self.rootdir
+        path = os.path.abspath(path)
+        absRoot = os.path.abspath(root)
+        absRootSlashedDir = os.path.join(absRoot,"")
+        if not path.startswith(absRootSlashedDir):
+            raise PathNotInRoot(path, absRootSlashedDir)
+        assert path != absRootSlashedDir, \
+            "file %s == root directory %s" % (path, absRootSlashedDir)
+        relpath = path[len(absRootSlashedDir):]
+        return relpath
+    def _u_abspath(self, path, root=None):
+        """
+        Return the absolute path from a path realtive to root.
+        >>> rcs = new()
+        >>> rcs._u_abspath(".be", "/a.b/c")
+        '/a.b/c/.be'
+        """
+        if root == None:
+            assert self.rootdir != None, "RCS not rooted"
+            root = self.rootdir
+        return os.path.abspath(os.path.join(root, path))
+    def _u_create_id(self, name, email=None):
+        """
+        >>> rcs = new()
+        >>> rcs._u_create_id("John Doe", "jdoe@example.com")
+        'John Doe <jdoe@example.com>'
+        >>> rcs._u_create_id("John Doe")
+        'John Doe'
+        """
+        assert len(name) > 0
+        if email == None or len(email) == 0:
+            return name
+        else:
+            return "%s <%s>" % (name, email)
+    def _u_parse_id(self, value):
+        """
+        >>> rcs = new()
+        >>> rcs._u_parse_id("John Doe <jdoe@example.com>")
+        ('John Doe', 'jdoe@example.com')
+        >>> rcs._u_parse_id("John Doe")
+        ('John Doe', None)
+        >>> try:
+        ...     rcs._u_parse_id("John Doe <jdoe@example.com><what?>")
+        ... except AssertionError:
+        ...     print "Invalid match"
+        Invalid match
+        """
+        emailexp = re.compile("(.*) <([^>]*)>(.*)")
+        match = emailexp.search(value)
+        if match == None:
+            email = None
+            name = value
+        else:
+            assert len(match.groups()) == 3
+            assert match.groups()[2] == "", match.groups()
+            email = match.groups()[1]
+            name = match.groups()[0]
+        assert name != None
+        assert len(name) > 0
+        return (name, email)
+    def _u_get_fallback_username(self):
+        name = None
+        for envariable in ["LOGNAME", "USERNAME"]:
+            if os.environ.has_key(envariable):
+                name = os.environ[envariable]
+                break
+        assert name != None
+        return name
+    def _u_get_fallback_email(self):
+        hostname = gethostname()
+        name = self._u_get_fallback_username()
+        return "%s@%s" % (name, hostname)
+    def _u_parse_commitfile(self, commitfile):
+        """
+        Split the commitfile created in self.commit() back into
+        summary and header lines.
+        """
+        f = file(commitfile, "rb")
+        summary = f.readline()
+        body = f.read()
+        body.lstrip('\n')
+        if len(body) == 0:
+            body = None
+        f.close
+        return (summary, body)
+        
+
+class RCStestCase(unittest.TestCase):
+    Class = RCS
+    def __init__(self, *args, **kwargs):
+        unittest.TestCase.__init__(self, *args, **kwargs)
+        self.dirname = None
+    def instantiateRCS(self):
+        return self.Class()
+    def setUp(self):
+        self.dir = Dir()
+        self.dirname = self.dir.path
+        self.rcs = self.instantiateRCS()
+    def tearDown(self):
+        del(self.rcs)
+        del(self.dirname)
+    def fullPath(self, path):
+        return os.path.join(self.dirname, path)
+    def assertPathExists(self, path):
+        fullpath = self.fullPath(path)
+        self.failUnless(os.path.exists(fullpath)==True,
+                        "path %s does not exist" % fullpath)
+    def uidTest(self):
+        user_id = self.rcs.get_user_id()
+        self.failUnless(user_id != None,
+                        "unable to get a user id")
+        user_idB = "John Doe <jdoe@example.com>"
+        if self.rcs.name in ["None", "hg"]:
+            self.assertRaises(SettingIDnotSupported, self.rcs.set_user_id,
+                              user_idB)
+        else:
+            self.rcs.set_user_id(user_idB)
+            self.failUnless(self.rcs.get_user_id() == user_idB,
+                            "user id not set correctly (was %s, is %s)" \
+                                % (user_id, self.rcs.get_user_id()))
+            self.failUnless(self.rcs.set_user_id(user_id) == None,
+                            "unable to restore user id %s" % user_id)
+            self.failUnless(self.rcs.get_user_id() == user_id,
+                            "unable to restore user id %s" % user_id)
+    def versionTest(self, path):
+        origpath = path
+        path = self.fullPath(path)
+        contentsA = "Lorem ipsum"
+        contentsB = "dolor sit amet"
+        self.rcs.set_file_contents(path,contentsA)
+        self.failUnless(self.rcs.get_file_contents(path)==contentsA,
+                        "File contents not set or read correctly")
+        revision = self.rcs.commit("Commit current status")
+        self.failUnless(self.rcs.get_file_contents(path)==contentsA,
+                        "Committing File contents not set or read correctly")
+        if self.rcs.versioned == True:
+            self.rcs.set_file_contents(path,contentsB)
+            self.failUnless(self.rcs.get_file_contents(path)==contentsB,
+                            "File contents not set correctly after commit")
+            contentsArev = self.rcs.get_file_contents(path, revision)
+            self.failUnless(contentsArev==contentsA, \
+                "Original file contents not saved in revision %s\n%s\n%s\n" \
+                                % (revision, contentsA, contentsArev))
+            dup = self.rcs.duplicate_repo(revision)
+            duppath = os.path.join(dup, origpath)
+            dupcont = file(duppath, "rb").read()
+            self.failUnless(dupcont == contentsA)
+            self.rcs.remove_duplicate_repo()
+    def testRun(self):
+        self.failUnless(self.rcs.installed() == True,
+                        "%s RCS not found" % self.Class.name)
+        if self.Class.name != "None":
+            self.failUnless(self.rcs.detect(self.dirname)==False,
+                            "Detected %s RCS before initializing" \
+                                % self.Class.name)
+        self.rcs.init(self.dirname)
+        self.failUnless(self.rcs.detect(self.dirname)==True,
+                        "Did not detect %s RCS after initializing" \
+                            % self.Class.name)
+        rp = os.path.realpath(self.rcs.rootdir)
+        dp = os.path.realpath(self.dirname)
+        self.failUnless(dp == rp or rp == None,
+                        "%s RCS root in wrong dir (%s %s)" \
+                            % (self.Class.name, dp, rp))
+        self.uidTest()
+        self.rcs.mkdir(self.fullPath('a'))
+        self.rcs.mkdir(self.fullPath('a/b'))
+        self.rcs.mkdir(self.fullPath('c'))
+        self.assertPathExists('a')
+        self.assertPathExists('a/b')
+        self.assertPathExists('c')
+        self.versionTest('a/text')
+        self.versionTest('a/b/text')
+        self.rcs.recursive_remove(self.fullPath('a'))
+
+unitsuite = unittest.TestLoader().loadTestsFromTestCase(RCStestCase)
+suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])