test_unix_events.py

"""Tests for unix_events.py."""

importcontextlib
importerrno
importio
importmultiprocessing
frommultiprocessing.utilimport_cleanup_testsasmultiprocessing_cleanup_tests
importos
importsignal
importsocket
importstat
importsys
importthreading
importtime
importunittest
fromunittestimportmock
importwarnings

fromtestimportsupport
fromtest.supportimportos_helper
fromtest.supportimportsocket_helper
fromtest.supportimportwait_process
fromtest.supportimporthashlib_helper

ifsys.platform=='win32':
raiseunittest.SkipTest('UNIX only')


importasyncio
fromasyncioimportlog
fromasyncioimportunix_events
fromtest.test_asyncioimportutilsastest_utils


deftearDownModule():
asyncio.set_event_loop_policy(None)


MOCK_ANY=mock.ANY


defEXITCODE(exitcode):
return32768+exitcode


defSIGNAL(signum):
ifnot1<=signum<=68:
raiseAssertionError(f'invalid signum {signum}')
return32768-signum


defclose_pipe_transport(transport):
# Don't call transport.close() because the event loop and the selector
# are mocked
iftransport._pipeisNone:
return
transport._pipe.close()
transport._pipe=None


@unittest.skipUnless(signal, 'Signals are not supported')
classSelectorEventLoopSignalTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=asyncio.SelectorEventLoop()
self.set_event_loop(self.loop)

deftest_check_signal(self):
self.assertRaises(
TypeError, self.loop._check_signal, '1')
self.assertRaises(
ValueError, self.loop._check_signal, signal.NSIG+1)

deftest_handle_signal_no_handler(self):
self.loop._handle_signal(signal.NSIG+1)

deftest_handle_signal_cancelled_handler(self):
h=asyncio.Handle(mock.Mock(), (),
loop=mock.Mock())
h.cancel()
self.loop._signal_handlers[signal.NSIG+1] =h
self.loop.remove_signal_handler=mock.Mock()
self.loop._handle_signal(signal.NSIG+1)
self.loop.remove_signal_handler.assert_called_with(signal.NSIG+1)

