bpo-32327: Convert asyncio functions documented as coroutines to coroutines. (#4872)

This commit is contained in:
Yury Selivanov 2017-12-14 20:53:26 -05:00 committed by GitHub
parent 41264f1cd4
commit 19a44f63c7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 170 additions and 195 deletions

View File

@ -269,9 +269,8 @@ Creating connections
socket type :py:data:`~socket.SOCK_STREAM`. *protocol_factory* must be a socket type :py:data:`~socket.SOCK_STREAM`. *protocol_factory* must be a
callable returning a :ref:`protocol <asyncio-protocol>` instance. callable returning a :ref:`protocol <asyncio-protocol>` instance.
This method is a :ref:`coroutine <coroutine>` which will try to This method will try to establish the connection in the background.
establish the connection in the background. When successful, the When successful, it returns a ``(transport, protocol)`` pair.
coroutine returns a ``(transport, protocol)`` pair.
The chronological synopsis of the underlying operation is as follows: The chronological synopsis of the underlying operation is as follows:
@ -344,9 +343,8 @@ Creating connections
:py:data:`~socket.SOCK_DGRAM`. *protocol_factory* must be a :py:data:`~socket.SOCK_DGRAM`. *protocol_factory* must be a
callable returning a :ref:`protocol <asyncio-protocol>` instance. callable returning a :ref:`protocol <asyncio-protocol>` instance.
This method is a :ref:`coroutine <coroutine>` which will try to This method will try to establish the connection in the background.
establish the connection in the background. When successful, the When successful, the it returns a ``(transport, protocol)`` pair.
coroutine returns a ``(transport, protocol)`` pair.
Options changing how the connection is created: Options changing how the connection is created:
@ -395,9 +393,8 @@ Creating connections
family is used to communicate between processes on the same machine family is used to communicate between processes on the same machine
efficiently. efficiently.
This method is a :ref:`coroutine <coroutine>` which will try to This method will try to establish the connection in the background.
establish the connection in the background. When successful, the When successful, the it returns a ``(transport, protocol)`` pair.
coroutine returns a ``(transport, protocol)`` pair.
*path* is the name of a UNIX domain socket, and is required unless a *sock* *path* is the name of a UNIX domain socket, and is required unless a *sock*
parameter is specified. Abstract UNIX sockets, :class:`str`, parameter is specified. Abstract UNIX sockets, :class:`str`,
@ -459,8 +456,6 @@ Creating listening connections
set this flag when being created. This option is not supported on set this flag when being created. This option is not supported on
Windows. Windows.
This method is a :ref:`coroutine <coroutine>`.
.. versionchanged:: 3.5 .. versionchanged:: 3.5
On Windows with :class:`ProactorEventLoop`, SSL/TLS is now supported. On Windows with :class:`ProactorEventLoop`, SSL/TLS is now supported.
@ -484,8 +479,6 @@ Creating listening connections
parameter is specified. Abstract UNIX sockets, :class:`str`, parameter is specified. Abstract UNIX sockets, :class:`str`,
:class:`bytes`, and :class:`~pathlib.Path` paths are supported. :class:`bytes`, and :class:`~pathlib.Path` paths are supported.
This method is a :ref:`coroutine <coroutine>`.
Availability: UNIX. Availability: UNIX.
.. versionchanged:: 3.7 .. versionchanged:: 3.7
@ -507,8 +500,7 @@ Creating listening connections
* *ssl* can be set to an :class:`~ssl.SSLContext` to enable SSL over the * *ssl* can be set to an :class:`~ssl.SSLContext` to enable SSL over the
accepted connections. accepted connections.
This method is a :ref:`coroutine <coroutine>`. When completed, the When completed it returns a ``(transport, protocol)`` pair.
coroutine returns a ``(transport, protocol)`` pair.
.. versionadded:: 3.5.3 .. versionadded:: 3.5.3
@ -565,7 +557,10 @@ Low-level socket operations
With :class:`SelectorEventLoop` event loop, the socket *sock* must be With :class:`SelectorEventLoop` event loop, the socket *sock* must be
non-blocking. non-blocking.
This method is a :ref:`coroutine <coroutine>`. .. versionchanged:: 3.7
Even though the method was always documented as a coroutine
method, before Python 3.7 it returned a :class:`Future`.
Since Python 3.7, this is an ``async def`` method.
.. coroutinemethod:: AbstractEventLoop.sock_recv_into(sock, buf) .. coroutinemethod:: AbstractEventLoop.sock_recv_into(sock, buf)
@ -578,8 +573,6 @@ Low-level socket operations
With :class:`SelectorEventLoop` event loop, the socket *sock* must be With :class:`SelectorEventLoop` event loop, the socket *sock* must be
non-blocking. non-blocking.
This method is a :ref:`coroutine <coroutine>`.
.. versionadded:: 3.7 .. versionadded:: 3.7
.. coroutinemethod:: AbstractEventLoop.sock_sendall(sock, data) .. coroutinemethod:: AbstractEventLoop.sock_sendall(sock, data)
@ -596,7 +589,10 @@ Low-level socket operations
With :class:`SelectorEventLoop` event loop, the socket *sock* must be With :class:`SelectorEventLoop` event loop, the socket *sock* must be
non-blocking. non-blocking.
This method is a :ref:`coroutine <coroutine>`. .. versionchanged:: 3.7
Even though the method was always documented as a coroutine
method, before Python 3.7 it returned an :class:`Future`.
Since Python 3.7, this is an ``async def`` method.
.. coroutinemethod:: AbstractEventLoop.sock_connect(sock, address) .. coroutinemethod:: AbstractEventLoop.sock_connect(sock, address)
@ -606,8 +602,6 @@ Low-level socket operations
With :class:`SelectorEventLoop` event loop, the socket *sock* must be With :class:`SelectorEventLoop` event loop, the socket *sock* must be
non-blocking. non-blocking.
This method is a :ref:`coroutine <coroutine>`.
.. versionchanged:: 3.5.2 .. versionchanged:: 3.5.2
``address`` no longer needs to be resolved. ``sock_connect`` ``address`` no longer needs to be resolved. ``sock_connect``
will try to check if the *address* is already resolved by calling will try to check if the *address* is already resolved by calling
@ -634,7 +628,10 @@ Low-level socket operations
The socket *sock* must be non-blocking. The socket *sock* must be non-blocking.
This method is a :ref:`coroutine <coroutine>`. .. versionchanged:: 3.7
Even though the method was always documented as a coroutine
method, before Python 3.7 it returned a :class:`Future`.
Since Python 3.7, this is an ``async def`` method.
.. seealso:: .. seealso::
@ -673,8 +670,6 @@ Use :class:`ProactorEventLoop` to support pipes on Windows.
With :class:`SelectorEventLoop` event loop, the *pipe* is set to With :class:`SelectorEventLoop` event loop, the *pipe* is set to
non-blocking mode. non-blocking mode.
This method is a :ref:`coroutine <coroutine>`.
.. coroutinemethod:: AbstractEventLoop.connect_write_pipe(protocol_factory, pipe) .. coroutinemethod:: AbstractEventLoop.connect_write_pipe(protocol_factory, pipe)
Register write pipe in eventloop. Register write pipe in eventloop.
@ -687,8 +682,6 @@ Use :class:`ProactorEventLoop` to support pipes on Windows.
With :class:`SelectorEventLoop` event loop, the *pipe* is set to With :class:`SelectorEventLoop` event loop, the *pipe* is set to
non-blocking mode. non-blocking mode.
This method is a :ref:`coroutine <coroutine>`.
.. seealso:: .. seealso::
The :meth:`AbstractEventLoop.subprocess_exec` and The :meth:`AbstractEventLoop.subprocess_exec` and
@ -738,8 +731,6 @@ pool of processes). By default, an event loop uses a thread pool executor
:ref:`Use functools.partial to pass keywords to the *func* :ref:`Use functools.partial to pass keywords to the *func*
<asyncio-pass-keywords>`. <asyncio-pass-keywords>`.
This method is a :ref:`coroutine <coroutine>`.
.. versionchanged:: 3.5.3 .. versionchanged:: 3.5.3
:meth:`BaseEventLoop.run_in_executor` no longer configures the :meth:`BaseEventLoop.run_in_executor` no longer configures the
``max_workers`` of the thread pool executor it creates, instead ``max_workers`` of the thread pool executor it creates, instead
@ -747,6 +738,11 @@ pool of processes). By default, an event loop uses a thread pool executor
(:class:`~concurrent.futures.ThreadPoolExecutor`) to set the (:class:`~concurrent.futures.ThreadPoolExecutor`) to set the
default. default.
.. versionchanged:: 3.7
Even though the method was always documented as a coroutine
method, before Python 3.7 it returned a :class:`Future`.
Since Python 3.7, this is an ``async def`` method.
.. method:: AbstractEventLoop.set_default_executor(executor) .. method:: AbstractEventLoop.set_default_executor(executor)
Set the default executor used by :meth:`run_in_executor`. Set the default executor used by :meth:`run_in_executor`.
@ -857,8 +853,6 @@ Server
Wait until the :meth:`close` method completes. Wait until the :meth:`close` method completes.
This method is a :ref:`coroutine <coroutine>`.
.. attribute:: sockets .. attribute:: sockets
List of :class:`socket.socket` objects the server is listening to, or List of :class:`socket.socket` objects the server is listening to, or

