cola/git.py

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