test_selector_events.py

"""Tests for selector_events.py"""

importcollections
importselectors
importsocket
importsys
importunittest
fromasyncioimportselector_events
fromunittestimportmock

try:
importssl
exceptImportError:
ssl=None

importasyncio
fromasyncio.selector_eventsimport (BaseSelectorEventLoop,
_SelectorDatagramTransport,
_SelectorSocketTransport,
_SelectorTransport)
fromtest.test_asyncioimportutilsastest_utils

MOCK_ANY=mock.ANY


deftearDownModule():
asyncio.set_event_loop_policy(None)


classTestBaseSelectorEventLoop(BaseSelectorEventLoop):

def_make_self_pipe(self):
self._ssock=mock.Mock()
self._csock=mock.Mock()
self._internal_fds+=1

def_close_self_pipe(self):
pass


deflist_to_buffer(l=()):
buffer=collections.deque()
buffer.extend((memoryview(i) foriinl))
returnbuffer



defclose_transport(transport):
# Don't call transport.close() because the event loop and the selector
# are mocked
iftransport._sockisNone:
return
transport._sock.close()
transport._sock=None


classBaseSelectorEventLoopTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.selector=mock.Mock()
self.selector.select.return_value= []
self.loop=TestBaseSelectorEventLoop(self.selector)
self.set_event_loop(self.loop)

deftest_make_socket_transport(self):
m=mock.Mock()
self.loop.add_reader=mock.Mock()
self.loop._ensure_fd_no_transport=mock.Mock()
transport=self.loop._make_socket_transport(m, asyncio.Protocol())
self.assertIsInstance(transport, _SelectorSocketTransport)
self.assertEqual(self.loop._ensure_fd_no_transport.call_count, 1)

# Calling repr() must not fail when the event loop is closed
self.loop.close()
repr(transport)

close_transport(transport)

@mock.patch('asyncio.selector_events.ssl', None)
@mock.patch('asyncio.sslproto.ssl', None)
deftest_make_ssl_transport_without_ssl_error(self):
m=mock.Mock()
self.loop.add_reader=mock.Mock()
self.loop.add_writer=mock.Mock()
self.loop.remove_reader=mock.Mock()
self.loop.remove_writer=mock.Mock()
self.loop._ensure_fd_no_transport=mock.Mock()
withself.assertRaises(RuntimeError):
self.loop._make_ssl_transport(m, m, m, m)
self.assertEqual(self.loop._ensure_fd_no_transport.call_count, 1)

deftest_close(self):
classEventLoop(BaseSelectorEventLoop):
def_make_self_pipe(self):
self._ssock=mock.Mock()
self._csock=mock.Mock()
self._internal_fds+=1

self.loop=EventLoop(self.selector)
self.set_event_loop(self.loop)

ssock=self.loop._ssock
ssock.fileno.return_value=7
csock=self.loop._csock
csock.fileno.return_value=1
remove_reader=self.loop._remove_reader=mock.Mock()

self.loop._selector.close()
self.loop._selector=selector=mock.Mock()
self.assertFalse(self.loop.is_closed())

self.loop.close()
self.assertTrue(self.loop.is_closed())
self.assertIsNone(self.loop._selector)
self.assertIsNone(self.loop._csock)
self.assertIsNone(self.loop._ssock)
selector.close.assert_called_with()
ssock.close.assert_called_with()
csock.close.assert_called_with()
remove_reader.assert_called_with(7)

# it should be possible to call close() more than once
self.loop.close()
self.loop.close()

# operation blocked when the loop is closed
f=self.loop.create_future()
self.assertRaises(RuntimeError, self.loop.run_forever)
self.assertRaises(RuntimeError, self.loop.run_until_complete, f)
fd=0
defcallback():
pass
self.assertRaises(RuntimeError, self.loop.add_reader, fd, callback)
self.assertRaises(RuntimeError, self.loop.add_writer, fd, callback)

deftest_close_no_selector(self):
self.loop.remove_reader=mock.Mock()
self.loop._selector.close()
self.loop._selector=None
self.loop.close()
self.assertIsNone(self.loop._selector)