View File

@ -157,20 +157,6 @@ def _ipaddr_info(host, port, family, type, proto):
return None return None
def _ensure_resolved(address, *, family=0, type=socket.SOCK_STREAM, proto=0,
flags=0, loop):
host, port = address[:2]
info = _ipaddr_info(host, port, family, type, proto)
if info is not None:
# "host" is already a resolved IP.
fut = loop.create_future()
fut.set_result([info])
return fut
else:
return loop.getaddrinfo(host, port, family=family, type=type,
proto=proto, flags=flags)
def _run_until_complete_cb(fut): def _run_until_complete_cb(fut):
exc = fut._exception exc = fut._exception
if isinstance(exc, BaseException) and not isinstance(exc, Exception): if isinstance(exc, BaseException) and not isinstance(exc, Exception):
@ -614,7 +600,7 @@ class BaseEventLoop(events.AbstractEventLoop):
self._write_to_self() self._write_to_self()
return handle return handle
def run_in_executor(self, executor, func, *args): async def run_in_executor(self, executor, func, *args):
self._check_closed() self._check_closed()
if self._debug: if self._debug:
self._check_callback(func, 'run_in_executor') self._check_callback(func, 'run_in_executor')
@ -623,7 +609,8 @@ class BaseEventLoop(events.AbstractEventLoop):
if executor is None: if executor is None:
executor = concurrent.futures.ThreadPoolExecutor() executor = concurrent.futures.ThreadPoolExecutor()
self._default_executor = executor self._default_executor = executor
return futures.wrap_future(executor.submit(func, *args), loop=self) return await futures.wrap_future(
executor.submit(func, *args), loop=self)
def set_default_executor(self, executor): def set_default_executor(self, executor):
self._default_executor = executor self._default_executor = executor
@ -652,17 +639,19 @@ class BaseEventLoop(events.AbstractEventLoop):
logger.debug(msg) logger.debug(msg)
return addrinfo return addrinfo
def getaddrinfo(self, host, port, *, async def getaddrinfo(self, host, port, *,
family=0, type=0, proto=0, flags=0): family=0, type=0, proto=0, flags=0):
if self._debug: if self._debug:
return self.run_in_executor(None, self._getaddrinfo_debug, getaddr_func = self._getaddrinfo_debug
host, port, family, type, proto, flags)
else: else:
return self.run_in_executor(None, socket.getaddrinfo, getaddr_func = socket.getaddrinfo
host, port, family, type, proto, flags)
def getnameinfo(self, sockaddr, flags=0): return await self.run_in_executor(
return self.run_in_executor(None, socket.getnameinfo, sockaddr, flags) None, getaddr_func, host, port, family, type, proto, flags)
async def getnameinfo(self, sockaddr, flags=0):
return await self.run_in_executor(
None, socket.getnameinfo, sockaddr, flags)
async def create_connection(self, protocol_factory, host=None, port=None, async def create_connection(self, protocol_factory, host=None, port=None,
*, ssl=None, family=0, *, ssl=None, family=0,
@ -703,25 +692,17 @@ class BaseEventLoop(events.AbstractEventLoop):
raise ValueError( raise ValueError(
'host/port and sock can not be specified at the same time') 'host/port and sock can not be specified at the same time')
f1 = _ensure_resolved((host, port), family=family, infos = await self._ensure_resolved(
type=socket.SOCK_STREAM, proto=proto, (host, port), family=family,
flags=flags, loop=self) type=socket.SOCK_STREAM, proto=proto, flags=flags, loop=self)
fs = [f1]
if local_addr is not None:
f2 = _ensure_resolved(local_addr, family=family,
type=socket.SOCK_STREAM, proto=proto,
flags=flags, loop=self)
fs.append(f2)
else:
f2 = None
await tasks.wait(fs, loop=self)
infos = f1.result()
if not infos: if not infos:
raise OSError('getaddrinfo() returned empty list') raise OSError('getaddrinfo() returned empty list')
if f2 is not None:
laddr_infos = f2.result() if local_addr is not None:
laddr_infos = await self._ensure_resolved(
local_addr, family=family,
type=socket.SOCK_STREAM, proto=proto,
flags=flags, loop=self)
if not laddr_infos: if not laddr_infos:
raise OSError('getaddrinfo() returned empty list') raise OSError('getaddrinfo() returned empty list')
@ -730,7 +711,7 @@ class BaseEventLoop(events.AbstractEventLoop):
try: try:
sock = socket.socket(family=family, type=type, proto=proto) sock = socket.socket(family=family, type=type, proto=proto)
sock.setblocking(False) sock.setblocking(False)
if f2 is not None: if local_addr is not None:
for _, _, _, _, laddr in laddr_infos: for _, _, _, _, laddr in laddr_infos:
try: try:
sock.bind(laddr) sock.bind(laddr)
@ -863,7 +844,7 @@ class BaseEventLoop(events.AbstractEventLoop):
assert isinstance(addr, tuple) and len(addr) == 2, ( assert isinstance(addr, tuple) and len(addr) == 2, (
'2-tuple is expected') '2-tuple is expected')
infos = await _ensure_resolved( infos = await self._ensure_resolved(
addr, family=family, type=socket.SOCK_DGRAM, addr, family=family, type=socket.SOCK_DGRAM,
proto=proto, flags=flags, loop=self) proto=proto, flags=flags, loop=self)
if not infos: if not infos:
@ -946,10 +927,22 @@ class BaseEventLoop(events.AbstractEventLoop):
return transport, protocol return transport, protocol
async def _ensure_resolved(self, address, *,
family=0, type=socket.SOCK_STREAM,
proto=0, flags=0, loop):
host, port = address[:2]
info = _ipaddr_info(host, port, family, type, proto)
if info is not None:
# "host" is already a resolved IP.
return [info]
else:
return await loop.getaddrinfo(host, port, family=family, type=type,
proto=proto, flags=flags)
async def _create_server_getaddrinfo(self, host, port, family, flags): async def _create_server_getaddrinfo(self, host, port, family, flags):
infos = await _ensure_resolved((host, port), family=family, infos = await self._ensure_resolved((host, port), family=family,
type=socket.SOCK_STREAM, type=socket.SOCK_STREAM,
flags=flags, loop=self) flags=flags, loop=self)
if not infos: if not infos:
raise OSError(f'getaddrinfo({host!r}) returned empty list') raise OSError(f'getaddrinfo({host!r}) returned empty list')
return infos return infos

View File

@ -432,20 +432,20 @@ class BaseProactorEventLoop(base_events.BaseEventLoop):
# Close the event loop # Close the event loop
super().close() super().close()
def sock_recv(self, sock, n): async def sock_recv(self, sock, n):
return self._proactor.recv(sock, n) return await self._proactor.recv(sock, n)
def sock_recv_into(self, sock, buf): async def sock_recv_into(self, sock, buf):
return self._proactor.recv_into(sock, buf) return await self._proactor.recv_into(sock, buf)
def sock_sendall(self, sock, data): async def sock_sendall(self, sock, data):
return self._proactor.send(sock, data) return await self._proactor.send(sock, data)
def sock_connect(self, sock, address): async def sock_connect(self, sock, address):
return self._proactor.connect(sock, address) return await self._proactor.connect(sock, address)
def sock_accept(self, sock): async def sock_accept(self, sock):
return self._proactor.accept(sock) return await self._proactor.accept(sock)
def _close_self_pipe(self): def _close_self_pipe(self):
if self._self_reading_future is not None: if self._self_reading_future is not None:

View File

@ -336,20 +336,18 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
self._ensure_fd_no_transport(fd) self._ensure_fd_no_transport(fd)
return self._remove_writer(fd) return self._remove_writer(fd)
def sock_recv(self, sock, n): async def sock_recv(self, sock, n):
"""Receive data from the socket. """Receive data from the socket.
The return value is a bytes object representing the data received. The return value is a bytes object representing the data received.
The maximum amount of data to be received at once is specified by The maximum amount of data to be received at once is specified by
nbytes. nbytes.
This method is a coroutine.
""" """
if self._debug and sock.gettimeout() != 0: if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking") raise ValueError("the socket must be non-blocking")
fut = self.create_future() fut = self.create_future()
self._sock_recv(fut, None, sock, n) self._sock_recv(fut, None, sock, n)
return fut return await fut
def _sock_recv(self, fut, registered_fd, sock, n): def _sock_recv(self, fut, registered_fd, sock, n):
# _sock_recv() can add itself as an I/O callback if the operation can't # _sock_recv() can add itself as an I/O callback if the operation can't
@ -372,19 +370,17 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
else: else:
fut.set_result(data) fut.set_result(data)
def sock_recv_into(self, sock, buf): async def sock_recv_into(self, sock, buf):
"""Receive data from the socket. """Receive data from the socket.
The received data is written into *buf* (a writable buffer). The received data is written into *buf* (a writable buffer).
The return value is the number of bytes written. The return value is the number of bytes written.
This method is a coroutine.
""" """
if self._debug and sock.gettimeout() != 0: if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking") raise ValueError("the socket must be non-blocking")
fut = self.create_future() fut = self.create_future()
self._sock_recv_into(fut, None, sock, buf) self._sock_recv_into(fut, None, sock, buf)
return fut return await fut
def _sock_recv_into(self, fut, registered_fd, sock, buf): def _sock_recv_into(self, fut, registered_fd, sock, buf):
# _sock_recv_into() can add itself as an I/O callback if the operation # _sock_recv_into() can add itself as an I/O callback if the operation
@ -408,7 +404,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
else: else:
fut.set_result(nbytes) fut.set_result(nbytes)
def sock_sendall(self, sock, data): async def sock_sendall(self, sock, data):
"""Send data to the socket. """Send data to the socket.
The socket must be connected to a remote socket. This method continues The socket must be connected to a remote socket. This method continues
@ -416,8 +412,6 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
error occurs. None is returned on success. On error, an exception is error occurs. None is returned on success. On error, an exception is
raised, and there is no way to determine how much data, if any, was raised, and there is no way to determine how much data, if any, was
successfully processed by the receiving end of the connection. successfully processed by the receiving end of the connection.
This method is a coroutine.
""" """
if self._debug and sock.gettimeout() != 0: if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking") raise ValueError("the socket must be non-blocking")
@ -426,7 +420,7 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
self._sock_sendall(fut, None, sock, data) self._sock_sendall(fut, None, sock, data)
else: else:
fut.set_result(None) fut.set_result(None)
return fut return await fut
def _sock_sendall(self, fut, registered_fd, sock, data): def _sock_sendall(self, fut, registered_fd, sock, data):
if registered_fd is not None: if registered_fd is not None:
@ -459,11 +453,9 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
raise ValueError("the socket must be non-blocking") raise ValueError("the socket must be non-blocking")
if not hasattr(socket, 'AF_UNIX') or sock.family != socket.AF_UNIX: if not hasattr(socket, 'AF_UNIX') or sock.family != socket.AF_UNIX:
resolved = base_events._ensure_resolved( resolved = await self._ensure_resolved(
address, family=sock.family, proto=sock.proto, loop=self) address, family=sock.family, proto=sock.proto, loop=self)
if not resolved.done(): _, _, _, _, address = resolved[0]
await resolved
_, _, _, _, address = resolved.result()[0]
fut = self.create_future() fut = self.create_future()
self._sock_connect(fut, sock, address) self._sock_connect(fut, sock, address)
@ -506,21 +498,19 @@ class BaseSelectorEventLoop(base_events.BaseEventLoop):
else: else:
fut.set_result(None) fut.set_result(None)
def sock_accept(self, sock): async def sock_accept(self, sock):
"""Accept a connection. """Accept a connection.
The socket must be bound to an address and listening for connections. The socket must be bound to an address and listening for connections.
The return value is a pair (conn, address) where conn is a new socket The return value is a pair (conn, address) where conn is a new socket
object usable to send and receive data on the connection, and address object usable to send and receive data on the connection, and address
is the address bound to the socket on the other end of the connection. is the address bound to the socket on the other end of the connection.
This method is a coroutine.
""" """
if self._debug and sock.gettimeout() != 0: if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking") raise ValueError("the socket must be non-blocking")
fut = self.create_future() fut = self.create_future()
self._sock_accept(fut, False, sock) self._sock_accept(fut, False, sock)
return fut return await fut
def _sock_accept(self, fut, registered, sock): def _sock_accept(self, fut, registered, sock):
fd = sock.fileno() fd = sock.fileno()

View File

@ -217,14 +217,6 @@ class BaseEventLoopTests(test_utils.TestCase):
self.loop.set_default_executor(executor) self.loop.set_default_executor(executor)
self.assertIs(executor, self.loop._default_executor) self.assertIs(executor, self.loop._default_executor)
def test_getnameinfo(self):
sockaddr = mock.Mock()
self.loop.run_in_executor = mock.Mock()
self.loop.getnameinfo(sockaddr)
self.assertEqual(
(None, socket.getnameinfo, sockaddr, 0),
self.loop.run_in_executor.call_args[0])
def test_call_soon(self): def test_call_soon(self):
def cb(): def cb():
pass pass
@ -345,26 +337,6 @@ class BaseEventLoopTests(test_utils.TestCase):
# check disabled if debug mode is disabled # check disabled if debug mode is disabled
test_thread(self.loop, False, create_loop=True) test_thread(self.loop, False, create_loop=True)
def test_run_once_in_executor_plain(self):
def cb():
pass
f = asyncio.Future(loop=self.loop)
executor = mock.Mock()
executor.submit.return_value = f
self.loop.set_default_executor(executor)
res = self.loop.run_in_executor(None, cb)
self.assertIs(f, res)
executor = mock.Mock()
executor.submit.return_value = f
res = self.loop.run_in_executor(executor, cb)
self.assertIs(f, res)
self.assertTrue(executor.submit.called)
f.cancel() # Don't complain about abandoned Future.
def test__run_once(self): def test__run_once(self):
h1 = asyncio.TimerHandle(time.monotonic() + 5.0, lambda: True, (), h1 = asyncio.TimerHandle(time.monotonic() + 5.0, lambda: True, (),
self.loop) self.loop)
@ -1007,6 +979,12 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
self.loop = asyncio.new_event_loop() self.loop = asyncio.new_event_loop()
self.set_event_loop(self.loop) self.set_event_loop(self.loop)
@mock.patch('socket.getnameinfo')
def test_getnameinfo(self, m_gai):
m_gai.side_effect = lambda *args: 42
r = self.loop.run_until_complete(self.loop.getnameinfo(('abc', 123)))
self.assertEqual(r, 42)
@patch_socket @patch_socket
def test_create_connection_multiple_errors(self, m_socket): def test_create_connection_multiple_errors(self, m_socket):
@ -1119,9 +1097,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
OSError, self.loop.run_until_complete, coro) OSError, self.loop.run_until_complete, coro)
def test_create_connection_connect_err(self): def test_create_connection_connect_err(self):
@asyncio.coroutine async def getaddrinfo(*args, **kw):
def getaddrinfo(*args, **kw):
yield from []
return [(2, 1, 6, '', ('107.6.106.82', 80))] return [(2, 1, 6, '', ('107.6.106.82', 80))]
def getaddrinfo_task(*args, **kwds): def getaddrinfo_task(*args, **kwds):
@ -1714,10 +1690,11 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
self.assertTrue(m_log.error.called) self.assertTrue(m_log.error.called)
self.assertFalse(sock.close.called) self.assertFalse(sock.close.called)
self.loop._remove_reader.assert_called_with(10) self.loop._remove_reader.assert_called_with(10)
self.loop.call_later.assert_called_with(constants.ACCEPT_RETRY_DELAY, self.loop.call_later.assert_called_with(
# self.loop._start_serving constants.ACCEPT_RETRY_DELAY,
mock.ANY, # self.loop._start_serving
MyProto, sock, None, None, mock.ANY) mock.ANY,
MyProto, sock, None, None, mock.ANY)
def test_call_coroutine(self): def test_call_coroutine(self):
@asyncio.coroutine @asyncio.coroutine
@ -1738,7 +1715,8 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
self.loop.call_at(self.loop.time() + 60, func) self.loop.call_at(self.loop.time() + 60, func)
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
self.loop.run_in_executor(None, func) self.loop.run_until_complete(
self.loop.run_in_executor(None, func))
@mock.patch('asyncio.base_events.logger') @mock.patch('asyncio.base_events.logger')
def test_log_slow_callbacks(self, m_logger): def test_log_slow_callbacks(self, m_logger):

