Merge pull request #1582 from gevent/combine-tests

Group some tests in the same process

docs/changes/1580.bugfix

+Fix destroying the libuv default loop and then using the default loop
+again.

docs/changes/1581.bugfix

+libuv loops that have watched children can now exit. Previously, the
+SIGCHLD watcher kept the loop alive even if there were no longer any
+watched children.

src/gevent/_ffi/loop.py

@@ -30,6 +30,12 @@ class _EVENTSType(object):
 EVENTS = GEVENT_CORE_EVENTS = _EVENTSType()
 
 
+class _DiscardedSet(frozenset):
+    __slots__ = ()
+
+    def discard(self, o):
+        "Does nothing."
+
 #####
 ## Note on CFFI objects, callbacks and the lifecycle of watcher objects
 #
@@ -383,6 +389,8 @@ class AbstractLoop(object):
     # whether they were the default loop.
     _default = None
 
+    _keepaliveset = _DiscardedSet()
+
     def __init__(self, ffi, lib, watchers, flags=None, default=None):
         self._ffi = ffi
         self._lib = lib
@@ -397,10 +405,8 @@ class AbstractLoop(object):
 
 
     def _init_loop_and_aux_watchers(self, flags=None, default=None):
-
         self._ptr = self._init_loop(flags, default)
 
-
         # self._check is a watcher that runs in each iteration of the
         # mainloop, just after the blocking call. It's point is to handle
         # signals. It doesn't run watchers or callbacks, it just exists to give
@@ -541,12 +547,13 @@ class AbstractLoop(object):
         raise NotImplementedError()
 
     def destroy(self):
-        if self._ptr:
+        ptr = self.ptr
+        if ptr:
             try:
-                if not self._can_destroy_loop(self._ptr):
+                if not self._can_destroy_loop(ptr):
                     return False
                 self._stop_aux_watchers()
-                self._destroy_loop(self._ptr)
+                self._destroy_loop(ptr)
             finally:
                 # not ffi.NULL, we don't want something that can be
                 # passed to C and crash later. This will create nice friendly
@@ -566,6 +573,7 @@ class AbstractLoop(object):
 
     @property
     def ptr(self):
+        # Use this when you need to be sure the pointer is valid.
         return self._ptr
 
     @property
@@ -650,7 +658,7 @@ class AbstractLoop(object):
 
     @property
     def default(self):
-        return self._default if self._ptr else False
+        return self._default if self.ptr else False
 
     @property
     def iteration(self):
@@ -730,9 +738,11 @@ class AbstractLoop(object):
         return cb
 
     def _format(self):
-        if not self._ptr:
+        ptr = self.ptr
+        if not ptr:
             return 'destroyed'
-        msg = self.backend
+        msg = "backend=" + self.backend
+        msg += ' ptr=' + str(ptr)
         if self.default:
             msg += ' default'
         msg += ' pending=%s' % self.pendingcnt
@@ -759,6 +769,6 @@ class AbstractLoop(object):
 
     @property
     def activecnt(self):
-        if not self._ptr:
+        if not self.ptr:
             raise ValueError('operation on destroyed loop')
         return 0

src/gevent/_ffi/watcher.py

@@ -268,7 +268,7 @@ class watcher(object):
         raise NotImplementedError()
 
     def _watcher_ffi_stop(self):
-        self._watcher_stop(self.loop._ptr, self._watcher)
+        self._watcher_stop(self.loop.ptr, self._watcher)
 
     def _watcher_ffi_ref(self):
         raise NotImplementedError()
@@ -345,25 +345,36 @@ class watcher(object):
                 # may fail if __init__ did; will be harmlessly printed
                 self.close()
 
+    __in_repr = False
+
     def __repr__(self):
-        formats = self._format()
-        result = "<%s at 0x%x%s" % (self.__class__.__name__, id(self), formats)
-        if self.pending:
-            result += " pending"
-        if self.callback is not None:
-            fself = getattr(self.callback, '__self__', None)
-            if fself is self:
-                result += " callback=<bound method %s of self>" % (self.callback.__name__)
-            else:
-                result += " callback=%r" % (self.callback, )
-        if self.args is not None:
-            result += " args=%r" % (self.args, )
-        if self.callback is None and self.args is None:
-            result += " stopped"
-        result += " watcher=%s" % (self._watcher)
-        result += " handle=%s" % (self._watcher_handle)
-        result += " ref=%s" % (self.ref)
-        return result + ">"
+        basic = "<%s at 0x%x" % (self.__class__.__name__, id(self))
+        if self.__in_repr:
+            return basic + '>'
+        # Running child watchers have been seen to have a
+        # recursive repr in ``self.args``, thanks to ``gevent.os.fork_and_watch``
+        # passing the watcher as an argument to its callback.
+        self.__in_repr = True
+        try:
+            result = '%s%s' % (basic, self._format())
+            if self.pending:
+                result += " pending"
+            if self.callback is not None:
+                fself = getattr(self.callback, '__self__', None)
+                if fself is self:
+                    result += " callback=<bound method %s of self>" % (self.callback.__name__)
+                else:
+                    result += " callback=%r" % (self.callback, )
+            if self.args is not None:
+                result += " args=%r" % (self.args, )
+            if self.callback is None and self.args is None:
+                result += " stopped"
+            result += " watcher=%s" % (self._watcher)
+            result += " handle=%s" % (self._watcher_handle)
+            result += " ref=%s" % (self.ref)
+            return result + ">"
+        finally:
+            self.__in_repr = False
 
     @property
     def _watcher_handle(self):

src/gevent/_queue.pxd

 cimport cython
 from gevent.__waiter cimport Waiter
 from gevent._event cimport Event
+from gevent.__hub_local cimport get_hub_noargs as get_hub
 
+cdef bint _greenlet_imported
 cdef _heappush
 cdef _heappop
 cdef _heapify
+cdef _Empty
+cdef _Full
+cdef Timeout
+cdef InvalidSwitchError
+
+cdef extern from "greenlet/greenlet.h":
+
+    ctypedef class greenlet.greenlet [object PyGreenlet]:
+        pass
+
+    # These are actually macros and so much be included
+    # (defined) in each .pxd, as are the two functions
+    # that call them.
+    greenlet PyGreenlet_GetCurrent()
+    void PyGreenlet_Import()
+
+cdef inline greenlet getcurrent():
+    return PyGreenlet_GetCurrent()
+
+cdef inline void greenlet_init():
+    global _greenlet_imported
+    if not _greenlet_imported:
+        PyGreenlet_Import()
+        _greenlet_imported = True
 
-@cython.final
-cdef _safe_remove(deq, item)
 
 @cython.final
-@cython.internal
-cdef class ItemWaiter(Waiter):
-    cdef readonly item
-    cdef readonly queue
+cdef _safe_remove(deq, item)
 
 cdef class Queue:
     cdef __weakref__
@@ -33,6 +54,7 @@ cdef class Queue:
     cpdef Py_ssize_t qsize(self)
     cpdef bint empty(self)
     cpdef bint full(self)
+    cpdef _create_queue(self, items=*)
 
     cpdef put(self, item, block=*, timeout=*)
     cpdef put_nowait(self, item)
@@ -46,6 +68,13 @@ cdef class Queue:
 
     cdef _schedule_unlock(self)
 
+@cython.final
+@cython.internal
+cdef class ItemWaiter(Waiter):
+    cdef readonly item
+    cdef readonly Queue queue
+
+
 @cython.final
 cdef class UnboundQueue(Queue):
     pass

src/gevent/libuv/_corecffi_source.c

@@ -100,14 +100,15 @@ static void _gevent_fs_poll_callback3(void* handlep, int status, const uv_stat_t
 static void gevent_uv_walk_callback_close(uv_handle_t* handle, void* arg)
 {
     if( handle && !uv_is_closing(handle) ) {
-	uv_close(handle, NULL);
+        uv_close(handle, NULL);
+        handle->data = NULL;
     }
 }
 
 static void gevent_close_all_handles(uv_loop_t* loop)
 {
     if (loop) {
-	uv_walk(loop, gevent_uv_walk_callback_close, NULL);
+        uv_walk(loop, gevent_uv_walk_callback_close, NULL);
     }
 }
 
@@ -155,7 +156,7 @@ static void* _gevent_uv_calloc(size_t count, size_t size)
     void* result;
     result = _gevent_uv_malloc(count * size);
     if(result) {
-	memset(result, 0, count * size);
+        memset(result, 0, count * size);
     }
     return result;
 }
@@ -163,9 +164,9 @@ static void* _gevent_uv_calloc(size_t count, size_t size)
 static void gevent_set_uv_alloc()
 {
     uv_replace_allocator(_gevent_uv_malloc,
-			 _gevent_uv_realloc,
-			 _gevent_uv_calloc,
-			 _gevent_uv_free);
+                         _gevent_uv_realloc,
+                         _gevent_uv_calloc,
+                         _gevent_uv_free);
 }
 
 #ifdef __clang__

