libbe/git.py

   1 # Copyright (C) 2008-2009 Ben Finney <ben+python@benfinney.id.au>
   2 #                         Chris Ball <cjb@laptop.org>
   3 #                         W. Trevor King <wking@drexel.edu>
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 2 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along
  16 # with this program; if not, write to the Free Software Foundation, Inc.,
  17 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18
  19 """
  20 Git backend.
  21 """
  22
  23 import os
  24 import re
  25 import sys
  26 import unittest
  27 import doctest
  28
  29 import vcs
  30
  31
  32 def new():
  33     return Git()
  34
  35 class Git(vcs.VCS):
  36     name="git"
  37     client="git"
  38     versioned=True
  39     def _vcs_version(self):
  40         status,output,error = self._u_invoke_client("--version")
  41         return output
  42     def _vcs_detect(self, path):
  43         if self._u_search_parent_directories(path, ".git") != None :
  44             return True
  45         return False
  46     def _vcs_root(self, path):
  47         """Find the root of the deepest repository containing path."""
  48         # Assume that nothing funny is going on; in particular, that we aren't
  49         # dealing with a bare repo.
  50         if os.path.isdir(path) != True:
  51             path = os.path.dirname(path)
  52         status,output,error = self._u_invoke_client("rev-parse", "--git-dir",
  53                                                     directory=path)
  54         gitdir = os.path.join(path, output.rstrip('\n'))
  55         dirname = os.path.abspath(os.path.dirname(gitdir))
  56         return dirname
  57     def _vcs_init(self, path):
  58         self._u_invoke_client("init", directory=path)
  59     def _vcs_get_user_id(self):
  60         status,output,error = \
  61             self._u_invoke_client("config", "user.name", expect=(0,1))
  62         if status == 0:
  63             name = output.rstrip('\n')
  64         else:
  65             name = ""
  66         status,output,error = \
  67             self._u_invoke_client("config", "user.email", expect=(0,1))
  68         if status == 0:
  69             email = output.rstrip('\n')
  70         else:
  71             email = ""
  72         if name != "" or email != "": # got something!
  73             # guess missing info, if necessary
  74             if name == "":
  75                 name = self._u_get_fallback_username()
  76             if email == "":
  77                 email = self._u_get_fallback_email()
  78             return self._u_create_id(name, email)
  79         return None # Git has no infomation
  80     def _vcs_set_user_id(self, value):
  81         name,email = self._u_parse_id(value)
  82         if email != None:
  83             self._u_invoke_client("config", "user.email", email)
  84         self._u_invoke_client("config", "user.name", name)
  85     def _vcs_add(self, path):
  86         if os.path.isdir(path):
  87             return
  88         self._u_invoke_client("add", path)
  89     def _vcs_remove(self, path):
  90         if not os.path.isdir(self._u_abspath(path)):
  91             self._u_invoke_client("rm", "-f", path)
  92     def _vcs_update(self, path):
  93         self._vcs_add(path)
  94     def _vcs_get_file_contents(self, path, revision=None, binary=False):
  95         if revision == None:
  96             return vcs.VCS._vcs_get_file_contents(self, path, revision, binary=binary)
  97         else:
  98             arg = "%s:%s" % (revision,path)
  99             status,output,error = self._u_invoke_client("show", arg)
 100             return output
 101     def _vcs_duplicate_repo(self, directory, revision=None):
 102         if revision==None:
 103             vcs.VCS._vcs_duplicate_repo(self, directory, revision)
 104         else:
 105             #self._u_invoke_client("archive", revision, directory) # makes tarball
 106             self._u_invoke_client("clone", "--no-checkout",".",directory)
 107             self._u_invoke_client("checkout", revision, directory=directory)
 108     def _vcs_commit(self, commitfile, allow_empty=False):
 109         args = ['commit', '--all', '--file', commitfile]
 110         if allow_empty == True:
 111             args.append("--allow-empty")
 112             status,output,error = self._u_invoke_client(*args)
 113         else:
 114             kwargs = {"expect":(0,1)}
 115             status,output,error = self._u_invoke_client(*args, **kwargs)
 116             strings = ["nothing to commit",
 117                        "nothing added to commit"]
 118             if self._u_any_in_string(strings, output) == True:
 119                 raise vcs.EmptyCommit()
 120         revision = None
 121         revline = re.compile("(.*) (.*)[:\]] (.*)")
 122         match = revline.search(output)
 123         assert match != None, output+error
 124         assert len(match.groups()) == 3
 125         revision = match.groups()[1]
 126         full_revision = self._vcs_revision_id(-1)
 127         assert full_revision.startswith(revision), \
 128             "Mismatched revisions:\n%s\n%s" % (revision, full_revision)
 129         return full_revision
 130     def _vcs_revision_id(self, index):
 131         args = ["rev-list", "--first-parent", "--reverse", "HEAD"]
 132         kwargs = {"expect":(0,128)}
 133         status,output,error = self._u_invoke_client(*args, **kwargs)
 134         if status == 128:
 135             if error.startswith("fatal: ambiguous argument 'HEAD': unknown "):
 136                 return None
 137             raise vcs.CommandError(args, status, stdout="", stderr=error)
 138         commits = output.splitlines()
 139         try:
 140             return commits[index]
 141         except IndexError:
 142             return None
 143
 144 \f
 145 vcs.make_vcs_testcase_subclasses(Git, sys.modules[__name__])
 146
 147 unitsuite = unittest.TestLoader().loadTestsFromModule(sys.modules[__name__])
 148 suite = unittest.TestSuite([unitsuite, doctest.DocTestSuite()])