View File

@ -1702,7 +1702,7 @@ class EventLoopTestsMixin:
def test_prompt_cancellation(self): def test_prompt_cancellation(self):
r, w = socket.socketpair() r, w = socket.socketpair()
r.setblocking(False) r.setblocking(False)
f = self.loop.sock_recv(r, 1) f = self.loop.create_task(self.loop.sock_recv(r, 1))
ov = getattr(f, 'ov', None) ov = getattr(f, 'ov', None)
if ov is not None: if ov is not None:
self.assertTrue(ov.pending) self.assertTrue(ov.pending)
@ -1819,7 +1819,8 @@ class EventLoopTestsMixin:
with self.assertRaises(RuntimeError): with self.assertRaises(RuntimeError):
self.loop.call_at(self.loop.time() + .0, func) self.loop.call_at(self.loop.time() + .0, func)
with self.assertRaises(RuntimeError): with self.assertRaises(RuntimeError):
self.loop.run_in_executor(None, func) self.loop.run_until_complete(
self.loop.run_in_executor(None, func))
with self.assertRaises(RuntimeError): with self.assertRaises(RuntimeError):
self.loop.create_task(coro) self.loop.create_task(coro)
with self.assertRaises(RuntimeError): with self.assertRaises(RuntimeError):

View File