deftest_read_from_self_tryagain(self):
self.loop._ssock.recv.side_effect=BlockingIOError
self.assertIsNone(self.loop._read_from_self())

deftest_read_from_self_exception(self):
self.loop._ssock.recv.side_effect=OSError
self.assertRaises(OSError, self.loop._read_from_self)

deftest_write_to_self_tryagain(self):
self.loop._csock.send.side_effect=BlockingIOError
withtest_utils.disable_logger():
self.assertIsNone(self.loop._write_to_self())

deftest_write_to_self_exception(self):
# _write_to_self() swallows OSError
self.loop._csock.send.side_effect=RuntimeError()
self.assertRaises(RuntimeError, self.loop._write_to_self)

@mock.patch('socket.getaddrinfo')
deftest_sock_connect_resolve_using_socket_params(self, m_gai):
addr= ('need-resolution.com', 8080)
forsock_typein [socket.SOCK_STREAM, socket.SOCK_DGRAM]:
withself.subTest(sock_type):
sock=test_utils.mock_nonblocking_socket(type=sock_type)

m_gai.side_effect= \
lambda*args: [(None, None, None, None, ('127.0.0.1', 0))]

con=self.loop.create_task(self.loop.sock_connect(sock, addr))
self.loop.run_until_complete(con)
m_gai.assert_called_with(
addr[0], addr[1], sock.family, sock.type, sock.proto, 0)

self.loop.run_until_complete(con)
sock.connect.assert_called_with(('127.0.0.1', 0))

deftest_add_reader(self):
self.loop._selector.get_map.return_value= {}
cb=lambda: True
self.loop.add_reader(1, cb)

self.assertTrue(self.loop._selector.register.called)
fd, mask, (r, w) =self.loop._selector.register.call_args[0]
self.assertEqual(1, fd)
self.assertEqual(selectors.EVENT_READ, mask)
self.assertEqual(cb, r._callback)
self.assertIsNone(w)

deftest_add_reader_existing(self):
reader=mock.Mock()
writer=mock.Mock()
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_WRITE, (reader, writer))}
cb=lambda: True
self.loop.add_reader(1, cb)

self.assertTrue(reader.cancel.called)
self.assertFalse(self.loop._selector.register.called)
self.assertTrue(self.loop._selector.modify.called)
fd, mask, (r, w) =self.loop._selector.modify.call_args[0]
self.assertEqual(1, fd)
self.assertEqual(selectors.EVENT_WRITE|selectors.EVENT_READ, mask)
self.assertEqual(cb, r._callback)
self.assertEqual(writer, w)

deftest_add_reader_existing_writer(self):
writer=mock.Mock()
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_WRITE, (None, writer))}
cb=lambda: True
self.loop.add_reader(1, cb)

self.assertFalse(self.loop._selector.register.called)
self.assertTrue(self.loop._selector.modify.called)
fd, mask, (r, w) =self.loop._selector.modify.call_args[0]
self.assertEqual(1, fd)
self.assertEqual(selectors.EVENT_WRITE|selectors.EVENT_READ, mask)
self.assertEqual(cb, r._callback)
self.assertEqual(writer, w)

deftest_remove_reader(self):
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (None, None))}
self.assertFalse(self.loop.remove_reader(1))

self.assertTrue(self.loop._selector.unregister.called)

deftest_remove_reader_read_write(self):
reader=mock.Mock()
writer=mock.Mock()
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_READ|selectors.EVENT_WRITE,
 (reader, writer))}
self.assertTrue(
self.loop.remove_reader(1))

self.assertFalse(self.loop._selector.unregister.called)
self.assertEqual(
 (1, selectors.EVENT_WRITE, (None, writer)),
self.loop._selector.modify.call_args[0])

deftest_remove_reader_unknown(self):
self.loop._selector.get_map.return_value= {}
self.assertFalse(
self.loop.remove_reader(1))

deftest_add_writer(self):
self.loop._selector.get_map.return_value= {}
cb=lambda: True
self.loop.add_writer(1, cb)

self.assertTrue(self.loop._selector.register.called)
fd, mask, (r, w) =self.loop._selector.register.call_args[0]
self.assertEqual(1, fd)
self.assertEqual(selectors.EVENT_WRITE, mask)
self.assertIsNone(r)
self.assertEqual(cb, w._callback)

