cola/utils.py

   1 # Copyright (C) 2007-2018 David Aguilar and contributors
   2 """This module provides miscellaneous utility functions."""
   3 from __future__ import division, absolute_import, unicode_literals
   4 import copy
   5 import os
   6 import random
   7 import re
   8 import shlex
   9 import sys
  10 import tempfile
  11 import time
  12 import traceback
  13
  14 from . import core
  15 from . import compat
  16
  17 random.seed(hash(time.time()))
  18
  19
  20 def asint(obj, default=0):
  21     """Make any value into an int, even if the cast fails"""
  22     try:
  23         value = int(obj)
  24     except (TypeError, ValueError):
  25         value = default
  26     return value
  27
  28
  29 def clamp(value, lo, hi):
  30     """Clamp a value to the specified range"""
  31     return min(hi, max(lo, value))
  32
  33
  34 def epoch_millis():
  35     return int(time.time() * 1000)
  36
  37
  38 def add_parents(paths):
  39     """Iterate over each item in the set and add its parent directories."""
  40     all_paths = set()
  41     for path in paths:
  42         while '//' in path:
  43             path = path.replace('//', '/')
  44         all_paths.add(path)
  45         if '/' in path:
  46             parent_dir = dirname(path)
  47             while parent_dir:
  48                 all_paths.add(parent_dir)
  49                 parent_dir = dirname(parent_dir)
  50     return all_paths
  51
  52
  53 def format_exception(e):
  54     exc_type, exc_value, exc_tb = sys.exc_info()
  55     details = traceback.format_exception(exc_type, exc_value, exc_tb)
  56     details = '\n'.join(map(core.decode, details))
  57     if hasattr(e, 'msg'):
  58         msg = e.msg
  59     else:
  60         msg = core.decode(repr(e))
  61     return (msg, details)
  62
  63
  64 def sublist(a, b):
  65     """Subtracts list b from list a and returns the resulting list."""
  66     # conceptually, c = a - b
  67     c = []
  68     for item in a:
  69         if item not in b:
  70             c.append(item)
  71     return c
  72
  73
  74 __grep_cache = {}
  75
  76
  77 def grep(pattern, items, squash=True):
  78     """Greps a list for items that match a pattern
  79
  80     :param squash: If only one item matches, return just that item
  81     :returns: List of matching items
  82
  83     """
  84     isdict = isinstance(items, dict)
  85     if pattern in __grep_cache:
  86         regex = __grep_cache[pattern]
  87     else:
  88         regex = __grep_cache[pattern] = re.compile(pattern)
  89     matched = []
  90     matchdict = {}
  91     for item in items:
  92         match = regex.match(item)
  93         if not match:
  94             continue
  95         groups = match.groups()
  96         if not groups:
  97             subitems = match.group(0)
  98         else:
  99             if len(groups) == 1:
 100                 subitems = groups[0]
 101             else:
 102                 subitems = list(groups)
 103         if isdict:
 104             matchdict[item] = items[item]
 105         else:
 106             matched.append(subitems)
 107
 108     if isdict:
 109         result = matchdict
 110     elif squash and len(matched) == 1:
 111         result = matched[0]
 112     else:
 113         result = matched
 114
 115     return result
 116
 117
 118 def basename(path):
 119     """
 120     An os.path.basename() implementation that always uses '/'
 121
 122     Avoid os.path.basename because git's output always
 123     uses '/' regardless of platform.
 124
 125     """
 126     return path.rsplit('/', 1)[-1]
 127
 128
 129 def strip_one(path):
 130     """Strip one level of directory"""
 131     return path.strip('/').split('/', 1)[-1]
 132
 133
 134 def dirname(path, current_dir=''):
 135     """
 136     An os.path.dirname() implementation that always uses '/'
 137
 138     Avoid os.path.dirname because git's output always
 139     uses '/' regardless of platform.
 140
 141     """
 142     while '//' in path:
 143         path = path.replace('//', '/')
 144     path_dirname = path.rsplit('/', 1)[0]
 145     if path_dirname == path:
 146         return current_dir
 147     return path.rsplit('/', 1)[0]
 148
 149
 150 def splitpath(path):
 151     """Split paths using '/' regardless of platform"""
 152     return path.split('/')
 153
 154
 155 def join(*paths):
 156     """Join paths using '/' regardless of platform"""
 157     return '/'.join(paths)
 158
 159
 160 def pathset(path):
 161     """Return all of the path components for the specified path
 162
 163     >>> pathset('foo/bar/baz') == ['foo', 'foo/bar', 'foo/bar/baz']
 164     True
 165
 166     """
 167     result = []
 168     parts = splitpath(path)
 169     prefix = ''
 170     for part in parts:
 171         result.append(prefix + part)
 172         prefix += part + '/'
 173
 174     return result
 175
 176
 177 def select_directory(paths):
 178     """Return the first directory in a list of paths"""
 179     if not paths:
 180         return core.getcwd()
 181
 182     for path in paths:
 183         if core.isdir(path):
 184             return path
 185
 186     return os.path.dirname(paths[0])
 187
 188
 189 def strip_prefix(prefix, string):
 190     """Return string, without the prefix. Blow up if string doesn't
 191     start with prefix."""
 192     assert string.startswith(prefix)
 193     return string[len(prefix) :]
 194
 195
 196 def sanitize(s):
 197     """Removes shell metacharacters from a string."""
 198     for c in """ \t!@#$%^&*()\\;,<>"'[]{}~|""":
 199         s = s.replace(c, '_')
 200     return s
 201
 202
 203 def tablength(word, tabwidth):
 204     """Return length of a word taking tabs into account
 205
 206     >>> tablength("\\t\\t\\t\\tX", 8)
 207     33
 208
 209     """
 210     return len(word.replace('\t', '')) + word.count('\t') * tabwidth
 211
 212
 213 def _shell_split_py2(s):
 214     """Python2 requires bytes inputs to shlex.split().  Returns [unicode]"""
 215     try:
 216         result = shlex.split(core.encode(s))
 217     except ValueError:
 218         result = core.encode(s).strip().split()
 219     # Decode to unicode strings
 220     return [core.decode(arg) for arg in result]
 221
 222
 223 def _shell_split_py3(s):
 224     """Python3 requires unicode inputs to shlex.split().  Converts to unicode"""
 225     try:
 226         result = shlex.split(s)
 227     except ValueError:
 228         result = core.decode(s).strip().split()
 229     # Already unicode
 230     return result
 231
 232
 233 def shell_split(s):
 234     if compat.PY2:
 235         # Encode before calling split()
 236         values = _shell_split_py2(s)
 237     else:
 238         # Python3 does not need the encode/decode dance
 239         values = _shell_split_py3(s)
 240     return values
 241
 242
 243 def tmp_filename(label, suffix=''):
 244     label = 'git-cola-' + label.replace('/', '-').replace('\\', '-')
 245     fd = tempfile.NamedTemporaryFile(prefix=label + '-', suffix=suffix)
 246     fd.close()
 247     return fd.name
 248
 249
 250 def is_linux():
 251     """Is this a linux machine?"""
 252     return sys.platform.startswith('linux')
 253
 254
 255 def is_debian():
 256     """Is it debian?"""
 257     return os.path.exists('/usr/bin/apt-get')
 258
 259
 260 def is_darwin():
 261     """Return True on OSX."""
 262     return sys.platform == 'darwin'
 263
 264
 265 def is_win32():
 266     """Return True on win32"""
 267     return sys.platform == 'win32' or sys.platform == 'cygwin'
 268
 269
 270 def expandpath(path):
 271     """Expand ~user/ and environment $variables"""
 272     path = os.path.expandvars(path)
 273     if path.startswith('~'):
 274         path = os.path.expanduser(path)
 275     return path
 276
 277
 278 class Group(object):
 279     """Operate on a collection of objects as a single unit"""
 280
 281     def __init__(self, *members):
 282         self._members = members
 283
 284     def __getattr__(self, name):
 285         """Return a function that relays calls to the group"""
 286
 287         def relay(*args, **kwargs):
 288             for member in self._members:
 289                 method = getattr(member, name)
 290                 method(*args, **kwargs)
 291
 292         setattr(self, name, relay)
 293         return relay
 294
 295
 296 class Proxy(object):
 297     """Wrap an object and override attributes"""
 298
 299     def __init__(self, obj, **overrides):
 300         self._obj = obj
 301         for k, v in overrides.items():
 302             setattr(self, k, v)
 303
 304     def __getattr__(self, name):
 305         return getattr(self._obj, name)
 306
 307
 308 def slice_fn(input_items, map_fn):
 309     """Slice input_items and call map_fn over every slice
 310
 311     This exists because of "errno: Argument list too long"
 312
 313     """
 314     # This comment appeared near the top of include/linux/binfmts.h
 315     # in the Linux source tree:
 316     #
 317     # /*
 318     #  * MAX_ARG_PAGES defines the number of pages allocated for arguments
 319     #  * and envelope for the new program. 32 should suffice, this gives
 320     #  * a maximum env+arg of 128kB w/4KB pages!
 321     #  */
 322     # #define MAX_ARG_PAGES 32
 323     #
 324     # 'size' is a heuristic to keep things highly performant by minimizing
 325     # the number of slices.  If we wanted it to run as few commands as
 326     # possible we could call "getconf ARG_MAX" and make a better guess,
 327     # but it's probably not worth the complexity (and the extra call to
 328     # getconf that we can't do on Windows anyways).
 329     #
 330     # In my testing, getconf ARG_MAX on Mac OS X Mountain Lion reported
 331     # 262144 and Debian/Linux-x86_64 reported 2097152.
 332     #
 333     # The hard-coded max_arg_len value is safely below both of these
 334     # real-world values.
 335
 336     # 4K pages x 32 MAX_ARG_PAGES
 337     max_arg_len = (32 * 4096) // 4  # allow plenty of space for the environment
 338     max_filename_len = 256
 339     size = max_arg_len // max_filename_len
 340
 341     status = 0
 342     outs = []
 343     errs = []
 344
 345     items = copy.copy(input_items)
 346     while items:
 347         stat, out, err = map_fn(items[:size])
 348         if stat < 0:
 349             status = min(stat, status)
 350         else:
 351             status = max(stat, status)
 352         outs.append(out)
 353         errs.append(err)
 354         items = items[size:]
 355
 356     return (status, '\n'.join(outs), '\n'.join(errs))
 357
 358
 359 class seq(object):
 360     def __init__(self, sequence):
 361         self.seq = sequence
 362
 363     def index(self, item, default=-1):
 364         try:
 365             idx = self.seq.index(item)
 366         except ValueError:
 367             idx = default
 368         return idx
 369
 370     def __getitem__(self, idx):
 371         return self.seq[idx]