src/gevent/libuv/loop.py

@@ -117,7 +117,7 @@ class loop(AbstractLoop):
         self._io_watchers = dict()
         self._fork_watchers = set()
         self._pid = os.getpid()
-        self._default = self._ptr == libuv.uv_default_loop()
+        self._default = (self._ptr == libuv.uv_default_loop())
         self._queued_callbacks = []
 
     def _queue_callback(self, watcher_ptr, revents):
@@ -148,8 +148,18 @@ class loop(AbstractLoop):
 
     _signal_idle = None
 
+    @property
+    def ptr(self):
+        if not self._ptr:
+            return None
+        if self._ptr and not self._ptr.data:
+            # Another instance of the Python loop destroyed
+            # the C loop. It was probably the default.
+            self._ptr = None
+        return self._ptr
+
     def _init_and_start_check(self):
-        libuv.uv_check_init(self._ptr, self._check)
+        libuv.uv_check_init(self.ptr, self._check)
         libuv.uv_check_start(self._check, libuv.python_check_callback)
         libuv.uv_unref(self._check)
 
@@ -167,7 +177,7 @@ class loop(AbstractLoop):
         # scheduled, this should also be the same and unnecessary?
         # libev does takes this basic approach on Windows.
         self._signal_idle = ffi.new("uv_timer_t*")
-        libuv.uv_timer_init(self._ptr, self._signal_idle)
+        libuv.uv_timer_init(self.ptr, self._signal_idle)
         self._signal_idle.data = self._handle_to_self
         sig_cb = ffi.cast('void(*)(uv_timer_t*)', libuv.python_check_callback)
         libuv.uv_timer_start(self._signal_idle,
@@ -210,12 +220,12 @@ class loop(AbstractLoop):
         super(loop, self)._run_callbacks()
 
     def _init_and_start_prepare(self):
-        libuv.uv_prepare_init(self._ptr, self._prepare)
+        libuv.uv_prepare_init(self.ptr, self._prepare)
         libuv.uv_prepare_start(self._prepare, libuv.python_prepare_callback)
         libuv.uv_unref(self._prepare)
 
     def _init_callback_timer(self):
-        libuv.uv_check_init(self._ptr, self._timer0)
+        libuv.uv_check_init(self.ptr, self._timer0)
 
     def _stop_callback_timer(self):
         libuv.uv_check_stop(self._timer0)
@@ -328,51 +338,64 @@ class loop(AbstractLoop):
         self._start_callback_timer()
         libuv.uv_ref(self._timer0)
 
-
     def _can_destroy_loop(self, ptr):
-        # We're being asked to destroy a loop that's,
-        # at the time it was constructed, was the default loop.
-        # If loop objects were constructed more than once,
-        # it may have already been destroyed, though.
-        # We track this in the data member.
-        return ptr.data
+        return ptr
 
-    def _destroy_loop(self, ptr):
-        ptr.data = ffi.NULL
-        libuv.uv_stop(ptr)
+    def __close_loop(self, ptr):
+        closed_failed = 1
 
-        libuv.gevent_close_all_handles(ptr)
+        while closed_failed:
+            closed_failed = libuv.uv_loop_close(ptr)
+            if not closed_failed:
+                break
 
-        closed_failed = libuv.uv_loop_close(ptr)
-        if closed_failed:
-            assert closed_failed == libuv.UV_EBUSY
+            if closed_failed != libuv.UV_EBUSY:
+                raise SystemError("Unknown close failure reason", closed_failed)
             # We already closed all the handles. Run the loop
             # once to let them be cut off from the loop.
             ran_has_more_callbacks = libuv.uv_run(ptr, libuv.UV_RUN_ONCE)
             if ran_has_more_callbacks:
                 libuv.uv_run(ptr, libuv.UV_RUN_NOWAIT)
-            closed_failed = libuv.uv_loop_close(ptr)
-            assert closed_failed == 0, closed_failed
 
-        # Destroy the native resources *after* we have closed
-        # the loop. If we do it before, walking the handles
-        # attached to the loop is likely to segfault.
 
-        libuv.gevent_zero_check(self._check)
-        libuv.gevent_zero_check(self._timer0)
-        libuv.gevent_zero_prepare(self._prepare)
-        libuv.gevent_zero_timer(self._signal_idle)
-        del self._check
-        del self._prepare
-        del self._signal_idle
-        del self._timer0
+    def _destroy_loop(self, ptr):
+        # We're being asked to destroy a loop that's, potentially, at
+        # the time it was constructed, was the default loop. If loop
+        # objects were constructed more than once, it may have already
+        # been destroyed, though. We track this in the data member.
+        data = ptr.data
+        ptr.data = ffi.NULL
+        try:
+            if data:
+                libuv.uv_stop(ptr)
+                libuv.gevent_close_all_handles(ptr)
+        finally:
+            ptr.data = ffi.NULL
+
+        try:
+            if data:
+                self.__close_loop(ptr)
+        finally:
+            # Destroy the native resources *after* we have closed
+            # the loop. If we do it before, walking the handles
+            # attached to the loop is likely to segfault.
+            # Note that these may have been closed already if the default loop was shared.
+            if data:
+                libuv.gevent_zero_check(self._check)
+                libuv.gevent_zero_check(self._timer0)
+                libuv.gevent_zero_prepare(self._prepare)
+                libuv.gevent_zero_timer(self._signal_idle)
+                libuv.gevent_zero_loop(ptr)
 
-        libuv.gevent_zero_loop(ptr)
+            del self._check
+            del self._prepare
+            del self._signal_idle
+            del self._timer0
 
-        # Destroy any watchers we're still holding on to.
-        del self._io_watchers
-        del self._fork_watchers
-        del self._child_watchers
+            # Destroy any watchers we're still holding on to.
+            del self._io_watchers
+            del self._fork_watchers
+            del self._child_watchers
 
 
     def debug(self):
@@ -402,7 +425,7 @@ class loop(AbstractLoop):
                                         libuv.uv_is_active(handle),
                                         libuv.uv_is_closing(handle)))
 