deftest_add_writer_existing(self):
reader=mock.Mock()
writer=mock.Mock()
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (reader, writer))}
cb=lambda: True
self.loop.add_writer(1, cb)

self.assertTrue(writer.cancel.called)
self.assertFalse(self.loop._selector.register.called)
self.assertTrue(self.loop._selector.modify.called)
fd, mask, (r, w) =self.loop._selector.modify.call_args[0]
self.assertEqual(1, fd)
self.assertEqual(selectors.EVENT_WRITE|selectors.EVENT_READ, mask)
self.assertEqual(reader, r)
self.assertEqual(cb, w._callback)

deftest_remove_writer(self):
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_WRITE, (None, None))}
self.assertFalse(self.loop.remove_writer(1))

self.assertTrue(self.loop._selector.unregister.called)

deftest_remove_writer_read_write(self):
reader=mock.Mock()
writer=mock.Mock()
self.loop._selector.get_map.return_value= {1: selectors.SelectorKey(
1, 1, selectors.EVENT_READ|selectors.EVENT_WRITE,
 (reader, writer))}
self.assertTrue(
self.loop.remove_writer(1))

self.assertFalse(self.loop._selector.unregister.called)
self.assertEqual(
 (1, selectors.EVENT_READ, (reader, None)),
self.loop._selector.modify.call_args[0])

deftest_remove_writer_unknown(self):
self.loop._selector.get_map.return_value= {}
self.assertFalse(
self.loop.remove_writer(1))

deftest_process_events_read(self):
reader=mock.Mock()
reader._cancelled=False

self.loop._add_callback=mock.Mock()
self.loop._process_events(
 [(selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (reader, None)),
selectors.EVENT_READ)])
self.assertTrue(self.loop._add_callback.called)
self.loop._add_callback.assert_called_with(reader)

deftest_process_events_read_cancelled(self):
reader=mock.Mock()
reader.cancelled=True

self.loop._remove_reader=mock.Mock()
self.loop._process_events(
 [(selectors.SelectorKey(
1, 1, selectors.EVENT_READ, (reader, None)),
selectors.EVENT_READ)])
self.loop._remove_reader.assert_called_with(1)

deftest_process_events_write(self):
writer=mock.Mock()
writer._cancelled=False

self.loop._add_callback=mock.Mock()
self.loop._process_events(
 [(selectors.SelectorKey(1, 1, selectors.EVENT_WRITE,
 (None, writer)),
selectors.EVENT_WRITE)])
self.loop._add_callback.assert_called_with(writer)

deftest_process_events_write_cancelled(self):
writer=mock.Mock()
writer.cancelled=True
self.loop._remove_writer=mock.Mock()

self.loop._process_events(
 [(selectors.SelectorKey(1, 1, selectors.EVENT_WRITE,
 (None, writer)),
selectors.EVENT_WRITE)])
self.loop._remove_writer.assert_called_with(1)

deftest_accept_connection_multiple(self):
sock=mock.Mock()
sock.accept.return_value= (mock.Mock(), mock.Mock())
backlog=100
# Mock the coroutine generation for a connection to prevent
# warnings related to un-awaited coroutines. _accept_connection2
# is an async function that is patched with AsyncMock. create_task
# creates a task out of coroutine returned by AsyncMock, so use
# asyncio.sleep(0) to ensure created tasks are complete to avoid
# task pending warnings.
mock_obj=mock.patch.object
withmock_obj(self.loop, '_accept_connection2') asaccept2_mock:
self.loop._accept_connection(
mock.Mock(), sock, backlog=backlog)
self.loop.run_until_complete(asyncio.sleep(0))
self.assertEqual(sock.accept.call_count, backlog)


classSelectorTransportTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()
self.protocol=test_utils.make_test_protocol(asyncio.Protocol)
self.sock=mock.Mock(socket.socket)
self.sock.fileno.return_value=7

defcreate_transport(self):
transport=_SelectorTransport(self.loop, self.sock, self.protocol,
None)
self.addCleanup(close_transport, transport)
returntransport

