"""Tests for events.py."""

import concurrent.futures

import contextlib

import functools

import io

import multiprocessing

import os

import platform

import re

import signal

import socket

try:

import ssl

except ImportError:

ssl = None

import subprocess

import sys

import threading

import time

import types

import errno

import unittest

from unittest import mock

import weakref

if sys.platform not in ('win32', 'vxworks'):

import tty

import asyncio

from asyncio import coroutines

from asyncio import events

from asyncio import selector_events

from multiprocessing.util import _cleanup_tests as multiprocessing_cleanup_tests

from test.test_asyncio import utils as test_utils

from test import support

from test.support import socket_helper

from test.support import threading_helper

from test.support import ALWAYS_EQ, LARGEST, SMALLEST

def tearDownModule():

asyncio.events._set_event_loop_policy(None)

def broken_unix_getsockname():

"""Return True if the platform is Mac OS 10.4 or older."""

if sys.platform.startswith("aix"):

return True

elif sys.platform != 'darwin':

return False

version = platform.mac_ver()[0]

version = tuple(map(int, version.split('.')))

return version < (10, 5)

def _test_get_event_loop_new_process__sub_proc():

async def doit():

return 'hello'

with contextlib.closing(asyncio.new_event_loop()) as loop:

asyncio.set_event_loop(loop)

return loop.run_until_complete(doit())

class CoroLike:

def send(self, v):

pass

def throw(self, *exc):

pass

def close(self):

pass

def __await__(self):

pass

class MyBaseProto(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)

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 MyProto(MyBaseProto):

def connection_made(self, transport):

super().connection_made(transport)

