test_events.py

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

importconcurrent.futures
importcontextlib
importfunctools
importio
importmultiprocessing
importos
importplatform
importre
importsignal
importsocket
try:
importssl
exceptImportError:
ssl=None
importsubprocess
importsys
importthreading
importtime
importtypes
importerrno
importunittest
fromunittestimportmock
importweakref
importwarnings
ifsys.platformnotin ('win32', 'vxworks'):
importtty

importasyncio
fromasyncioimportcoroutines
fromasyncioimportevents
fromasyncioimportselector_events
frommultiprocessing.utilimport_cleanup_testsasmultiprocessing_cleanup_tests
fromtest.test_asyncioimportutilsastest_utils
fromtestimportsupport
fromtest.supportimportsocket_helper
fromtest.supportimportthreading_helper
fromtest.supportimportALWAYS_EQ, LARGEST, SMALLEST


deftearDownModule():
asyncio.set_event_loop_policy(None)


defbroken_unix_getsockname():
"""Return True if the platform is Mac OS 10.4 or older."""
ifsys.platform.startswith("aix"):
returnTrue
elifsys.platform!='darwin':
returnFalse
version=platform.mac_ver()[0]
version=tuple(map(int, version.split('.')))
returnversion< (10, 5)


def_test_get_event_loop_new_process__sub_proc():
asyncdefdoit():
return'hello'

withcontextlib.closing(asyncio.new_event_loop()) asloop:
asyncio.set_event_loop(loop)
returnloop.run_until_complete(doit())


classCoroLike:
defsend(self, v):
pass

defthrow(self, *exc):
pass

defclose(self):
pass

def__await__(self):
pass


classMyBaseProto(asyncio.Protocol):
connected=None
done=None

def__init__(self, loop=None):
self.transport=None
self.state='INITIAL'
self.nbytes=0
ifloopisnotNone:
self.connected=loop.create_future()
self.done=loop.create_future()

def_assert_state(self, *expected):
ifself.statenotinexpected:
raiseAssertionError(f'state: {self.state!r}, expected: {expected!r}')

defconnection_made(self, transport):
self.transport=transport
self._assert_state('INITIAL')
self.state='CONNECTED'
ifself.connected:
self.connected.set_result(None)

defdata_received(self, data):
self._assert_state('CONNECTED')
self.nbytes+=len(data)

defeof_received(self):
self._assert_state('CONNECTED')
self.state='EOF'

defconnection_lost(self, exc):
self._assert_state('CONNECTED', 'EOF')
self.state='CLOSED'
ifself.done:
self.done.set_result(None)