deftest_ctor(self):
tr=self.create_transport()
self.assertIs(tr._loop, self.loop)
self.assertIs(tr._sock, self.sock)
self.assertIs(tr._sock_fd, 7)

deftest_abort(self):
tr=self.create_transport()
tr._force_close=mock.Mock()

tr.abort()
tr._force_close.assert_called_with(None)

deftest_close(self):
tr=self.create_transport()
tr.close()

self.assertTrue(tr.is_closing())
self.assertEqual(1, self.loop.remove_reader_count[7])
self.protocol.connection_lost(None)
self.assertEqual(tr._conn_lost, 1)

tr.close()
self.assertEqual(tr._conn_lost, 1)
self.assertEqual(1, self.loop.remove_reader_count[7])

deftest_close_write_buffer(self):
tr=self.create_transport()
tr._buffer.extend(b'data')
tr.close()

self.assertFalse(self.loop.readers)
test_utils.run_briefly(self.loop)
self.assertFalse(self.protocol.connection_lost.called)

deftest_force_close(self):
tr=self.create_transport()
tr._buffer.extend(b'1')
self.loop._add_reader(7, mock.sentinel)
self.loop._add_writer(7, mock.sentinel)
tr._force_close(None)

self.assertTrue(tr.is_closing())
self.assertEqual(tr._buffer, list_to_buffer())
self.assertFalse(self.loop.readers)
self.assertFalse(self.loop.writers)

# second close should not remove reader
tr._force_close(None)
self.assertFalse(self.loop.readers)
self.assertEqual(1, self.loop.remove_reader_count[7])

@mock.patch('asyncio.log.logger.error')
deftest_fatal_error(self, m_exc):
exc=OSError()
tr=self.create_transport()
tr._force_close=mock.Mock()
tr._fatal_error(exc)

m_exc.assert_not_called()

tr._force_close.assert_called_with(exc)

@mock.patch('asyncio.log.logger.error')
deftest_fatal_error_custom_exception(self, m_exc):
classMyError(Exception):
pass
exc=MyError()
tr=self.create_transport()
tr._force_close=mock.Mock()
tr._fatal_error(exc)

m_exc.assert_called_with(
test_utils.MockPattern(
'Fatal error on transport\nprotocol:.*\ntransport:.*'),
exc_info=(MyError, MOCK_ANY, MOCK_ANY))

tr._force_close.assert_called_with(exc)

deftest_connection_lost(self):
exc=OSError()
tr=self.create_transport()
self.assertIsNotNone(tr._protocol)
self.assertIsNotNone(tr._loop)
tr._call_connection_lost(exc)

self.protocol.connection_lost.assert_called_with(exc)
self.sock.close.assert_called_with()
self.assertIsNone(tr._sock)

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

deftest__add_reader(self):
tr=self.create_transport()
tr._buffer.extend(b'1')
tr._add_reader(7, mock.sentinel)
self.assertTrue(self.loop.readers)

tr._force_close(None)

self.assertTrue(tr.is_closing())
self.assertFalse(self.loop.readers)

# can not add readers after closing
tr._add_reader(7, mock.sentinel)
self.assertFalse(self.loop.readers)


classSelectorSocketTransportTests(test_utils.TestCase):

defsetUp(self):
super().setUp()
self.loop=self.new_test_loop()
self.protocol=test_utils.make_test_protocol(asyncio.Protocol)
self.sock=mock.Mock(socket.socket)
self.sock_fd=self.sock.fileno.return_value=7

defsocket_transport(self, waiter=None, sendmsg=False):
transport=_SelectorSocketTransport(self.loop, self.sock,
self.protocol, waiter=waiter)
ifsendmsg:
transport._write_ready=transport._write_sendmsg
else:
transport._write_ready=transport._write_send
self.addCleanup(close_transport, transport)
returntransport

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

self.loop.assert_reader(7, tr._read_ready)
test_utils.run_briefly(self.loop)
self.protocol.connection_made.assert_called_with(tr)

deftest_ctor_with_waiter(self):
waiter=self.loop.create_future()
self.socket_transport(waiter=waiter)
self.loop.run_until_complete(waiter)

self.assertIsNone(waiter.result())

