| import socket |
| import asyncio |
| import sys |
| import unittest |
| |
| from asyncio import proactor_events |
| from itertools import cycle, islice |
| from unittest.mock import Mock |
| from test.test_asyncio import utils as test_utils |
| from test import support |
| from test.support import socket_helper |
| |
| if socket_helper.tcp_blackhole(): |
| raise unittest.SkipTest('Not relevant to ProactorEventLoop') |
| |
| |
| def tearDownModule(): |
| asyncio.events._set_event_loop_policy(None) |
| |
| |
| class MyProto(asyncio.Protocol): |
| connected = None |
| done = None |
| |
| def __init__(self, loop=None): |
| self.transport = None |
| self.state = 'INITIAL' |
| self.nbytes = 0 |
| if loop is not None: |
| self.connected = loop.create_future() |
| self.done = loop.create_future() |
| |
| def _assert_state(self, *expected): |
| if self.state not in expected: |
| raise AssertionError(f'state: {self.state!r}, expected: {expected!r}') |
| |
| def connection_made(self, transport): |
| self.transport = transport |
| self._assert_state('INITIAL') |
| self.state = 'CONNECTED' |
| if self.connected: |
| self.connected.set_result(None) |
| transport.write(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n') |
| |
| def data_received(self, data): |
| self._assert_state('CONNECTED') |
| self.nbytes += len(data) |
| |
| def eof_received(self): |
| self._assert_state('CONNECTED') |
| self.state = 'EOF' |
| |
| def connection_lost(self, exc): |
| self._assert_state('CONNECTED', 'EOF') |
| self.state = 'CLOSED' |
| if self.done: |
| self.done.set_result(None) |
| |
| |
| class BaseSockTestsMixin: |
| |
| def create_event_loop(self): |
| raise NotImplementedError |
| |
| def setUp(self): |
| self.loop = self.create_event_loop() |
| self.set_event_loop(self.loop) |
| super().setUp() |
| |
| def tearDown(self): |
| # just in case if we have transport close callbacks |
| if not self.loop.is_closed(): |
| test_utils.run_briefly(self.loop) |
| |
| self.doCleanups() |
| support.gc_collect() |
| super().tearDown() |
| |
| def _basetest_sock_client_ops(self, httpd, sock): |
| if not isinstance(self.loop, proactor_events.BaseProactorEventLoop): |
| # in debug mode, socket operations must fail |
| # if the socket is not in blocking mode |
| self.loop.set_debug(True) |
| sock.setblocking(True) |
| with self.assertRaises(ValueError): |
| self.loop.run_until_complete( |
| self.loop.sock_connect(sock, httpd.address)) |
| with self.assertRaises(ValueError): |
| self.loop.run_until_complete( |
| self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n')) |
| with self.assertRaises(ValueError): |
| self.loop.run_until_complete( |
| self.loop.sock_recv(sock, 1024)) |
| with self.assertRaises(ValueError): |
| self.loop.run_until_complete( |
| self.loop.sock_recv_into(sock, bytearray())) |
| with self.assertRaises(ValueError): |
| self.loop.run_until_complete( |
| self.loop.sock_accept(sock)) |
| |
| # test in non-blocking mode |
| sock.setblocking(False) |
| self.loop.run_until_complete( |
| self.loop.sock_connect(sock, httpd.address)) |
| self.loop.run_until_complete( |
| self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n')) |
| data = self.loop.run_until_complete( |
| self.loop.sock_recv(sock, 1024)) |
| # consume data |
| self.loop.run_until_complete( |
| self.loop.sock_recv(sock, 1024)) |
| sock.close() |
| self.assertStartsWith(data, b'HTTP/1.0 200 OK') |
| |
| def _basetest_sock_recv_into(self, httpd, sock): |
| # same as _basetest_sock_client_ops, but using sock_recv_into |
| sock.setblocking(False) |
| self.loop.run_until_complete( |
| self.loop.sock_connect(sock, httpd.address)) |
| self.loop.run_until_complete( |
| self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n')) |
| data = bytearray(1024) |
| with memoryview(data) as buf: |
| nbytes = self.loop.run_until_complete( |
| self.loop.sock_recv_into(sock, buf[:1024])) |
| # consume data |
| self.loop.run_until_complete( |
| self.loop.sock_recv_into(sock, buf[nbytes:])) |
| sock.close() |
| self.assertStartsWith(data, b'HTTP/1.0 200 OK') |
| |
| def test_sock_client_ops(self): |
| with test_utils.run_test_server() as httpd: |
| sock = socket.socket() |
| self._basetest_sock_client_ops(httpd, sock) |
| sock = socket.socket() |
| self._basetest_sock_recv_into(httpd, sock) |
| |
| async def _basetest_sock_recv_racing(self, httpd, sock): |
| sock.setblocking(False) |
| await self.loop.sock_connect(sock, httpd.address) |
| |
| task = asyncio.create_task(self.loop.sock_recv(sock, 1024)) |
| await asyncio.sleep(0) |
| task.cancel() |
| |
| asyncio.create_task( |
| self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n')) |
| data = await self.loop.sock_recv(sock, 1024) |
| # consume data |
| await self.loop.sock_recv(sock, 1024) |
| |
| self.assertStartsWith(data, b'HTTP/1.0 200 OK') |
| |
| async def _basetest_sock_recv_into_racing(self, httpd, sock): |
| sock.setblocking(False) |
| await self.loop.sock_connect(sock, httpd.address) |
| |
| data = bytearray(1024) |
| with memoryview(data) as buf: |
| task = asyncio.create_task( |
| self.loop.sock_recv_into(sock, buf[:1024])) |
| await asyncio.sleep(0) |
| task.cancel() |
| |
| task = asyncio.create_task( |
| self.loop.sock_sendall(sock, b'GET / HTTP/1.0\r\n\r\n')) |
| nbytes = await self.loop.sock_recv_into(sock, buf[:1024]) |
| # consume data |
| await self.loop.sock_recv_into(sock, buf[nbytes:]) |
| self.assertStartsWith(data, b'HTTP/1.0 200 OK') |
| |
| await task |
| |
| async def _basetest_sock_send_racing(self, listener, sock): |
| listener.bind(('127.0.0.1', 0)) |
| listener.listen(1) |
| |
| # make connection |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024) |
| sock.setblocking(False) |
| task = asyncio.create_task( |
| self.loop.sock_connect(sock, listener.getsockname())) |
| await asyncio.sleep(0) |
| server = listener.accept()[0] |
| server.setblocking(False) |
| |
| with server: |
| await task |
| |
| # fill the buffer until sending 5 chars would block |
| size = 8192 |
| while size >= 4: |
| with self.assertRaises(BlockingIOError): |
| while True: |
| sock.send(b' ' * size) |
| size = int(size / 2) |
| |
| # cancel a blocked sock_sendall |
| task = asyncio.create_task( |
| self.loop.sock_sendall(sock, b'hello')) |
| await asyncio.sleep(0) |
| task.cancel() |
| |
| # receive everything that is not a space |
| async def recv_all(): |
| rv = b'' |
| while True: |
| buf = await self.loop.sock_recv(server, 8192) |
| if not buf: |
| return rv |
| rv += buf.strip() |
| task = asyncio.create_task(recv_all()) |
| |
| # immediately make another sock_sendall call |
| await self.loop.sock_sendall(sock, b'world') |
| sock.shutdown(socket.SHUT_WR) |
| data = await task |
| # ProactorEventLoop could deliver hello, so endswith is necessary |
| self.assertEndsWith(data, b'world') |
| |
| # After the first connect attempt before the listener is ready, |
| # the socket needs time to "recover" to make the next connect call. |
| # On Linux, a second retry will do. On Windows, the waiting time is |
| # unpredictable; and on FreeBSD the socket may never come back |
| # because it's a loopback address. Here we'll just retry for a few |
| # times, and have to skip the test if it's not working. See also: |
| # https://stackoverflow.com/a/54437602/3316267 |
| # https://lists.freebsd.org/pipermail/freebsd-current/2005-May/049876.html |
| async def _basetest_sock_connect_racing(self, listener, sock): |
| listener.bind(('127.0.0.1', 0)) |
| addr = listener.getsockname() |
| sock.setblocking(False) |
| |
| task = asyncio.create_task(self.loop.sock_connect(sock, addr)) |
| await asyncio.sleep(0) |
| task.cancel() |
| |
| listener.listen(1) |
| |
| skip_reason = "Max retries reached" |
| for i in range(128): |
| try: |
| await self.loop.sock_connect(sock, addr) |
| except ConnectionRefusedError as e: |
| skip_reason = e |
| except OSError as e: |
| skip_reason = e |
| |
| # Retry only for this error: |
| # [WinError 10022] An invalid argument was supplied |
| if getattr(e, 'winerror', 0) != 10022: |
| break |
| else: |
| # success |
| return |
| |
| self.skipTest(skip_reason) |
| |
| def test_sock_client_racing(self): |
| with test_utils.run_test_server() as httpd: |
| sock = socket.socket() |
| with sock: |
| self.loop.run_until_complete(asyncio.wait_for( |
| self._basetest_sock_recv_racing(httpd, sock), 10)) |
| sock = socket.socket() |
| with sock: |
| self.loop.run_until_complete(asyncio.wait_for( |
| self._basetest_sock_recv_into_racing(httpd, sock), 10)) |
| listener = socket.socket() |
| sock = socket.socket() |
| with listener, sock: |
| self.loop.run_until_complete(asyncio.wait_for( |
| self._basetest_sock_send_racing(listener, sock), 10)) |
| |
| def test_sock_client_connect_racing(self): |
| listener = socket.socket() |
| sock = socket.socket() |
| with listener, sock: |
| self.loop.run_until_complete(asyncio.wait_for( |
| self._basetest_sock_connect_racing(listener, sock), 10)) |
| |
| async def _basetest_huge_content(self, address): |
| sock = socket.socket() |
| sock.setblocking(False) |
| DATA_SIZE = 10_000_00 |
| |
| chunk = b'0123456789' * (DATA_SIZE // 10) |
| |
| await self.loop.sock_connect(sock, address) |
| await self.loop.sock_sendall(sock, |
| (b'POST /loop HTTP/1.0\r\n' + |
| b'Content-Length: %d\r\n' % DATA_SIZE + |
| b'\r\n')) |
| |
| task = asyncio.create_task(self.loop.sock_sendall(sock, chunk)) |
| |
| data = await self.loop.sock_recv(sock, DATA_SIZE) |
| # HTTP headers size is less than MTU, |
| # they are sent by the first packet always |
| self.assertStartsWith(data, b'HTTP/1.0 200 OK') |
| while data.find(b'\r\n\r\n') == -1: |
| data += await self.loop.sock_recv(sock, DATA_SIZE) |
| # Strip headers |
| headers = data[:data.index(b'\r\n\r\n') + 4] |
| data = data[len(headers):] |
| |
| size = DATA_SIZE |
| checker = cycle(b'0123456789') |
| |
| expected = bytes(islice(checker, len(data))) |
| self.assertEqual(data, expected) |
| size -= len(data) |
| |
| while True: |
| data = await self.loop.sock_recv(sock, DATA_SIZE) |
| if not data: |
| break |
| expected = bytes(islice(checker, len(data))) |
| self.assertEqual(data, expected) |
| size -= len(data) |
| self.assertEqual(size, 0) |
| |
| await task |
| sock.close() |
| |
| def test_huge_content(self): |
| with test_utils.run_test_server() as httpd: |
| self.loop.run_until_complete( |
| self._basetest_huge_content(httpd.address)) |
| |
| async def _basetest_huge_content_recvinto(self, address): |
| sock = socket.socket() |
| sock.setblocking(False) |
| DATA_SIZE = 10_000_00 |
| |
| chunk = b'0123456789' * (DATA_SIZE // 10) |
| |
| await self.loop.sock_connect(sock, address) |
| await self.loop.sock_sendall(sock, |
| (b'POST /loop HTTP/1.0\r\n' + |
| b'Content-Length: %d\r\n' % DATA_SIZE + |
| b'\r\n')) |
| |
| task = asyncio.create_task(self.loop.sock_sendall(sock, chunk)) |
| |
| array = bytearray(DATA_SIZE) |
| buf = memoryview(array) |
| |
| nbytes = await self.loop.sock_recv_into(sock, buf) |
| data = bytes(buf[:nbytes]) |
| # HTTP headers size is less than MTU, |
| # they are sent by the first packet always |
| self.assertStartsWith(data, b'HTTP/1.0 200 OK') |
| while data.find(b'\r\n\r\n') == -1: |
| nbytes = await self.loop.sock_recv_into(sock, buf) |
| data = bytes(buf[:nbytes]) |
| # Strip headers |
| headers = data[:data.index(b'\r\n\r\n') + 4] |
| data = data[len(headers):] |
| |
| size = DATA_SIZE |
| checker = cycle(b'0123456789') |
| |
| expected = bytes(islice(checker, len(data))) |
| self.assertEqual(data, expected) |
| size -= len(data) |
| |
| while True: |
| nbytes = await self.loop.sock_recv_into(sock, buf) |
| data = buf[:nbytes] |
| if not data: |
| break |
| expected = bytes(islice(checker, len(data))) |
| self.assertEqual(data, expected) |
| size -= len(data) |
| self.assertEqual(size, 0) |
| |
| await task |
| sock.close() |
| |
| def test_huge_content_recvinto(self): |
| with test_utils.run_test_server() as httpd: |
| self.loop.run_until_complete( |
| self._basetest_huge_content_recvinto(httpd.address)) |
| |
| async def _basetest_datagram_recvfrom(self, server_address): |
| # Happy path, sock.sendto() returns immediately |
| data = b'\x01' * 4096 |
| with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: |
| sock.setblocking(False) |
| await self.loop.sock_sendto(sock, data, server_address) |
| received_data, from_addr = await self.loop.sock_recvfrom( |
| sock, 4096) |
| self.assertEqual(received_data, data) |
| self.assertEqual(from_addr, server_address) |
| |
| def test_recvfrom(self): |
| with test_utils.run_udp_echo_server() as server_address: |
| self.loop.run_until_complete( |
| self._basetest_datagram_recvfrom(server_address)) |
| |
| async def _basetest_datagram_recvfrom_into(self, server_address): |
| # Happy path, sock.sendto() returns immediately |
| with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: |
| sock.setblocking(False) |
| |
| buf = bytearray(4096) |
| data = b'\x01' * 4096 |
| await self.loop.sock_sendto(sock, data, server_address) |
| num_bytes, from_addr = await self.loop.sock_recvfrom_into( |
| sock, buf) |
| self.assertEqual(num_bytes, 4096) |
| self.assertEqual(buf, data) |
| self.assertEqual(from_addr, server_address) |
| |
| buf = bytearray(8192) |
| await self.loop.sock_sendto(sock, data, server_address) |
| num_bytes, from_addr = await self.loop.sock_recvfrom_into( |
| sock, buf, 4096) |
| self.assertEqual(num_bytes, 4096) |
| self.assertEqual(buf[:4096], data[:4096]) |
| self.assertEqual(from_addr, server_address) |
| |
| def test_recvfrom_into(self): |
| with test_utils.run_udp_echo_server() as server_address: |
| self.loop.run_until_complete( |
| self._basetest_datagram_recvfrom_into(server_address)) |
| |
| async def _basetest_datagram_sendto_blocking(self, server_address): |
| # Sad path, sock.sendto() raises BlockingIOError |
| # This involves patching sock.sendto() to raise BlockingIOError but |
| # sendto() is not used by the proactor event loop |
| data = b'\x01' * 4096 |
| with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: |
| sock.setblocking(False) |
| mock_sock = Mock(sock) |
| mock_sock.gettimeout = sock.gettimeout |
| mock_sock.sendto.configure_mock(side_effect=BlockingIOError) |
| mock_sock.fileno = sock.fileno |
| self.loop.call_soon( |
| lambda: setattr(mock_sock, 'sendto', sock.sendto) |
| ) |
| await self.loop.sock_sendto(mock_sock, data, server_address) |
| |
| received_data, from_addr = await self.loop.sock_recvfrom( |
| sock, 4096) |
| self.assertEqual(received_data, data) |
| self.assertEqual(from_addr, server_address) |
| |
| def test_sendto_blocking(self): |
| if sys.platform == 'win32': |
| if isinstance(self.loop, asyncio.ProactorEventLoop): |
| raise unittest.SkipTest('Not relevant to ProactorEventLoop') |
| |
| with test_utils.run_udp_echo_server() as server_address: |
| self.loop.run_until_complete( |
| self._basetest_datagram_sendto_blocking(server_address)) |
| |
| @socket_helper.skip_unless_bind_unix_socket |
| def test_unix_sock_client_ops(self): |
| with test_utils.run_test_unix_server() as httpd: |
| sock = socket.socket(socket.AF_UNIX) |
| self._basetest_sock_client_ops(httpd, sock) |
| sock = socket.socket(socket.AF_UNIX) |
| self._basetest_sock_recv_into(httpd, sock) |
| |
| def test_sock_client_fail(self): |
| # Make sure that we will get an unused port |
| address = None |
| try: |
| s = socket.socket() |
| s.bind(('127.0.0.1', 0)) |
| address = s.getsockname() |
| finally: |
| s.close() |
| |
| sock = socket.socket() |
| sock.setblocking(False) |
| with self.assertRaises(ConnectionRefusedError): |
| self.loop.run_until_complete( |
| self.loop.sock_connect(sock, address)) |
| sock.close() |
| |
| def test_sock_accept(self): |
| listener = socket.socket() |
| listener.setblocking(False) |
| listener.bind(('127.0.0.1', 0)) |
| listener.listen(1) |
| client = socket.socket() |
| client.connect(listener.getsockname()) |
| |
| f = self.loop.sock_accept(listener) |
| conn, addr = self.loop.run_until_complete(f) |
| self.assertEqual(conn.gettimeout(), 0) |
| self.assertEqual(addr, client.getsockname()) |
| self.assertEqual(client.getpeername(), listener.getsockname()) |
| client.close() |
| conn.close() |
| listener.close() |
| |
| def test_cancel_sock_accept(self): |
| listener = socket.socket() |
| listener.setblocking(False) |
| listener.bind(('127.0.0.1', 0)) |
| listener.listen(1) |
| sockaddr = listener.getsockname() |
| f = asyncio.wait_for(self.loop.sock_accept(listener), 0.1) |
| with self.assertRaises(asyncio.TimeoutError): |
| self.loop.run_until_complete(f) |
| |
| listener.close() |
| client = socket.socket() |
| client.setblocking(False) |
| f = self.loop.sock_connect(client, sockaddr) |
| with self.assertRaises(ConnectionRefusedError): |
| self.loop.run_until_complete(f) |
| |
| client.close() |
| |
| def test_create_connection_sock(self): |
| with test_utils.run_test_server() as httpd: |
| sock = None |
| infos = self.loop.run_until_complete( |
| self.loop.getaddrinfo( |
| *httpd.address, type=socket.SOCK_STREAM)) |
| for family, type, proto, cname, address in infos: |
| try: |
| sock = socket.socket(family=family, type=type, proto=proto) |
| sock.setblocking(False) |
| self.loop.run_until_complete( |
| self.loop.sock_connect(sock, address)) |
| except BaseException: |
| pass |
| else: |
| break |
| else: |
| self.fail('Can not create socket.') |
| |
| f = self.loop.create_connection( |
| lambda: MyProto(loop=self.loop), sock=sock) |
| tr, pr = self.loop.run_until_complete(f) |
| self.assertIsInstance(tr, asyncio.Transport) |
| self.assertIsInstance(pr, asyncio.Protocol) |
| self.loop.run_until_complete(pr.done) |
| self.assertGreater(pr.nbytes, 0) |
| tr.close() |
| |
| |
| if sys.platform == 'win32': |
| |
| class SelectEventLoopTests(BaseSockTestsMixin, |
| test_utils.TestCase): |
| |
| def create_event_loop(self): |
| return asyncio.SelectorEventLoop() |
| |
| |
| class ProactorEventLoopTests(BaseSockTestsMixin, |
| test_utils.TestCase): |
| |
| def create_event_loop(self): |
| return asyncio.ProactorEventLoop() |
| |
| |
| async def _basetest_datagram_send_to_non_listening_address(self, |
| recvfrom): |
| # see: |
| # https://github.com/python/cpython/issues/91227 |
| # https://github.com/python/cpython/issues/88906 |
| # https://bugs.python.org/issue47071 |
| # https://bugs.python.org/issue44743 |
| # The Proactor event loop would fail to receive datagram messages |
| # after sending a message to an address that wasn't listening. |
| |
| def create_socket(): |
| sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| sock.setblocking(False) |
| sock.bind(('127.0.0.1', 0)) |
| return sock |
| |
| socket_1 = create_socket() |
| addr_1 = socket_1.getsockname() |
| |
| socket_2 = create_socket() |
| addr_2 = socket_2.getsockname() |
| |
| # creating and immediately closing this to try to get an address |
| # that is not listening |
| socket_3 = create_socket() |
| addr_3 = socket_3.getsockname() |
| socket_3.shutdown(socket.SHUT_RDWR) |
| socket_3.close() |
| |
| socket_1_recv_task = self.loop.create_task(recvfrom(socket_1)) |
| socket_2_recv_task = self.loop.create_task(recvfrom(socket_2)) |
| await asyncio.sleep(0) |
| |
| await self.loop.sock_sendto(socket_1, b'a', addr_2) |
| self.assertEqual(await socket_2_recv_task, b'a') |
| |
| await self.loop.sock_sendto(socket_2, b'b', addr_1) |
| self.assertEqual(await socket_1_recv_task, b'b') |
| socket_1_recv_task = self.loop.create_task(recvfrom(socket_1)) |
| await asyncio.sleep(0) |
| |
| # this should send to an address that isn't listening |
| await self.loop.sock_sendto(socket_1, b'c', addr_3) |
| self.assertEqual(await socket_1_recv_task, b'') |
| socket_1_recv_task = self.loop.create_task(recvfrom(socket_1)) |
| await asyncio.sleep(0) |
| |
| # socket 1 should still be able to receive messages after sending |
| # to an address that wasn't listening |
| socket_2.sendto(b'd', addr_1) |
| self.assertEqual(await socket_1_recv_task, b'd') |
| |
| socket_1.shutdown(socket.SHUT_RDWR) |
| socket_1.close() |
| socket_2.shutdown(socket.SHUT_RDWR) |
| socket_2.close() |
| |
| |
| def test_datagram_send_to_non_listening_address_recvfrom(self): |
| async def recvfrom(socket): |
| data, _ = await self.loop.sock_recvfrom(socket, 4096) |
| return data |
| |
| self.loop.run_until_complete( |
| self._basetest_datagram_send_to_non_listening_address( |
| recvfrom)) |
| |
| |
| def test_datagram_send_to_non_listening_address_recvfrom_into(self): |
| async def recvfrom_into(socket): |
| buf = bytearray(4096) |
| length, _ = await self.loop.sock_recvfrom_into(socket, buf, |
| 4096) |
| return buf[:length] |
| |
| self.loop.run_until_complete( |
| self._basetest_datagram_send_to_non_listening_address( |
| recvfrom_into)) |
| |
| else: |
| import selectors |
| |
| if hasattr(selectors, 'KqueueSelector'): |
| class KqueueEventLoopTests(BaseSockTestsMixin, |
| test_utils.TestCase): |
| |
| def create_event_loop(self): |
| return asyncio.SelectorEventLoop( |
| selectors.KqueueSelector()) |
| |
| if hasattr(selectors, 'EpollSelector'): |
| class EPollEventLoopTests(BaseSockTestsMixin, |
| test_utils.TestCase): |
| |
| def create_event_loop(self): |
| return asyncio.SelectorEventLoop(selectors.EpollSelector()) |
| |
| if hasattr(selectors, 'PollSelector'): |
| class PollEventLoopTests(BaseSockTestsMixin, |
| test_utils.TestCase): |
| |
| def create_event_loop(self): |
| return asyncio.SelectorEventLoop(selectors.PollSelector()) |
| |
| # Should always exist. |
| class SelectEventLoopTests(BaseSockTestsMixin, |
| test_utils.TestCase): |
| |
| def create_event_loop(self): |
| return asyncio.SelectorEventLoop(selectors.SelectSelector()) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
