diff --git a/tests/test_pipes.py b/tests/test_pipes.py index 50b9d4b6..9ff567e4 100644 --- a/tests/test_pipes.py +++ b/tests/test_pipes.py @@ -1,8 +1,8 @@ import asyncio import io import os +import socket -from asyncio import test_utils from uvloop import _testbase as tb @@ -81,11 +81,11 @@ def connect(): self.loop.run_until_complete(connect()) os.write(wpipe, b'1') - test_utils.run_until(self.loop, lambda: proto.nbytes >= 1) + tb.run_until(self.loop, lambda: proto.nbytes >= 1) self.assertEqual(1, proto.nbytes) os.write(wpipe, b'2345') - test_utils.run_until(self.loop, lambda: proto.nbytes >= 5) + tb.run_until(self.loop, lambda: proto.nbytes >= 5) self.assertEqual(['INITIAL', 'CONNECTED'], proto.state) self.assertEqual(5, proto.nbytes) @@ -114,11 +114,11 @@ def connect(): self.loop.run_until_complete(connect()) os.write(slave, b'1') - test_utils.run_until(self.loop, lambda: proto.nbytes) + tb.run_until(self.loop, lambda: proto.nbytes) self.assertEqual(1, proto.nbytes) os.write(slave, b'2345') - test_utils.run_until(self.loop, lambda: proto.nbytes >= 5) + tb.run_until(self.loop, lambda: proto.nbytes >= 5) self.assertEqual(['INITIAL', 'CONNECTED'], proto.state) self.assertEqual(5, proto.nbytes) @@ -158,11 +158,11 @@ def reader(data): data += chunk return len(data) - test_utils.run_until(self.loop, lambda: reader(data) >= 1) + tb.run_until(self.loop, lambda: reader(data) >= 1) self.assertEqual(b'1', data) transport.write(b'2345') - test_utils.run_until(self.loop, lambda: reader(data) >= 5) + tb.run_until(self.loop, lambda: reader(data) >= 5) self.assertEqual(b'12345', data) self.assertEqual('CONNECTED', proto.state) @@ -177,7 +177,7 @@ def reader(data): self.assertEqual('CLOSED', proto.state) def test_write_pipe_disconnect_on_close(self): - rsock, wsock = test_utils.socketpair() + rsock, wsock = socket.socketpair() rsock.setblocking(False) pipeobj = io.open(wsock.detach(), 'wb', 1024) @@ -223,12 +223,12 @@ def reader(data): data += chunk return len(data) - test_utils.run_until(self.loop, lambda: reader(data) >= 1, + tb.run_until(self.loop, lambda: reader(data) >= 1, timeout=10) self.assertEqual(b'1', data) transport.write(b'2345') - test_utils.run_until(self.loop, lambda: reader(data) >= 5, + tb.run_until(self.loop, lambda: reader(data) >= 5, timeout=10) self.assertEqual(b'12345', data) self.assertEqual('CONNECTED', proto.state) diff --git a/tests/test_process.py b/tests/test_process.py index 51e1970a..fced8562 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -9,7 +9,6 @@ import time import unittest -from asyncio import test_utils from uvloop import _testbase as tb @@ -504,7 +503,7 @@ def cancel_make_transport(): # ignore the log: # "Exception during subprocess creation, kill the subprocess" - with test_utils.disable_logger(): + with tb.disable_logger(): self.loop.run_until_complete(cancel_make_transport()) def test_cancel_post_init(self): @@ -522,9 +521,9 @@ def cancel_make_transport(): # ignore the log: # "Exception during subprocess creation, kill the subprocess" - with test_utils.disable_logger(): + with tb.disable_logger(): self.loop.run_until_complete(cancel_make_transport()) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) class Test_UV_Process(_TestProcess, tb.UVTestCase): diff --git a/tests/test_tasks.py b/tests/test_tasks.py index b237ab67..f2a7689a 100644 --- a/tests/test_tasks.py +++ b/tests/test_tasks.py @@ -2,7 +2,6 @@ import asyncio -from asyncio import test_utils from uvloop import _testbase as tb @@ -30,59 +29,6 @@ def format_coroutine(qualname, state, src, source_traceback, generator=False): class _TestTasks: - def test_task_repr(self): - self.loop.set_debug(False) - - @asyncio.coroutine - def notmuch(): - yield from [] - return 'abc' - - # test coroutine function - self.assertEqual(notmuch.__name__, 'notmuch') - self.assertEqual(notmuch.__qualname__, - '_TestTasks.test_task_repr..notmuch') - self.assertEqual(notmuch.__module__, __name__) - - filename, lineno = test_utils.get_function_source(notmuch) - src = "%s:%s" % (filename, lineno) - - # test coroutine object - gen = notmuch() - coro_qualname = '_TestTasks.test_task_repr..notmuch' - self.assertEqual(gen.__name__, 'notmuch') - self.assertEqual(gen.__qualname__, coro_qualname) - - # test pending Task - t = asyncio.Task(gen, loop=self.loop) - t.add_done_callback(Dummy()) - - coro = format_coroutine(coro_qualname, 'running', src, - t._source_traceback, generator=True) - self.assertEqual(repr(t), - '()]>' % coro) - - # test canceling Task - t.cancel() # Does not take immediate effect! - self.assertEqual(repr(t), - '()]>' % coro) - - # test canceled Task - self.assertRaises(asyncio.CancelledError, - self.loop.run_until_complete, t) - coro = format_coroutine(coro_qualname, 'done', src, - t._source_traceback) - self.assertEqual(repr(t), - '' % coro) - - # test finished Task - t = asyncio.Task(notmuch(), loop=self.loop) - self.loop.run_until_complete(t) - coro = format_coroutine(coro_qualname, 'done', src, - t._source_traceback) - self.assertEqual(repr(t), - "" % coro) - def test_task_basics(self): @asyncio.coroutine def outer(): @@ -109,7 +55,7 @@ def task(): return 12 t = self.create_task(task()) - test_utils.run_briefly(self.loop) # start coro + tb.run_briefly(self.loop) # start coro t.cancel() self.assertRaises( asyncio.CancelledError, self.loop.run_until_complete, t) @@ -126,7 +72,7 @@ def task(): return 12 t = self.create_task(task()) - test_utils.run_briefly(self.loop) # start task + tb.run_briefly(self.loop) # start task f.cancel() with self.assertRaises(asyncio.CancelledError): self.loop.run_until_complete(t) @@ -143,7 +89,7 @@ def task(): t = self.create_task(task()) self.assertEqual(asyncio.Task.all_tasks(loop=self.loop), {t}) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) f.cancel() t.cancel() @@ -168,10 +114,10 @@ def task(): return 42 t = self.create_task(task()) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertIs(t._fut_waiter, fut1) # White-box test. fut1.set_result(None) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertIs(t._fut_waiter, fut2) # White-box test. t.cancel() self.assertTrue(fut2.cancelled()) @@ -195,14 +141,14 @@ def task(): return res t = self.create_task(task()) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertIs(t._fut_waiter, fut1) # White-box test. fut1.set_result(None) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertIs(t._fut_waiter, fut2) # White-box test. t.cancel() self.assertTrue(fut2.cancelled()) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertIs(t._fut_waiter, fut3) # White-box test. fut3.set_result(42) res = self.loop.run_until_complete(t) @@ -232,7 +178,8 @@ def notmutch(): raise BaseException() task = self.create_task(notmutch()) - self.assertRaises(BaseException, task._step) + with self.assertRaises(BaseException): + tb.run_briefly(self.loop) self.assertTrue(task.done()) self.assertIsInstance(task.exception(), BaseException) @@ -245,7 +192,7 @@ def __init__(self, *args, **kwds): self.cb_added = False super().__init__(*args, **kwds) - def add_done_callback(self, fn): + def add_done_callback(self, fn, context=None): self.cb_added = True super().add_done_callback(fn) @@ -258,12 +205,12 @@ def wait_for_future(): result = yield from fut t = self.create_task(wait_for_future()) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertTrue(fut.cb_added) res = object() fut.set_result(res) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertIs(res, result) self.assertTrue(t.done()) self.assertIsNone(t.result()) @@ -356,7 +303,7 @@ def outer(): proof += 10 f = asyncio.ensure_future(outer(), loop=self.loop) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) f.cancel() self.loop.run_until_complete(f) self.assertEqual(proof, 101) @@ -381,12 +328,12 @@ def outer(): proof += 100 f = asyncio.ensure_future(outer(), loop=self.loop) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) f.cancel() self.assertRaises( asyncio.CancelledError, self.loop.run_until_complete, f) waiter.set_result(None) - test_utils.run_briefly(self.loop) + tb.run_briefly(self.loop) self.assertEqual(proof, 1) diff --git a/tests/test_udp.py b/tests/test_udp.py index b3f79e8b..1e5be9b5 100644 --- a/tests/test_udp.py +++ b/tests/test_udp.py @@ -3,7 +3,6 @@ import unittest import sys -from asyncio import test_utils from uvloop import _testbase as tb @@ -77,9 +76,9 @@ def datagram_received(self, data, addr): self.assertIs(client.transport, transport) transport.sendto(b'xxx') - test_utils.run_until(self.loop, lambda: server.nbytes) + tb.run_until(self.loop, lambda: server.nbytes) self.assertEqual(3, server.nbytes) - test_utils.run_until(self.loop, lambda: client.nbytes) + tb.run_until(self.loop, lambda: client.nbytes) # received self.assertEqual(8, client.nbytes) diff --git a/tests/test_unix.py b/tests/test_unix.py index 879b8a2f..a77311e9 100644 --- a/tests/test_unix.py +++ b/tests/test_unix.py @@ -5,6 +5,7 @@ import tempfile import time import unittest +import sys from uvloop import _testbase as tb @@ -362,6 +363,7 @@ async def client(): self.assertIn(excs[0].__class__, (BrokenPipeError, ConnectionResetError)) + @unittest.skipUnless(sys.version_info < (3, 7), 'Python version must be < 3.7') def test_transport_unclosed_warning(self): async def test(sock): return await self.loop.create_unix_connection( diff --git a/uvloop/_testbase.py b/uvloop/_testbase.py index 22ebda08..69be1cba 100644 --- a/uvloop/_testbase.py +++ b/uvloop/_testbase.py @@ -483,3 +483,46 @@ def _handle_client(self, sock): @property def addr(self): return self._sock.getsockname() + + +############################################################################### +# A few helpers from asyncio/tests/testutils.py +############################################################################### + + +def run_briefly(loop): + async def once(): + pass + gen = once() + t = loop.create_task(gen) + # Don't log a warning if the task is not done after run_until_complete(). + # It occurs if the loop is stopped or if a task raises a BaseException. + t._log_destroy_pending = False + try: + loop.run_until_complete(t) + finally: + gen.close() + + +def run_until(loop, pred, timeout=30): + deadline = time.time() + timeout + while not pred(): + if timeout is not None: + timeout = deadline - time.time() + if timeout <= 0: + raise asyncio.futures.TimeoutError() + loop.run_until_complete(asyncio.tasks.sleep(0.001, loop=loop)) + + +@contextlib.contextmanager +def disable_logger(): + """Context manager to disable asyncio logger. + + For example, it can be used to ignore warnings in debug mode. + """ + old_level = asyncio.log.logger.level + try: + asyncio.log.logger.setLevel(logging.CRITICAL + 1) + yield + finally: + asyncio.log.logger.setLevel(old_level) diff --git a/uvloop/includes/stdlib.pxi b/uvloop/includes/stdlib.pxi index c09cec57..94c2dd7c 100644 --- a/uvloop/includes/stdlib.pxi +++ b/uvloop/includes/stdlib.pxi @@ -38,10 +38,10 @@ cdef aio_iscoroutinefunction = asyncio.iscoroutinefunction cdef aio_BaseProtocol = asyncio.BaseProtocol cdef aio_Protocol = asyncio.Protocol cdef aio_SSLProtocol = asyncio.sslproto.SSLProtocol -cdef aio_debug_wrapper = asyncio.coroutines.debug_wrapper cdef aio_isfuture = getattr(asyncio, 'isfuture', None) cdef aio_get_running_loop = getattr(asyncio, '_get_running_loop', None) cdef aio_set_running_loop = getattr(asyncio, '_set_running_loop', None) +cdef aio_debug_wrapper = getattr(asyncio.coroutines, 'debug_wrapper', None) cdef col_deque = collections.deque cdef col_Iterable = collections.Iterable diff --git a/uvloop/loop.pxd b/uvloop/loop.pxd index c715e028..8afacfa9 100644 --- a/uvloop/loop.pxd +++ b/uvloop/loop.pxd @@ -29,11 +29,14 @@ ctypedef object (*method2_t)(object, object, object) ctypedef object (*method3_t)(object, object, object, object) + cdef class Loop: cdef: uv.uv_loop_t *uvloop - bint _coroutine_wrapper_set + + bint _coroutine_debug_set + int _coroutine_origin_tracking_saved_depth public slow_callback_duration @@ -194,7 +197,7 @@ cdef class Loop: cdef _read_from_self(self) cdef _process_self_data(self, data) - cdef _set_coroutine_wrapper(self, bint enabled) + cdef _set_coroutine_debug(self, bint enabled) cdef _print_debug_info(self) diff --git a/uvloop/loop.pyx b/uvloop/loop.pyx index 44292dc8..7a5e0dc4 100644 --- a/uvloop/loop.pyx +++ b/uvloop/loop.pyx @@ -152,7 +152,7 @@ cdef class Loop: self._signal_handlers = {} self._listening_signals = False - self._coroutine_wrapper_set = False + self._coroutine_debug_set = False if hasattr(sys, 'get_asyncgen_hooks'): # Python >= 3.6 @@ -970,32 +970,45 @@ cdef class Loop: if err < 0: raise convert_error(-errno.errno) - cdef _set_coroutine_wrapper(self, bint enabled): + cdef _set_coroutine_debug(self, bint enabled): enabled = bool(enabled) - if self._coroutine_wrapper_set == enabled: + if self._coroutine_debug_set == enabled: return - wrapper = aio_debug_wrapper - current_wrapper = sys_get_coroutine_wrapper() - - if enabled: - if current_wrapper not in (None, wrapper): - warnings.warn( - "loop.set_debug(True): cannot set debug coroutine " - "wrapper; another wrapper is already set %r" % - current_wrapper, RuntimeWarning) + if sys_version_info >= (3, 7, 0): + if enabled: + self._coroutine_origin_tracking_saved_depth = ( + sys.get_coroutine_origin_tracking_depth()) + sys.set_coroutine_origin_tracking_depth( + DEBUG_STACK_DEPTH) else: - sys_set_coroutine_wrapper(wrapper) - self._coroutine_wrapper_set = True + sys.set_coroutine_origin_tracking_depth( + self._coroutine_origin_tracking_saved_depth) + + self._coroutine_debug_set = enabled else: - if current_wrapper not in (None, wrapper): - warnings.warn( - "loop.set_debug(False): cannot unset debug coroutine " - "wrapper; another wrapper was set %r" % - current_wrapper, RuntimeWarning) + wrapper = aio_debug_wrapper + current_wrapper = sys_get_coroutine_wrapper() + + if enabled: + if current_wrapper not in (None, wrapper): + warnings.warn( + "loop.set_debug(True): cannot set debug coroutine " + "wrapper; another wrapper is already set %r" % + current_wrapper, RuntimeWarning) + else: + sys_set_coroutine_wrapper(wrapper) + self._coroutine_debug_set = True else: - sys_set_coroutine_wrapper(None) - self._coroutine_wrapper_set = False + if current_wrapper not in (None, wrapper): + warnings.warn( + "loop.set_debug(False): cannot unset debug coroutine " + "wrapper; another wrapper was set %r" % + current_wrapper, RuntimeWarning) + else: + sys_set_coroutine_wrapper(None) + self._coroutine_debug_set = False + cdef _create_server(self, system.sockaddr *addr, object protocol_factory, @@ -1151,7 +1164,7 @@ cdef class Loop: self.is_closed(), self.get_debug()) - def call_soon(self, callback, *args): + def call_soon(self, callback, *args, context=None): """Arrange for a callback to be called as soon as possible. This operates as a FIFO queue: callbacks are called in the @@ -1168,7 +1181,7 @@ cdef class Loop: else: return self._call_soon(callback, None) - def call_soon_threadsafe(self, callback, *args): + def call_soon_threadsafe(self, callback, *args, context=None): """Like call_soon(), but thread-safe.""" if not args: args = None @@ -1176,7 +1189,7 @@ cdef class Loop: self.handler_async.send() return handle - def call_later(self, delay, callback, *args): + def call_later(self, delay, callback, *args, context=None): """Arrange for a callback to be called at a given time. Return a Handle: an opaque object with a cancel() method that @@ -1213,7 +1226,7 @@ cdef class Loop: else: return self._call_later(when, callback, args) - def call_at(self, when, callback, *args): + def call_at(self, when, callback, *args, context=None): """Like call_later(), but uses an absolute time. Absolute time corresponds to the event loop's time() method. @@ -1251,7 +1264,7 @@ cdef class Loop: # loop.stop() was called right before loop.run_forever(). # This is how asyncio loop behaves. mode = uv.UV_RUN_NOWAIT - self._set_coroutine_wrapper(self._debug) + self._set_coroutine_debug(self._debug) if self._asyncgens is not None: old_agen_hooks = sys.get_asyncgen_hooks() sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook, @@ -1259,7 +1272,7 @@ cdef class Loop: try: self._run(mode) finally: - self._set_coroutine_wrapper(False) + self._set_coroutine_debug(False) if self._asyncgens is not None: sys.set_asyncgen_hooks(*old_agen_hooks) @@ -1280,7 +1293,7 @@ cdef class Loop: def set_debug(self, enabled): self._debug = bool(enabled) if self.is_running(): - self._set_coroutine_wrapper(self._debug) + self.call_soon_threadsafe(self._set_coroutine_debug, self, self._debug) def is_running(self): """Return whether the event loop is currently running."""