@ -483,27 +483,6 @@ class BaseProactorEventLoopTests(test_utils.TestCase):
self.loop.close() self.loop.close()
self.assertFalse(self.loop._close_self_pipe.called) self.assertFalse(self.loop._close_self_pipe.called)
def test_sock_recv(self):
self.loop.sock_recv(self.sock, 1024)
self.proactor.recv.assert_called_with(self.sock, 1024)
def test_sock_recv_into(self):
buf = bytearray(10)
self.loop.sock_recv_into(self.sock, buf)
self.proactor.recv_into.assert_called_with(self.sock, buf)
def test_sock_sendall(self):
self.loop.sock_sendall(self.sock, b'data')
self.proactor.send.assert_called_with(self.sock, b'data')
def test_sock_connect(self):
self.loop.sock_connect(self.sock, ('1.2.3.4', 123))
self.proactor.connect.assert_called_with(self.sock, ('1.2.3.4', 123))
def test_sock_accept(self):
self.loop.sock_accept(self.sock)
self.proactor.accept.assert_called_with(self.sock)
def test_make_socket_transport(self): def test_make_socket_transport(self):
tr = self.loop._make_socket_transport(self.sock, asyncio.Protocol()) tr = self.loop._make_socket_transport(self.sock, asyncio.Protocol())
self.assertIsInstance(tr, _ProactorSocketTransport) self.assertIsInstance(tr, _ProactorSocketTransport)