transport.write(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n')

class MyDatagramProto(asyncio.DatagramProtocol):

done = None

def __init__(self, loop=None):

self.state = 'INITIAL'

self.nbytes = 0

if loop is not None:

self.done = loop.create_future()

def _assert_state(self, expected):

if self.state != 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 = 'INITIALIZED'

def datagram_received(self, data, addr):

self._assert_state('INITIALIZED')

self.nbytes += len(data)

def error_received(self, exc):

self._assert_state('INITIALIZED')

def connection_lost(self, exc):

self._assert_state('INITIALIZED')

self.state = 'CLOSED'

if self.done:

self.done.set_result(None)

class MyReadPipeProto(asyncio.Protocol):

done = None

def __init__(self, loop=None):

self.state = ['INITIAL']

self.nbytes = 0

self.transport = None

if loop is not None:

self.done = loop.create_future()

def _assert_state(self, expected):

if self.state != 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.append('CONNECTED')

def data_received(self, data):

self._assert_state(['INITIAL', 'CONNECTED'])

self.nbytes += len(data)

def eof_received(self):

self._assert_state(['INITIAL', 'CONNECTED'])

self.state.append('EOF')

def connection_lost(self, exc):

if 'EOF' not in self.state:

self.state.append('EOF') # It is okay if EOF is missed.

self._assert_state(['INITIAL', 'CONNECTED', 'EOF'])

self.state.append('CLOSED')

if self.done:

self.done.set_result(None)

class MyWritePipeProto(asyncio.BaseProtocol):

done = None

def __init__(self, loop=None):

self.state = 'INITIAL'

self.transport = None

if loop is not None:

self.done = loop.create_future()

def _assert_state(self, expected):

if self.state != 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'

def connection_lost(self, exc):

self._assert_state('CONNECTED')

self.state = 'CLOSED'

if self.done:

self.done.set_result(None)

class MySubprocessProtocol(asyncio.SubprocessProtocol):

def __init__(self, loop):

self.state = 'INITIAL'

self.transport = None

self.connected = loop.create_future()

self.completed = loop.create_future()

self.disconnects = {fd: loop.create_future() for fd in range(3)}

self.data = {1: b'', 2: b''}

self.returncode = None

self.got_data = {1: asyncio.Event(),

2: asyncio.Event()}

def _assert_state(self, expected):

if self.state != 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'

self.connected.set_result(None)

def connection_lost(self, exc):

self._assert_state('CONNECTED')

self.state = 'CLOSED'

self.completed.set_result(None)

def pipe_data_received(self, fd, data):

self._assert_state('CONNECTED')

self.data[fd] += data

self.got_data[fd].set()

def pipe_connection_lost(self, fd, exc):

self._assert_state('CONNECTED')

if exc:

self.disconnects[fd].set_exception(exc)

else:

self.disconnects[fd].set_result(exc)

def process_exited(self):

self._assert_state('CONNECTED')

self.returncode = self.transport.get_returncode()

class EventLoopTestsMixin:

def setUp(self):

super().setUp()

self.loop = self.create_event_loop()

self.set_event_loop(self.loop)

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()

@unittest.expectedFailure # TODO: RUSTPYTHON; - RuntimeWarning for unawaited coroutine not triggered

def test_run_until_complete_nesting(self):

async def coro1():

await asyncio.sleep(0)

async def coro2():

self.assertTrue(self.loop.is_running())

self.loop.run_until_complete(coro1())

with self.assertWarnsRegex(

RuntimeWarning,

r"coroutine \S+ was never awaited"

):

self.assertRaises(

RuntimeError, self.loop.run_until_complete, coro2())

# Note: because of the default Windows timing granularity of

# 15.6 msec, we use fairly long sleep times here (~100 msec).

def test_run_until_complete(self):

delay = 0.100

t0 = self.loop.time()

self.loop.run_until_complete(asyncio.sleep(delay))

dt = self.loop.time() - t0

self.assertGreaterEqual(dt, delay - test_utils.CLOCK_RES)

def test_run_until_complete_stopped(self):

async def cb():

self.loop.stop()

await asyncio.sleep(0.1)

task = cb()

self.assertRaises(RuntimeError,

self.loop.run_until_complete, task)

def test_call_later(self):

results = []

def callback(arg):

results.append(arg)

self.loop.stop()

self.loop.call_later(0.1, callback, 'hello world')

self.loop.run_forever()

self.assertEqual(results, ['hello world'])

def test_call_soon(self):

results = []

def callback(arg1, arg2):

results.append((arg1, arg2))

self.loop.stop()

self.loop.call_soon(callback, 'hello', 'world')

self.loop.run_forever()

self.assertEqual(results, [('hello', 'world')])

def test_call_soon_threadsafe(self):

results = []

lock = threading.Lock()

def callback(arg):

results.append(arg)

if len(results) >= 2:

self.loop.stop()

def run_in_thread():

self.loop.call_soon_threadsafe(callback, 'hello')

lock.release()

lock.acquire()

t = threading.Thread(target=run_in_thread)

t.start()

with lock:

self.loop.call_soon(callback, 'world')

self.loop.run_forever()

t.join()

self.assertEqual(results, ['hello', 'world'])

def test_call_soon_threadsafe_handle_block_check_cancelled(self):

results = []

callback_started = threading.Event()

callback_finished = threading.Event()

def callback(arg):

callback_started.set()

results.append(arg)

time.sleep(1)

callback_finished.set()

def run_in_thread():

handle = self.loop.call_soon_threadsafe(callback, 'hello')

self.assertIsInstance(handle, events._ThreadSafeHandle)

callback_started.wait()

# callback started so it should block checking for cancellation

# until it finishes

self.assertFalse(handle.cancelled())

self.assertTrue(callback_finished.is_set())

self.loop.call_soon_threadsafe(self.loop.stop)

t = threading.Thread(target=run_in_thread)

t.start()

self.loop.run_forever()

t.join()

self.assertEqual(results, ['hello'])

def test_call_soon_threadsafe_handle_block_cancellation(self):

results = []

callback_started = threading.Event()

callback_finished = threading.Event()

def callback(arg):

callback_started.set()

results.append(arg)

time.sleep(1)

callback_finished.set()

def run_in_thread():

handle = self.loop.call_soon_threadsafe(callback, 'hello')

self.assertIsInstance(handle, events._ThreadSafeHandle)

callback_started.wait()

# callback started so it cannot be cancelled from other thread until

# it finishes

handle.cancel()

self.assertTrue(callback_finished.is_set())

self.loop.call_soon_threadsafe(self.loop.stop)

t = threading.Thread(target=run_in_thread)

t.start()

self.loop.run_forever()

t.join()

self.assertEqual(results, ['hello'])

def test_call_soon_threadsafe_handle_cancel_same_thread(self):

results = []

callback_started = threading.Event()

callback_finished = threading.Event()

fut = concurrent.futures.Future()

def callback(arg):

callback_started.set()

handle = fut.result()

handle.cancel()

results.append(arg)

callback_finished.set()

self.loop.stop()

def run_in_thread():

handle = self.loop.call_soon_threadsafe(callback, 'hello')

fut.set_result(handle)

self.assertIsInstance(handle, events._ThreadSafeHandle)

callback_started.wait()

# callback cancels itself from same thread so it has no effect

# it runs to completion

self.assertTrue(handle.cancelled())

self.assertTrue(callback_finished.is_set())

self.loop.call_soon_threadsafe(self.loop.stop)

t = threading.Thread(target=run_in_thread)

t.start()

self.loop.run_forever()

t.join()

self.assertEqual(results, ['hello'])

def test_call_soon_threadsafe_handle_cancel_other_thread(self):

results = []

ev = threading.Event()

callback_finished = threading.Event()

def callback(arg):

results.append(arg)

callback_finished.set()

self.loop.stop()

def run_in_thread():

handle = self.loop.call_soon_threadsafe(callback, 'hello')

# handle can be cancelled from other thread if not started yet

self.assertIsInstance(handle, events._ThreadSafeHandle)

handle.cancel()

self.assertTrue(handle.cancelled())

self.assertFalse(callback_finished.is_set())

ev.set()

self.loop.call_soon_threadsafe(self.loop.stop)

# block the main loop until the callback is added and cancelled in the

# other thread

self.loop.call_soon(ev.wait)

t = threading.Thread(target=run_in_thread)

t.start()

self.loop.run_forever()

t.join()

self.assertEqual(results, [])

self.assertFalse(callback_finished.is_set())

def test_call_soon_threadsafe_same_thread(self):

results = []

def callback(arg):

results.append(arg)

if len(results) >= 2:

self.loop.stop()

self.loop.call_soon_threadsafe(callback, 'hello')

self.loop.call_soon(callback, 'world')

self.loop.run_forever()

self.assertEqual(results, ['hello', 'world'])

def test_run_in_executor(self):

def run(arg):

return (arg, threading.get_ident())

f2 = self.loop.run_in_executor(None, run, 'yo')

res, thread_id = self.loop.run_until_complete(f2)

self.assertEqual(res, 'yo')

self.assertNotEqual(thread_id, threading.get_ident())

def test_run_in_executor_cancel(self):

called = False

def patched_call_soon(*args):

nonlocal called

called = True

def run():

time.sleep(0.05)

f2 = self.loop.run_in_executor(None, run)

f2.cancel()

self.loop.run_until_complete(

self.loop.shutdown_default_executor())

self.loop.close()

self.loop.call_soon = patched_call_soon

self.loop.call_soon_threadsafe = patched_call_soon

time.sleep(0.4)

self.assertFalse(called)

def test_reader_callback(self):

r, w = socket.socketpair()

r.setblocking(False)

bytes_read = bytearray()

def reader():

try:

data = r.recv(1024)

except BlockingIOError:

# Spurious readiness notifications are possible

# at least on Linux -- see man select.

return

if data:

bytes_read.extend(data)

else:

self.assertTrue(self.loop.remove_reader(r.fileno()))

r.close()

self.loop.add_reader(r.fileno(), reader)

self.loop.call_soon(w.send, b'abc')

test_utils.run_until(self.loop, lambda: len(bytes_read) >= 3)

self.loop.call_soon(w.send, b'def')

test_utils.run_until(self.loop, lambda: len(bytes_read) >= 6)

self.loop.call_soon(w.close)

self.loop.call_soon(self.loop.stop)

self.loop.run_forever()

self.assertEqual(bytes_read, b'abcdef')

def test_writer_callback(self):

r, w = socket.socketpair()

w.setblocking(False)

def writer(data):

w.send(data)

self.loop.stop()

data = b'x' * 1024

self.loop.add_writer(w.fileno(), writer, data)

self.loop.run_forever()

self.assertTrue(self.loop.remove_writer(w.fileno()))

self.assertFalse(self.loop.remove_writer(w.fileno()))

w.close()

read = r.recv(len(data) * 2)

r.close()

self.assertEqual(read, data)

@unittest.expectedFailure # TODO: RUSTPYTHON; - signal handler implementation differs

@unittest.skipUnless(hasattr(signal, 'SIGKILL'), 'No SIGKILL')

def test_add_signal_handler(self):

caught = 0

def my_handler():

nonlocal caught

caught += 1

# Check error behavior first.

self.assertRaises(

TypeError, self.loop.add_signal_handler, 'boom', my_handler)

self.assertRaises(

TypeError, self.loop.remove_signal_handler, 'boom')

self.assertRaises(

ValueError, self.loop.add_signal_handler, signal.NSIG+1,

my_handler)

self.assertRaises(

ValueError, self.loop.remove_signal_handler, signal.NSIG+1)

self.assertRaises(

ValueError, self.loop.add_signal_handler, 0, my_handler)

self.assertRaises(

ValueError, self.loop.remove_signal_handler, 0)

self.assertRaises(

ValueError, self.loop.add_signal_handler, -1, my_handler)

self.assertRaises(

ValueError, self.loop.remove_signal_handler, -1)

self.assertRaises(

RuntimeError, self.loop.add_signal_handler, signal.SIGKILL,

my_handler)

# Removing SIGKILL doesn't raise, since we don't call signal().

self.assertFalse(self.loop.remove_signal_handler(signal.SIGKILL))

# Now set a handler and handle it.

self.loop.add_signal_handler(signal.SIGINT, my_handler)

os.kill(os.getpid(), signal.SIGINT)

test_utils.run_until(self.loop, lambda: caught)

# Removing it should restore the default handler.

self.assertTrue(self.loop.remove_signal_handler(signal.SIGINT))

self.assertEqual(signal.getsignal(signal.SIGINT),

signal.default_int_handler)

# Removing again returns False.

self.assertFalse(self.loop.remove_signal_handler(signal.SIGINT))

@unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')

@unittest.skipUnless(hasattr(signal, 'setitimer'),

'need signal.setitimer()')

def test_signal_handling_while_selecting(self):

# Test with a signal actually arriving during a select() call.

caught = 0

def my_handler():

nonlocal caught

caught += 1

self.loop.stop()

self.loop.add_signal_handler(signal.SIGALRM, my_handler)

signal.setitimer(signal.ITIMER_REAL, 0.01, 0) # Send SIGALRM once.

self.loop.call_later(60, self.loop.stop)

self.loop.run_forever()

self.assertEqual(caught, 1)

@unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')

@unittest.skipUnless(hasattr(signal, 'setitimer'),

'need signal.setitimer()')

def test_signal_handling_args(self):

some_args = (42,)

caught = 0

def my_handler(*args):

nonlocal caught

caught += 1

self.assertEqual(args, some_args)

self.loop.stop()

self.loop.add_signal_handler(signal.SIGALRM, my_handler, *some_args)

signal.setitimer(signal.ITIMER_REAL, 0.1, 0) # Send SIGALRM once.

self.loop.call_later(60, self.loop.stop)

self.loop.run_forever()

self.assertEqual(caught, 1)

def _basetest_create_connection(self, connection_fut, check_sockname=True):

tr, pr = self.loop.run_until_complete(connection_fut)

self.assertIsInstance(tr, asyncio.Transport)

self.assertIsInstance(pr, asyncio.Protocol)

self.assertIs(pr.transport, tr)

if check_sockname:

self.assertIsNotNone(tr.get_extra_info('sockname'))

self.loop.run_until_complete(pr.done)

self.assertGreater(pr.nbytes, 0)

tr.close()

def test_create_connection(self):

with test_utils.run_test_server() as httpd:

conn_fut = self.loop.create_connection(

lambda: MyProto(loop=self.loop), *httpd.address)

self._basetest_create_connection(conn_fut)

@socket_helper.skip_unless_bind_unix_socket

def test_create_unix_connection(self):

# Issue #20682: On Mac OS X Tiger, getsockname() returns a

# zero-length address for UNIX socket.

check_sockname = not broken_unix_getsockname()

with test_utils.run_test_unix_server() as httpd:

conn_fut = self.loop.create_unix_connection(

lambda: MyProto(loop=self.loop), httpd.address)

self._basetest_create_connection(conn_fut, check_sockname)

def check_ssl_extra_info(self, client, check_sockname=True,

peername=None, peercert={}):

if check_sockname:

self.assertIsNotNone(client.get_extra_info('sockname'))

if peername:

self.assertEqual(peername,

client.get_extra_info('peername'))

else:

self.assertIsNotNone(client.get_extra_info('peername'))

self.assertEqual(peercert,

client.get_extra_info('peercert'))

# test SSL cipher

cipher = client.get_extra_info('cipher')

self.assertIsInstance(cipher, tuple)

self.assertEqual(len(cipher), 3, cipher)

self.assertIsInstance(cipher[0], str)

self.assertIsInstance(cipher[1], str)

self.assertIsInstance(cipher[2], int)

# test SSL object

sslobj = client.get_extra_info('ssl_object')

self.assertIsNotNone(sslobj)

self.assertEqual(sslobj.compression(),

client.get_extra_info('compression'))

self.assertEqual(sslobj.cipher(),

client.get_extra_info('cipher'))

self.assertEqual(sslobj.getpeercert(),

client.get_extra_info('peercert'))

self.assertEqual(sslobj.compression(),

client.get_extra_info('compression'))

def _basetest_create_ssl_connection(self, connection_fut,

check_sockname=True,

peername=None):

tr, pr = self.loop.run_until_complete(connection_fut)

self.assertIsInstance(tr, asyncio.Transport)

self.assertIsInstance(pr, asyncio.Protocol)

self.assertTrue('ssl' in tr.__class__.__name__.lower())

self.check_ssl_extra_info(tr, check_sockname, peername)

self.loop.run_until_complete(pr.done)

self.assertGreater(pr.nbytes, 0)

tr.close()

def _test_create_ssl_connection(self, httpd, create_connection,

check_sockname=True, peername=None):

conn_fut = create_connection(ssl=test_utils.dummy_ssl_context())

self._basetest_create_ssl_connection(conn_fut, check_sockname,

peername)

# ssl.Purpose was introduced in Python 3.4

if hasattr(ssl, 'Purpose'):

def _dummy_ssl_create_context(purpose=ssl.Purpose.SERVER_AUTH, *,

cafile=None, capath=None,

cadata=None):

"""

A ssl.create_default_context() replacement that doesn't enable

cert validation.

"""

self.assertEqual(purpose, ssl.Purpose.SERVER_AUTH)

return test_utils.dummy_ssl_context()

# With ssl=True, ssl.create_default_context() should be called

with mock.patch('ssl.create_default_context',

side_effect=_dummy_ssl_create_context) as m:

conn_fut = create_connection(ssl=True)

self._basetest_create_ssl_connection(conn_fut, check_sockname,

peername)

self.assertEqual(m.call_count, 1)

# With the real ssl.create_default_context(), certificate

# validation will fail

with self.assertRaises(ssl.SSLError) as cm:

conn_fut = create_connection(ssl=True)

# Ignore the "SSL handshake failed" log in debug mode

with test_utils.disable_logger():

self._basetest_create_ssl_connection(conn_fut, check_sockname,

peername)

self.assertEqual(cm.exception.reason, 'CERTIFICATE_VERIFY_FAILED')

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_ssl_connection(self):

with test_utils.run_test_server(use_ssl=True) as httpd:

create_connection = functools.partial(

self.loop.create_connection,

lambda: MyProto(loop=self.loop),

*httpd.address)

self._test_create_ssl_connection(httpd, create_connection,

peername=httpd.address)

@socket_helper.skip_unless_bind_unix_socket

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_ssl_unix_connection(self):

# Issue #20682: On Mac OS X Tiger, getsockname() returns a

# zero-length address for UNIX socket.

check_sockname = not broken_unix_getsockname()

with test_utils.run_test_unix_server(use_ssl=True) as httpd:

create_connection = functools.partial(

self.loop.create_unix_connection,

lambda: MyProto(loop=self.loop), httpd.address,

server_hostname='127.0.0.1')

self._test_create_ssl_connection(httpd, create_connection,

check_sockname,

peername=httpd.address)

def test_create_connection_local_addr(self):

with test_utils.run_test_server() as httpd:

port = socket_helper.find_unused_port()

f = self.loop.create_connection(

lambda: MyProto(loop=self.loop),

*httpd.address, local_addr=(httpd.address[0], port))

tr, pr = self.loop.run_until_complete(f)

expected = pr.transport.get_extra_info('sockname')[1]

self.assertEqual(port, expected)

tr.close()

@socket_helper.skip_if_tcp_blackhole

def test_create_connection_local_addr_skip_different_family(self):

# See https://github.com/python/cpython/issues/86508

port1 = socket_helper.find_unused_port()

port2 = socket_helper.find_unused_port()

getaddrinfo_orig = self.loop.getaddrinfo

async def getaddrinfo(host, port, *args, **kwargs):

if port == port2:

return [(socket.AF_INET6, socket.SOCK_STREAM, 0, '', ('::1', 0, 0, 0)),

(socket.AF_INET, socket.SOCK_STREAM, 0, '', ('127.0.0.1', 0))]

return await getaddrinfo_orig(host, port, *args, **kwargs)

self.loop.getaddrinfo = getaddrinfo

f = self.loop.create_connection(

lambda: MyProto(loop=self.loop),

'localhost', port1, local_addr=('localhost', port2))

with self.assertRaises(OSError):

self.loop.run_until_complete(f)

@socket_helper.skip_if_tcp_blackhole

def test_create_connection_local_addr_nomatch_family(self):

# See https://github.com/python/cpython/issues/86508

port1 = socket_helper.find_unused_port()

port2 = socket_helper.find_unused_port()

getaddrinfo_orig = self.loop.getaddrinfo

async def getaddrinfo(host, port, *args, **kwargs):

if port == port2:

return [(socket.AF_INET6, socket.SOCK_STREAM, 0, '', ('::1', 0, 0, 0))]

return await getaddrinfo_orig(host, port, *args, **kwargs)

self.loop.getaddrinfo = getaddrinfo

f = self.loop.create_connection(

lambda: MyProto(loop=self.loop),

'localhost', port1, local_addr=('localhost', port2))

with self.assertRaises(OSError):

self.loop.run_until_complete(f)

def test_create_connection_local_addr_in_use(self):

with test_utils.run_test_server() as httpd:

f = self.loop.create_connection(

lambda: MyProto(loop=self.loop),

*httpd.address, local_addr=httpd.address)

with self.assertRaises(OSError) as cm:

self.loop.run_until_complete(f)

self.assertEqual(cm.exception.errno, errno.EADDRINUSE)

self.assertIn(str(httpd.address), cm.exception.strerror)

def test_connect_accepted_socket(self, server_ssl=None, client_ssl=None):

loop = self.loop

class MyProto(MyBaseProto):

def connection_lost(self, exc):

super().connection_lost(exc)

loop.call_soon(loop.stop)

def data_received(self, data):

super().data_received(data)

self.transport.write(expected_response)

lsock = socket.create_server(('127.0.0.1', 0), backlog=1)

addr = lsock.getsockname()

message = b'test data'

response = None

expected_response = b'roger'

def client():

nonlocal response

try:

csock = socket.socket()

if client_ssl is not None:

csock = client_ssl.wrap_socket(csock)

csock.connect(addr)

csock.sendall(message)

response = csock.recv(99)

csock.close()

except Exception as exc:

print(

"Failure in client thread in test_connect_accepted_socket",

exc)

thread = threading.Thread(target=client, daemon=True)

thread.start()

conn, _ = lsock.accept()

proto = MyProto(loop=loop)

proto.loop = loop

loop.run_until_complete(

loop.connect_accepted_socket(

(lambda: proto), conn, ssl=server_ssl))

loop.run_forever()

proto.transport.close()

lsock.close()

threading_helper.join_thread(thread)

self.assertFalse(thread.is_alive())

self.assertEqual(proto.state, 'CLOSED')

self.assertEqual(proto.nbytes, len(message))

self.assertEqual(response, expected_response)

@unittest.skipIf(ssl is None, 'No ssl module')

def test_ssl_connect_accepted_socket(self):

server_context = test_utils.simple_server_sslcontext()

client_context = test_utils.simple_client_sslcontext()

self.test_connect_accepted_socket(server_context, client_context)

def test_connect_accepted_socket_ssl_timeout_for_plain_socket(self):

sock = socket.socket()

self.addCleanup(sock.close)

coro = self.loop.connect_accepted_socket(

MyProto, sock, ssl_handshake_timeout=support.LOOPBACK_TIMEOUT)

with self.assertRaisesRegex(

ValueError,

'ssl_handshake_timeout is only meaningful with ssl'):

self.loop.run_until_complete(coro)

@mock.patch('asyncio.base_events.socket')

def create_server_multiple_hosts(self, family, hosts, mock_sock):

async def getaddrinfo(host, port, *args, **kw):

if family == socket.AF_INET:

return [(family, socket.SOCK_STREAM, 6, '', (host, port))]

else:

return [(family, socket.SOCK_STREAM, 6, '', (host, port, 0, 0))]

def getaddrinfo_task(*args, **kwds):

return self.loop.create_task(getaddrinfo(*args, **kwds))

unique_hosts = set(hosts)

if family == socket.AF_INET:

mock_sock.socket().getsockbyname.side_effect = [

(host, 80) for host in unique_hosts]

else:

mock_sock.socket().getsockbyname.side_effect = [

(host, 80, 0, 0) for host in unique_hosts]

self.loop.getaddrinfo = getaddrinfo_task

self.loop._start_serving = mock.Mock()

self.loop._stop_serving = mock.Mock()

f = self.loop.create_server(lambda: MyProto(self.loop), hosts, 80)

server = self.loop.run_until_complete(f)

self.addCleanup(server.close)

server_hosts = {sock.getsockbyname()[0] for sock in server.sockets}

self.assertEqual(server_hosts, unique_hosts)

def test_create_server_multiple_hosts_ipv4(self):

self.create_server_multiple_hosts(socket.AF_INET,

['1.2.3.4', '5.6.7.8', '1.2.3.4'])

def test_create_server_multiple_hosts_ipv6(self):

self.create_server_multiple_hosts(socket.AF_INET6,

['::1', '::2', '::1'])

def test_create_server(self):

proto = MyProto(self.loop)

f = self.loop.create_server(lambda: proto, '0.0.0.0', 0)

server = self.loop.run_until_complete(f)

self.assertEqual(len(server.sockets), 1)

sock = server.sockets[0]

host, port = sock.getsockname()

self.assertEqual(host, '0.0.0.0')

client = socket.socket()

client.connect(('127.0.0.1', port))

client.sendall(b'xxx')

self.loop.run_until_complete(proto.connected)

self.assertEqual('CONNECTED', proto.state)

test_utils.run_until(self.loop, lambda: proto.nbytes > 0)

self.assertEqual(3, proto.nbytes)

# extra info is available

self.assertIsNotNone(proto.transport.get_extra_info('sockname'))

self.assertEqual('127.0.0.1',

proto.transport.get_extra_info('peername')[0])

# close connection

proto.transport.close()

self.loop.run_until_complete(proto.done)

self.assertEqual('CLOSED', proto.state)

# the client socket must be closed after to avoid ECONNRESET upon

# recv()/send() on the serving socket

client.close()

# close server

server.close()

def test_create_server_trsock(self):

proto = MyProto(self.loop)

f = self.loop.create_server(lambda: proto, '0.0.0.0', 0)

server = self.loop.run_until_complete(f)

self.assertEqual(len(server.sockets), 1)

sock = server.sockets[0]

self.assertIsInstance(sock, asyncio.trsock.TransportSocket)

host, port = sock.getsockname()

self.assertEqual(host, '0.0.0.0')

dup = sock.dup()

self.addCleanup(dup.close)