Refactor test__queue for easier debugging. Solve a couple issues with leaktests and FFI

src/gevent/_ffi/loop.py

@@ -531,11 +531,20 @@ class AbstractLoop(object):
         pass
 
     def now(self):
+        "Return the loop's notion of the current time."
         raise NotImplementedError()
 
-    def update(self):
+    def update_now(self):
+        "Update the loop's notion of the current time."
         raise NotImplementedError()
 
+    def update(self):
+        import warnings
+        warnings.warn("'update' is deprecated; use 'update_now'",
+                      DeprecationWarning,
+                      stacklevel=2)
+        self.update_now()
+
     def __repr__(self):
         return '<%s at 0x%x %s>' % (self.__class__.__name__, id(self), self._format())
 

src/gevent/_ffi/watcher.py

@@ -443,12 +443,9 @@ class TimerMixin(object):
             # good idea."
             # 1.3 changed the default for this to False. Note that
             # starting Timeout objects internally still sets this to true.
-            self._update_now()
+            self.loop.update()
         super(TimerMixin, self).start(callback, *args)
 
-    def _update_now(self):
-        raise NotImplementedError()
-
     def again(self, callback, *args, **kw):
         raise NotImplementedError()
 

src/gevent/libev/corecext.ppyx

@@ -420,10 +420,12 @@ cdef public class loop [object PyGeventLoopObject, type PyGeventLoop_Type]:
         CHECK_LOOP2(self)
         return libev.ev_now(self._ptr)
 
-    def update(self):
+    def update_now(self):
         CHECK_LOOP2(self)
         libev.ev_now_update(self._ptr)
 
+    update = update_now # Old name, deprecated.
+
     def __repr__(self):
         return '<%s at 0x%x %s>' % (self.__class__.__name__, id(self), self._format())
 

src/gevent/libev/corecffi.py

@@ -334,7 +334,7 @@ class loop(AbstractLoop):
     def now(self):
         return libev.ev_now(self._ptr)
 
-    def update(self):
+    def update_now(self):
         libev.ev_now_update(self._ptr)
 
     def __repr__(self):

src/gevent/libuv/loop.py

@@ -389,7 +389,7 @@ class loop(AbstractLoop):
     def now(self):
         return libuv.uv_now(self._ptr)
 
-    def update(self):
+    def update_now(self):
         libuv.uv_update_time(self._ptr)
 
     @property

src/gevent/queue.py

@@ -75,6 +75,8 @@ class Queue(object):
        previously anyway, but that wasn't the case for PyPy.
     """
 
+    _warn_depth = 2
+
     def __init__(self, maxsize=None, items=None):
         if maxsize is not None and maxsize <= 0:
             self.maxsize = None
@@ -83,7 +85,7 @@ class Queue(object):
                 warnings.warn(
                     'Queue(0) now equivalent to Queue(None); if you want a channel, use Channel',
                     DeprecationWarning,
-                    stacklevel=2)
+                    stacklevel=self._warn_depth)
         else:
             self.maxsize = maxsize
         # Explicitly maintain order for getters and putters that block
@@ -425,6 +427,8 @@ class JoinableQueue(Queue):
     :meth:`task_done` and :meth:`join` methods.
     """
 
