cola/git.py

   1 from __future__ import absolute_import, division, print_function, unicode_literals
   2 from functools import partial
   3 import errno
   4 import os
   5 from os.path import join
   6 import subprocess
   7 import threading
   8 import time
   9
  10 from . import core
  11 from .compat import int_types
  12 from .compat import ustr
  13 from .compat import WIN32
  14 from .decorators import memoize
  15 from .interaction import Interaction
  16
  17
  18 GIT_COLA_TRACE = core.getenv('GIT_COLA_TRACE', '')
  19 GIT = core.getenv('GIT_COLA_GIT', 'git')
  20 STATUS = 0
  21 STDOUT = 1
  22 STDERR = 2
  23
  24 # Object ID / SHA1-related constants
  25 # Git's empty tree is a built-in constant object name.
  26 EMPTY_TREE_OID = '4b825dc642cb6eb9a060e54bf8d69288fbee4904'
  27 # Git's diff machinery returns zeroes for modified files whose content exists
  28 # in the worktree only.
  29 MISSING_BLOB_OID = '0000000000000000000000000000000000000000'
  30 # Git's SHA-1 object IDs are 40 characters  long.
  31 # This will need to change when Git moves away from SHA-1.
  32 # When that happens we'll have to detect and update this at runtime in
  33 # order to support both old and new git.
  34 OID_LENGTH = 40
  35
  36 _index_lock = threading.Lock()
  37
  38
  39 def dashify(s):
  40     return s.replace('_', '-')
  41
  42
  43 def is_git_dir(git_dir):
  44     """From git's setup.c:is_git_directory()."""
  45     result = False
  46     if git_dir:
  47         headref = join(git_dir, 'HEAD')
  48
  49         if (
  50             core.isdir(git_dir)
  51             and (
  52                 core.isdir(join(git_dir, 'objects'))
  53                 and core.isdir(join(git_dir, 'refs'))
  54             )
  55             or (
  56                 core.isfile(join(git_dir, 'gitdir'))
  57                 and core.isfile(join(git_dir, 'commondir'))
  58             )
  59         ):
  60
  61             result = core.isfile(headref) or (
  62                 core.islink(headref) and core.readlink(headref).startswith('refs/')
  63             )
  64         else:
  65             result = is_git_file(git_dir)
  66
  67     return result
  68
  69
  70 def is_git_file(f):
  71     return core.isfile(f) and os.path.basename(f) == '.git'
  72
  73
  74 def is_git_worktree(d):
  75     return is_git_dir(join(d, '.git'))
  76
  77
  78 def is_git_repository(path):
  79     return is_git_worktree(path) or is_git_dir(path)
  80
  81
  82 def read_git_file(path):
  83     """Read the path from a .git-file
  84
  85     `None` is returned when <path> is not a .git-file.
  86
  87     """
  88     result = None
  89     if path and is_git_file(path):
  90         header = 'gitdir: '
  91         data = core.read(path).strip()
  92         if data.startswith(header):
  93             result = data[len(header) :]
  94             if result and not os.path.isabs(result):
  95                 path_folder = os.path.dirname(path)
  96                 repo_relative = join(path_folder, result)
  97                 result = os.path.normpath(repo_relative)
  98     return result
  99
 100
 101 class Paths(object):
 102     """Git repository paths of interest"""
 103
 104     def __init__(self, git_dir=None, git_file=None, worktree=None, common_dir=None):
 105         if git_dir and not is_git_dir(git_dir):
 106             git_dir = None
 107         self.git_dir = git_dir
 108         self.git_file = git_file
 109         self.worktree = worktree
 110         self.common_dir = common_dir
 111
 112     def get(self, path):
 113         ceiling_dirs = set()
 114         ceiling = core.getenv('GIT_CEILING_DIRECTORIES')
 115         if ceiling:
 116             ceiling_dirs.update([x for x in ceiling.split(':') if x])
 117
 118         if path:
 119             path = core.abspath(path)
 120
 121         if not self.git_dir or not self.worktree:
 122             # Search for a .git directory
 123             while path:
 124                 if path in ceiling_dirs:
 125                     break
 126                 if is_git_dir(path):
 127                     if not self.git_dir:
 128                         self.git_dir = path
 129                     basename = os.path.basename(path)
 130                     if not self.worktree and basename == '.git':
 131                         self.worktree = os.path.dirname(path)
 132                 # We are either in a bare repository, or someone set GIT_DIR
 133                 # but did not set GIT_WORK_TREE.
 134                 if self.git_dir:
 135                     if not self.worktree:
 136                         basename = os.path.basename(self.git_dir)
 137                         if basename == '.git':
 138                             self.worktree = os.path.dirname(self.git_dir)
 139                         elif path and not is_git_dir(path):
 140                             self.worktree = path
 141                     break
 142                 gitpath = join(path, '.git')
 143                 if is_git_dir(gitpath):
 144                     if not self.git_dir:
 145                         self.git_dir = gitpath
 146                     if not self.worktree:
 147                         self.worktree = path
 148                     break
 149                 path, dummy = os.path.split(path)
 150                 if not dummy:
 151                     break
 152
 153         if self.git_dir:
 154             git_dir_path = read_git_file(self.git_dir)
 155             if git_dir_path:
 156                 self.git_file = self.git_dir
 157                 self.git_dir = git_dir_path
 158
 159                 commondir_file = join(git_dir_path, 'commondir')
 160                 if core.exists(commondir_file):
 161                     common_path = core.read(commondir_file).strip()
 162                     if common_path:
 163                         if os.path.isabs(common_path):
 164                             common_dir = common_path
 165                         else:
 166                             common_dir = join(git_dir_path, common_path)
 167                             common_dir = os.path.normpath(common_dir)
 168                         self.common_dir = common_dir
 169         # usage: Paths().get()
 170         return self
 171
 172
 173 def find_git_directory(path):
 174     """Perform Git repository discovery"""
 175     return Paths(
 176         git_dir=core.getenv('GIT_DIR'), worktree=core.getenv('GIT_WORK_TREE')
 177     ).get(path)
 178
 179
 180 class Git(object):
 181     """
 182     The Git class manages communication with the Git binary
 183     """
 184
 185     def __init__(self):
 186         self.paths = Paths()
 187
 188         self._valid = {}  #: Store the result of is_git_dir() for performance
 189         self.set_worktree(core.getcwd())
 190
 191     # pylint: disable=no-self-use
 192     def is_git_repository(self, path):
 193         return is_git_repository(path)
 194
 195     def getcwd(self):
 196         """Return the working directory used by git()"""
 197         return self.paths.worktree or self.paths.git_dir
 198
 199     def set_worktree(self, path):
 200         path = core.decode(path)
 201         self.paths = find_git_directory(path)
 202         return self.paths.worktree
 203
 204     def worktree(self):
 205         if not self.paths.worktree:
 206             path = core.abspath(core.getcwd())
 207             self.paths = find_git_directory(path)
 208         return self.paths.worktree
 209
 210     def is_valid(self):
 211         """Is this a valid git repository?
 212
 213         Cache the result to avoid hitting the filesystem.
 214
 215         """
 216         git_dir = self.paths.git_dir
 217         try:
 218             valid = bool(git_dir) and self._valid[git_dir]
 219         except KeyError:
 220             valid = self._valid[git_dir] = is_git_dir(git_dir)
 221
 222         return valid
 223
 224     def git_path(self, *paths):
 225         result = None
 226         if self.paths.git_dir:
 227             result = join(self.paths.git_dir, *paths)
 228         if result and self.paths.common_dir and not core.exists(result):
 229             common_result = join(self.paths.common_dir, *paths)
 230             if core.exists(common_result):
 231                 result = common_result
 232         return result
 233
 234     def git_dir(self):
 235         if not self.paths.git_dir:
 236             path = core.abspath(core.getcwd())
 237             self.paths = find_git_directory(path)
 238         return self.paths.git_dir
 239
 240     def __getattr__(self, name):
 241         git_cmd = partial(self.git, name)
 242         setattr(self, name, git_cmd)
 243         return git_cmd
 244
 245     @staticmethod
 246     def execute(
 247         command,
 248         _cwd=None,
 249         _decode=True,
 250         _encoding=None,
 251         _raw=False,
 252         _stdin=None,
 253         _stderr=subprocess.PIPE,
 254         _stdout=subprocess.PIPE,
 255         _readonly=False,
 256         _no_win32_startupinfo=False,
 257     ):
 258         """
 259         Execute a command and returns its output
 260
 261         :param command: argument list to execute.
 262         :param _cwd: working directory, defaults to the current directory.
 263         :param _decode: whether to decode output, defaults to True.
 264         :param _encoding: default encoding, defaults to None (utf-8).
 265         :param _readonly: avoid taking the index lock. Assume the command is read-only.
 266         :param _raw: do not strip trailing whitespace.
 267         :param _stdin: optional stdin filehandle.
 268         :returns (status, out, err): exit status, stdout, stderr
 269
 270         """
 271         # Allow the user to have the command executed in their working dir.
 272         if not _cwd:
 273             _cwd = core.getcwd()
 274
 275         extra = {}
 276
 277         if hasattr(os, 'setsid'):
 278             # SSH uses the SSH_ASKPASS variable only if the process is really
 279             # detached from the TTY (stdin redirection and setting the
 280             # SSH_ASKPASS environment variable is not enough).  To detach a
 281             # process from the console it should fork and call os.setsid().
 282             extra['preexec_fn'] = os.setsid
 283
 284         start_time = time.time()
 285
 286         # Start the process
 287         # Guard against thread-unsafe .git/index.lock files
 288         if not _readonly:
 289             _index_lock.acquire()
 290         try:
 291             status, out, err = core.run_command(
 292                 command,
 293                 cwd=_cwd,
 294                 encoding=_encoding,
 295                 stdin=_stdin,
 296                 stdout=_stdout,
 297                 stderr=_stderr,
 298                 no_win32_startupinfo=_no_win32_startupinfo,
 299                 **extra
 300             )
 301         finally:
 302             # Let the next thread in
 303             if not _readonly:
 304                 _index_lock.release()
 305
 306         end_time = time.time()
 307         elapsed_time = abs(end_time - start_time)
 308
 309         if not _raw and out is not None:
 310             out = core.UStr(out.rstrip('\n'), out.encoding)
 311
 312         cola_trace = GIT_COLA_TRACE
 313         if cola_trace == 'trace':
 314             msg = 'trace: %.3fs: %s' % (elapsed_time, core.list2cmdline(command))
 315             Interaction.log_status(status, msg, '')
 316         elif cola_trace == 'full':
 317             if out or err:
 318                 core.print_stderr(
 319                     "# %.3fs: %s -> %d: '%s' '%s'"
 320                     % (elapsed_time, ' '.join(command), status, out, err)
 321                 )
 322             else:
 323                 core.print_stderr(
 324                     '# %.3fs: %s -> %d' % (elapsed_time, ' '.join(command), status)
 325                 )
 326         elif cola_trace:
 327             core.print_stderr('# %.3fs: %s' % (elapsed_time, ' '.join(command)))
 328
 329         # Allow access to the command's status code
 330         return (status, out, err)
 331
 332     def git(self, cmd, *args, **kwargs):
 333         # Handle optional arguments prior to calling transform_kwargs
 334         # otherwise they'll end up in args, which is bad.
 335         _kwargs = dict(_cwd=self.getcwd())
 336         execute_kwargs = (
 337             '_cwd',
 338             '_decode',
 339             '_encoding',
 340             '_stdin',
 341             '_stdout',
 342             '_stderr',
 343             '_raw',
 344             '_readonly',
 345             '_no_win32_startupinfo',
 346         )
 347
 348         for kwarg in execute_kwargs:
 349             if kwarg in kwargs:
 350                 _kwargs[kwarg] = kwargs.pop(kwarg)
 351
 352         # Prepare the argument list
 353         git_args = [
 354             GIT,
 355             '-c',
 356             'diff.suppressBlankEmpty=false',
 357             '-c',
 358             'log.showSignature=false',
 359             dashify(cmd),
 360         ]
 361         opt_args = transform_kwargs(**kwargs)
 362         call = git_args + opt_args
 363         call.extend(args)
 364         try:
 365             result = self.execute(call, **_kwargs)
 366         except OSError as e:
 367             if WIN32 and e.errno == errno.ENOENT:
 368                 # see if git exists at all. on win32 it can fail with ENOENT in
 369                 # case of argv overflow. we should be safe from that but use
 370                 # defensive coding for the worst-case scenario. On UNIX
 371                 # we have ENAMETOOLONG but that doesn't exist on Windows.
 372                 if _git_is_installed():
 373                     raise e
 374                 _print_win32_git_hint()
 375             result = (1, '', "error: unable to execute '%s'" % GIT)
 376         return result
 377
 378
 379 def _git_is_installed():
 380     """Return True if git is installed"""
 381     # On win32 Git commands can fail with ENOENT in case of argv overflow. we
 382     # should be safe from that but use defensive coding for the worst-case
 383     # scenario. On UNIX we have ENAMETOOLONG but that doesn't exist on
 384     # Windows.
 385     try:
 386         status, _, _ = Git.execute([GIT, '--version'])
 387         result = status == 0
 388     except OSError:
 389         result = False
 390     return result
 391
 392
 393 def transform_kwargs(**kwargs):
 394     """Transform kwargs into git command line options
 395
 396     Callers can assume the following behavior:
 397
 398     Passing foo=None ignores foo, so that callers can
 399     use default values of None that are ignored unless
 400     set explicitly.
 401
 402     Passing foo=False ignore foo, for the same reason.
 403
 404     Passing foo={string-or-number} results in ['--foo=<value>']
 405     in the resulting arguments.
 406
 407     """
 408     args = []
 409     types_to_stringify = (ustr, float, str) + int_types
 410
 411     for k, v in kwargs.items():
 412         if len(k) == 1:
 413             dashes = '-'
 414             eq = ''
 415         else:
 416             dashes = '--'
 417             eq = '='
 418         # isinstance(False, int) is True, so we have to check bool first
 419         if isinstance(v, bool):
 420             if v:
 421                 args.append('%s%s' % (dashes, dashify(k)))
 422             # else: pass  # False is ignored; flag=False inhibits --flag
 423         elif isinstance(v, types_to_stringify):
 424             args.append('%s%s%s%s' % (dashes, dashify(k), eq, v))
 425
 426     return args
 427
 428
 429 def win32_git_error_hint():
 430     return (
 431         '\n'
 432         'NOTE: If you have Git installed in a custom location, e.g.\n'
 433         'C:\\Tools\\Git, then you can create a file at\n'
 434         '~/.config/git-cola/git-bindir with following text\n'
 435         'and git-cola will add the specified location to your $PATH\n'
 436         'automatically when starting cola:\n'
 437         '\n'
 438         r'C:\Tools\Git\bin'
 439     )
 440
 441
 442 @memoize
 443 def _print_win32_git_hint():
 444     hint = '\n' + win32_git_error_hint() + '\n'
 445     core.print_stderr("error: unable to execute 'git'" + hint)
 446
 447
 448 def create():
 449     """Create Git instances
 450
 451     >>> git = create()
 452     >>> status, out, err = git.version()
 453     >>> 'git' == out[:3].lower()
 454     True
 455
 456     """
 457     return Git()