# util.py - Mercurial utility functions and platform specfic implementations##  Copyright 2005 K. Thananchayan <thananck@yahoo.com>#  Copyright 2005-2007 Matt Mackall <mpm@selenic.com>#  Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com>## This software may be used and distributed according to the terms of the# GNU General Public License version 2, incorporated herein by reference. """Mercurial utility functions and platform specfic implementations. This contains helper routines that are independent of the SCM core andhide platform-specific details from the core.""" from i18n import _import error, osutilimport cStringIO, errno, re, shutil, sys, tempfile, tracebackimport os, stat, time, calendar, random, textwrapimport imp # Python compatibility def sha1(s):    return _fastsha1(s) def _fastsha1(s):    # This function will import sha1 from hashlib or sha (whichever is    # available) and overwrite itself with it on the first call.    # Subsequent calls will go directly to the imported function.    try:        from hashlib import sha1 as _sha1    except ImportError:        from sha import sha as _sha1    global _fastsha1, sha1    _fastsha1 = sha1 = _sha1    return _sha1(s) import subprocessclosefds = os.name == 'posix'def popen2(cmd):    # Setting bufsize to -1 lets the system decide the buffer size.    # The default for bufsize is 0, meaning unbuffered. This leads to    # poor performance on Mac OS X: http://bugs.python.org/issue4194    p = subprocess.Popen(cmd, shell=True, bufsize=-1,                         close_fds=closefds,                         stdin=subprocess.PIPE, stdout=subprocess.PIPE)    return p.stdin, p.stdoutdef popen3(cmd):    p = subprocess.Popen(cmd, shell=True, bufsize=-1,                         close_fds=closefds,                         stdin=subprocess.PIPE, stdout=subprocess.PIPE,                         stderr=subprocess.PIPE)    return p.stdin, p.stdout, p.stderr def version():    """Return version information if available."""    try:        import __version__        return __version__.version    except ImportError:        return 'unknown' # used by parsedatedefaultdateformats = (    '%Y-%m-%d %H:%M:%S',    '%Y-%m-%d %I:%M:%S%p',    '%Y-%m-%d %H:%M',    '%Y-%m-%d %I:%M%p',    '%Y-%m-%d',    '%m-%d',    '%m/%d',    '%m/%d/%y',    '%m/%d/%Y',    '%a %b %d %H:%M:%S %Y',    '%a %b %d %I:%M:%S%p %Y',    '%a, %d %b %Y %H:%M:%S',        #  GNU coreutils "/bin/date --rfc-2822"    '%b %d %H:%M:%S %Y',    '%b %d %I:%M:%S%p %Y',    '%b %d %H:%M:%S',    '%b %d %I:%M:%S%p',    '%b %d %H:%M',    '%b %d %I:%M%p',    '%b %d %Y',    '%b %d',    '%H:%M:%S',    '%I:%M:%SP',    '%H:%M',    '%I:%M%p',) extendeddateformats = defaultdateformats + (    "%Y",    "%Y-%m",    "%b",    "%b %Y",    ) def cachefunc(func):    '''cache the result of function calls'''    # XXX doesn't handle keywords args    cache = {}    if func.func_code.co_argcount == 1:        # we gain a small amount of time because        # we don't need to pack/unpack the list        def f(arg):            if arg not in cache:                cache[arg] = func(arg)            return cache[arg]    else:        def f(*args):            if args not in cache:                cache[args] = func(*args)            return cache[args]     return f def lrucachefunc(func):    '''cache most recent results of function calls'''    cache = {}    order = []    if func.func_code.co_argcount == 1:        def f(arg):            if arg not in cache:                if len(cache) > 20:                    del cache[order.pop(0)]                cache[arg] = func(arg)            else:                order.remove(arg)            order.append(arg)            return cache[arg]    else:        def f(*args):            if args not in cache:                if len(cache) > 20:                    del cache[order.pop(0)]                cache[args] = func(*args)            else:                order.remove(args)            order.append(args)            return cache[args]     return f class propertycache(object):    def __init__(self, func):        self.func = func        self.name = func.__name__    def __get__(self, obj, type=None):        result = self.func(obj)        setattr(obj, self.name, result)        return result def pipefilter(s, cmd):    '''filter string S through command CMD, returning its output'''    p = subprocess.Popen(cmd, shell=True, close_fds=closefds,                         stdin=subprocess.PIPE, stdout=subprocess.PIPE)    pout, perr = p.communicate(s)    return pout def tempfilter(s, cmd):    '''filter string S through a pair of temporary files with CMD.    CMD is used as a template to create the real command to be run,    with the strings INFILE and OUTFILE replaced by the real names of    the temporary files generated.'''    inname, outname = None, None    try:        infd, inname = tempfile.mkstemp(prefix='hg-filter-in-')        fp = os.fdopen(infd, 'wb')        fp.write(s)        fp.close()        outfd, outname = tempfile.mkstemp(prefix='hg-filter-out-')        os.close(outfd)        cmd = cmd.replace('INFILE', inname)        cmd = cmd.replace('OUTFILE', outname)        code = os.system(cmd)        if sys.platform == 'OpenVMS' and code & 1:            code = 0        if code: raise Abort(_("command '%s' failed: %s") %                             (cmd, explain_exit(code)))        return open(outname, 'rb').read()    finally:        try:            if inname: os.unlink(inname)        except: pass        try:            if outname: os.unlink(outname)        except: pass filtertable = {    'tempfile:': tempfilter,    'pipe:': pipefilter,    } def filter(s, cmd):    "filter a string through a command that transforms its input to its output"    for name, fn in filtertable.iteritems():        if cmd.startswith(name):            return fn(s, cmd[len(name):].lstrip())    return pipefilter(s, cmd) def binary(s):    """return true if a string is binary data"""    return bool(s and '\0' in s) def increasingchunks(source, min=1024, max=65536):    '''return no less than min bytes per chunk while data remains,    doubling min after each chunk until it reaches max'''    def log2(x):        if not x:            return 0        i = 0        while x:            x >>= 1            i += 1        return i - 1     buf = []    blen = 0    for chunk in source:        buf.append(chunk)        blen += len(chunk)        if blen >= min:            if min < max:                min = min << 1                nmin = 1 << log2(blen)                if nmin > min:                    min = nmin                if min > max:                    min = max            yield ''.join(buf)            blen = 0            buf = []    if buf:        yield ''.join(buf) Abort = error.Abort def always(fn): return Truedef never(fn): return False def pathto(root, n1, n2):    '''return the relative path from one place to another.    root should use os.sep to separate directories    n1 should use os.sep to separate directories    n2 should use "/" to separate directories    returns an os.sep-separated path.     If n1 is a relative path, it's assumed it's    relative to root.    n2 should always be relative to root.    '''    if not n1: return localpath(n2)    if os.path.isabs(n1):        if os.path.splitdrive(root)[0] != os.path.splitdrive(n1)[0]:            return os.path.join(root, localpath(n2))        n2 = '/'.join((pconvert(root), n2))    a, b = splitpath(n1), n2.split('/')    a.reverse()    b.reverse()    while a and b and a[-1] == b[-1]:        a.pop()        b.pop()    b.reverse()    return os.sep.join((['..'] * len(a)) + b) or '.' def canonpath(root, cwd, myname):    """return the canonical path of myname, given cwd and root"""    if root == os.sep:        rootsep = os.sep    elif endswithsep(root):        rootsep = root    else:        rootsep = root + os.sep    name = myname    if not os.path.isabs(name):        name = os.path.join(root, cwd, name)    name = os.path.normpath(name)    audit_path = path_auditor(root)    if name != rootsep and name.startswith(rootsep):        name = name[len(rootsep):]        audit_path(name)        return pconvert(name)    elif name == root:        return ''    else:        # Determine whether `name' is in the hierarchy at or beneath `root',        # by iterating name=dirname(name) until that causes no change (can't        # check name == '/', because that doesn't work on windows).  For each        # `name', compare dev/inode numbers.  If they match, the list `rel'        # holds the reversed list of components making up the relative file        # name we want.        root_st = os.stat(root)        rel = []        while True:            try:                name_st = os.stat(name)            except OSError:                break            if samestat(name_st, root_st):                if not rel:                    # name was actually the same as root (maybe a symlink)                    return ''                rel.reverse()                name = os.path.join(*rel)                audit_path(name)                return pconvert(name)            dirname, basename = os.path.split(name)            rel.append(basename)            if dirname == name:                break            name = dirname         raise Abort('%s not under root' % myname) _hgexecutable = None def main_is_frozen():    """return True if we are a frozen executable.     The code supports py2exe (most common, Windows only) and tools/freeze    (portable, not much used).    """    return (hasattr(sys, "frozen") or # new py2exe            hasattr(sys, "importers") or # old py2exe            imp.is_frozen("__main__")) # tools/freeze def hgexecutable():    """return location of the 'hg' executable.     Defaults to $HG or 'hg' in the search path.    """    if _hgexecutable is None:        hg = os.environ.get('HG')        if hg:            set_hgexecutable(hg)        elif main_is_frozen():            set_hgexecutable(sys.executable)        else:            set_hgexecutable(find_exe('hg') or 'hg')    return _hgexecutable def set_hgexecutable(path):    """set location of the 'hg' executable"""    global _hgexecutable    _hgexecutable = path def system(cmd, environ={}, cwd=None, onerr=None, errprefix=None):    '''enhanced shell command execution.    run with environment maybe modified, maybe in different dir.     if command fails and onerr is None, return status.  if ui object,    print error message and return status, else raise onerr object as    exception.'''    def py2shell(val):        'convert python object into string that is useful to shell'        if val is None or val is False:            return '0'        if val is True:            return '1'        return str(val)    oldenv = {}    for k in environ:        oldenv[k] = os.environ.get(k)    if cwd is not None:        oldcwd = os.getcwd()    origcmd = cmd    if os.name == 'nt':        cmd = '"%s"' % cmd    try:        for k, v in environ.iteritems():            os.environ[k] = py2shell(v)        os.environ['HG'] = hgexecutable()        if cwd is not None and oldcwd != cwd:            os.chdir(cwd)        rc = os.system(cmd)        if sys.platform == 'OpenVMS' and rc & 1:            rc = 0        if rc and onerr:            errmsg = '%s %s' % (os.path.basename(origcmd.split(None, 1)[0]),                                explain_exit(rc)[0])            if errprefix:                errmsg = '%s: %s' % (errprefix, errmsg)            try:                onerr.warn(errmsg + '\n')            except AttributeError:                raise onerr(errmsg)        return rc    finally:        for k, v in oldenv.iteritems():            if v is None:                del os.environ[k]            else:                os.environ[k] = v        if cwd is not None and oldcwd != cwd:            os.chdir(oldcwd) def checksignature(func):    '''wrap a function with code to check for calling errors'''    def check(*args, **kwargs):        try:            return func(*args, **kwargs)        except TypeError:            if len(traceback.extract_tb(sys.exc_info()[2])) == 1:                raise error.SignatureError            raise     return check # os.path.lexists is not available on python2.3def lexists(filename):    "test whether a file with this name exists. does not follow symlinks"    try:        os.lstat(filename)    except:        return False    return True def rename(src, dst):    """forcibly rename a file"""    try:        os.rename(src, dst)    except OSError, err: # FIXME: check err (EEXIST ?)         # On windows, rename to existing file is not allowed, so we        # must delete destination first. But if a file is open, unlink        # schedules it for delete but does not delete it. Rename        # happens immediately even for open files, so we rename        # destination to a temporary name, then delete that. Then        # rename is safe to do.        # The temporary name is chosen at random to avoid the situation        # where a file is left lying around from a previous aborted run.        # The usual race condition this introduces can't be avoided as        # we need the name to rename into, and not the file itself. Due        # to the nature of the operation however, any races will at worst        # lead to the rename failing and the current operation aborting.         def tempname(prefix):            for tries in xrange(10):                temp = '%s-%08x' % (prefix, random.randint(0, 0xffffffff))                if not os.path.exists(temp):                    return temp            raise IOError, (errno.EEXIST, "No usable temporary filename found")         temp = tempname(dst)        os.rename(dst, temp)        os.unlink(temp)        os.rename(src, dst) def unlink(f):    """unlink and remove the directory if it is empty"""    os.unlink(f)    # try removing directories that might now be empty    try:        os.removedirs(os.path.dirname(f))    except OSError:        pass def copyfile(src, dest):    "copy a file, preserving mode and atime/mtime"    if os.path.islink(src):        try:            os.unlink(dest)        except:            pass        os.symlink(os.readlink(src), dest)    else:        try:            shutil.copyfile(src, dest)            shutil.copystat(src, dest)        except shutil.Error, inst:            raise Abort(str(inst)) def copyfiles(src, dst, hardlink=None):    """Copy a directory tree using hardlinks if possible"""     if hardlink is None:        hardlink = (os.stat(src).st_dev ==                    os.stat(os.path.dirname(dst)).st_dev)     if os.path.isdir(src):        os.mkdir(dst)        for name, kind in osutil.listdir(src):            srcname = os.path.join(src, name)            dstname = os.path.join(dst, name)            copyfiles(srcname, dstname, hardlink)    else:        if hardlink:            try:                os_link(src, dst)            except (IOError, OSError):                hardlink = False                shutil.copy(src, dst)        else:            shutil.copy(src, dst) class path_auditor(object):    '''ensure that a filesystem path contains no banned components.    the following properties of a path are checked: 

Lines 1–500 of 1284. Load more