@mock.patch('asyncio.unix_events.signal')
deftest_add_signal_handler_setup_error(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals
m_signal.set_wakeup_fd.side_effect=ValueError

self.assertRaises(
RuntimeError,
self.loop.add_signal_handler,
signal.SIGINT, lambda: True)

@mock.patch('asyncio.unix_events.signal')
deftest_add_signal_handler_coroutine_error(self, m_signal):
m_signal.NSIG=signal.NSIG

asyncdefsimple_coroutine():
pass

# callback must not be a coroutine function
coro_func=simple_coroutine
coro_obj=coro_func()
self.addCleanup(coro_obj.close)
forfuncin (coro_func, coro_obj):
self.assertRaisesRegex(
TypeError, 'coroutines cannot be used with add_signal_handler',
self.loop.add_signal_handler,
signal.SIGINT, func)

@mock.patch('asyncio.unix_events.signal')
deftest_add_signal_handler(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals

cb=lambda: True
self.loop.add_signal_handler(signal.SIGHUP, cb)
h=self.loop._signal_handlers.get(signal.SIGHUP)
self.assertIsInstance(h, asyncio.Handle)
self.assertEqual(h._callback, cb)

@mock.patch('asyncio.unix_events.signal')
deftest_add_signal_handler_install_error(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals

defset_wakeup_fd(fd):
iffd==-1:
raiseValueError()
m_signal.set_wakeup_fd=set_wakeup_fd

classErr(OSError):
errno=errno.EFAULT
m_signal.signal.side_effect=Err

self.assertRaises(
Err,
self.loop.add_signal_handler,
signal.SIGINT, lambda: True)

@mock.patch('asyncio.unix_events.signal')
@mock.patch('asyncio.base_events.logger')
deftest_add_signal_handler_install_error2(self, m_logging, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals

classErr(OSError):
errno=errno.EINVAL
m_signal.signal.side_effect=Err

self.loop._signal_handlers[signal.SIGHUP] =lambda: True
self.assertRaises(
RuntimeError,
self.loop.add_signal_handler,
signal.SIGINT, lambda: True)
self.assertFalse(m_logging.info.called)
self.assertEqual(1, m_signal.set_wakeup_fd.call_count)

@mock.patch('asyncio.unix_events.signal')
@mock.patch('asyncio.base_events.logger')
deftest_add_signal_handler_install_error3(self, m_logging, m_signal):
classErr(OSError):
errno=errno.EINVAL
m_signal.signal.side_effect=Err
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals

self.assertRaises(
RuntimeError,
self.loop.add_signal_handler,
signal.SIGINT, lambda: True)
self.assertFalse(m_logging.info.called)
self.assertEqual(2, m_signal.set_wakeup_fd.call_count)

@mock.patch('asyncio.unix_events.signal')
deftest_remove_signal_handler(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals

self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

self.assertTrue(
self.loop.remove_signal_handler(signal.SIGHUP))
self.assertTrue(m_signal.set_wakeup_fd.called)
self.assertTrue(m_signal.signal.called)
self.assertEqual(
 (signal.SIGHUP, m_signal.SIG_DFL), m_signal.signal.call_args[0])

@mock.patch('asyncio.unix_events.signal')
deftest_remove_signal_handler_2(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.SIGINT=signal.SIGINT
m_signal.valid_signals=signal.valid_signals

self.loop.add_signal_handler(signal.SIGINT, lambda: True)
self.loop._signal_handlers[signal.SIGHUP] =object()
m_signal.set_wakeup_fd.reset_mock()

self.assertTrue(
self.loop.remove_signal_handler(signal.SIGINT))
self.assertFalse(m_signal.set_wakeup_fd.called)
self.assertTrue(m_signal.signal.called)
self.assertEqual(
 (signal.SIGINT, m_signal.default_int_handler),
m_signal.signal.call_args[0])

@mock.patch('asyncio.unix_events.signal')
@mock.patch('asyncio.base_events.logger')
deftest_remove_signal_handler_cleanup_error(self, m_logging, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

m_signal.set_wakeup_fd.side_effect=ValueError

self.loop.remove_signal_handler(signal.SIGHUP)
self.assertTrue(m_logging.info)

@mock.patch('asyncio.unix_events.signal')
deftest_remove_signal_handler_error(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

m_signal.signal.side_effect=OSError

self.assertRaises(
OSError, self.loop.remove_signal_handler, signal.SIGHUP)

@mock.patch('asyncio.unix_events.signal')
deftest_remove_signal_handler_error2(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

classErr(OSError):
errno=errno.EINVAL
m_signal.signal.side_effect=Err

self.assertRaises(
RuntimeError, self.loop.remove_signal_handler, signal.SIGHUP)

@mock.patch('asyncio.unix_events.signal')
deftest_close(self, m_signal):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals

self.loop.add_signal_handler(signal.SIGHUP, lambda: True)
self.loop.add_signal_handler(signal.SIGCHLD, lambda: True)

self.assertEqual(len(self.loop._signal_handlers), 2)

m_signal.set_wakeup_fd.reset_mock()

self.loop.close()

self.assertEqual(len(self.loop._signal_handlers), 0)
m_signal.set_wakeup_fd.assert_called_once_with(-1)

@mock.patch('asyncio.unix_events.sys')
@mock.patch('asyncio.unix_events.signal')
deftest_close_on_finalizing(self, m_signal, m_sys):
m_signal.NSIG=signal.NSIG
m_signal.valid_signals=signal.valid_signals
self.loop.add_signal_handler(signal.SIGHUP, lambda: True)

self.assertEqual(len(self.loop._signal_handlers), 1)
m_sys.is_finalizing.return_value=True
m_signal.signal.reset_mock()

withself.assertWarnsRegex(ResourceWarning,
"skipping signal handlers removal"):
self.loop.close()

self.assertEqual(len(self.loop._signal_handlers), 0)
self.assertFalse(m_signal.signal.called)


@unittest.skipUnless(hasattr(socket, 'AF_UNIX'),
'UNIX Sockets are not supported')
classSelectorEventLoopUnixSocketTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=asyncio.SelectorEventLoop()
self.set_event_loop(self.loop)

@socket_helper.skip_unless_bind_unix_socket
deftest_create_unix_server_existing_path_sock(self):
withtest_utils.unix_socket_path() aspath:
sock=socket.socket(socket.AF_UNIX)
sock.bind(path)
sock.listen(1)
sock.close()

coro=self.loop.create_unix_server(lambda: None, path)
srv=self.loop.run_until_complete(coro)
srv.close()
self.loop.run_until_complete(srv.wait_closed())

@socket_helper.skip_unless_bind_unix_socket
deftest_create_unix_server_pathlike(self):
withtest_utils.unix_socket_path() aspath:
path=os_helper.FakePath(path)
srv_coro=self.loop.create_unix_server(lambda: None, path)
srv=self.loop.run_until_complete(srv_coro)
srv.close()
self.loop.run_until_complete(srv.wait_closed())

deftest_create_unix_connection_pathlike(self):
withtest_utils.unix_socket_path() aspath:
path=os_helper.FakePath(path)
coro=self.loop.create_unix_connection(lambda: None, path)
withself.assertRaises(FileNotFoundError):
# If path-like object weren't supported, the exception would be
# different.
self.loop.run_until_complete(coro)

deftest_create_unix_server_existing_path_nonsock(self):
path=test_utils.gen_unix_socket_path()
self.addCleanup(os_helper.unlink, path)
# create the file
open(path, "wb").close()

coro=self.loop.create_unix_server(lambda: None, path)
withself.assertRaisesRegex(OSError,
'Address.*is already in use'):
self.loop.run_until_complete(coro)

deftest_create_unix_server_ssl_bool(self):
coro=self.loop.create_unix_server(lambda: None, path='spam',
ssl=True)
withself.assertRaisesRegex(TypeError,
'ssl argument must be an SSLContext'):
self.loop.run_until_complete(coro)

deftest_create_unix_server_nopath_nosock(self):
coro=self.loop.create_unix_server(lambda: None, path=None)
withself.assertRaisesRegex(ValueError,
'path was not specified, and no sock'):
self.loop.run_until_complete(coro)

deftest_create_unix_server_path_inetsock(self):
sock=socket.socket()
withsock:
coro=self.loop.create_unix_server(lambda: None, path=None,
sock=sock)
withself.assertRaisesRegex(ValueError,
'A UNIX Domain Stream.*was expected'):
self.loop.run_until_complete(coro)

deftest_create_unix_server_path_dgram(self):
sock=socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
withsock:
coro=self.loop.create_unix_server(lambda: None, path=None,
sock=sock)
withself.assertRaisesRegex(ValueError,
'A UNIX Domain Stream.*was expected'):
self.loop.run_until_complete(coro)

@unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'),
'no socket.SOCK_NONBLOCK (linux only)')
@socket_helper.skip_unless_bind_unix_socket
deftest_create_unix_server_path_stream_bittype(self):
fn=test_utils.gen_unix_socket_path()
self.addCleanup(os_helper.unlink, fn)

sock=socket.socket(socket.AF_UNIX,
socket.SOCK_STREAM|socket.SOCK_NONBLOCK)
withsock:
sock.bind(fn)
coro=self.loop.create_unix_server(lambda: None, path=None,
sock=sock)
srv=self.loop.run_until_complete(coro)
srv.close()
self.loop.run_until_complete(srv.wait_closed())

deftest_create_unix_server_ssl_timeout_with_plain_sock(self):
coro=self.loop.create_unix_server(lambda: None, path='spam',
ssl_handshake_timeout=1)
withself.assertRaisesRegex(
ValueError,
'ssl_handshake_timeout is only meaningful with ssl'):
self.loop.run_until_complete(coro)

deftest_create_unix_connection_path_inetsock(self):
sock=socket.socket()
withsock:
coro=self.loop.create_unix_connection(lambda: None,
sock=sock)
withself.assertRaisesRegex(ValueError,
'A UNIX Domain Stream.*was expected'):
self.loop.run_until_complete(coro)

@mock.patch('asyncio.unix_events.socket')
deftest_create_unix_server_bind_error(self, m_socket):
# Ensure that the socket is closed on any bind error
sock=mock.Mock()
m_socket.socket.return_value=sock

sock.bind.side_effect=OSError
coro=self.loop.create_unix_server(lambda: None, path="/test")
withself.assertRaises(OSError):
self.loop.run_until_complete(coro)
self.assertTrue(sock.close.called)

sock.bind.side_effect=MemoryError
coro=self.loop.create_unix_server(lambda: None, path="/test")
withself.assertRaises(MemoryError):
self.loop.run_until_complete(coro)
self.assertTrue(sock.close.called)

deftest_create_unix_connection_path_sock(self):
coro=self.loop.create_unix_connection(
lambda: None, os.devnull, sock=object())
withself.assertRaisesRegex(ValueError, 'path and sock can not be'):
self.loop.run_until_complete(coro)

deftest_create_unix_connection_nopath_nosock(self):
coro=self.loop.create_unix_connection(
lambda: None, None)
withself.assertRaisesRegex(ValueError,
'no path and sock were specified'):
self.loop.run_until_complete(coro)

deftest_create_unix_connection_nossl_serverhost(self):
coro=self.loop.create_unix_connection(
lambda: None, os.devnull, server_hostname='spam')
withself.assertRaisesRegex(ValueError,
'server_hostname is only meaningful'):
self.loop.run_until_complete(coro)

deftest_create_unix_connection_ssl_noserverhost(self):
coro=self.loop.create_unix_connection(
lambda: None, os.devnull, ssl=True)

withself.assertRaisesRegex(
ValueError, 'you have to pass server_hostname when using ssl'):

self.loop.run_until_complete(coro)

deftest_create_unix_connection_ssl_timeout_with_plain_sock(self):
coro=self.loop.create_unix_connection(lambda: None, path='spam',
ssl_handshake_timeout=1)
withself.assertRaisesRegex(
ValueError,
'ssl_handshake_timeout is only meaningful with ssl'):
self.loop.run_until_complete(coro)


@unittest.skipUnless(hasattr(os, 'sendfile'),
'sendfile is not supported')
classSelectorEventLoopUnixSockSendfileTests(test_utils.TestCase):
DATA=b"12345abcde"*16*1024# 160 KiB

classMyProto(asyncio.Protocol):

def__init__(self, loop):
self.started=False
self.closed=False
self.data=bytearray()
self.fut=loop.create_future()
self.transport=None
self._ready=loop.create_future()

defconnection_made(self, transport):
self.started=True
self.transport=transport
self._ready.set_result(None)

defdata_received(self, data):
self.data.extend(data)

defconnection_lost(self, exc):
self.closed=True
self.fut.set_result(None)

asyncdefwait_closed(self):
awaitself.fut

@classmethod
defsetUpClass(cls):
withopen(os_helper.TESTFN, 'wb') asfp:
fp.write(cls.DATA)
super().setUpClass()

@classmethod
deftearDownClass(cls):
os_helper.unlink(os_helper.TESTFN)
super().tearDownClass()

defsetUp(self):
self.loop=asyncio.new_event_loop()
self.set_event_loop(self.loop)
self.file=open(os_helper.TESTFN, 'rb')
self.addCleanup(self.file.close)
super().setUp()

defmake_socket(self, cleanup=True):
sock=socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(False)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024)
ifcleanup:
self.addCleanup(sock.close)
returnsock

defrun_loop(self, coro):
returnself.loop.run_until_complete(coro)

defprepare(self):
sock=self.make_socket()
proto=self.MyProto(self.loop)
port=socket_helper.find_unused_port()
srv_sock=self.make_socket(cleanup=False)
srv_sock.bind((socket_helper.HOST, port))
server=self.run_loop(self.loop.create_server(
lambda: proto, sock=srv_sock))
self.run_loop(self.loop.sock_connect(sock, (socket_helper.HOST, port)))
self.run_loop(proto._ready)

defcleanup():
proto.transport.close()
self.run_loop(proto.wait_closed())

server.close()
self.run_loop(server.wait_closed())

self.addCleanup(cleanup)

returnsock, proto

deftest_sock_sendfile_not_available(self):
sock, proto=self.prepare()
withmock.patch('asyncio.unix_events.os', spec=[]):
withself.assertRaisesRegex(asyncio.SendfileNotAvailableError,
"os[.]sendfile[(][)] is not available"):
self.run_loop(self.loop._sock_sendfile_native(sock, self.file,
0, None))
self.assertEqual(self.file.tell(), 0)

deftest_sock_sendfile_not_a_file(self):
sock, proto=self.prepare()
f=object()
withself.assertRaisesRegex(asyncio.SendfileNotAvailableError,
"not a regular file"):
self.run_loop(self.loop._sock_sendfile_native(sock, f,
0, None))
self.assertEqual(self.file.tell(), 0)

deftest_sock_sendfile_iobuffer(self):
sock, proto=self.prepare()
f=io.BytesIO()
withself.assertRaisesRegex(asyncio.SendfileNotAvailableError,
"not a regular file"):
self.run_loop(self.loop._sock_sendfile_native(sock, f,
0, None))
self.assertEqual(self.file.tell(), 0)

deftest_sock_sendfile_not_regular_file(self):
sock, proto=self.prepare()
f=mock.Mock()
f.fileno.return_value=-1
withself.assertRaisesRegex(asyncio.SendfileNotAvailableError,
"not a regular file"):
self.run_loop(self.loop._sock_sendfile_native(sock, f,
0, None))
self.assertEqual(self.file.tell(), 0)

deftest_sock_sendfile_cancel1(self):
sock, proto=self.prepare()

fut=self.loop.create_future()
fileno=self.file.fileno()
self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
0, None, len(self.DATA), 0)
fut.cancel()
withcontextlib.suppress(asyncio.CancelledError):
self.run_loop(fut)
withself.assertRaises(KeyError):
self.loop._selector.get_key(sock)

deftest_sock_sendfile_cancel2(self):
sock, proto=self.prepare()

fut=self.loop.create_future()
fileno=self.file.fileno()
self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
0, None, len(self.DATA), 0)
fut.cancel()
self.loop._sock_sendfile_native_impl(fut, sock.fileno(), sock, fileno,
0, None, len(self.DATA), 0)
withself.assertRaises(KeyError):
self.loop._selector.get_key(sock)

deftest_sock_sendfile_blocking_error(self):
sock, proto=self.prepare()

fileno=self.file.fileno()
fut=mock.Mock()
fut.cancelled.return_value=False
withmock.patch('os.sendfile', side_effect=BlockingIOError()):
self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
0, None, len(self.DATA), 0)
key=self.loop._selector.get_key(sock)
self.assertIsNotNone(key)
fut.add_done_callback.assert_called_once_with(mock.ANY)

deftest_sock_sendfile_os_error_first_call(self):
sock, proto=self.prepare()

fileno=self.file.fileno()
fut=self.loop.create_future()
withmock.patch('os.sendfile', side_effect=OSError()):
self.loop._sock_sendfile_native_impl(fut, None, sock, fileno,
0, None, len(self.DATA), 0)
withself.assertRaises(KeyError):
self.loop._selector.get_key(sock)
exc=fut.exception()
self.assertIsInstance(exc, asyncio.SendfileNotAvailableError)
self.assertEqual(0, self.file.tell())

deftest_sock_sendfile_os_error_next_call(self):
sock, proto=self.prepare()

fileno=self.file.fileno()
fut=self.loop.create_future()
err=OSError()
withmock.patch('os.sendfile', side_effect=err):
self.loop._sock_sendfile_native_impl(fut, sock.fileno(),
sock, fileno,
1000, None, len(self.DATA),
1000)
withself.assertRaises(KeyError):
self.loop._selector.get_key(sock)
exc=fut.exception()
self.assertIs(exc, err)
self.assertEqual(1000, self.file.tell())

deftest_sock_sendfile_exception(self):
sock, proto=self.prepare()

fileno=self.file.fileno()
fut=self.loop.create_future()
err=asyncio.SendfileNotAvailableError()
withmock.patch('os.sendfile', side_effect=err):
self.loop._sock_sendfile_native_impl(fut, sock.fileno(),
sock, fileno,
1000, None, len(self.DATA),
1000)
withself.assertRaises(KeyError):
self.loop._selector.get_key(sock)
exc=fut.exception()
self.assertIs(exc, err)
self.assertEqual(1000, self.file.tell())


classUnixReadPipeTransportTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()
self.protocol=test_utils.make_test_protocol(asyncio.Protocol)
self.pipe=mock.Mock(spec_set=io.RawIOBase)
self.pipe.fileno.return_value=5

blocking_patcher=mock.patch('os.set_blocking')
blocking_patcher.start()
self.addCleanup(blocking_patcher.stop)

fstat_patcher=mock.patch('os.fstat')
m_fstat=fstat_patcher.start()
st=mock.Mock()
st.st_mode=stat.S_IFIFO
m_fstat.return_value=st
self.addCleanup(fstat_patcher.stop)

defread_pipe_transport(self, waiter=None):
transport=unix_events._UnixReadPipeTransport(self.loop, self.pipe,
self.protocol,
waiter=waiter)
self.addCleanup(close_pipe_transport, transport)
returntransport

deftest_ctor(self):
waiter=self.loop.create_future()
tr=self.read_pipe_transport(waiter=waiter)
self.loop.run_until_complete(waiter)

self.protocol.connection_made.assert_called_with(tr)
self.loop.assert_reader(5, tr._read_ready)
self.assertIsNone(waiter.result())

@mock.patch('os.read')
deftest__read_ready(self, m_read):
tr=self.read_pipe_transport()
m_read.return_value=b'data'
tr._read_ready()

m_read.assert_called_with(5, tr.max_size)
self.protocol.data_received.assert_called_with(b'data')

@mock.patch('os.read')
deftest__read_ready_eof(self, m_read):
tr=self.read_pipe_transport()
m_read.return_value=b''
tr._read_ready()

m_read.assert_called_with(5, tr.max_size)
self.assertFalse(self.loop.readers)
test_utils.run_briefly(self.loop)
self.protocol.eof_received.assert_called_with()
self.protocol.connection_lost.assert_called_with(None)

@mock.patch('os.read')
deftest__read_ready_blocked(self, m_read):
tr=self.read_pipe_transport()
m_read.side_effect=BlockingIOError
tr._read_ready()

m_read.assert_called_with(5, tr.max_size)
test_utils.run_briefly(self.loop)
self.assertFalse(self.protocol.data_received.called)

@mock.patch('asyncio.log.logger.error')
@mock.patch('os.read')
deftest__read_ready_error(self, m_read, m_logexc):
tr=self.read_pipe_transport()
err=OSError()
m_read.side_effect=err
tr._close=mock.Mock()
tr._read_ready()

m_read.assert_called_with(5, tr.max_size)
tr._close.assert_called_with(err)
m_logexc.assert_called_with(
test_utils.MockPattern(
'Fatal read error on pipe transport'
'\nprotocol:.*\ntransport:.*'),
exc_info=(OSError, MOCK_ANY, MOCK_ANY))

@mock.patch('os.read')
deftest_pause_reading(self, m_read):
tr=self.read_pipe_transport()
m=mock.Mock()
self.loop.add_reader(5, m)
tr.pause_reading()
self.assertFalse(self.loop.readers)

@mock.patch('os.read')
deftest_resume_reading(self, m_read):
tr=self.read_pipe_transport()
tr.pause_reading()
tr.resume_reading()
self.loop.assert_reader(5, tr._read_ready)

@mock.patch('os.read')
deftest_close(self, m_read):
tr=self.read_pipe_transport()
tr._close=mock.Mock()
tr.close()
tr._close.assert_called_with(None)

@mock.patch('os.read')
deftest_close_already_closing(self, m_read):
tr=self.read_pipe_transport()
tr._closing=True
tr._close=mock.Mock()
tr.close()
self.assertFalse(tr._close.called)

@mock.patch('os.read')
deftest__close(self, m_read):
tr=self.read_pipe_transport()
err=object()
tr._close(err)
self.assertTrue(tr.is_closing())
self.assertFalse(self.loop.readers)
test_utils.run_briefly(self.loop)
self.protocol.connection_lost.assert_called_with(err)

deftest__call_connection_lost(self):
tr=self.read_pipe_transport()
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)

err=None
tr._call_connection_lost(err)
self.protocol.connection_lost.assert_called_with(err)
self.pipe.close.assert_called_with()

self.assertIsNone(tr._protocol)
self.assertIsNone(tr._loop)

deftest__call_connection_lost_with_err(self):
tr=self.read_pipe_transport()
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)

err=OSError()
tr._call_connection_lost(err)
self.protocol.connection_lost.assert_called_with(err)
self.pipe.close.assert_called_with()

self.assertIsNone(tr._protocol)
self.assertIsNone(tr._loop)

deftest_pause_reading_on_closed_pipe(self):
tr=self.read_pipe_transport()
tr.close()
test_utils.run_briefly(self.loop)
self.assertIsNone(tr._loop)
tr.pause_reading()

deftest_pause_reading_on_paused_pipe(self):
tr=self.read_pipe_transport()
tr.pause_reading()
# the second call should do nothing
tr.pause_reading()

deftest_resume_reading_on_closed_pipe(self):
tr=self.read_pipe_transport()
tr.close()
test_utils.run_briefly(self.loop)
self.assertIsNone(tr._loop)
tr.resume_reading()

deftest_resume_reading_on_paused_pipe(self):
tr=self.read_pipe_transport()
# the pipe is not paused
# resuming should do nothing
tr.resume_reading()


classUnixWritePipeTransportTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()
self.protocol=test_utils.make_test_protocol(asyncio.BaseProtocol)
self.pipe=mock.Mock(spec_set=io.RawIOBase)
self.pipe.fileno.return_value=5

blocking_patcher=mock.patch('os.set_blocking')
blocking_patcher.start()
self.addCleanup(blocking_patcher.stop)

fstat_patcher=mock.patch('os.fstat')
m_fstat=fstat_patcher.start()
st=mock.Mock()
st.st_mode=stat.S_IFSOCK
m_fstat.return_value=st
self.addCleanup(fstat_patcher.stop)

defwrite_pipe_transport(self, waiter=None):
transport=unix_events._UnixWritePipeTransport(self.loop, self.pipe,
self.protocol,
waiter=waiter)
self.addCleanup(close_pipe_transport, transport)
returntransport

deftest_ctor(self):
waiter=self.loop.create_future()
tr=self.write_pipe_transport(waiter=waiter)
self.loop.run_until_complete(waiter)

self.protocol.connection_made.assert_called_with(tr)
self.loop.assert_reader(5, tr._read_ready)
self.assertEqual(None, waiter.result())

deftest_can_write_eof(self):
tr=self.write_pipe_transport()
self.assertTrue(tr.can_write_eof())

@mock.patch('os.write')
deftest_write(self, m_write):
tr=self.write_pipe_transport()
m_write.return_value=4
tr.write(b'data')
m_write.assert_called_with(5, b'data')
self.assertFalse(self.loop.writers)
self.assertEqual(bytearray(), tr._buffer)

@mock.patch('os.write')
deftest_write_no_data(self, m_write):
tr=self.write_pipe_transport()
tr.write(b'')
self.assertFalse(m_write.called)
self.assertFalse(self.loop.writers)
self.assertEqual(bytearray(b''), tr._buffer)

@mock.patch('os.write')
deftest_write_partial(self, m_write):
tr=self.write_pipe_transport()
m_write.return_value=2
tr.write(b'data')
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual(bytearray(b'ta'), tr._buffer)

@mock.patch('os.write')
deftest_write_buffer(self, m_write):
tr=self.write_pipe_transport()
self.loop.add_writer(5, tr._write_ready)
tr._buffer=bytearray(b'previous')
tr.write(b'data')
self.assertFalse(m_write.called)
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual(bytearray(b'previousdata'), tr._buffer)

@mock.patch('os.write')
deftest_write_again(self, m_write):
tr=self.write_pipe_transport()
m_write.side_effect=BlockingIOError()
tr.write(b'data')
m_write.assert_called_with(5, bytearray(b'data'))
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual(bytearray(b'data'), tr._buffer)

@mock.patch('asyncio.unix_events.logger')
@mock.patch('os.write')
deftest_write_err(self, m_write, m_log):
tr=self.write_pipe_transport()
err=OSError()
m_write.side_effect=err
tr._fatal_error=mock.Mock()
tr.write(b'data')
m_write.assert_called_with(5, b'data')
self.assertFalse(self.loop.writers)
self.assertEqual(bytearray(), tr._buffer)
tr._fatal_error.assert_called_with(
err,
'Fatal write error on pipe transport')
self.assertEqual(1, tr._conn_lost)

tr.write(b'data')
self.assertEqual(2, tr._conn_lost)
tr.write(b'data')
tr.write(b'data')
tr.write(b'data')
tr.write(b'data')
# This is a bit overspecified. :-(
m_log.warning.assert_called_with(
'pipe closed by peer or os.write(pipe, data) raised exception.')
tr.close()

@mock.patch('os.write')
deftest_write_close(self, m_write):
tr=self.write_pipe_transport()
tr._read_ready() # pipe was closed by peer

tr.write(b'data')
self.assertEqual(tr._conn_lost, 1)
tr.write(b'data')
self.assertEqual(tr._conn_lost, 2)

deftest__read_ready(self):
tr=self.write_pipe_transport()
tr._read_ready()
self.assertFalse(self.loop.readers)
self.assertFalse(self.loop.writers)
self.assertTrue(tr.is_closing())
test_utils.run_briefly(self.loop)
self.protocol.connection_lost.assert_called_with(None)

@mock.patch('os.write')
deftest__write_ready(self, m_write):
tr=self.write_pipe_transport()
self.loop.add_writer(5, tr._write_ready)
tr._buffer=bytearray(b'data')
m_write.return_value=4
tr._write_ready()
self.assertFalse(self.loop.writers)
self.assertEqual(bytearray(), tr._buffer)

@mock.patch('os.write')
deftest__write_ready_partial(self, m_write):
tr=self.write_pipe_transport()
self.loop.add_writer(5, tr._write_ready)
tr._buffer=bytearray(b'data')
m_write.return_value=3
tr._write_ready()
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual(bytearray(b'a'), tr._buffer)

@mock.patch('os.write')
deftest__write_ready_again(self, m_write):
tr=self.write_pipe_transport()
self.loop.add_writer(5, tr._write_ready)
tr._buffer=bytearray(b'data')
m_write.side_effect=BlockingIOError()
tr._write_ready()
m_write.assert_called_with(5, bytearray(b'data'))
self.loop.assert_writer(5, tr._write_ready)
self.assertEqual(bytearray(b'data'), tr._buffer)

@mock.patch('os.write')