classMyProto(MyBaseProto):
defconnection_made(self, transport):
super().connection_made(transport)
transport.write(b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n')


classMyDatagramProto(asyncio.DatagramProtocol):
done=None

def__init__(self, loop=None):
self.state='INITIAL'
self.nbytes=0
ifloopisnotNone:
self.done=loop.create_future()

def_assert_state(self, expected):
ifself.state!=expected:
raiseAssertionError(f'state: {self.state!r}, expected: {expected!r}')

defconnection_made(self, transport):
self.transport=transport
self._assert_state('INITIAL')
self.state='INITIALIZED'

defdatagram_received(self, data, addr):
self._assert_state('INITIALIZED')
self.nbytes+=len(data)

deferror_received(self, exc):
self._assert_state('INITIALIZED')

defconnection_lost(self, exc):
self._assert_state('INITIALIZED')
self.state='CLOSED'
ifself.done:
self.done.set_result(None)


classMyReadPipeProto(asyncio.Protocol):
done=None

def__init__(self, loop=None):
self.state= ['INITIAL']
self.nbytes=0
self.transport=None
ifloopisnotNone:
self.done=loop.create_future()

def_assert_state(self, expected):
ifself.state!=expected:
raiseAssertionError(f'state: {self.state!r}, expected: {expected!r}')

defconnection_made(self, transport):
self.transport=transport
self._assert_state(['INITIAL'])
self.state.append('CONNECTED')

defdata_received(self, data):
self._assert_state(['INITIAL', 'CONNECTED'])
self.nbytes+=len(data)

defeof_received(self):
self._assert_state(['INITIAL', 'CONNECTED'])
self.state.append('EOF')

defconnection_lost(self, exc):
if'EOF'notinself.state:
self.state.append('EOF') # It is okay if EOF is missed.
self._assert_state(['INITIAL', 'CONNECTED', 'EOF'])
self.state.append('CLOSED')
ifself.done:
self.done.set_result(None)


classMyWritePipeProto(asyncio.BaseProtocol):
done=None

def__init__(self, loop=None):
self.state='INITIAL'
self.transport=None
ifloopisnotNone:
self.done=loop.create_future()

def_assert_state(self, expected):
ifself.state!=expected:
raiseAssertionError(f'state: {self.state!r}, expected: {expected!r}')

defconnection_made(self, transport):
self.transport=transport
self._assert_state('INITIAL')
self.state='CONNECTED'

defconnection_lost(self, exc):
self._assert_state('CONNECTED')
self.state='CLOSED'
ifself.done:
self.done.set_result(None)


classMySubprocessProtocol(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() forfdinrange(3)}
self.data= {1: b'', 2: b''}
self.returncode=None
self.got_data= {1: asyncio.Event(),
2: asyncio.Event()}

def_assert_state(self, expected):
ifself.state!=expected:
raiseAssertionError(f'state: {self.state!r}, expected: {expected!r}')

defconnection_made(self, transport):
self.transport=transport
self._assert_state('INITIAL')
self.state='CONNECTED'
self.connected.set_result(None)

defconnection_lost(self, exc):
self._assert_state('CONNECTED')
self.state='CLOSED'
self.completed.set_result(None)

defpipe_data_received(self, fd, data):
self._assert_state('CONNECTED')
self.data[fd] +=data
self.got_data[fd].set()

defpipe_connection_lost(self, fd, exc):
self._assert_state('CONNECTED')
ifexc:
self.disconnects[fd].set_exception(exc)
else:
self.disconnects[fd].set_result(exc)

defprocess_exited(self):
self._assert_state('CONNECTED')
self.returncode=self.transport.get_returncode()


classEventLoopTestsMixin:

defsetUp(self):
super().setUp()
self.loop=self.create_event_loop()
self.set_event_loop(self.loop)

deftearDown(self):
# just in case if we have transport close callbacks
ifnotself.loop.is_closed():
test_utils.run_briefly(self.loop)

self.doCleanups()
support.gc_collect()
super().tearDown()

deftest_run_until_complete_nesting(self):
asyncdefcoro1():
awaitasyncio.sleep(0)

asyncdefcoro2():
self.assertTrue(self.loop.is_running())
self.loop.run_until_complete(coro1())

withself.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).

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

deftest_run_until_complete_stopped(self):

asyncdefcb():
self.loop.stop()
awaitasyncio.sleep(0.1)
task=cb()
self.assertRaises(RuntimeError,
self.loop.run_until_complete, task)

deftest_call_later(self):
results= []