+    _warn_depth = 3
+
     def __init__(self, maxsize=None, items=None, unfinished_tasks=None):
         """
 

src/greentest/greentest.py

@@ -270,6 +270,11 @@ def wrap_refcount(method):
 
     # Some builtin things that we ignore
     IGNORED_TYPES = (tuple, dict, types.FrameType, types.TracebackType)
+    try:
+        callback_kind = gevent.core.callback
+    except AttributeError:
+        # Must be using FFI.
+        from gevent._ffi.callback import callback as callback_kind
 
     def type_hist():
         import collections
@@ -278,7 +283,7 @@ def wrap_refcount(method):
             k = type(x)
             if k in IGNORED_TYPES:
                 continue
-            if k == gevent.core.callback and x.callback is None and x.args is None:
+            if k == callback_kind and x.callback is None and x.args is None:
                 # these represent callbacks that have been stopped, but
                 # the event loop hasn't cycled around to run them. The only
                 # known cause of this is killing greenlets before they get a chance
@@ -488,6 +493,13 @@ class TestCase(TestCaseMetaClass("NewBase", (BaseTestCase,), {})):
 
     def setUp(self):
         super(TestCase, self).setUp()
+        # Especially if we're running in leakcheck mode, where
+        # the same test gets executed repeatedly, we need to update the
+        # current time. Tests don't always go through the full event loop,
+        # 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()
         self.close_on_teardown = []
 
     def tearDown(self):
@@ -606,20 +618,6 @@ class TestCase(TestCaseMetaClass("NewBase", (BaseTestCase,), {})):
             self.assertLessEqual(delay, max_time)
             self.assertGreaterEqual(delay, min_time)
 
-    if not hasattr(BaseTestCase, 'assertIsNot'):
-        # Methods added in 3.1, backport for 2.7
-        def assertIs(self, expr1, expr2, msg=None):
-            """Just like self.assertTrue(a is b), but with a nicer default message."""
-            if expr1 is not expr2:
-                standardMsg = '%s is not %s' % (safe_repr(expr1),
-                                                safe_repr(expr2))
-                self.fail(self._formatMessage(msg, standardMsg))
-
-        def assertIsNot(self, expr1, expr2, msg=None):
-            """Just like self.assertTrue(a is not b), but with a nicer default message."""
-            if expr1 is expr2:
-                standardMsg = 'unexpectedly identical: %s' % (safe_repr(expr1),)
-                self.fail(self._formatMessage(msg, standardMsg))
 
     def assertMonkeyPatchedFuncSignatures(self, mod_name, func_names=(), exclude=()):
         # We use inspect.getargspec because it's the only thing available
@@ -739,12 +737,10 @@ class _DelayWaitMixin(object):
     def test_outer_timeout_is_not_lost(self):
         timeout = gevent.Timeout.start_new(0.001, ref=False)
         try:
-            try:
+            with self.assertRaises(gevent.Timeout) as exc:
                 result = self.wait(timeout=1)
-            except gevent.Timeout as ex:
-                assert ex is timeout, (ex, timeout)
-            else:
-                raise AssertionError('must raise Timeout (returned %r)' % (result, ))
+
+            self.assertIs(exc.exception, timeout)
         finally:
             timeout.cancel()
 
@@ -775,7 +771,8 @@ class GenericGetTestCase(_DelayWaitMixin, TestCase):
         pass
 
     def test_raises_timeout_number(self):
-        self.assertRaises(self.Timeout, self._wait_and_check, timeout=0.01)
+        with self.assertRaises(self.Timeout):
+            self._wait_and_check(timeout=0.01)
         # get raises Timeout after timeout expired
         self.cleanup()
 
@@ -784,7 +781,7 @@ class GenericGetTestCase(_DelayWaitMixin, TestCase):
         try:
             self._wait_and_check(timeout=timeout)
         except gevent.Timeout as ex:
-            assert ex is timeout, (ex, timeout)
+            self.assertIs(ex, timeout)
         self.cleanup()
 
     def test_raises_timeout_Timeout_exc_customized(self):
@@ -793,7 +790,7 @@ class GenericGetTestCase(_DelayWaitMixin, TestCase):
         try:
             self._wait_and_check(timeout=timeout)
         except RuntimeError as ex:
-            assert ex is error, (ex, error)
+            self.assertIs(ex, error)
         self.cleanup()
 
 

src/greentest/test__queue.py

@@ -107,7 +107,7 @@ class TestQueue(TestCase):
 
         sendings = ['1', '2', '3', '4']
         evts = [AsyncResult() for x in sendings]
-        for i, x in enumerate(sendings):
+        for i, _ in enumerate(sendings):
             gevent.spawn(waiter, q, evts[i])  # XXX use waitall for them
 
         gevent.sleep(0.01)  # get 'em all waiting
@@ -244,22 +244,24 @@ class TestChannel(TestCase):
         self.assertEqual(['waiting', 'sending hello', 'hello', 'sending world', 'world', 'sent world'], events)
         g.get()
 
-    def test_task_done(self):
-        channel = queue.JoinableQueue(0)
-        X = object()
-        gevent.spawn(channel.put, X)
-        result = channel.get()
-        assert result is X, (result, X)
-        assert channel.unfinished_tasks == 1, channel.unfinished_tasks
-        channel.task_done()
-        assert channel.unfinished_tasks == 0, channel.unfinished_tasks
-
     def test_iterable(self):
         channel = queue.Channel()
         gevent.spawn(channel.put, StopIteration)
         r = list(channel)
         self.assertEqual(r, [])
 
+class TestJoinableQueue(TestCase):
+
+    def test_task_done(self):
+        channel = queue.JoinableQueue()
+        X = object()
+        gevent.spawn(channel.put, X)
+        result = channel.get()
+        self.assertIs(result, X)
+        self.assertEqual(1, channel.unfinished_tasks)
+        channel.task_done()
+        self.assertEqual(0, channel.unfinished_tasks)
+
 
 class TestNoWait(TestCase):
 
@@ -370,44 +372,62 @@ class TestJoinEmpty(TestCase):
         q.join()
 
 
-def make_get_interrupt(queue_type):
-
-    class TestGetInterrupt(GenericGetTestCase):
+class TestGetInterrupt(GenericGetTestCase):
 
     Timeout = Empty
 
+    kind = queue.Queue
+
     def wait(self, timeout):
-            return queue_type().get(timeout=timeout)
+        return self._makeOne().get(timeout=timeout)
 
-    TestGetInterrupt.__name__ += '_' + queue_type.__name__
-    return TestGetInterrupt
+    def _makeOne(self):
+        return self.kind()
 
+class TestGetInterruptJoinableQueue(TestGetInterrupt):
+    kind = queue.JoinableQueue
 
-for queue_type in [queue.Queue, queue.JoinableQueue, queue.LifoQueue, queue.PriorityQueue, queue.Channel]:
-    klass = make_get_interrupt(queue_type)
-    globals()[klass.__name__] = klass
-del klass, queue_type
+class TestGetInterruptLifoQueue(TestGetInterrupt):
+    kind = queue.LifoQueue
 
+class TestGetInterruptPriorityQueue(TestGetInterrupt):
+    kind = queue.PriorityQueue
 
-def make_put_interrupt(queue):
+class TestGetInterruptChannel(TestGetInterrupt):
+    kind = queue.Channel
 
-    class TestPutInterrupt(GenericGetTestCase):
 
+class TestPutInterrupt(GenericGetTestCase):
+    kind = queue.Queue
     Timeout = Full
 
+    def setUp(self):
+        super(TestPutInterrupt, self).setUp()
+        self.queue = self._makeOne()
+
     def wait(self, timeout):
-            while not queue.full():
-                queue.put(1)
-            return queue.put(2, timeout=timeout)
+        while not self.queue.full():
+            self.queue.put(1)
+        return self.queue.put(2, timeout=timeout)
+
+    def _makeOne(self):
+        return self.kind(1)
+
+
+class TestPutInterruptJoinableQueue(TestPutInterrupt):
+    kind = queue.JoinableQueue
+
+class TestPutInterruptLifoQueue(TestPutInterrupt):
+    kind = queue.LifoQueue
 
-    TestPutInterrupt.__name__ += '_' + queue.__class__.__name__
-    return TestPutInterrupt
+class TestPutInterruptPriorityQueue(TestPutInterrupt):
+    kind = queue.PriorityQueue
 
+class TestPutInterruptChannel(TestPutInterrupt):
+    kind = queue.Channel
 
-for obj in [queue.Queue(1), queue.JoinableQueue(1), queue.LifoQueue(1), queue.PriorityQueue(1), queue.Channel()]:
-    klass = make_put_interrupt(obj)
-    globals()[klass.__name__] = klass
-del klass, obj
+    def _makeOne(self):
+        return self.kind()
 
 
 del GenericGetTestCase

src/greentest/util.py

@@ -110,7 +110,7 @@ def getname(command, env=None, setenv=None):
     env.update(setenv or {})
 
     for key, value in sorted(env.items()):
-        if key.startswith('GEVENT_') or key.startswith('GEVENTARES_'):
+        if key.startswith('GEVENT'):
             result.append('%s=%s' % (key, value))
 
     if isinstance(command, six.string_types):