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