deftest_pause_resume_reading(self):
tr=self.socket_transport()
test_utils.run_briefly(self.loop)
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())
self.loop.assert_reader(7, tr._read_ready)

tr.pause_reading()
tr.pause_reading()
self.assertTrue(tr._paused)
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)

tr.resume_reading()
tr.resume_reading()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())
self.loop.assert_reader(7, tr._read_ready)

tr.close()
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)

deftest_pause_reading_connection_made(self):
tr=self.socket_transport()
self.protocol.connection_made.side_effect=lambda_: tr.pause_reading()
test_utils.run_briefly(self.loop)
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)

tr.resume_reading()
self.assertTrue(tr.is_reading())
self.loop.assert_reader(7, tr._read_ready)

tr.close()
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)


deftest_read_eof_received_error(self):
transport=self.socket_transport()
transport.close=mock.Mock()
transport._fatal_error=mock.Mock()

self.loop.call_exception_handler=mock.Mock()

self.protocol.eof_received.side_effect=LookupError()

self.sock.recv.return_value=b''
transport._read_ready()

self.protocol.eof_received.assert_called_with()
self.assertTrue(transport._fatal_error.called)

deftest_data_received_error(self):
transport=self.socket_transport()
transport._fatal_error=mock.Mock()

self.loop.call_exception_handler=mock.Mock()
self.protocol.data_received.side_effect=LookupError()

self.sock.recv.return_value=b'data'
transport._read_ready()

self.assertTrue(transport._fatal_error.called)
self.assertTrue(self.protocol.data_received.called)

deftest_read_ready(self):
transport=self.socket_transport()

self.sock.recv.return_value=b'data'
transport._read_ready()

self.protocol.data_received.assert_called_with(b'data')

deftest_read_ready_eof(self):
transport=self.socket_transport()
transport.close=mock.Mock()

self.sock.recv.return_value=b''
transport._read_ready()

self.protocol.eof_received.assert_called_with()
transport.close.assert_called_with()

deftest_read_ready_eof_keep_open(self):
transport=self.socket_transport()
transport.close=mock.Mock()

self.sock.recv.return_value=b''
self.protocol.eof_received.return_value=True
transport._read_ready()

self.protocol.eof_received.assert_called_with()
self.assertFalse(transport.close.called)

@mock.patch('logging.exception')
deftest_read_ready_tryagain(self, m_exc):
self.sock.recv.side_effect=BlockingIOError

transport=self.socket_transport()
transport._fatal_error=mock.Mock()
transport._read_ready()

self.assertFalse(transport._fatal_error.called)

@mock.patch('logging.exception')
deftest_read_ready_tryagain_interrupted(self, m_exc):
self.sock.recv.side_effect=InterruptedError

transport=self.socket_transport()
transport._fatal_error=mock.Mock()
transport._read_ready()

self.assertFalse(transport._fatal_error.called)

@mock.patch('logging.exception')
deftest_read_ready_conn_reset(self, m_exc):
err=self.sock.recv.side_effect=ConnectionResetError()

transport=self.socket_transport()
transport._force_close=mock.Mock()
withtest_utils.disable_logger():
transport._read_ready()
transport._force_close.assert_called_with(err)

@mock.patch('logging.exception')
deftest_read_ready_err(self, m_exc):
err=self.sock.recv.side_effect=OSError()

transport=self.socket_transport()
transport._fatal_error=mock.Mock()
transport._read_ready()

transport._fatal_error.assert_called_with(
err,
'Fatal read error on socket transport')

deftest_write(self):
data=b'data'
self.sock.send.return_value=len(data)

transport=self.socket_transport()
transport.write(data)
self.sock.send.assert_called_with(data)

deftest_write_bytearray(self):
data=bytearray(b'data')
self.sock.send.return_value=len(data)

transport=self.socket_transport()
transport.write(data)
self.sock.send.assert_called_with(data)
self.assertEqual(data, bytearray(b'data')) # Hasn't been mutated.

deftest_write_memoryview(self):
data=memoryview(b'data')
self.sock.send.return_value=len(data)

transport=self.socket_transport()
transport.write(data)
self.sock.send.assert_called_with(data)

