more py2.5 compatibility patches

rhodecode/lib/compat.py

@@ -1,585 +1,594 b''
 # -*- coding: utf-8 -*-
 """
     rhodecode.lib.compat
     ~~~~~~~~~~~~~~~~~~~~
 
     Python backward compatibility functions and common libs
 
 
     :created_on: Oct 7, 2011
     :author: marcink
     :copyright: (C) 2010-2010 Marcin Kuzminski <[e-mail]>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
 from rhodecode import __platform__, PLATFORM_WIN, __py_version__
 
 #==============================================================================
 # json
 #==============================================================================
 from rhodecode.lib.ext_json import json
 
 
 #==============================================================================
 # izip_longest
 #==============================================================================
 try:
     from itertools import izip_longest
 except ImportError:
     import itertools
 
     def izip_longest(*args, **kwds):
         fillvalue = kwds.get("fillvalue")
 
         def sentinel(counter=([fillvalue] * (len(args) - 1)).pop):
             yield counter()  # yields the fillvalue, or raises IndexError
 
         fillers = itertools.repeat(fillvalue)
         iters = [itertools.chain(it, sentinel(), fillers)
                     for it in args]
         try:
             for tup in itertools.izip(*iters):
                 yield tup
         except IndexError:
             pass
 
 
 #==============================================================================
 # OrderedDict
 #==============================================================================
 
 # Python Software Foundation License
 
 # XXX: it feels like using the class with "is" and "is not" instead of "==" and
 # "!=" should be faster.
 class _Nil(object):
 
     def __repr__(self):
         return "nil"
 
     def __eq__(self, other):
         if (isinstance(other, _Nil)):
             return True
         else:
             return NotImplemented
 
     def __ne__(self, other):
         if (isinstance(other, _Nil)):
             return False
         else:
             return NotImplemented
 
 _nil = _Nil()
 
 
 class _odict(object):
     """Ordered dict data structure, with O(1) complexity for dict operations
     that modify one element.
 
     Overwriting values doesn't change their original sequential order.
     """
 
     def _dict_impl(self):
         return None
 
     def __init__(self, data=(), **kwds):
         """This doesn't accept keyword initialization as normal dicts to avoid
         a trap - inside a function or method the keyword args are accessible
         only as a dict, without a defined order, so their original order is
         lost.
         """
         if kwds:
             raise TypeError("__init__() of ordered dict takes no keyword "
                             "arguments to avoid an ordering trap.")
         self._dict_impl().__init__(self)
         # If you give a normal dict, then the order of elements is undefined
         if hasattr(data, "iteritems"):
             for key, val in data.iteritems():
                 self[key] = val
         else:
             for key, val in data:
                 self[key] = val
 
     # Double-linked list header
     def _get_lh(self):
         dict_impl = self._dict_impl()
         if not hasattr(self, '_lh'):
             dict_impl.__setattr__(self, '_lh', _nil)
         return dict_impl.__getattribute__(self, '_lh')
 
     def _set_lh(self, val):
         self._dict_impl().__setattr__(self, '_lh', val)
 
     lh = property(_get_lh, _set_lh)
 
     # Double-linked list tail
     def _get_lt(self):
         dict_impl = self._dict_impl()
         if not hasattr(self, '_lt'):
             dict_impl.__setattr__(self, '_lt', _nil)
         return dict_impl.__getattribute__(self, '_lt')
 
     def _set_lt(self, val):
         self._dict_impl().__setattr__(self, '_lt', val)
 
     lt = property(_get_lt, _set_lt)
 
     def __getitem__(self, key):
         return self._dict_impl().__getitem__(self, key)[1]
 
     def __setitem__(self, key, val):
         dict_impl = self._dict_impl()
         try:
             dict_impl.__getitem__(self, key)[1] = val
         except KeyError:
             new = [dict_impl.__getattribute__(self, 'lt'), val, _nil]
             dict_impl.__setitem__(self, key, new)
             if dict_impl.__getattribute__(self, 'lt') == _nil:
                 dict_impl.__setattr__(self, 'lh', key)
             else:
                 dict_impl.__getitem__(
                     self, dict_impl.__getattribute__(self, 'lt'))[2] = key
             dict_impl.__setattr__(self, 'lt', key)
 
     def __delitem__(self, key):
         dict_impl = self._dict_impl()
         pred, _, succ = self._dict_impl().__getitem__(self, key)
         if pred == _nil:
             dict_impl.__setattr__(self, 'lh', succ)
         else:
             dict_impl.__getitem__(self, pred)[2] = succ
         if succ == _nil:
             dict_impl.__setattr__(self, 'lt', pred)
         else:
             dict_impl.__getitem__(self, succ)[0] = pred
         dict_impl.__delitem__(self, key)
 
     def __contains__(self, key):
         return key in self.keys()
 
     def __len__(self):
         return len(self.keys())
 
     def __str__(self):
         pairs = ("%r: %r" % (k, v) for k, v in self.iteritems())
         return "{%s}" % ", ".join(pairs)
 
     def __repr__(self):
         if self:
             pairs = ("(%r, %r)" % (k, v) for k, v in self.iteritems())
             return "odict([%s])" % ", ".join(pairs)
         else:
             return "odict()"
 
     def get(self, k, x=None):
         if k in self:
             return self._dict_impl().__getitem__(self, k)[1]
         else:
             return x
 
     def __iter__(self):
         dict_impl = self._dict_impl()
         curr_key = dict_impl.__getattribute__(self, 'lh')
         while curr_key != _nil:
             yield curr_key
             curr_key = dict_impl.__getitem__(self, curr_key)[2]
 
     iterkeys = __iter__
 
     def keys(self):
         return list(self.iterkeys())
 
     def itervalues(self):
         dict_impl = self._dict_impl()
         curr_key = dict_impl.__getattribute__(self, 'lh')
         while curr_key != _nil:
             _, val, curr_key = dict_impl.__getitem__(self, curr_key)
             yield val
 
     def values(self):
         return list(self.itervalues())
 
     def iteritems(self):
         dict_impl = self._dict_impl()
         curr_key = dict_impl.__getattribute__(self, 'lh')
         while curr_key != _nil:
             _, val, next_key = dict_impl.__getitem__(self, curr_key)
             yield curr_key, val
             curr_key = next_key
 
     def items(self):
         return list(self.iteritems())
 
     def sort(self, cmp=None, key=None, reverse=False):
         items = [(k, v) for k, v in self.items()]
         if cmp is not None:
             items = sorted(items, cmp=cmp)
         elif key is not None:
             items = sorted(items, key=key)
         else:
             items = sorted(items, key=lambda x: x[1])
         if reverse:
             items.reverse()
         self.clear()
         self.__init__(items)
 
     def clear(self):
         dict_impl = self._dict_impl()
         dict_impl.clear(self)
         dict_impl.__setattr__(self, 'lh', _nil)
         dict_impl.__setattr__(self, 'lt', _nil)
 
     def copy(self):
         return self.__class__(self)
 
     def update(self, data=(), **kwds):
         if kwds:
             raise TypeError("update() of ordered dict takes no keyword "
                             "arguments to avoid an ordering trap.")
         if hasattr(data, "iteritems"):
             data = data.iteritems()
         for key, val in data:
             self[key] = val
 
     def setdefault(self, k, x=None):
         try:
             return self[k]
         except KeyError:
             self[k] = x
             return x
 
     def pop(self, k, x=_nil):
         try:
             val = self[k]
             del self[k]
             return val
         except KeyError:
             if x == _nil:
                 raise
             return x
 
     def popitem(self):
         try:
             dict_impl = self._dict_impl()
             key = dict_impl.__getattribute__(self, 'lt')
             return key, self.pop(key)
         except KeyError:
             raise KeyError("'popitem(): ordered dictionary is empty'")
 
     def riterkeys(self):
         """To iterate on keys in reversed order.
         """
         dict_impl = self._dict_impl()
         curr_key = dict_impl.__getattribute__(self, 'lt')
         while curr_key != _nil:
             yield curr_key
             curr_key = dict_impl.__getitem__(self, curr_key)[0]
 
     __reversed__ = riterkeys
 
     def rkeys(self):
         """List of the keys in reversed order.
         """
         return list(self.riterkeys())
 
     def ritervalues(self):
         """To iterate on values in reversed order.
         """
         dict_impl = self._dict_impl()
         curr_key = dict_impl.__getattribute__(self, 'lt')
         while curr_key != _nil:
             curr_key, val, _ = dict_impl.__getitem__(self, curr_key)
             yield val
 
     def rvalues(self):
         """List of the values in reversed order.
         """
         return list(self.ritervalues())
 
     def riteritems(self):
         """To iterate on (key, value) in reversed order.
         """
         dict_impl = self._dict_impl()
         curr_key = dict_impl.__getattribute__(self, 'lt')
         while curr_key != _nil:
             pred_key, val, _ = dict_impl.__getitem__(self, curr_key)
             yield curr_key, val
             curr_key = pred_key
 
     def ritems(self):
         """List of the (key, value) in reversed order.
         """
         return list(self.riteritems())
 
     def firstkey(self):
         if self:
             return self._dict_impl().__getattribute__(self, 'lh')
         else:
             raise KeyError("'firstkey(): ordered dictionary is empty'")
 
     def lastkey(self):
         if self:
             return self._dict_impl().__getattribute__(self, 'lt')
         else:
             raise KeyError("'lastkey(): ordered dictionary is empty'")
 
     def as_dict(self):
         return self._dict_impl()(self.items())
 
     def _repr(self):
         """_repr(): low level repr of the whole data contained in the odict.
         Useful for debugging.
         """
         dict_impl = self._dict_impl()
         form = "odict low level repr lh,lt,data: %r, %r, %s"
         return form % (dict_impl.__getattribute__(self, 'lh'),
                        dict_impl.__getattribute__(self, 'lt'),
                        dict_impl.__repr__(self))
 
 
 class OrderedDict(_odict, dict):
 
     def _dict_impl(self):
         return dict
 
 
 #==============================================================================
 # OrderedSet
 #==============================================================================
 from sqlalchemy.util import OrderedSet
 
 
 #==============================================================================
 # kill FUNCTIONS
 #==============================================================================
 if __platform__ in PLATFORM_WIN:
     import ctypes
 
     def kill(pid, sig):
         """kill function for Win32"""
         kernel32 = ctypes.windll.kernel32
         handle = kernel32.OpenProcess(1, 0, pid)
         return (0 != kernel32.TerminateProcess(handle, 0))
 
 else:
     kill = os.kill
 
 
 #==============================================================================
 # itertools.product
 #==============================================================================
 
 try:
     from itertools import product
 except ImportError:
     def product(*args, **kwds):
         # product('ABCD', 'xy') --> Ax Ay Bx By Cx Cy Dx Dy
         # product(range(2), repeat=3) --> 000 001 010 011 100 101 110 111
         pools = map(tuple, args) * kwds.get('repeat', 1)
         result = [[]]
         for pool in pools:
             result = [x + [y] for x in result for y in pool]
         for prod in result:
             yield tuple(prod)
 
 
 #==============================================================================
 # BytesIO
 #==============================================================================
 
 try:
     from io import BytesIO
 except ImportError:
     from cStringIO import StringIO as BytesIO
 
 
 #==============================================================================
+# bytes
+#==============================================================================
+if __py_version__ >= (2, 6):
+    _bytes = bytes
+else:
+    # in py2.6 bytes is a synonim for str
+    _bytes = str
+
+#==============================================================================
 # deque
 #==============================================================================
 
 if __py_version__ >= (2, 6):
     from collections import deque
 else:
     #need to implement our own deque with maxlen
     class deque(object):
 
-        def __init__(self, iterable=(), maxlen=-1):
+        def __init__(self, iterable=(), maxlen= -1):
             if not hasattr(self, 'data'):
                 self.left = self.right = 0
                 self.data = {}
-            self.maxlen = maxlen
+            self.maxlen = maxlen or -1
             self.extend(iterable)
 
         def append(self, x):
             self.data[self.right] = x
             self.right += 1
             if self.maxlen != -1 and len(self) > self.maxlen:
                 self.popleft()
 
         def appendleft(self, x):
             self.left -= 1
             self.data[self.left] = x
             if self.maxlen != -1 and len(self) > self.maxlen:
                 self.pop()
 
         def pop(self):
             if self.left == self.right:
                 raise IndexError('cannot pop from empty deque')
             self.right -= 1
             elem = self.data[self.right]
             del self.data[self.right]
             return elem
 
         def popleft(self):
             if self.left == self.right:
                 raise IndexError('cannot pop from empty deque')
             elem = self.data[self.left]
             del self.data[self.left]
             self.left += 1
             return elem
 
         def clear(self):
             self.data.clear()
             self.left = self.right = 0
 
         def extend(self, iterable):
             for elem in iterable:
                 self.append(elem)
 
         def extendleft(self, iterable):
             for elem in iterable:
                 self.appendleft(elem)
 
         def rotate(self, n=1):
             if self:
                 n %= len(self)
                 for i in xrange(n):
                     self.appendleft(self.pop())
 
         def __getitem__(self, i):
             if i < 0:
                 i += len(self)
             try:
                 return self.data[i + self.left]
             except KeyError:
                 raise IndexError
 
         def __setitem__(self, i, value):
             if i < 0:
                 i += len(self)
             try:
                 self.data[i + self.left] = value
             except KeyError:
                 raise IndexError
 
         def __delitem__(self, i):
             size = len(self)
             if not (-size <= i < size):
                 raise IndexError
             data = self.data
             if i < 0:
                 i += size
             for j in xrange(self.left + i, self.right - 1):
                 data[j] = data[j + 1]
             self.pop()
 
         def __len__(self):
             return self.right - self.left
 
         def __cmp__(self, other):
             if type(self) != type(other):
                 return cmp(type(self), type(other))
             return cmp(list(self), list(other))
 
         def __repr__(self, _track=[]):
             if id(self) in _track:
                 return '...'
             _track.append(id(self))
             r = 'deque(%r, maxlen=%s)' % (list(self), self.maxlen)
             _track.remove(id(self))
             return r
 
         def __getstate__(self):
             return (tuple(self),)
 
         def __setstate__(self, s):
             self.__init__(s[0])
 
         def __hash__(self):
             raise TypeError
 
         def __copy__(self):
             return self.__class__(self)
 
         def __deepcopy__(self, memo={}):
             from copy import deepcopy
             result = self.__class__()
             memo[id(self)] = result
             result.__init__(deepcopy(tuple(self), memo))
             return result
 
 
 #==============================================================================
 # threading.Event
 #==============================================================================
 
 if __py_version__ >= (2, 6):
-    from threading import Event
+    from threading import Event, Thread
 else:
-    from threading import _Verbose, Condition, Lock
+    from threading import _Verbose, Condition, Lock, Thread
 
     def Event(*args, **kwargs):
         return _Event(*args, **kwargs)
 
     class _Event(_Verbose):
 
         # After Tim Peters' event class (without is_posted())
 
         def __init__(self, verbose=None):
             _Verbose.__init__(self, verbose)
             self.__cond = Condition(Lock())
             self.__flag = False
 
         def isSet(self):
             return self.__flag
 
         is_set = isSet
 
         def set(self):
             self.__cond.acquire()
             try:
                 self.__flag = True
                 self.__cond.notify_all()
             finally:
                 self.__cond.release()
 
         def clear(self):
             self.__cond.acquire()
             try:
                 self.__flag = False
             finally:
                 self.__cond.release()
 
         def wait(self, timeout=None):
             self.__cond.acquire()
             try:
                 if not self.__flag:
                     self.__cond.wait(timeout)
             finally:
                 self.__cond.release()
 
 
 

rhodecode/lib/subprocessio.py

@@ -1,409 +1,409 b''
 '''
 Module provides a class allowing to wrap communication over subprocess.Popen
 input, output, error streams into a meaningfull, non-blocking, concurrent
 stream processor exposing the output data as an iterator fitting to be a
 return value passed by a WSGI applicaiton to a WSGI server per PEP 3333.
 
 Copyright (c) 2011  Daniel Dotsenko <[e-mail]>
 
 This file is part of git_http_backend.py Project.
 
 git_http_backend.py Project is free software: you can redistribute it and/or
 modify it under the terms of the GNU Lesser General Public License as
 published by the Free Software Foundation, either version 2.1 of the License,
 or (at your option) any later version.
 
 git_http_backend.py Project is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 GNU Lesser General Public License for more details.
 
 You should have received a copy of the GNU Lesser General Public License
 along with git_http_backend.py Project.
 If not, see <http://www.gnu.org/licenses/>.
 '''
 import os
 import subprocess
-import threading
-from rhodecode.lib.compat import deque, Event
+from rhodecode.lib.compat import deque, Event, Thread, _bytes
 
 
-class StreamFeeder(threading.Thread):
+class StreamFeeder(Thread):
     """
     Normal writing into pipe-like is blocking once the buffer is filled.
     This thread allows a thread to seep data from a file-like into a pipe
     without blocking the main thread.
     We close inpipe once the end of the source stream is reached.
     """
     def __init__(self, source):
         super(StreamFeeder, self).__init__()
         self.daemon = True
         filelike = False
-        self.bytes = bytes()
-        if type(source) in (type(''), bytes, bytearray):  # string-like
-            self.bytes = bytes(source)
+        self.bytes = _bytes()
+        if type(source) in (type(''), _bytes, bytearray):  # string-like
+            self.bytes = _bytes(source)
         else:  # can be either file pointer or file-like
             if type(source) in (int, long):  # file pointer it is
                 ## converting file descriptor (int) stdin into file-like
                 try:
                     source = os.fdopen(source, 'rb', 16384)
                 except Exception:
                     pass
             # let's see if source is file-like by now
             try:
                 filelike = source.read
             except Exception:
                 pass
         if not filelike and not self.bytes:
             raise TypeError("StreamFeeder's source object must be a readable "
                             "file-like, a file descriptor, or a string-like.")
         self.source = source
         self.readiface, self.writeiface = os.pipe()
 
     def run(self):
         t = self.writeiface
         if self.bytes:
             os.write(t, self.bytes)
         else:
             s = self.source
             b = s.read(4096)
             while b:
                 os.write(t, b)
                 b = s.read(4096)
         os.close(t)
 
     @property
     def output(self):
         return self.readiface
 
 
-class InputStreamChunker(threading.Thread):
+class InputStreamChunker(Thread):
     def __init__(self, source, target, buffer_size, chunk_size):
 
         super(InputStreamChunker, self).__init__()
 
         self.daemon = True  # die die die.
 
         self.source = source
         self.target = target
         self.chunk_count_max = int(buffer_size / chunk_size) + 1
         self.chunk_size = chunk_size
 
         self.data_added = Event()
         self.data_added.clear()
 
         self.keep_reading = Event()
         self.keep_reading.set()
 
         self.EOF = Event()
         self.EOF.clear()
 
         self.go = Event()
         self.go.set()
 
     def stop(self):
         self.go.clear()
         self.EOF.set()
         try:
             # this is not proper, but is done to force the reader thread let
             # go of the input because, if successful, .close() will send EOF
             # down the pipe.
             self.source.close()
         except:
             pass
 
     def run(self):
         s = self.source
         t = self.target
         cs = self.chunk_size
         ccm = self.chunk_count_max
         kr = self.keep_reading
         da = self.data_added
         go = self.go
         b = s.read(cs)
+
         while b and go.is_set():
             if len(t) > ccm:
                 kr.clear()
                 kr.wait(2)
 #                # this only works on 2.7.x and up
 #                if not kr.wait(10):
 #                    raise Exception("Timed out while waiting for input to be read.")
                 # instead we'll use this
                 if len(t) > ccm + 3:
                     raise IOError("Timed out while waiting for input from subprocess.")
             t.append(b)
             da.set()
             b = s.read(cs)
         self.EOF.set()
         da.set()  # for cases when done but there was no input.
 
 
 class BufferedGenerator():
     '''
     Class behaves as a non-blocking, buffered pipe reader.
     Reads chunks of data (through a thread)
     from a blocking pipe, and attaches these to an array (Deque) of chunks.
     Reading is halted in the thread when max chunks is internally buffered.
     The .next() may operate in blocking or non-blocking fashion by yielding
     '' if no data is ready
     to be sent or by not returning until there is some data to send
     When we get EOF from underlying source pipe we raise the marker to raise
     StopIteration after the last chunk of data is yielded.
     '''
 
     def __init__(self, source, buffer_size=65536, chunk_size=4096,
                  starting_values=[], bottomless=False):
 
         if bottomless:
             maxlen = int(buffer_size / chunk_size)
         else:
             maxlen = None
 
         self.data = deque(starting_values, maxlen)
 
         self.worker = InputStreamChunker(source, self.data, buffer_size,
                                          chunk_size)
         if starting_values:
             self.worker.data_added.set()
         self.worker.start()
 
     ####################
     # Generator's methods
     ####################
 
     def __iter__(self):
         return self
 
     def next(self):
         while not len(self.data) and not self.worker.EOF.is_set():
             self.worker.data_added.clear()
             self.worker.data_added.wait(0.2)
         if len(self.data):
             self.worker.keep_reading.set()
-            return bytes(self.data.popleft())
+            return _bytes(self.data.popleft())
         elif self.worker.EOF.is_set():
             raise StopIteration
 
     def throw(self, type, value=None, traceback=None):
         if not self.worker.EOF.is_set():
             raise type(value)
 
     def start(self):
         self.worker.start()
 
     def stop(self):
         self.worker.stop()
 
     def close(self):
         try:
             self.worker.stop()
             self.throw(GeneratorExit)
         except (GeneratorExit, StopIteration):
             pass
 
     def __del__(self):
         self.close()
 
     ####################
     # Threaded reader's infrastructure.
     ####################
     @property
     def input(self):
         return self.worker.w
 
     @property
     def data_added_event(self):
         return self.worker.data_added
 
     @property
     def data_added(self):
         return self.worker.data_added.is_set()
 
     @property
     def reading_paused(self):
         return not self.worker.keep_reading.is_set()
 
     @property
     def done_reading_event(self):
         '''
         Done_reding does not mean that the iterator's buffer is empty.
         Iterator might have done reading from underlying source, but the read
         chunks might still be available for serving through .next() method.
 
         @return An Event class instance.
         '''
         return self.worker.EOF
 
     @property
     def done_reading(self):
         '''
         Done_reding does not mean that the iterator's buffer is empty.
         Iterator might have done reading from underlying source, but the read
         chunks might still be available for serving through .next() method.
 
         @return An Bool value.
         '''
         return self.worker.EOF.is_set()
 
     @property
     def length(self):
         '''
         returns int.
 
         This is the lenght of the que of chunks, not the length of
         the combined contents in those chunks.
 
         __len__() cannot be meaningfully implemented because this
         reader is just flying throuh a bottomless pit content and
         can only know the lenght of what it already saw.
 
         If __len__() on WSGI server per PEP 3333 returns a value,
         the responce's length will be set to that. In order not to
         confuse WSGI PEP3333 servers, we will not implement __len__
         at all.
         '''
         return len(self.data)
 
     def prepend(self, x):
         self.data.appendleft(x)
 
     def append(self, x):
         self.data.append(x)
 
     def extend(self, o):
         self.data.extend(o)
 
     def __getitem__(self, i):
         return self.data[i]
 
 
 class SubprocessIOChunker(object):
     '''
     Processor class wrapping handling of subprocess IO.
 
     In a way, this is a "communicate()" replacement with a twist.
 
     - We are multithreaded. Writing in and reading out, err are all sep threads.
     - We support concurrent (in and out) stream processing.
     - The output is not a stream. It's a queue of read string (bytes, not unicode)
       chunks. The object behaves as an iterable. You can "for chunk in obj:" us.
     - We are non-blocking in more respects than communicate()
       (reading from subprocess out pauses when internal buffer is full, but
        does not block the parent calling code. On the flip side, reading from
        slow-yielding subprocess may block the iteration until data shows up. This
        does not block the parallel inpipe reading occurring parallel thread.)
 
     The purpose of the object is to allow us to wrap subprocess interactions into
     and interable that can be passed to a WSGI server as the application's return
     value. Because of stream-processing-ability, WSGI does not have to read ALL
     of the subprocess's output and buffer it, before handing it to WSGI server for
     HTTP response. Instead, the class initializer reads just a bit of the stream
     to figure out if error ocurred or likely to occur and if not, just hands the
     further iteration over subprocess output to the server for completion of HTTP
     response.
 
     The real or perceived subprocess error is trapped and raised as one of
     EnvironmentError family of exceptions
 
     Example usage:
     #    try:
     #        answer = SubprocessIOChunker(
     #            cmd,
     #            input,
     #            buffer_size = 65536,
     #            chunk_size = 4096
     #            )
     #    except (EnvironmentError) as e:
     #        print str(e)
     #        raise e
     #
     #    return answer
 
 
     '''
     def __init__(self, cmd, inputstream=None, buffer_size=65536,
                  chunk_size=4096, starting_values=[], **kwargs):
         '''
         Initializes SubprocessIOChunker
 
         :param cmd: A Subprocess.Popen style "cmd". Can be string or array of strings
         :param inputstream: (Default: None) A file-like, string, or file pointer.
         :param buffer_size: (Default: 65536) A size of total buffer per stream in bytes.
         :param chunk_size: (Default: 4096) A max size of a chunk. Actual chunk may be smaller.
         :param starting_values: (Default: []) An array of strings to put in front of output que.
         '''
 
         if inputstream:
             input_streamer = StreamFeeder(inputstream)
             input_streamer.start()
             inputstream = input_streamer.output
 
         if isinstance(cmd, (list, tuple)):
             cmd = ' '.join(cmd)
 
         _shell = kwargs.get('shell') or True
         kwargs['shell'] = _shell
         _p = subprocess.Popen(cmd,
             bufsize=-1,
             stdin=inputstream,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             **kwargs
             )
 
         bg_out = BufferedGenerator(_p.stdout, buffer_size, chunk_size, starting_values)
         bg_err = BufferedGenerator(_p.stderr, 16000, 1, bottomless=True)
 
         while not bg_out.done_reading and not bg_out.reading_paused and not bg_err.length:
             # doing this until we reach either end of file, or end of buffer.
             bg_out.data_added_event.wait(1)
             bg_out.data_added_event.clear()
 
         # at this point it's still ambiguous if we are done reading or just full buffer.
         # Either way, if error (returned by ended process, or implied based on
         # presence of stuff in stderr output) we error out.
         # Else, we are happy.
         _returncode = _p.poll()
         if _returncode or (_returncode == None and bg_err.length):
             try:
                 _p.terminate()
             except:
                 pass
             bg_out.stop()
             bg_err.stop()
             err = '%s' % ''.join(bg_err)
             raise EnvironmentError("Subprocess exited due to an error:\n" + err)
 
         self.process = _p
         self.output = bg_out
         self.error = bg_err
 
     def __iter__(self):
         return self
 
     def next(self):
         if self.process.poll():
             err = '%s' % ''.join(self.error)
             raise EnvironmentError("Subprocess exited due to an error:\n" + err)
         return self.output.next()
 
     def throw(self, type, value=None, traceback=None):
         if self.output.length or not self.output.done_reading:
             raise type(value)
 
     def close(self):
         try:
             self.process.terminate()
         except:
             pass
         try:
             self.output.close()
         except:
             pass
         try:
             self.error.close()
         except:
             pass
 
     def __del__(self):
         self.close()

rhodecode/tests/test_libs.py

@@ -1,177 +1,177 b''
 # -*- coding: utf-8 -*-
 """
     rhodecode.tests.test_libs
     ~~~~~~~~~~~~~~~~~~~~~~~~~
 
 
     Package for testing various lib/helper functions in rhodecode
 
     :created_on: Jun 9, 2011
     :copyright: (C) 2011-2012 Marcin Kuzminski <[e-mail]>
     :license: GPLv3, see COPYING for more details.
 """
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
+from __future__ import with_statement
 import unittest
 import datetime
 import hashlib
 import mock
 from rhodecode.tests import *
 
 proto = 'http'
 TEST_URLS = [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
      '%s://domain.org' % proto),
     ('%s://user:[e-mail]:8080' % proto, ['%s://' % proto, 'domain.org',
                                                 '8080'],
      '%s://domain.org:8080' % proto),
 ]
 
 proto = 'https'
 TEST_URLS += [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
      '%s://domain.org' % proto),
     ('%s://user:[e-mail]:8080' % proto, ['%s://' % proto, 'domain.org',
                                                 '8080'],
      '%s://domain.org:8080' % proto),
 ]
 
 
 class TestLibs(unittest.TestCase):
 
     def test_uri_filter(self):
         from rhodecode.lib.utils2 import uri_filter
 
         for url in TEST_URLS:
             self.assertEqual(uri_filter(url[0]), url[1])
 
     def test_credentials_filter(self):
         from rhodecode.lib.utils2 import credentials_filter
 
         for url in TEST_URLS:
             self.assertEqual(credentials_filter(url[0]), url[2])
 
     def test_str2bool(self):
         from rhodecode.lib.utils2 import str2bool
         test_cases = [
             ('t', True),
             ('true', True),
             ('y', True),
             ('yes', True),
             ('on', True),
             ('1', True),
             ('Y', True),
             ('yeS', True),
             ('Y', True),
             ('TRUE', True),
             ('T', True),
             ('False', False),
             ('F', False),
             ('FALSE', False),
             ('0', False),
             ('-1', False),
             ('', False), ]
 
         for case in test_cases:
             self.assertEqual(str2bool(case[0]), case[1])
 
     def test_mention_extractor(self):
         from rhodecode.lib.utils2 import extract_mentioned_users
         sample = (
             "@first hi there @marcink here's my email [e-mail] "
             "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
             "@MARCIN    @maRCiN @2one_more22 @john please see this http://org.pl "
             "@marian.user just do it @marco-polo and next extract @marco_polo "
             "user.dot  hej ! not-needed [e-mail]"
         )
 
         s = sorted([
         'first', 'marcink', 'lukaszb', 'one_more22', 'MARCIN', 'maRCiN', 'john',
         'marian.user', 'marco-polo', 'marco_polo'
         ], key=lambda k: k.lower())
         self.assertEqual(s, extract_mentioned_users(sample))
 
     def test_age(self):
         import calendar
         from rhodecode.lib.utils2 import age
         n = datetime.datetime.now()
         delt = lambda *args, **kwargs: datetime.timedelta(*args, **kwargs)
         self.assertEqual(age(n), u'just now')
         self.assertEqual(age(n - delt(seconds=1)), u'1 second ago')
         self.assertEqual(age(n - delt(seconds=60 * 2)), u'2 minutes ago')
         self.assertEqual(age(n - delt(hours=1)), u'1 hour ago')
         self.assertEqual(age(n - delt(hours=24)), u'1 day ago')
         self.assertEqual(age(n - delt(hours=24 * 5)), u'5 days ago')
         self.assertEqual(age(n - delt(hours=24 * (calendar.mdays[n.month-1] + 2))),
                          u'1 month and 2 days ago')
         self.assertEqual(age(n - delt(hours=24 * 400)), u'1 year and 1 month ago')
 
     def test_tag_exctrator(self):
         sample = (
             "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
             "[requires] [stale] [see<>=>] [see => http://url.com]"
             "[requires => url] [lang => python] [just a tag]"
             "[,d] [ => ULR ] [obsolete] [desc]]"
         )
         from rhodecode.lib.helpers import desc_stylize
         res = desc_stylize(sample)
         self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)
         self.assertTrue('<div class="metatag" tag="obsolete">obsolete</div>' in res)
         self.assertTrue('<div class="metatag" tag="stale">stale</div>' in res)
         self.assertTrue('<div class="metatag" tag="lang">python</div>' in res)
         self.assertTrue('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>' in res)
         self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)
 
     def test_alternative_gravatar(self):
         from rhodecode.lib.helpers import gravatar_url
         _md5 = lambda s: hashlib.md5(s).hexdigest()
 
         def fake_conf(**kwargs):
             from pylons import config
             config['app_conf'] = {}
             config['app_conf']['use_gravatar'] = True
             config['app_conf'].update(kwargs)
             return config
         fake = fake_conf(alternative_gravatar_url='http://test.com/{email}')
         with mock.patch('pylons.config', fake):
             grav = gravatar_url(email_address='[e-mail]', size=24)
             assert grav == 'http://test.com/[e-mail]'
 
         fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}')
         with mock.patch('pylons.config', fake):
             em = '[e-mail]'
             grav = gravatar_url(email_address=em, size=24)
             assert grav == 'http://test.com/%s' % (_md5(em))
 
         fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}/{size}')
         with mock.patch('pylons.config', fake):
             em = '[e-mail]'
             grav = gravatar_url(email_address=em, size=24)
             assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)