-        libuv.uv_walk(self._ptr,
+        libuv.uv_walk(self.ptr,
                       ffi.callback("void(*)(uv_handle_t*,void*)",
                                    walk),
                       ffi.NULL)
@@ -416,7 +439,7 @@ class loop(AbstractLoop):
         pass
 
     def break_(self, how=None):
-        libuv.uv_stop(self._ptr)
+        libuv.uv_stop(self.ptr)
 
     def reinit(self):
         # TODO: How to implement? We probably have to simply
@@ -440,7 +463,7 @@ class loop(AbstractLoop):
         # (https://github.com/joyent/libuv/issues/1405)
 
         # In 1.12, the uv_loop_fork function was added (by gevent!)
-        libuv.uv_loop_fork(self._ptr)
+        libuv.uv_loop_fork(self.ptr)
 
     _prepare_ran_callbacks = False
 
@@ -540,14 +563,14 @@ class loop(AbstractLoop):
         # libuv's now is expressed as an integer number of
         # milliseconds, so to get it compatible with time.time units
         # that this method is supposed to return, we have to divide by 1000.0
-        now = libuv.uv_now(self._ptr)
+        now = libuv.uv_now(self.ptr)
         return now / 1000.0
 
     def update_now(self):
-        libuv.uv_update_time(self._ptr)
+        libuv.uv_update_time(self.ptr)
 
     def fileno(self):
-        if self._ptr:
+        if self.ptr:
             fd = libuv.uv_backend_fd(self._ptr)
             if fd >= 0:
                 return fd
@@ -563,8 +586,10 @@ class loop(AbstractLoop):
             return
 
         self._sigchld_watcher = ffi.new('uv_signal_t*')
-        libuv.uv_signal_init(self._ptr, self._sigchld_watcher)
+        libuv.uv_signal_init(self.ptr, self._sigchld_watcher)
         self._sigchld_watcher.data = self._handle_to_self
+        # Don't let this keep the loop alive
+        libuv.uv_unref(self._sigchld_watcher)
 
         libuv.uv_signal_start(self._sigchld_watcher,
                               libuv.python_sigchld_callback,
@@ -612,7 +637,7 @@ class loop(AbstractLoop):
         except ValueError:
             pass
 
-        # Now's a good time to clean up any dead lists we don't need
+        # Now's a good time to clean up any dead watchers we don't need
         # anymore
         for pid in list(self._child_watchers):
             if not self._child_watchers[pid]:

src/gevent/libuv/watcher.py

@@ -474,22 +474,25 @@ class _SimulatedWithAsyncMixin(object):
         return self._async.active
 
     def start(self, cb, *args):
+        assert self._async is not None
         self._register_loop_callback()
         self.callback = cb
         self.args = args
         self._async.start(cb, *args)
-        #watcher.start(self, cb, *args)
 
     def stop(self):
         self._unregister_loop_callback()
         self.callback = None
         self.args = None
-        self._async.stop()
+        if self._async is not None:
+            # If we're stop() after close().
+            # That should be allowed.
+            self._async.stop()
 
     def close(self):
         if self._async is not None:
             a = self._async
-            #self._async = None
+            self._async = None
             a.close()
 
     def _register_loop_callback(self):
@@ -503,9 +506,7 @@ class _SimulatedWithAsyncMixin(object):
 class fork(_SimulatedWithAsyncMixin,
            _base.ForkMixin,
            watcher):
-    # We'll have to implement this one completely manually
-    # Right now it doesn't matter much since libuv doesn't survive
-    # a fork anyway. (That's a work in progress)
+    # We'll have to implement this one completely manually.
     _watcher_skip_ffi = False
 
     def _register_loop_callback(self):
@@ -619,7 +620,7 @@ class timer(_base.TimerMixin, watcher):
     _again = False
 
     def _watcher_ffi_init(self, args):
-        self._watcher_init(self.loop._ptr, self._watcher)
+        self._watcher_init(self.loop.ptr, self._watcher)
         self._after, self._repeat = args
         if self._after and self._after < 0.001:
             import warnings
@@ -674,7 +675,7 @@ class stat(_base.StatMixin, watcher):
         return data
 
     def _watcher_ffi_init(self, args):
-        return self._watcher_init(self.loop._ptr, self._watcher)
+        return self._watcher_init(self.loop.ptr, self._watcher)
 
     MIN_STAT_INTERVAL = 0.1074891 # match libev; 0.0 is default
 
@@ -707,7 +708,7 @@ class signal(_base.SignalMixin, watcher):
     _watcher_callback_name = '_gevent_signal_callback1'
 
     def _watcher_ffi_init(self, args):
-        self._watcher_init(self.loop._ptr, self._watcher)
+        self._watcher_init(self.loop.ptr, self._watcher)
         self.ref = False # libev doesn't ref these by default
 
 

src/gevent/os.py

@@ -277,9 +277,11 @@ if hasattr(os, 'fork'):
             # just not waited on yet.
             now = time.time()
             oldest_allowed = now - timeout
-            dead = [pid for pid, val
-                    in _watched_children.items()
-                    if isinstance(val, tuple) and val[2] < oldest_allowed]
+            dead = [
+                pid for pid, val
+                in _watched_children.items()
+                if isinstance(val, tuple) and val[2] < oldest_allowed
+            ]
             for pid in dead:
                 del _watched_children[pid]
 
@@ -296,7 +298,10 @@ if hasattr(os, 'fork'):
             :func:`os.waitpid`. Some combinations of *options* may not
             be supported cooperatively (as of 1.1 that includes
             WUNTRACED). Using a *pid* of 0 to request waiting on only processes
-            from the current process group is not cooperative.
+            from the current process group is not cooperative. A *pid* of -1
+            to wait for any child is non-blocking, but may or may not
+            require a trip around the event loop, depending on whether any children
+            have already terminated but not been waited on.
 
             Availability: POSIX.
 
@@ -316,12 +321,19 @@ if hasattr(os, 'fork'):
             if pid <= 0:
                 # magic functions for multiple children.
                 if pid == -1:
-                    # Any child. If we have one that we're watching and that finished,
-                    # we will use that one. Otherwise, let the OS take care of it.
+                    # Any child. If we have one that we're watching
+                    # and that finished, we will use that one,
+                    # preferring the oldest. Otherwise, let the OS
+                    # take care of it.
+                    finished_at = None
                     for k, v in _watched_children.items():
-                        if isinstance(v, tuple):
+                        if (
+                                isinstance(v, tuple)
+                                and (finished_at is None or v[2] < finished_at)
+                        ):
                             pid = k
-                            break
+                            finished_at = v[2]
+
                 if pid <= 0:
                     # We didn't have one that was ready. If there are
                     # no funky options set, and the pid was -1

src/gevent/queue.py

@@ -38,12 +38,14 @@ if sys.version_info[0] == 2:
     import Queue as __queue__ # python 3: pylint:disable=import-error
 else:
     import queue as __queue__ # python 2: pylint:disable=import-error
-Full = __queue__.Full
-Empty = __queue__.Empty
+# We re-export these exceptions to client modules.
+# But we also want fast access to them from Cython with a cdef,
+# and we do that with the _ definition.
+_Full = Full = __queue__.Full
+_Empty = Empty = __queue__.Empty
 
 from gevent.timeout import Timeout
 from gevent._hub_local import get_hub_noargs as get_hub
-from greenlet import getcurrent
 from gevent.exceptions import InvalidSwitchError
 
 __all__ = []
@@ -71,6 +73,8 @@ def _safe_remove(deq, item):
 
 import gevent._waiter
 locals()['Waiter'] = gevent._waiter.Waiter
+locals()['getcurrent'] = __import__('greenlet').getcurrent
+locals()['greenlet_init'] = lambda: None
 
 class ItemWaiter(Waiter): # pylint:disable=undefined-variable
     # pylint:disable=assigning-non-slot
@@ -256,7 +260,7 @@ class Queue(object):
             self._put(item)
             if self.getters:
                 self._schedule_unlock()
-        elif self.hub is getcurrent():
+        elif self.hub is getcurrent(): # pylint:disable=undefined-variable
             # We're in the mainloop, so we cannot wait; we can switch to other greenlets though.
             # Check if possible to get a free slot in the queue.
             while self.getters and self.qsize() and self.qsize() >= self._maxsize:
@@ -290,13 +294,14 @@ class Queue(object):
         """
         self.put(item, False)
 
+
     def __get_or_peek(self, method, block, timeout):
         # Internal helper method. The `method` should be either
         # self._get when called from self.get() or self._peek when
         # called from self.peek(). Call this after the initial check
         # to see if there are items in the queue.
 
-        if self.hub is getcurrent():
+        if self.hub is getcurrent(): # pylint:disable=undefined-variable
             # special case to make get_nowait() or peek_nowait() runnable in the mainloop greenlet
             # there are no items in the queue; try to fix the situation by unlocking putters
             while self.putters:
@@ -305,12 +310,12 @@ class Queue(object):
                 self.putters.popleft().put_and_switch()
                 if self.qsize():
                     return method()
-            raise Empty()
+            raise Empty
 
         if not block:
             # We can't block, we're not the hub, and we have nothing
             # to return. No choice...
-            raise Empty()
+            raise Empty
 
         waiter = Waiter() # pylint:disable=undefined-variable
         timeout = Timeout._start_new_or_dummy(timeout, Empty)
@@ -362,7 +367,8 @@ class Queue(object):
         (*timeout* is ignored in that case).
         """
         if self.qsize():
-            # XXX: Why doesn't this schedule an unlock like get() does?
+            # This doesn't schedule an unlock like get() does because we're not
+            # actually making any space.
             return self._peek()
 
         return self.__get_or_peek(self._peek, block, timeout)
@@ -604,7 +610,7 @@ class Channel(object):
         return True
 
     def put(self, item, block=True, timeout=None):
-        if self.hub is getcurrent():
+        if self.hub is getcurrent(): # pylint:disable=undefined-variable
             if self.getters:
                 getter = self.getters.popleft()
                 getter.switch(item)
@@ -634,7 +640,7 @@ class Channel(object):
         self.put(item, False)
 
     def get(self, block=True, timeout=None):
-        if self.hub is getcurrent():
+        if self.hub is getcurrent(): # pylint:disable=undefined-variable
             if self.putters:
                 item, putter = self.putters.popleft()
                 self.hub.loop.run_callback(putter.switch, putter)
@@ -681,5 +687,11 @@ class Channel(object):
 
     next = __next__ # Py2
 
+def _init():
+    greenlet_init() # pylint:disable=undefined-variable
+
+_init()
+
+
 from gevent._util import import_c_accel
 import_c_accel(globals(), 'gevent._queue')

src/gevent/testing/errorhandler.py

@@ -17,7 +17,7 @@
 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 # THE SOFTWARE.
-
+from __future__ import print_function
 from functools import wraps
 
 
@@ -29,7 +29,6 @@ def wrap_error_fatal(method):
     def wrapper(self, *args, **kwargs):
         # XXX should also be able to do gevent.SYSTEM_ERROR = object
         # which is a global default to all hubs
-
         gevent.get_hub().SYSTEM_ERROR = object
         try:
             return method(self, *args, **kwargs)

src/gevent/testing/hub.py

@@ -25,6 +25,8 @@ from gevent.hub import Hub
 from .exception import ExpectedException
 
 class QuietHub(Hub):
+    _resolver = None
+    _threadpool = None
 
     EXPECTED_TEST_ERROR = (ExpectedException,)
 

src/gevent/testing/patched_tests_setup.py

@@ -1247,7 +1247,7 @@ disabled_tests += [
     'test_ssl.BasicSocketTests.test_openssl_version'
 ]
 
-if TRAVIS and OSX:
+if OSX:
 
     disabled_tests += [
         # This sometimes produces OSError: Errno 40: Message too long

src/gevent/testing/skipping.py

@@ -32,6 +32,8 @@ def _do_not_skip(reason):
     return _identity
 
 
+skipOnMac = _do_not_skip
+skipOnMacOnCI = _do_not_skip
 skipOnWindows = _do_not_skip
 skipOnAppVeyor = _do_not_skip
 skipOnCI = _do_not_skip
@@ -62,6 +64,8 @@ skipOnLibev = _do_not_skip
 if sysinfo.WIN:
     skipOnWindows = unittest.skip
 
+if sysinfo.OSX:
+    skipOnMac = unittest.skip
 
 if sysinfo.RUNNING_ON_APPVEYOR:
     # See comments scattered around about timeouts and the timer
@@ -76,6 +80,8 @@ if sysinfo.RUNNING_ON_APPVEYOR:
 
 if sysinfo.RUNNING_ON_CI:
     skipOnCI = unittest.skip
+    if sysinfo.OSX:
+        skipOnMacOnCI = unittest.skip
 
 if sysinfo.RUNNING_ON_MANYLINUX:
     skipOnManylinux = unittest.skip

src/gevent/testing/testcase.py

@@ -29,6 +29,7 @@ import gevent
 from gevent._util import LazyOnClass
 from gevent._compat import perf_counter
 from gevent._compat import get_clock_info
+from gevent._hub_local import get_hub_if_exists
 
 from . import sysinfo
 from . import params
@@ -146,12 +147,24 @@ class StringAssertMixin(object):
 class TestTimeout(gevent.Timeout):
     _expire_info = ''
 
-    def __init__(self, timeout):
-        gevent.Timeout.__init__(self, timeout, 'test timed out\n', ref=False)
+    def __init__(self, timeout, method='Not Given'):
+        gevent.Timeout.__init__(
+            self,
+            timeout,
+            '%r: test timed out\n' % (method,),
+            ref=False
+        )
 
     def _on_expiration(self, prev_greenlet, ex):
         from gevent.util import format_run_info
-        self._expire_info = '\n'.join(format_run_info())
+        loop = gevent.get_hub().loop
+        debug_info = 'N/A'
+        if hasattr(loop, 'debug'):
+            debug_info = [str(s) for s in loop.debug()]
+        run_info = format_run_info()
+        self._expire_info = 'Loop Debug:\n%s\nRun Info:\n%s' % (
+            '\n'.join(debug_info), '\n'.join(run_info)
+        )
         gevent.Timeout._on_expiration(self, prev_greenlet, ex)
 
     def __str__(self):
@@ -165,7 +178,7 @@ def _wrap_timeout(timeout, method):
 
     @wraps(method)
     def wrapper(self, *args, **kwargs):
-        with TestTimeout(timeout):
+        with TestTimeout(timeout, method):
             return method(self, *args, **kwargs)
 
     return wrapper
@@ -277,7 +290,9 @@ class TestCase(TestCaseMetaClass("NewBase",
         # so that doesn't always happen. test__pool.py:TestPoolYYY.test_async
         # tends to show timeouts that are too short if we don't.
         # XXX: Should some core part of the loop call this?
-        gevent.get_hub().loop.update_now()
+        hub = get_hub_if_exists()
+        if hub and hub.loop:
+            hub.loop.update_now()
         self.close_on_teardown = []
         self.addCleanup(self._tearDownCloseOnTearDown)
 

src/gevent/testing/testrunner.py

 #!/usr/bin/env python
 from __future__ import print_function, absolute_import, division
 
+import re
 import sys
 import os
 import glob
 import traceback
 import importlib
-from datetime import timedelta
 
+from contextlib import contextmanager
+from datetime import timedelta
 from multiprocessing.pool import ThreadPool
 from multiprocessing import cpu_count
+
+from gevent._util import Lazy
+
 from . import util
 from .resources import parse_resources
 from .resources import setup_resources
@@ -244,106 +249,269 @@ class TravisFoldingRunner(object):
     def __call__(self):
         return self._runner()
 
-def discover(
-        tests=None, ignore_files=None,
-        ignored=(), coverage=False,
-        package=None,
-        configured_ignore_coverage=(),
-        configured_test_options=None,
-):
-    # pylint:disable=too-many-locals,too-many-branches
-    configured_test_options = configured_test_options or {}
-    olddir = os.getcwd()
-    ignore = set(ignored or ())
-
-    if ignore_files:
-        ignore_files = ignore_files.split(',')
-        for f in ignore_files:
-            ignore.update(set(load_list_from_file(f, package)))
-
-    if coverage:
-        ignore.update(configured_ignore_coverage)
-
-    if package:
-        package_dir = _dir_from_package_name(package)
-        # We need to glob relative names, our config is based on filenames still
-        os.chdir(package_dir)
-
-    if not tests:
-        tests = set(glob.glob('test_*.py')) - set(['test_support.py'])
-    else:
-        tests = set(tests)
-
-    if ignore:
-        # Always ignore the designated list, even if tests were specified
-        # on the command line. This fixes a nasty interaction with test__threading_vs_settrace.py
-        # being run under coverage when 'grep -l subprocess test*py' is used to list the tests
-        # to run.
-        tests -= ignore
-    tests = sorted(tests)
-
-    to_process = []
-    to_import = []
-
-    for filename in tests:
-        # Support either 'gevent.tests.foo' or 'gevent/tests/foo.py'
-        if filename.startswith('gevent.tests'):
-            # XXX: How does this interact with 'package'? Probably not well
-            qualified_name = module_name = filename
-            filename = filename[len('gevent.tests') + 1:]
-            filename = filename.replace('.', os.sep) + '.py'
-        else:
-            module_name = os.path.splitext(filename)[0]
-            qualified_name = package + '.' + module_name if package else module_name
-        with open(os.path.abspath(filename), 'rb') as f:
-            # Some of the test files (e.g., test__socket_dns) are
-            # UTF8 encoded. Depending on the environment, Python 3 may
-            # try to decode those as ASCII, which fails with UnicodeDecodeError.
-            # Thus, be sure to open and compare in binary mode.
-            # Open the absolute path to make errors more clear,
-            # but we can't store the absolute path, our configuration is based on
-            # relative file names.
-            contents = f.read()
-        if b'TESTRUNNER' in contents: # test__monkey_patching.py
-            # XXX: Rework this to avoid importing.
-            to_import.append(qualified_name)
-        else:
-            # XXX: For simple python module tests, try this with `runpy.run_module`,
-            # very similar to the way we run things for monkey patching.
-            # The idea here is that we can perform setup ahead of time (e.g., setup_resources())
-            # in each test without having to do it manually or force calls or modifications to those
-            # tests.
 
+class Discovery(object):
+    package_dir = None
+    package = None
+
+    def __init__(
+            self,
+            tests=None,
+            ignore_files=None,
+            ignored=(),
+            coverage=False,
+            package=None,
+            config=None,
+    ):
+        self.config = config or {}
+        self.ignore = set(ignored or ())
+        self.tests = tests
+        self.configured_test_options = config.get('TEST_FILE_OPTIONS', set())
+
+        if ignore_files:
+            ignore_files = ignore_files.split(',')
+            for f in ignore_files:
+                self.ignore.update(set(load_list_from_file(f, package)))
+
+        if coverage:
+            self.ignore.update(config.get('IGNORE_COVERAGE', set()))
+
+        if package:
+            self.package = package
+            self.package_dir = _dir_from_package_name(package)
+
+    class Discovered(object):
+        def __init__(self, package, configured_test_options, ignore, config):
+            self.orig_dir = os.getcwd()
+            self.configured_run_alone = config['RUN_ALONE']
+            self.configured_failing_tests = config['FAILING_TESTS']
+            self.package = package
+            self.configured_test_options = configured_test_options
+            self.ignore = ignore
+
+            self.to_import = []
+            self.std_monkey_patch_files = []
+            self.no_monkey_patch_files = []
+
+            self.commands = []
+
+        @staticmethod
+        def __makes_simple_monkey_patch(
+                contents,
+                _patch_present=re.compile(br'[^#].*patch_all\(\)'),
+                _patch_indented=re.compile(br'    .*patch_all\(\)')
+        ):
+            return (
+                # A non-commented patch_all() call is present
+                bool(_patch_present.search(contents))
+                # that is not indented (because that implies its not at the top-level,
+                # so some preconditions are being set)
+                and not _patch_indented.search(contents)
+            )
+
+        @staticmethod
+        def __file_allows_monkey_combine(contents):
+            return b'testrunner-no-monkey-combine' not in contents
+
+        @staticmethod
+        def __file_allows_combine(contents):
+            return b'testrunner-no-combine' not in contents
+
+        @staticmethod
+        def __calls_unittest_main_toplevel(
+                contents,
+                _greentest_main=re.compile(br'    greentest.main\(\)'),
+                _unittest_main=re.compile(br'    unittest.main\(\)'),
+                _import_main=re.compile(br'from gevent.testing import.*main'),
+                _main=re.compile(br'    main\(\)'),
+        ):
+            # TODO: Add a check that this comes in a line directly after
+            # if __name__ == __main__.
+            return (
+                _greentest_main.search(contents)
+                or _unittest_main.search(contents)
+                or (_import_main.search(contents) and _main.search(contents))
+            )
+
+        def __has_config(self, filename):
+            return (
+                RUN_LEAKCHECKS
+                or filename in self.configured_test_options
+                or filename in self.configured_run_alone
+                or matches(self.configured_failing_tests, filename)
+            )
+
+        def __can_monkey_combine(self, filename, contents):
+            return (
+                not self.__has_config(filename)
+                and self.__makes_simple_monkey_patch(contents)
+                and self.__file_allows_monkey_combine(contents)
+                and self.__file_allows_combine(contents)
+                and self.__calls_unittest_main_toplevel(contents)
+            )
+
+        @staticmethod
+        def __makes_no_monkey_patch(contents, _patch_present=re.compile(br'[^#].*patch_\w*\(')):
+            return not _patch_present.search(contents)
+
+        def __can_nonmonkey_combine(self, filename, contents):
+            return (
+                not self.__has_config(filename)
+                and self.__makes_no_monkey_patch(contents)
+                and self.__file_allows_combine(contents)
+                and self.__calls_unittest_main_toplevel(contents)
+            )
+
+        def __begin_command(self):
             cmd = [sys.executable, '-u']
             if PYPY and PY2:
                 # Doesn't seem to be an env var for this
                 cmd.extend(('-X', 'track-resources'))
-            if package:
-                # Using a package is the best way to work with coverage 5
-                # when we specify 'source = <package>'
-                cmd.append('-m' + qualified_name)
+            return cmd
+
+
+        def __add_test(self, qualified_name, filename, contents):
+            if b'TESTRUNNER' in contents: # test__monkey_patching.py
+                # XXX: Rework this to avoid importing.
+                # XXX: Rework this to allow test combining (it could write the files out and return
+                # them directly; we would use 'python -m gevent.monkey --module unittest ...)
+                self.to_import.append(qualified_name)
+            elif self.__can_monkey_combine(filename, contents):
+                self.std_monkey_patch_files.append(qualified_name if self.package else filename)
+            elif self.__can_nonmonkey_combine(filename, contents):
+                self.no_monkey_patch_files.append(qualified_name if self.package else filename)
             else:
-                cmd.append(filename)
-
-            options = DEFAULT_RUN_OPTIONS.copy()
-            options.update(configured_test_options.get(filename, {}))
-            to_process.append((cmd, options))
-
-    os.chdir(olddir)
-    # When we actually execute, do so from the original directory,
-    # this helps find setup.py
-    for qualified_name in to_import:
-        module = importlib.import_module(qualified_name)
-        for cmd, options in module.TESTRUNNER():
-            if remove_options(cmd)[-1] in ignore:
-                continue
-            to_process.append((cmd, options))
+                # XXX: For simple python module tests, try this with
+                # `runpy.run_module`, very similar to the way we run
+                # things for monkey patching. The idea here is that we
+                # can perform setup ahead of time (e.g.,
+                # setup_resources()) in each test without having to do
+                # it manually or force calls or modifications to those
+                # tests.
+                cmd = self.__begin_command()
+                if self.package:
+                    # Using a package is the best way to work with coverage 5
+                    # when we specify 'source = <package>'
+                    cmd.append('-m' + qualified_name)
+                else:
+                    cmd.append(filename)
+
+                options = DEFAULT_RUN_OPTIONS.copy()
+                options.update(self.configured_test_options.get(filename, {}))
+                self.commands.append((cmd, options))
+
+        @staticmethod
+        def __remove_options(lst):
+            return [x for x in lst if x and not x.startswith('-')]
+
+        def __expand_imports(self):
+            for qualified_name in self.to_import:
+                module = importlib.import_module(qualified_name)
+                for cmd, options in module.TESTRUNNER():
+                    if self.__remove_options(cmd)[-1] in self.ignore:
+                        continue
+                    self.commands.append((cmd, options))
+            del self.to_import[:]
+
+        def __combine_commands(self, files, group_size=5):
+            if not files:
+                return
+
+            from itertools import groupby
+            cnt = [0, 0]
+            def make_group(_):
+                if cnt[0] > group_size:
+                    cnt[0] = 0
+                    cnt[1] += 1
+                cnt[0] += 1
+                return cnt[1]
+
+            for _, group in groupby(files, make_group):
+
+                cmd = self.__begin_command()
+                cmd.append('-m')
+                cmd.append('unittest')
+                # cmd.append('-v')
+                for name in group:
+                    cmd.append(name)
+                self.commands.insert(0, (cmd, DEFAULT_RUN_OPTIONS.copy()))
+
+            del files[:]
+
+
+        def visit_file(self, filename):
+            # Support either 'gevent.tests.foo' or 'gevent/tests/foo.py'
+            if filename.startswith('gevent.tests'):
+                # XXX: How does this interact with 'package'? Probably not well
+                qualified_name = module_name = filename
+                filename = filename[len('gevent.tests') + 1:]
+                filename = filename.replace('.', os.sep) + '.py'
+            else:
+                module_name = os.path.splitext(filename)[0]
+                qualified_name = self.package + '.' + module_name if self.package else module_name
+
+            with open(os.path.abspath(filename), 'rb') as f:
+                # Some of the test files (e.g., test__socket_dns) are
+                # UTF8 encoded. Depending on the environment, Python 3 may
+                # try to decode those as ASCII, which fails with UnicodeDecodeError.
+                # Thus, be sure to open and compare in binary mode.
+                # Open the absolute path to make errors more clear,
+                # but we can't store the absolute path, our configuration is based on
+                # relative file names.
+                contents = f.read()
+
+            self.__add_test(qualified_name, filename, contents)
+
+        def visit_files(self, filenames):
+            for filename in filenames:
+                self.visit_file(filename)
+            with Discovery._in_dir(self.orig_dir):
+                self.__expand_imports()
+            self.__combine_commands(self.std_monkey_patch_files)
+            self.__combine_commands(self.no_monkey_patch_files)
+
+    @staticmethod
+    @contextmanager
+    def _in_dir(package_dir):
+        olddir = os.getcwd()
+        if package_dir:
+            os.chdir(package_dir)
+        try:
+            yield
+        finally:
+            os.chdir(olddir)
 
-    return to_process
+    @Lazy
+    def discovered(self):
+        tests = self.tests
+        discovered = self.Discovered(self.package, self.configured_test_options,
+                                     self.ignore, self.config)
 
+        # We need to glob relative names, our config is based on filenames still
+        with self._in_dir(self.package_dir):
+            if not tests:
+                tests = set(glob.glob('test_*.py')) - set(['test_support.py'])
+            else:
+                tests = set(tests)
 
-def remove_options(lst):
-    return [x for x in lst if x and not x.startswith('-')]
+            if self.ignore:
+                # Always ignore the designated list, even if tests
+                # were specified on the command line. This fixes a
+                # nasty interaction with
+                # test__threading_vs_settrace.py being run under
+                # coverage when 'grep -l subprocess test*py' is used
+                # to list the tests to run.
+                tests -= self.ignore
+            tests = sorted(tests)
+            discovered.visit_files(tests)
+
+        return discovered
+
+    def __iter__(self):
+        return iter(self.discovered.commands) # pylint:disable=no-member
+
+    def __len__(self):
+        return len(self.discovered.commands) # pylint:disable=no-member
 
 def load_list_from_file(filename, package):
     result = []
@@ -369,6 +537,8 @@ def matches(possibilities, command, include_flaky=True):
         # XXX: This could be much better. Our command needs better structure.
         if command.endswith(' ' + line) or command.endswith(line.replace(".py", '')):
             return True
+        if ' '  not in command and command == line:
+            return True
     return False
 
 
@@ -559,7 +729,6 @@ def main():
     FAILING_TESTS = []
     IGNORED_TESTS = []
     RUN_ALONE = []
-    TEST_FILE_OPTIONS = {}
 
     coverage = False
     if options.coverage or os.environ.get("GEVENTTEST_COVERAGE"):
@@ -596,18 +765,14 @@ def main():
         FAILING_TESTS = config['FAILING_TESTS']
         IGNORED_TESTS = config['IGNORED_TESTS']
         RUN_ALONE = config['RUN_ALONE']
-        TEST_FILE_OPTIONS = config['TEST_FILE_OPTIONS']
-        IGNORE_COVERAGE = config['IGNORE_COVERAGE']
-
 
-    tests = discover(
+    tests = Discovery(
         options.tests,
         ignore_files=options.ignore,
         ignored=IGNORED_TESTS,
         coverage=coverage,
         package=options.package,
-        configured_ignore_coverage=IGNORE_COVERAGE,
-        configured_test_options=TEST_FILE_OPTIONS,
+        config=config,
     )
     if options.discover:
         for cmd, options in tests:

src/gevent/tests/known_failures.py

@@ -364,7 +364,6 @@ class Definitions(DefinitionsBase):
         it seems highly load dependent. Observed with both libev and libuv.
         """,
         when=TRAVIS & (PYPY | OSX),
-        run_alone=ALWAYS,
         # This often takes much longer on PyPy on CI.
         options={'timeout': (CI & PYPY, 180)},
     )
@@ -438,11 +437,11 @@ class Definitions(DefinitionsBase):
         """,
     )
 
-    test__pool = test__queue = RunAlone(
+    test__pool = RunAlone(
         """
         On a heavily loaded box, these can all take upwards of 200s.
         """,
-        when=CI & LEAKTEST | PY3
+        when=CI & LEAKTEST
     )
 
     test_socket = RunAlone(
@@ -453,7 +452,7 @@ class Definitions(DefinitionsBase):
 
     test__refcount = Ignored(
         "Sometimes fails to connect for no reason",
-        when=(CI & OSX) | (CI & PYPY),
+        when=(CI & OSX) | (CI & PYPY) | APPVEYOR,
         ignore_coverage=PYPY
     )
 

src/gevent/tests/test__all__.py

@@ -110,7 +110,6 @@ class AbstractTestMixin(object):
             except ImportError:
                 if modname in modules.OPTIONAL_MODULES:
                     msg = "Unable to import %s" % modname
-                    warnings.warn(msg) # make the testrunner print it
                     raise unittest.SkipTest(msg)
                 raise
 

src/gevent/tests/test__ares_timeout.py

@@ -29,7 +29,8 @@ class TestTimeout(greentest.TestCase):
             while True:
                 listener.recvfrom(10000)
 
-        gevent.spawn(reader)
+        greader = gevent.spawn(reader)
+        self._close_on_teardown(greader.kill)
 
         r = Resolver(servers=[address[0]], timeout=0.001, tries=1,
                      udp_port=address[-1])

src/gevent/tests/test__backdoor.py

@@ -158,4 +158,4 @@ class Test(greentest.TestCase):
 
 
 if __name__ == '__main__':
-    greentest.main()
+    greentest.main() # pragma: testrunner-no-combine

src/gevent/tests/test__core_async.py

@@ -7,18 +7,25 @@ try:
 except ImportError:
     import _thread as thread
 
+from gevent import testing as greentest
 
-hub = gevent.get_hub()
-watcher = hub.loop.async_()
+class Test(greentest.TestCase):
+    def test(self):
+        hub = gevent.get_hub()
+        watcher = hub.loop.async_()
 
-# BWC for <3.7: This should still be an attribute
-assert hasattr(hub.loop, 'async')
+        # BWC for <3.7: This should still be an attribute
+        assert hasattr(hub.loop, 'async')
 
-gevent.spawn_later(0.1, thread.start_new_thread, watcher.send, ())
+        gevent.spawn_later(0.1, thread.start_new_thread, watcher.send, ())
 
-start = time.time()
+        start = time.time()
 
-with gevent.Timeout(1.0): # Large timeout for appveyor
-    hub.wait(watcher)
+        with gevent.Timeout(1.0): # Large timeout for appveyor
+            hub.wait(watcher)
 
-print('Watcher %r reacted after %.6f seconds' % (watcher, time.time() - start - 0.1))
+        print('Watcher %r reacted after %.6f seconds' % (watcher, time.time() - start - 0.1))
+
+
+if __name__ == '__main__':
+    greentest.main()

src/gevent/tests/test__core_callback.py

 import gevent
 from gevent.hub import get_hub
 
-called = []
+from gevent import testing as greentest
 
+class Test(greentest.TestCase):
 
-def f():
-    called.append(1)
+    def test(self):
+        loop = get_hub().loop
+        called = []
 
+        def f():
+            called.append(1)
 
-def main():
-    loop = get_hub().loop
-    x = loop.run_callback(f)
+        x = loop.run_callback(f)
 
-    assert x, x
-    gevent.sleep(0)
-    assert called == [1], called
-    assert not x, (x, bool(x))
+        assert x, x
+        gevent.sleep(0)
+        assert called == [1], called
+        assert not x, (x, bool(x))
 
-    x = loop.run_callback(f)
-    assert x, x
-    x.stop()
-    assert not x, x
-    gevent.sleep(0)
-    assert called == [1], called
-    assert not x, x
+        x = loop.run_callback(f)
+        assert x, x
+        x.stop()
+        assert not x, x
+        gevent.sleep(0)
+        assert called == [1], called
+        assert not x, x
 
 
 if __name__ == '__main__':
-    called[:] = []
-    main()
+    greentest.main()

src/gevent/tests/test__destroy.py

@@ -48,4 +48,4 @@ class TestDestroyHub(unittest.TestCase):
         hub.destroy()
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main() # pragma: testrunner-no-combine

src/gevent/tests/test__destroy_default_loop.py

@@ -5,6 +5,20 @@ import unittest
 
 class TestDestroyDefaultLoop(unittest.TestCase):
 
+    def tearDown(self):
+        self._reset_hub()
+        super(TestDestroyDefaultLoop, self).tearDown()
+
+    def _reset_hub(self):
+        from gevent._hub_local import set_hub
+        from gevent._hub_local import set_loop
+        from gevent._hub_local import get_hub_if_exists
+        hub = get_hub_if_exists()
+        if hub is not None:
+            hub.destroy(destroy_loop=True)
+        set_hub(None)
+        set_loop(None)
+
     def test_destroy_gc(self):
         # Issue 1098: destroying the default loop
         # while using the C extension could crash
@@ -19,6 +33,7 @@ class TestDestroyDefaultLoop(unittest.TestCase):
         loop = gevent.config.loop(default=True)
         self.assertTrue(loop.default)
         # Destroy it
+
         loop.destroy()
         # It no longer claims to be the default
         self.assertFalse(loop.default)
@@ -31,10 +46,7 @@ class TestDestroyDefaultLoop(unittest.TestCase):
         # crash only happened when that greenlet object
         # was collected at exit time, which was most common
         # in CPython 3.5)
-        from gevent._hub_local import set_hub
-        set_hub(None)
-
-
+        self._reset_hub()
 
     def test_destroy_two(self):
         # Get two new loop object, but using the default
@@ -50,6 +62,10 @@ class TestDestroyDefaultLoop(unittest.TestCase):
 
         # Destroy the second. This doesn't crash.
         loop2.destroy()
+        self.assertFalse(loop2.default)
+        self.assertFalse(loop2.ptr)
+        self._reset_hub()
+        self.assertTrue(gevent.get_hub().loop.ptr)
 
 
 if __name__ == '__main__':

src/gevent/tests/test__example_portforwarder.py

 from __future__ import print_function, absolute_import
-from gevent import monkey; monkey.patch_all(subprocess=True)
+from gevent import monkey; monkey.patch_all()
 import signal
 
 import socket

src/gevent/tests/test__example_udp_client.py

 from gevent import monkey
-monkey.patch_all(subprocess=True)
+monkey.patch_all()
 
 
 from gevent.server import DatagramServer
@@ -30,4 +30,6 @@ class Test_udp_client(util.TestServer):
 
 
 if __name__ == '__main__':
-    main()
+    # Running this following test__example_portforwarder on Appveyor
+    # doesn't work in the same process for some reason.
+    main() # pragma: testrunner-no-combine

src/gevent/tests/test__greenio.py

@@ -21,7 +21,8 @@ import sys
 import gevent
 from gevent import socket
 
-from gevent.testing import TestCase, main, tcp_listener
+from gevent import testing as greentest
+from gevent.testing import TestCase, tcp_listener
 from gevent.testing import gc_collect_if_needed
 from gevent.testing import skipOnPyPy
 from gevent.testing import params
@@ -142,4 +143,4 @@ class TestGreenIo(TestCase):
 
 
 if __name__ == '__main__':
-    main()
+    greentest.main()

src/gevent/tests/test__issue112.py

 import sys
 import unittest
+import threading
+import gevent
+import gevent.monkey
+gevent.monkey.patch_all()
+
 
 @unittest.skipUnless(
     sys.version_info[0] == 2,
@@ -8,11 +13,6 @@ import unittest
 class Test(unittest.TestCase):
 
     def test(self):
-        import threading
-        import gevent.monkey
-        gevent.monkey.patch_all()
-        import gevent
-
         self.assertIs(threading._sleep, gevent.sleep)
 
 if __name__ == '__main__':

src/gevent/tests/test__issue230.py

@@ -4,6 +4,8 @@ gevent.monkey.patch_all()
 import socket
 import multiprocessing
 
+from gevent import testing as greentest
+
 # Make sure that using the resolver in a forked process
 # doesn't hang forever.
 
@@ -12,12 +14,14 @@ def block():
     socket.getaddrinfo('localhost', 8001)
 
 
-def main():
-    socket.getaddrinfo('localhost', 8001)
 
-    p = multiprocessing.Process(target=block)
-    p.start()
-    p.join()
+class Test(greentest.TestCase):
+    def test(self):
+        socket.getaddrinfo('localhost', 8001)
+
+        p = multiprocessing.Process(target=block)
+        p.start()
+        p.join()
 
 if __name__ == '__main__':
-    main()
+    greentest.main()

src/gevent/tests/test__issue467.py

 import gevent
+from gevent import testing as greentest
+
 #import socket # on windows
 
 # iwait should not raise `LoopExit: This operation would block forever`
@@ -13,23 +15,26 @@ def worker(i):
         raise ValueError(i)
     return i
 
-
-def main():
-    finished = 0
-    # Wait on a group that includes one that will already be
-    # done, plus some that will finish as we watch
-    done_worker = gevent.spawn(worker, "done")
-    gevent.joinall((done_worker,))
-
-    workers = [gevent.spawn(worker, i) for i in range(3)]
-    workers.append(done_worker)
-    for _ in gevent.iwait(workers):
-        finished += 1
-        # Simulate doing something that causes greenlets to switch;
-        # a non-zero timeout is crucial
-        gevent.sleep(0.01)
-
-    assert finished == 4
+class Test(greentest.TestCase):
+    def test(self):
+        finished = 0
+        # Wait on a group that includes one that will already be
+        # done, plus some that will finish as we watch
+        done_worker = gevent.spawn(worker, "done")
+        gevent.joinall((done_worker,))
+
+        workers = [gevent.spawn(worker, i) for i in range(3)]
+        workers.append(done_worker)
+        for _ in gevent.iwait(workers):
+            finished += 1
+            # Simulate doing something that causes greenlets to switch;
+            # a non-zero timeout is crucial
+            try:
+                gevent.sleep(0.01)
+            except ValueError as ex:
+                self.assertEqual(ex.args[0], 2)
+
+        self.assertEqual(finished, 4)
 
 if __name__ == '__main__':
-    main()
+    greentest.main()

src/gevent/tests/test__issue600.py

@@ -10,7 +10,7 @@ import sys
 from multiprocessing import Process
 from subprocess import Popen, PIPE
 
-import gevent.testing as greentest
+from gevent import testing as greentest
 
 def f(sleep_sec):
     gevent.sleep(sleep_sec)

src/gevent/tests/test__issue639.py

 # Test idle
 import gevent
-gevent.sleep()
-gevent.idle()
+
+from gevent import testing as greentest
+
+class Test(greentest.TestCase):
+    def test(self):
+        gevent.sleep()
+        gevent.idle()
+
+if __name__ == '__main__':
+    greentest.main()

src/gevent/tests/test__joinall.py

 import gevent
 
+from gevent import testing as greentest
 
-def func():
-    pass
 
+class Test(greentest.TestCase):
 
-a = gevent.spawn(func)
-b = gevent.spawn(func)
-gevent.joinall([a, b, a])
+    def test(self):
+
+        def func():
+            pass
+
+
+        a = gevent.spawn(func)
+        b = gevent.spawn(func)
+        gevent.joinall([a, b, a])
+
+
+if __name__ == '__main__':
+    greentest.main()

src/gevent/tests/test__loop_callback.py

-from gevent.core import loop
+from gevent import get_hub
+from gevent import testing as greentest
 
-count = 0
+class Test(greentest.TestCase):
+    def test(self):
+        count = [0]
 
+        def incr():
+            count[0] += 1
 
-def incr():
-    global count
-    count += 1
+        loop = get_hub().loop
+        loop.run_callback(incr)
+        loop.run()
+        self.assertEqual(count, [1])
 
-loop = loop()
-loop.run_callback(incr)
-loop.run()
-assert count == 1, count
+
+if __name__ == '__main__':
+    greentest.main()

src/gevent/tests/test__memleak.py

 import sys
 import unittest
 
-from gevent.testing import TestCase, main
+from gevent.testing import TestCase
 import gevent
 from gevent.timeout import Timeout
 
@@ -53,4 +53,4 @@ class TestQueue(TestCase): # pragma: no cover
 
 
 if __name__ == '__main__':
-    main()
+    unittest.main()

src/gevent/tests/test__monkey.py

-from subprocess import Popen
-
 from gevent import monkey
 monkey.patch_all()
 
@@ -79,6 +77,7 @@ class TestMonkey(SubscriberCleanupMixin, unittest.TestCase):
                 self.assertTrue(monkey.is_object_patched(modname, objname))
 
     def test_patch_subprocess_twice(self):
+        Popen = monkey.get_original('subprocess', 'Popen')
         self.assertNotIn('gevent', repr(Popen))
         self.assertIs(Popen, monkey.get_original('subprocess', 'Popen'))
         monkey.patch_subprocess()

src/gevent/tests/test__os.py

@@ -160,11 +160,13 @@ class TestForkAndWatch(greentest.TestCase):
         pid = os.fork_and_watch()
         if pid:
             os.waitpid(-1, 0)
-            # Can't assert on what the pid actually was,
+            # Can't assert on what the found pid actually was,
             # our testrunner may have spawned multiple children.
             os._reap_children(0) # make the leakchecker happy
         else: # pragma: no cover
             gevent.sleep(2)
+            # The test framework will catch a regular SystemExit
+            # from sys.exit(), we need to just kill the process.
             os._exit(0)
 
     def test_waitpid_wrong_neg(self):

src/gevent/tests/test__pywsgi.py

@@ -22,7 +22,7 @@ from __future__ import print_function
 
 from gevent import monkey
 
-monkey.patch_all(thread=False)
+monkey.patch_all()
 
 from contextlib import contextmanager
 try:

src/gevent/tests/test__queue.py

 import unittest
 
 import gevent.testing as greentest
-from gevent.testing import TestCase, main
+from gevent.testing import TestCase
 import gevent
 from gevent.hub import get_hub, LoopExit
 from gevent import util
@@ -24,7 +24,8 @@ class TestQueue(TestCase):
     def test_peek_empty(self):
         q = queue.Queue()
         # No putters waiting, in the main loop: LoopExit
-        self.assertRaises(LoopExit, q.peek)
+        with self.assertRaises(LoopExit):
+            q.peek()
 
         def waiter(q):
             self.assertRaises(Empty, q.peek, timeout=0.01)
@@ -323,6 +324,8 @@ class TestNoWait(TestCase):
         assert q.empty(), q
 
     def test_get_nowait_unlock_channel(self):
+        # get_nowait runs fine in the hub, and
+        # it switches to a waiting putter if needed.
         result = []
         q = queue.Channel()
         p = gevent.spawn(q.put, 5)
@@ -330,19 +333,21 @@ class TestNoWait(TestCase):
         def store_result(func, *args):
             result.append(func(*args))
 
-        assert q.empty(), q
-        assert q.full(), q
+        self.assertTrue(q.empty())
+        self.assertTrue(q.full())
+
         gevent.sleep(0.001)
-        assert q.empty(), q
-        assert q.full(), q
+        self.assertTrue(q.empty())
+        self.assertTrue(q.full())
+
         get_hub().loop.run_callback(store_result, q.get_nowait)
         gevent.sleep(0.001)
-        assert q.empty(), q
-        assert q.full(), q
-        assert result == [5], result
-        assert p.ready(), p
-        assert p.dead, p
-        assert q.empty(), q
+        self.assertTrue(q.empty())
+        self.assertTrue(q.full())
+        self.assertEqual(result, [5])
+        self.assertTrue(p.ready())
+        self.assertTrue(p.dead)
+        self.assertTrue(q.empty())
 
     # put_nowait must work from the mainloop
     def test_put_nowait_unlock(self):
@@ -462,4 +467,4 @@ del AbstractGenericGetTestCase
 
 
 if __name__ == '__main__':
-    main()
+    greentest.main()

src/gevent/tests/test__real_greenlet.py

@@ -6,24 +6,29 @@ Fails with PyPy 2.2.1
 """
 from __future__ import print_function
 import sys
-import greenlet
 
+from gevent import testing as greentest
 
-print('Your greenlet version: %s' % (getattr(greenlet, '__version__', None), ))
+class Test(greentest.TestCase):
 
+    def test(self):
+        import greenlet
 
-result = []
+        print('Your greenlet version: %s' % (getattr(greenlet, '__version__', None), ))
 
+        result = []
 
-def func():
-    result.append(repr(sys.exc_info()))
+        def func():
+            result.append(repr(sys.exc_info()))
 
+        g = greenlet.greenlet(func)
+        try:
+            1 / 0
+        except ZeroDivisionError:
+            g.switch()
 
-g = greenlet.greenlet(func)
-try:
-    1 / 0
-except ZeroDivisionError:
-    g.switch()
 
+        self.assertEqual(result, ['(None, None, None)'])
 
-assert result == ['(None, None, None)'], result
+if __name__ == '__main__':
+    greentest.main()

src/gevent/tests/test__refcount_core.py

+import sys
 import weakref
 
+from gevent import testing as greentest
 
-class Dummy:
+
+class Dummy(object):
     def __init__(self):
         __import__('gevent.core')
 
-try:
-    assert weakref.ref(Dummy())() is None
+@greentest.skipIf(weakref.ref(Dummy())() is not None,
+                  "Relies on refcounting for fast weakref cleanup")
+class Test(greentest.TestCase):
+    def test(self):
+        from gevent import socket
+        s = socket.socket()
+        r = weakref.ref(s)
+        s.close()
+        del s
+        self.assertIsNone(r())
+
+assert weakref.ref(Dummy())() is None or hasattr(sys, 'pypy_version_info')
 
-    from gevent import socket
-    s = socket.socket()
-    r = weakref.ref(s)
-    s.close()
-    del s
-    assert r() is None
-except AssertionError: # pragma: no cover
-    import sys
-    if hasattr(sys, 'pypy_version_info'):
-        # PyPy uses a non refcounted GC which may defer
-        # the collection of the weakref, unlike CPython
-        pass
-    else:
-        raise
+if __name__ == '__main__':
+    greentest.main()

src/gevent/tests/test__server.py

@@ -140,7 +140,10 @@ class TestCase(greentest.TestCase):
                 conn.close()
 
         ex = exc.exception
-        self.assertIn(ex.args[0], (errno.ECONNREFUSED, errno.EADDRNOTAVAIL), ex)
+        self.assertIn(ex.args[0],
+                      (errno.ECONNREFUSED, errno.EADDRNOTAVAIL,
+                       errno.ECONNRESET, errno.ECONNABORTED),
+                      (ex, ex.args))
 
     def assert500(self):
         self.Settings.assert500(self)

src/gevent/tests/test__socket.py

@@ -22,6 +22,7 @@ from gevent.testing import support
 from gevent.testing import params
 from gevent.testing.sockets import tcp_listener
 from gevent.testing.skipping import skipWithoutExternalNetwork
+from gevent.testing.skipping import skipOnMacOnCI
 
 # we use threading on purpose so that we can test both regular and
 # gevent sockets with the same code
@@ -49,7 +50,6 @@ class Thread(_Thread):
 
 
 class TestTCP(greentest.TestCase):
-
     __timeout__ = None
     TIMEOUT_ERROR = socket.timeout
     long_data = ", ".join([str(x) for x in range(20000)])
@@ -210,7 +210,9 @@ class TestTCP(greentest.TestCase):
 
         if match_data is None:
             match_data = self.long_data
-        self.assertEqual(read_data, [match_data])
+        read_data = read_data[0].split(b',')
+        match_data = match_data.split(b',')
+        self.assertEqual(read_data, match_data)
 
     def test_sendall_str(self):
         self._test_sendall(self.long_data)
@@ -219,6 +221,7 @@ class TestTCP(greentest.TestCase):
         def test_sendall_unicode(self):
             self._test_sendall(six.text_type(self.long_data))
 
+    @skipOnMacOnCI("Sometimes fails for no apparent reason (buffering?)")
     def test_sendall_array(self):
         data = array.array("B", self.long_data)
         self._test_sendall(data)

src/gevent/tests/test__systemerror.py

@@ -41,6 +41,8 @@ class Test(greentest.TestCase):
             raise AssertionError('must raise KeyboardInterrupt')
 
     def test_keyboard_interrupt_stderr_patched(self):
+        # XXX: This one non-top-level call prevents us from being
+        # run in a process with other tests.
         from gevent import monkey
         monkey.patch_sys(stdin=False, stdout=False, stderr=True)
         try:

src/gevent/tests/test__threading.py

 """
 Tests specifically for the monkey-patched threading module.
 """
-from gevent import monkey; monkey.patch_all()
+from gevent import monkey; monkey.patch_all() # pragma: testrunner-no-monkey-combine
 import gevent.hub
 
 # check that the locks initialized by 'threading' did not init the hub

src/gevent/tests/test__threading_before_monkey.py

@@ -3,7 +3,7 @@
 
 import threading
 from gevent import monkey
-monkey.patch_all()
+monkey.patch_all() # pragma: testrunner-no-monkey-combine
 
 import gevent.testing as greentest
 

src/gevent/tests/test__threading_holding_lock_while_monkey.py

@@ -5,4 +5,4 @@ import threading
 # in python code, this used to throw RuntimeErro("Cannot release un-acquired lock")
 # See https://github.com/gevent/gevent/issues/615
 with threading.RLock():
-    monkey.patch_all()
+    monkey.patch_all() # pragma: testrunner-no-monkey-combine

src/gevent/tests/test__threading_monkey_in_thread.py

@@ -25,7 +25,7 @@ class Test(greentest.TestCase):
 
         def target():
             tcurrent = threading.current_thread()
-            monkey.patch_all()
+            monkey.patch_all() # pragma: testrunner-no-monkey-combine
             tcurrent2 = threading.current_thread()
             self.assertIsNot(tcurrent, current)
             # We get a dummy thread now

src/gevent/tests/test__threading_native_before_monkey.py

@@ -52,6 +52,6 @@ if __name__ == '__main__':
 
     # Only patch after we're running
     from gevent import monkey
-    monkey.patch_all()
+    monkey.patch_all() # pragma: testrunner-no-monkey-combine
 
     greentest.main()

src/gevent/tests/test__threading_vs_settrace.py

@@ -7,7 +7,7 @@ import gevent.testing as greentest
 
 script = """
 from gevent import monkey
-monkey.patch_all()
+monkey.patch_all() # pragma: testrunner-no-monkey-combine
 import sys, os, threading, time
 
 

src/gevent/tests/test__util.py

@@ -86,7 +86,8 @@ class TestTree(greentest.TestCase):
         # so perhaps we need a GC?
         for _ in range(3):
             gc.collect()
-
+        gevent.get_hub().resolver = None # Reset resolver, don't need to see it
+        gevent.get_hub().threadpool = None # ditto the pool
         glets = []
         l = MyLocal(42)
         assert l
@@ -135,6 +136,10 @@ class TestTree(greentest.TestCase):
     def _normalize_tree_format(self, value):
         import re
         hexobj = re.compile('0x[0123456789abcdef]+L?', re.I)
+
+        hub_repr = repr(gevent.get_hub())
+        value = value.replace(hub_repr, "<HUB>")
+
         value = hexobj.sub('X', value)
         value = value.replace('epoll', 'select')
         value = value.replace('select', 'default')
@@ -142,6 +147,7 @@ class TestTree(greentest.TestCase):
         value = re.compile(' fileno=.').sub('', value)
         value = value.replace('ref=-1', 'ref=0')
         value = value.replace("type.current_tree", 'GreenletTree.current_tree')
+        value = value.replace('gevent.tests.__main__.MyLocal', '__main__.MyLocal')
         return value
 
     @greentest.ignores_leakcheck
@@ -159,26 +165,26 @@ class TestTree(greentest.TestCase):
  :    Greenlet Locals:
  :      Local <class '__main__.MyLocal'> at X
  :        {'foo': 42}
- +--- <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ +--- <HUB>
  :          Parent: <greenlet.greenlet object at X>
  +--- <Greenlet "Greenlet-1" at X: t2>; finished with value <Greenlet "CustomName-0" at 0x
- :          Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :          Parent: <HUB>
  |    +--- <Greenlet "CustomName-0" at X: t1>; finished with exception ExpectedException()
- :                Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :                Parent: <HUB>
  +--- <Greenlet "Greenlet-2" at X: t2>; finished with value <Greenlet "CustomName-4" at 0x
- :          Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :          Parent: <HUB>
  |    +--- <Greenlet "CustomName-4" at X: t1>; finished with exception ExpectedException()
- :                Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :                Parent: <HUB>
  +--- <Greenlet "Greenlet-3" at X: t3>; finished with value <Greenlet "Greenlet-5" at X
- :          Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :          Parent: <HUB>
  :          Spawn Tree Locals
  :          {'stl': 'STL'}
  |    +--- <Greenlet "Greenlet-5" at X: t2>; finished with value <Greenlet "CustomName-6" at 0x
- :                Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :                Parent: <HUB>
  |         +--- <Greenlet "CustomName-6" at X: t1>; finished with exception ExpectedException()
- :                      Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+ :                      Parent: <HUB>
  +--- <Greenlet "Greenlet-7" at X: <bound method GreenletTree.current_tree of <class 'gevent.util.GreenletTree'>>>; finished with value <gevent.util.GreenletTree obje
-            Parent: <QuietHub '' at X default default pending=0 ref=0 thread_ident=X>
+            Parent: <HUB>
         """.strip()
         self.assertEqual(expected, value)