deftest_write_no_data(self):
transport=self.socket_transport()
transport._buffer.append(memoryview(b'data'))
transport.write(b'')
self.assertFalse(self.sock.send.called)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)

deftest_write_buffer(self):
transport=self.socket_transport()
transport._buffer.append(b'data1')
transport.write(b'data2')
self.assertFalse(self.sock.send.called)
self.assertEqual(list_to_buffer([b'data1', b'data2']),
transport._buffer)

deftest_write_partial(self):
data=b'data'
self.sock.send.return_value=2

transport=self.socket_transport()
transport.write(data)

self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

deftest_write_partial_bytearray(self):
data=bytearray(b'data')
self.sock.send.return_value=2

transport=self.socket_transport()
transport.write(data)

self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'ta']), transport._buffer)
self.assertEqual(data, bytearray(b'data')) # Hasn't been mutated.

deftest_write_partial_memoryview(self):
data=memoryview(b'data')
self.sock.send.return_value=2

transport=self.socket_transport()
transport.write(data)

self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

deftest_write_partial_none(self):
data=b'data'
self.sock.send.return_value=0
self.sock.fileno.return_value=7

transport=self.socket_transport()
transport.write(data)

self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)

deftest_write_tryagain(self):
self.sock.send.side_effect=BlockingIOError

data=b'data'
transport=self.socket_transport()
transport.write(data)

self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)

deftest_write_sendmsg_no_data(self):
self.sock.sendmsg=mock.Mock()
self.sock.sendmsg.return_value=0
transport=self.socket_transport(sendmsg=True)
transport._buffer.append(memoryview(b'data'))
transport.write(b'')
self.assertFalse(self.sock.sendmsg.called)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)

@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
deftest_writelines_sendmsg_full(self):
data=memoryview(b'data')
self.sock.sendmsg=mock.Mock()
self.sock.sendmsg.return_value=len(data)

transport=self.socket_transport(sendmsg=True)
transport.writelines([data])
self.assertTrue(self.sock.sendmsg.called)
self.assertFalse(self.loop.writers)

@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
deftest_writelines_sendmsg_partial(self):
data=memoryview(b'data')
self.sock.sendmsg=mock.Mock()
self.sock.sendmsg.return_value=2

transport=self.socket_transport(sendmsg=True)
transport.writelines([data])
self.assertTrue(self.sock.sendmsg.called)
self.assertTrue(self.loop.writers)

deftest_writelines_send_full(self):
data=memoryview(b'data')
self.sock.send.return_value=len(data)
self.sock.send.fileno.return_value=7

transport=self.socket_transport()
transport.writelines([data])
self.assertTrue(self.sock.send.called)
self.assertFalse(self.loop.writers)

deftest_writelines_send_partial(self):
data=memoryview(b'data')
self.sock.send.return_value=2
self.sock.send.fileno.return_value=7

transport=self.socket_transport()
transport.writelines([data])
self.assertTrue(self.sock.send.called)
self.assertTrue(self.loop.writers)

deftest_writelines_pauses_protocol(self):
data=memoryview(b'data')
self.sock.send.return_value=2
self.sock.send.fileno.return_value=7

transport=self.socket_transport()
transport._high_water=1
transport.writelines([data])
self.assertTrue(self.protocol.pause_writing.called)
self.assertTrue(self.sock.send.called)
self.assertTrue(self.loop.writers)

@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
deftest_write_sendmsg_full(self):
data=memoryview(b'data')
self.sock.sendmsg=mock.Mock()
self.sock.sendmsg.return_value=len(data)