View File

@ -176,9 +176,15 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
sock = test_utils.mock_nonblocking_socket() sock = test_utils.mock_nonblocking_socket()
self.loop._sock_recv = mock.Mock() self.loop._sock_recv = mock.Mock()
f = self.loop.sock_recv(sock, 1024) f = self.loop.create_task(self.loop.sock_recv(sock, 1024))
self.assertIsInstance(f, asyncio.Future) self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
self.loop._sock_recv.assert_called_with(f, None, sock, 1024)
self.assertEqual(self.loop._sock_recv.call_args[0][1:],
(None, sock, 1024))
f.cancel()
with self.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f)
def test_sock_recv_reconnection(self): def test_sock_recv_reconnection(self):
sock = mock.Mock() sock = mock.Mock()
@ -188,7 +194,11 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
self.loop.add_reader = mock.Mock() self.loop.add_reader = mock.Mock()
self.loop.remove_reader = mock.Mock() self.loop.remove_reader = mock.Mock()
fut = self.loop.sock_recv(sock, 1024) fut = self.loop.create_task(
self.loop.sock_recv(sock, 1024))
self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
callback = self.loop.add_reader.call_args[0][1] callback = self.loop.add_reader.call_args[0][1]
params = self.loop.add_reader.call_args[0][2:] params = self.loop.add_reader.call_args[0][2:]
@ -198,6 +208,8 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
sock.recv.side_effect = OSError(9) sock.recv.side_effect = OSError(9)
callback(*params) callback(*params)
self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
self.assertIsInstance(fut.exception(), OSError) self.assertIsInstance(fut.exception(), OSError)
self.assertEqual((10,), self.loop.remove_reader.call_args[0]) self.assertEqual((10,), self.loop.remove_reader.call_args[0])
@ -245,18 +257,26 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
sock = test_utils.mock_nonblocking_socket() sock = test_utils.mock_nonblocking_socket()
self.loop._sock_sendall = mock.Mock() self.loop._sock_sendall = mock.Mock()
f = self.loop.sock_sendall(sock, b'data') f = self.loop.create_task(
self.assertIsInstance(f, asyncio.Future) self.loop.sock_sendall(sock, b'data'))
self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
self.assertEqual( self.assertEqual(
(f, None, sock, b'data'), (None, sock, b'data'),
self.loop._sock_sendall.call_args[0]) self.loop._sock_sendall.call_args[0][1:])
f.cancel()
with self.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f)
def test_sock_sendall_nodata(self): def test_sock_sendall_nodata(self):
sock = test_utils.mock_nonblocking_socket() sock = test_utils.mock_nonblocking_socket()
self.loop._sock_sendall = mock.Mock() self.loop._sock_sendall = mock.Mock()
f = self.loop.sock_sendall(sock, b'') f = self.loop.create_task(self.loop.sock_sendall(sock, b''))
self.assertIsInstance(f, asyncio.Future) self.loop.run_until_complete(asyncio.sleep(0, loop=self.loop))
self.assertTrue(f.done()) self.assertTrue(f.done())
self.assertIsNone(f.result()) self.assertIsNone(f.result())
self.assertFalse(self.loop._sock_sendall.called) self.assertFalse(self.loop._sock_sendall.called)
@ -269,7 +289,10 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
self.loop.add_writer = mock.Mock() self.loop.add_writer = mock.Mock()
self.loop.remove_writer = mock.Mock() self.loop.remove_writer = mock.Mock()
fut = self.loop.sock_sendall(sock, b'data') fut = self.loop.create_task(self.loop.sock_sendall(sock, b'data'))
self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
callback = self.loop.add_writer.call_args[0][1] callback = self.loop.add_writer.call_args[0][1]
params = self.loop.add_writer.call_args[0][2:] params = self.loop.add_writer.call_args[0][2:]
@ -279,6 +302,8 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
sock.send.side_effect = OSError(9) sock.send.side_effect = OSError(9)
callback(*params) callback(*params)
self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
self.assertIsInstance(fut.exception(), OSError) self.assertIsInstance(fut.exception(), OSError)
self.assertEqual((10,), self.loop.remove_writer.call_args[0]) self.assertEqual((10,), self.loop.remove_writer.call_args[0])
@ -402,17 +427,17 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
def test_sock_connect_resolve_using_socket_params(self, m_gai): def test_sock_connect_resolve_using_socket_params(self, m_gai):
addr = ('need-resolution.com', 8080) addr = ('need-resolution.com', 8080)
sock = test_utils.mock_nonblocking_socket() sock = test_utils.mock_nonblocking_socket()
m_gai.side_effect = (None, None, None, None, ('127.0.0.1', 0))
m_gai._is_coroutine = False m_gai.side_effect = \
lambda *args: [(None, None, None, None, ('127.0.0.1', 0))]
con = self.loop.create_task(self.loop.sock_connect(sock, addr)) con = self.loop.create_task(self.loop.sock_connect(sock, addr))
while not m_gai.called: self.loop.run_until_complete(con)
self.loop._run_once()
m_gai.assert_called_with( m_gai.assert_called_with(
addr[0], addr[1], sock.family, sock.type, sock.proto, 0) addr[0], addr[1], sock.family, sock.type, sock.proto, 0)
con.cancel() self.loop.run_until_complete(con)
with self.assertRaises(asyncio.CancelledError): sock.connect.assert_called_with(('127.0.0.1', 0))
self.loop.run_until_complete(con)
def test__sock_connect(self): def test__sock_connect(self):
f = asyncio.Future(loop=self.loop) f = asyncio.Future(loop=self.loop)
@ -487,10 +512,15 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
sock = test_utils.mock_nonblocking_socket() sock = test_utils.mock_nonblocking_socket()
self.loop._sock_accept = mock.Mock() self.loop._sock_accept = mock.Mock()
f = self.loop.sock_accept(sock) f = self.loop.create_task(self.loop.sock_accept(sock))
self.assertIsInstance(f, asyncio.Future) self.loop.run_until_complete(asyncio.sleep(0.01, loop=self.loop))
self.assertEqual(
(f, False, sock), self.loop._sock_accept.call_args[0]) self.assertFalse(self.loop._sock_accept.call_args[0][1])
self.assertIs(self.loop._sock_accept.call_args[0][2], sock)
f.cancel()
with self.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f)
def test__sock_accept(self): def test__sock_accept(self):
f = asyncio.Future(loop=self.loop) f = asyncio.Future(loop=self.loop)