defcallback(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'])

deftest_call_soon(self):
results= []

defcallback(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')])

deftest_call_soon_threadsafe(self):
results= []
lock=threading.Lock()

defcallback(arg):
results.append(arg)
iflen(results) >=2:
self.loop.stop()

defrun_in_thread():
self.loop.call_soon_threadsafe(callback, 'hello')
lock.release()

lock.acquire()
t=threading.Thread(target=run_in_thread)
t.start()

withlock:
self.loop.call_soon(callback, 'world')
self.loop.run_forever()
t.join()
self.assertEqual(results, ['hello', 'world'])

deftest_call_soon_threadsafe_same_thread(self):
results= []

defcallback(arg):
results.append(arg)
iflen(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'])

deftest_run_in_executor(self):
defrun(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())

deftest_run_in_executor_cancel(self):
called=False

defpatched_call_soon(*args):
nonlocalcalled
called=True

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

deftest_reader_callback(self):
r, w=socket.socketpair()
r.setblocking(False)
bytes_read=bytearray()

defreader():
try:
data=r.recv(1024)
exceptBlockingIOError:
# Spurious readiness notifications are possible
# at least on Linux -- see man select.
return
ifdata:
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')

deftest_writer_callback(self):
r, w=socket.socketpair()
w.setblocking(False)

defwriter(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.skipUnless(hasattr(signal, 'SIGKILL'), 'No SIGKILL')
deftest_add_signal_handler(self):
caught=0

defmy_handler():
nonlocalcaught
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()')
deftest_signal_handling_while_selecting(self):
# Test with a signal actually arriving during a select() call.
caught=0

defmy_handler():
nonlocalcaught
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()')
deftest_signal_handling_args(self):
some_args= (42,)
caught=0

defmy_handler(*args):
nonlocalcaught
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)
ifcheck_sockname:
self.assertIsNotNone(tr.get_extra_info('sockname'))
self.loop.run_until_complete(pr.done)
self.assertGreater(pr.nbytes, 0)
tr.close()

deftest_create_connection(self):
withtest_utils.run_test_server() ashttpd:
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
deftest_create_unix_connection(self):
# Issue #20682: On Mac OS X Tiger, getsockname() returns a
# zero-length address for UNIX socket.
check_sockname=notbroken_unix_getsockname()

withtest_utils.run_test_unix_server() ashttpd:
conn_fut=self.loop.create_unix_connection(
lambda: MyProto(loop=self.loop), httpd.address)
self._basetest_create_connection(conn_fut, check_sockname)

defcheck_ssl_extra_info(self, client, check_sockname=True,
peername=None, peercert={}):
ifcheck_sockname:
self.assertIsNotNone(client.get_extra_info('sockname'))
ifpeername:
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'intr.__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
ifhasattr(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)
returntest_utils.dummy_ssl_context()

# With ssl=True, ssl.create_default_context() should be called
withmock.patch('ssl.create_default_context',
side_effect=_dummy_ssl_create_context) asm:
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
withself.assertRaises(ssl.SSLError) ascm:
conn_fut=create_connection(ssl=True)
# Ignore the "SSL handshake failed" log in debug mode
withtest_utils.disable_logger():
self._basetest_create_ssl_connection(conn_fut, check_sockname,
peername)

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

@unittest.skipIf(sslisNone, 'No ssl module')
deftest_create_ssl_connection(self):
withtest_utils.run_test_server(use_ssl=True) ashttpd:
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(sslisNone, 'No ssl module')
deftest_create_ssl_unix_connection(self):
# Issue #20682: On Mac OS X Tiger, getsockname() returns a
# zero-length address for UNIX socket.
check_sockname=notbroken_unix_getsockname()

withtest_utils.run_test_unix_server(use_ssl=True) ashttpd:
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)

deftest_create_connection_local_addr(self):
withtest_utils.run_test_server() ashttpd:
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
deftest_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

asyncdefgetaddrinfo(host, port, *args, **kwargs):
ifport==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))]
returnawaitgetaddrinfo_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))

withself.assertRaises(OSError):
self.loop.run_until_complete(f)

@socket_helper.skip_if_tcp_blackhole
deftest_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

asyncdefgetaddrinfo(host, port, *args, **kwargs):
ifport==port2:
return [(socket.AF_INET6, socket.SOCK_STREAM, 0, '', ('::1', 0, 0, 0))]
returnawaitgetaddrinfo_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))

withself.assertRaises(OSError):
self.loop.run_until_complete(f)

deftest_create_connection_local_addr_in_use(self):
withtest_utils.run_test_server() ashttpd:
f=self.loop.create_connection(
lambda: MyProto(loop=self.loop),
*httpd.address, local_addr=httpd.address)
withself.assertRaises(OSError) ascm:
self.loop.run_until_complete(f)
self.assertEqual(cm.exception.errno, errno.EADDRINUSE)
self.assertIn(str(httpd.address), cm.exception.strerror)

deftest_connect_accepted_socket(self, server_ssl=None, client_ssl=None):
loop=self.loop

classMyProto(MyBaseProto):

defconnection_lost(self, exc):
super().connection_lost(exc)
loop.call_soon(loop.stop)

defdata_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'

defclient():
nonlocalresponse
try:
csock=socket.socket()
ifclient_sslisnotNone:
csock=client_ssl.wrap_socket(csock)
csock.connect(addr)
csock.sendall(message)
response=csock.recv(99)
csock.close()
exceptExceptionasexc:
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(sslisNone, 'No ssl module')
deftest_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)

deftest_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)
withself.assertRaisesRegex(
ValueError,
'ssl_handshake_timeout is only meaningful with ssl'):
self.loop.run_until_complete(coro)