transport=self.socket_transport(sendmsg=True)
transport._buffer.append(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.assertTrue(self.sock.sendmsg.called)
self.assertFalse(self.loop.writers)

@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
deftest_write_sendmsg_partial(self):

data=memoryview(b'data')
self.sock.sendmsg=mock.Mock()
# Sent partial data
self.sock.sendmsg.return_value=2

transport=self.socket_transport(sendmsg=True)
transport._buffer.append(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.assertTrue(self.sock.sendmsg.called)
self.assertTrue(self.loop.writers)
self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
deftest_write_sendmsg_half_buffer(self):
data= [memoryview(b'data1'), memoryview(b'data2')]
self.sock.sendmsg=mock.Mock()
# Sent partial data
self.sock.sendmsg.return_value=2

transport=self.socket_transport(sendmsg=True)
transport._buffer.extend(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.assertTrue(self.sock.sendmsg.called)
self.assertTrue(self.loop.writers)
self.assertEqual(list_to_buffer([b'ta1', b'data2']), transport._buffer)

@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
deftest_write_sendmsg_OSError(self):
data=memoryview(b'data')
self.sock.sendmsg=mock.Mock()
err=self.sock.sendmsg.side_effect=OSError()

transport=self.socket_transport(sendmsg=True)
transport._fatal_error=mock.Mock()
transport._buffer.extend(data)
# Calls _fatal_error and clears the buffer
transport._write_ready()
self.assertTrue(self.sock.sendmsg.called)
self.assertFalse(self.loop.writers)
self.assertEqual(list_to_buffer([]), transport._buffer)
transport._fatal_error.assert_called_with(
err,
'Fatal write error on socket transport')

@mock.patch('asyncio.selector_events.logger')
deftest_write_exception(self, m_log):
err=self.sock.send.side_effect=OSError()

data=b'data'
transport=self.socket_transport()
transport._fatal_error=mock.Mock()
transport.write(data)
transport._fatal_error.assert_called_with(
err,
'Fatal write error on socket transport')
transport._conn_lost=1

self.sock.reset_mock()
transport.write(data)
self.assertFalse(self.sock.send.called)
self.assertEqual(transport._conn_lost, 2)
transport.write(data)
transport.write(data)
transport.write(data)
transport.write(data)
m_log.warning.assert_called_with('socket.send() raised exception.')

deftest_write_str(self):
transport=self.socket_transport()
self.assertRaises(TypeError, transport.write, 'str')

deftest_write_closing(self):
transport=self.socket_transport()
transport.close()
self.assertEqual(transport._conn_lost, 1)
transport.write(b'data')
self.assertEqual(transport._conn_lost, 2)

deftest_write_ready(self):
data=b'data'
self.sock.send.return_value=len(data)

transport=self.socket_transport()
transport._buffer.append(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.assertTrue(self.sock.send.called)
self.assertFalse(self.loop.writers)

deftest_write_ready_closing(self):
data=memoryview(b'data')
self.sock.send.return_value=len(data)

transport=self.socket_transport()
transport._closing=True
transport._buffer.append(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.assertTrue(self.sock.send.called)
self.assertFalse(self.loop.writers)
self.sock.close.assert_called_with()
self.protocol.connection_lost.assert_called_with(None)

@unittest.skipIf(sys.flags.optimize, "Assertions are disabled in optimized mode")
deftest_write_ready_no_data(self):
transport=self.socket_transport()
# This is an internal error.
self.assertRaises(AssertionError, transport._write_ready)

deftest_write_ready_partial(self):
data=memoryview(b'data')
self.sock.send.return_value=2

transport=self.socket_transport()
transport._buffer.append(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'ta']), transport._buffer)

deftest_write_ready_partial_none(self):
data=b'data'
self.sock.send.return_value=0

transport=self.socket_transport()
transport._buffer.append(data)
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()
self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)

deftest_write_ready_tryagain(self):
self.sock.send.side_effect=BlockingIOError

transport=self.socket_transport()
buffer=list_to_buffer([b'data1', b'data2'])
transport._buffer=buffer
self.loop._add_writer(7, transport._write_ready)
transport._write_ready()

self.loop.assert_writer(7, transport._write_ready)
self.assertEqual(buffer, transport._buffer)

deftest_write_ready_exception(self):
err=self.sock.send.side_effect=OSError()

transport=self.socket_transport()
transport._fatal_error=mock.Mock()
transport._buffer.extend(b'data')
transport._write_ready()
transport._fatal_error.assert_called_with(
err,
'Fatal write error on socket transport')

deftest_write_eof(self):
tr=self.socket_transport()
self.assertTrue(tr.can_write_eof())
tr.write_eof()
self.sock.shutdown.assert_called_with(socket.SHUT_WR)
tr.write_eof()
self.assertEqual(self.sock.shutdown.call_count, 1)
tr.close()