View File

@ -2545,19 +2545,26 @@ class RunCoroutineThreadsafeTests(test_utils.TestCase):
def test_run_coroutine_threadsafe_task_factory_exception(self): def test_run_coroutine_threadsafe_task_factory_exception(self):
"""Test coroutine submission from a tread to an event loop """Test coroutine submission from a tread to an event loop
when the task factory raise an exception.""" when the task factory raise an exception."""
# Schedule the target
future = self.loop.run_in_executor( def task_factory(loop, coro):
None, lambda: self.target(advance_coro=True)) raise NameError
# Set corrupted task factory
self.loop.set_task_factory(lambda loop, coro: wrong_name) run = self.loop.create_task(
self.loop.run_in_executor(
None, lambda: self.target(advance_coro=True)))
# Set exception handler # Set exception handler
callback = test_utils.MockCallback() callback = test_utils.MockCallback()
self.loop.set_exception_handler(callback) self.loop.set_exception_handler(callback)
# Set corrupted task factory
self.loop.set_task_factory(task_factory)
# Run event loop # Run event loop
with self.assertRaises(NameError) as exc_context: with self.assertRaises(NameError) as exc_context:
self.loop.run_until_complete(future) self.loop.run_until_complete(run)
# Check exceptions # Check exceptions
self.assertIn('wrong_name', exc_context.exception.args[0])
self.assertEqual(len(callback.call_args_list), 1) self.assertEqual(len(callback.call_args_list), 1)
(loop, context), kwargs = callback.call_args (loop, context), kwargs = callback.call_args
self.assertEqual(context['exception'], exc_context.exception) self.assertEqual(context['exception'], exc_context.exception)

View File

@ -39,7 +39,7 @@ class ProactorTests(test_utils.TestCase):
def test_close(self): def test_close(self):
a, b = socket.socketpair() a, b = socket.socketpair()
trans = self.loop._make_socket_transport(a, asyncio.Protocol()) trans = self.loop._make_socket_transport(a, asyncio.Protocol())
f = asyncio.ensure_future(self.loop.sock_recv(b, 100)) f = asyncio.ensure_future(self.loop.sock_recv(b, 100), loop=self.loop)
trans.close() trans.close()
self.loop.run_until_complete(f) self.loop.run_until_complete(f)
self.assertEqual(f.result(), b'') self.assertEqual(f.result(), b'')

View File

@ -0,0 +1,3 @@
Convert asyncio functions that were documented as coroutines to coroutines.
Affected functions: loop.sock_sendall, loop.sock_recv, loop.sock_accept,
loop.run_in_executor, loop.getaddrinfo, loop.getnameinfo.