@mock.patch('asyncio.base_events.socket')
defcreate_server_multiple_hosts(self, family, hosts, mock_sock):
asyncdefgetaddrinfo(host, port, *args, **kw):
iffamily==socket.AF_INET:
return [(family, socket.SOCK_STREAM, 6, '', (host, port))]
else:
return [(family, socket.SOCK_STREAM, 6, '', (host, port, 0, 0))]

defgetaddrinfo_task(*args, **kwds):
returnself.loop.create_task(getaddrinfo(*args, **kwds))

unique_hosts=set(hosts)

iffamily==socket.AF_INET:
mock_sock.socket().getsockbyname.side_effect= [
 (host, 80) forhostinunique_hosts]
else:
mock_sock.socket().getsockbyname.side_effect= [
 (host, 80, 0, 0) forhostinunique_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] forsockinserver.sockets}
self.assertEqual(server_hosts, unique_hosts)

deftest_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'])

deftest_create_server_multiple_hosts_ipv6(self):
self.create_server_multiple_hosts(socket.AF_INET6,
 ['::1', '::2', '::1'])

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

deftest_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)
self.assertIsInstance(dup, socket.socket)
self.assertFalse(sock.get_inheritable())
withself.assertRaises(ValueError):
sock.settimeout(1)
sock.settimeout(0)
self.assertEqual(sock.gettimeout(), 0)
withself.assertRaises(ValueError):
sock.setblocking(True)
sock.setblocking(False)
server.close()


@unittest.skipUnless(hasattr(socket, 'SO_REUSEPORT'), 'No SO_REUSEPORT')
deftest_create_server_reuse_port(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.assertFalse(
sock.getsockopt(
socket.SOL_SOCKET, socket.SO_REUSEPORT))
server.close()

test_utils.run_briefly(self.loop)

proto=MyProto(self.loop)
f=self.loop.create_server(
lambda: proto, '0.0.0.0', 0, reuse_port=True)
server=self.loop.run_until_complete(f)
self.assertEqual(len(server.sockets), 1)
sock=server.sockets[0]
self.assertTrue(
sock.getsockopt(
socket.SOL_SOCKET, socket.SO_REUSEPORT))
server.close()

def_make_unix_server(self, factory, **kwargs):
path=test_utils.gen_unix_socket_path()
self.addCleanup(lambda: os.path.exists(path) andos.unlink(path))

f=self.loop.create_unix_server(factory, path, **kwargs)
server=self.loop.run_until_complete(f)

returnserver, path

@socket_helper.skip_unless_bind_unix_socket
deftest_create_unix_server(self):
proto=MyProto(loop=self.loop)
server, path=self._make_unix_server(lambda: proto)
self.assertEqual(len(server.sockets), 1)

client=socket.socket(socket.AF_UNIX)
client.connect(path)
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)

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

@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
deftest_create_unix_server_path_socket_error(self):
proto=MyProto(loop=self.loop)
sock=socket.socket()
withsock:
f=self.loop.create_unix_server(lambda: proto, '/test', sock=sock)
withself.assertRaisesRegex(ValueError,
'path and sock can not be specified '
'at the same time'):
self.loop.run_until_complete(f)

def_create_ssl_context(self, certfile, keyfile=None):
sslcontext=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
sslcontext.options|=ssl.OP_NO_SSLv2
sslcontext.load_cert_chain(certfile, keyfile)
returnsslcontext

def_make_ssl_server(self, factory, certfile, keyfile=None):
sslcontext=self._create_ssl_context(certfile, keyfile)

f=self.loop.create_server(factory, '127.0.0.1', 0, ssl=sslcontext)
server=self.loop.run_until_complete(f)

sock=server.sockets[0]
host, port=sock.getsockname()
self.assertEqual(host, '127.0.0.1')
returnserver, host, port

def_make_ssl_unix_server(self, factory, certfile, keyfile=None):
sslcontext=self._create_ssl_context(certfile, keyfile)
returnself._make_unix_server(factory, ssl=sslcontext)

@unittest.skipIf(sslisNone, 'No ssl module')
deftest_create_server_ssl(self):
proto=MyProto(loop=self.loop)
server, host, port=self._make_ssl_server(
lambda: proto, test_utils.ONLYCERT, test_utils.ONLYKEY)

f_c=self.loop.create_connection(MyBaseProto, host, port,
ssl=test_utils.dummy_ssl_context())
client, pr=self.loop.run_until_complete(f_c)

client.write(b'xxx')