summaryrefslogtreecommitdiff
path: root/Lib/test
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test')
-rw-r--r--Lib/test/test_asyncio/test_buffered_proto.py85
-rw-r--r--Lib/test/test_asyncio/test_proactor_events.py197
-rw-r--r--Lib/test/test_asyncio/test_selector_events.py171
3 files changed, 450 insertions, 3 deletions
diff --git a/Lib/test/test_asyncio/test_buffered_proto.py b/Lib/test/test_asyncio/test_buffered_proto.py
new file mode 100644
index 0000000000..22f9269e81
--- /dev/null
+++ b/Lib/test/test_asyncio/test_buffered_proto.py
@@ -0,0 +1,85 @@
+import asyncio
+import unittest
+
+from test.test_asyncio import functional as func_tests
+
+
+class ReceiveStuffProto(asyncio.BufferedProtocol):
+ def __init__(self, cb, con_lost_fut):
+ self.cb = cb
+ self.con_lost_fut = con_lost_fut
+
+ def get_buffer(self):
+ self.buffer = bytearray(100)
+ return self.buffer
+
+ def buffer_updated(self, nbytes):
+ self.cb(self.buffer[:nbytes])
+
+ def connection_lost(self, exc):
+ if exc is None:
+ self.con_lost_fut.set_result(None)
+ else:
+ self.con_lost_fut.set_exception(exc)
+
+
+class BaseTestBufferedProtocol(func_tests.FunctionalTestCaseMixin):
+
+ def new_loop(self):
+ raise NotImplementedError
+
+ def test_buffered_proto_create_connection(self):
+
+ NOISE = b'12345678+' * 1024
+
+ async def client(addr):
+ data = b''
+
+ def on_buf(buf):
+ nonlocal data
+ data += buf
+ if data == NOISE:
+ tr.write(b'1')
+
+ conn_lost_fut = self.loop.create_future()
+
+ tr, pr = await self.loop.create_connection(
+ lambda: ReceiveStuffProto(on_buf, conn_lost_fut), *addr)
+
+ await conn_lost_fut
+
+ async def on_server_client(reader, writer):
+ writer.write(NOISE)
+ await reader.readexactly(1)
+ writer.close()
+ await writer.wait_closed()
+
+ srv = self.loop.run_until_complete(
+ asyncio.start_server(
+ on_server_client, '127.0.0.1', 0))
+
+ addr = srv.sockets[0].getsockname()
+ self.loop.run_until_complete(
+ asyncio.wait_for(client(addr), 5, loop=self.loop))
+
+ srv.close()
+ self.loop.run_until_complete(srv.wait_closed())
+
+
+class BufferedProtocolSelectorTests(BaseTestBufferedProtocol,
+ unittest.TestCase):
+
+ def new_loop(self):
+ return asyncio.SelectorEventLoop()
+
+
+@unittest.skipUnless(hasattr(asyncio, 'ProactorEventLoop'), 'Windows only')
+class BufferedProtocolProactorTests(BaseTestBufferedProtocol,
+ unittest.TestCase):
+
+ def new_loop(self):
+ return asyncio.ProactorEventLoop()
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py
index 82bff28d7d..f627dfce0e 100644
--- a/Lib/test/test_asyncio/test_proactor_events.py
+++ b/Lib/test/test_asyncio/test_proactor_events.py
@@ -44,12 +44,12 @@ class ProactorSocketTransportTests(test_utils.TestCase):
test_utils.run_briefly(self.loop)
self.assertIsNone(fut.result())
self.protocol.connection_made(tr)
- self.proactor.recv.assert_called_with(self.sock, 4096)
+ self.proactor.recv.assert_called_with(self.sock, 32768)
def test_loop_reading(self):
tr = self.socket_transport()
tr._loop_reading()
- self.loop._proactor.recv.assert_called_with(self.sock, 4096)
+ self.loop._proactor.recv.assert_called_with(self.sock, 32768)
self.assertFalse(self.protocol.data_received.called)
self.assertFalse(self.protocol.eof_received.called)
@@ -60,7 +60,7 @@ class ProactorSocketTransportTests(test_utils.TestCase):
tr = self.socket_transport()
tr._read_fut = res
tr._loop_reading(res)
- self.loop._proactor.recv.assert_called_with(self.sock, 4096)
+ self.loop._proactor.recv.assert_called_with(self.sock, 32768)
self.protocol.data_received.assert_called_with(b'data')
def test_loop_reading_no_data(self):
@@ -444,6 +444,197 @@ class ProactorSocketTransportTests(test_utils.TestCase):
self.assertFalse(self.protocol.pause_writing.called)
+class ProactorSocketTransportBufferedProtoTests(test_utils.TestCase):
+
+ def setUp(self):
+ super().setUp()
+ self.loop = self.new_test_loop()
+ self.addCleanup(self.loop.close)
+ self.proactor = mock.Mock()
+ self.loop._proactor = self.proactor
+
+ self.protocol = test_utils.make_test_protocol(asyncio.BufferedProtocol)
+ self.buf = mock.Mock()
+ self.protocol.get_buffer.side_effect = lambda: self.buf
+
+ self.sock = mock.Mock(socket.socket)
+
+ def socket_transport(self, waiter=None):
+ transport = _ProactorSocketTransport(self.loop, self.sock,
+ self.protocol, waiter=waiter)
+ self.addCleanup(close_transport, transport)
+ return transport
+
+ def test_ctor(self):
+ fut = asyncio.Future(loop=self.loop)
+ tr = self.socket_transport(waiter=fut)
+ test_utils.run_briefly(self.loop)
+ self.assertIsNone(fut.result())
+ self.protocol.connection_made(tr)
+ self.proactor.recv_into.assert_called_with(self.sock, self.buf)
+
+ def test_loop_reading(self):
+ tr = self.socket_transport()
+ tr._loop_reading()
+ self.loop._proactor.recv_into.assert_called_with(self.sock, self.buf)
+ self.assertTrue(self.protocol.get_buffer.called)
+ self.assertFalse(self.protocol.buffer_updated.called)
+ self.assertFalse(self.protocol.eof_received.called)
+
+ def test_get_buffer_error(self):
+ transport = self.socket_transport()
+ transport._fatal_error = mock.Mock()
+
+ self.loop.call_exception_handler = mock.Mock()
+ self.protocol.get_buffer.side_effect = LookupError()
+
+ transport._loop_reading()
+
+ self.assertTrue(transport._fatal_error.called)
+ self.assertTrue(self.protocol.get_buffer.called)
+ self.assertFalse(self.protocol.buffer_updated.called)
+
+ def test_buffer_updated_error(self):
+ transport = self.socket_transport()
+ transport._fatal_error = mock.Mock()
+
+ self.loop.call_exception_handler = mock.Mock()
+ self.protocol.buffer_updated.side_effect = LookupError()
+
+ res = asyncio.Future(loop=self.loop)
+ res.set_result(10)
+ transport._read_fut = res
+ transport._loop_reading(res)
+
+ self.assertTrue(transport._fatal_error.called)
+ self.assertFalse(self.protocol.get_buffer.called)
+ self.assertTrue(self.protocol.buffer_updated.called)
+
+ def test_loop_eof_received_error(self):
+ res = asyncio.Future(loop=self.loop)
+ res.set_result(0)
+
+ self.protocol.eof_received.side_effect = LookupError()
+
+ tr = self.socket_transport()
+ tr._fatal_error = mock.Mock()
+
+ tr.close = mock.Mock()
+ tr._read_fut = res
+ tr._loop_reading(res)
+ self.assertFalse(self.loop._proactor.recv_into.called)
+ self.assertTrue(self.protocol.eof_received.called)
+ self.assertTrue(tr._fatal_error.called)
+
+ def test_loop_reading_data(self):
+ res = asyncio.Future(loop=self.loop)
+ res.set_result(4)
+
+ tr = self.socket_transport()
+ tr._read_fut = res
+ tr._loop_reading(res)
+ self.loop._proactor.recv_into.assert_called_with(self.sock, self.buf)
+ self.protocol.buffer_updated.assert_called_with(4)
+
+ def test_loop_reading_no_data(self):
+ res = asyncio.Future(loop=self.loop)
+ res.set_result(0)
+
+ tr = self.socket_transport()
+ self.assertRaises(AssertionError, tr._loop_reading, res)
+
+ tr.close = mock.Mock()
+ tr._read_fut = res
+ tr._loop_reading(res)
+ self.assertFalse(self.loop._proactor.recv_into.called)
+ self.assertTrue(self.protocol.eof_received.called)
+ self.assertTrue(tr.close.called)
+
+ def test_loop_reading_aborted(self):
+ err = self.loop._proactor.recv_into.side_effect = \
+ ConnectionAbortedError()
+
+ tr = self.socket_transport()
+ tr._fatal_error = mock.Mock()
+ tr._loop_reading()
+ tr._fatal_error.assert_called_with(
+ err, 'Fatal read error on pipe transport')
+
+ def test_loop_reading_aborted_closing(self):
+ self.loop._proactor.recv.side_effect = ConnectionAbortedError()
+
+ tr = self.socket_transport()
+ tr._closing = True
+ tr._fatal_error = mock.Mock()
+ tr._loop_reading()
+ self.assertFalse(tr._fatal_error.called)
+
+ def test_loop_reading_aborted_is_fatal(self):
+ self.loop._proactor.recv_into.side_effect = ConnectionAbortedError()
+ tr = self.socket_transport()
+ tr._closing = False
+ tr._fatal_error = mock.Mock()
+ tr._loop_reading()
+ self.assertTrue(tr._fatal_error.called)
+
+ def test_loop_reading_conn_reset_lost(self):
+ err = self.loop._proactor.recv_into.side_effect = ConnectionResetError()
+
+ tr = self.socket_transport()
+ tr._closing = False
+ tr._fatal_error = mock.Mock()
+ tr._force_close = mock.Mock()
+ tr._loop_reading()
+ self.assertFalse(tr._fatal_error.called)
+ tr._force_close.assert_called_with(err)
+
+ def test_loop_reading_exception(self):
+ err = self.loop._proactor.recv_into.side_effect = OSError()
+
+ tr = self.socket_transport()
+ tr._fatal_error = mock.Mock()
+ tr._loop_reading()
+ tr._fatal_error.assert_called_with(
+ err, 'Fatal read error on pipe transport')
+
+ def test_pause_resume_reading(self):
+ tr = self.socket_transport()
+ futures = []
+ for msg in [10, 20, 30, 40, 0]:
+ f = asyncio.Future(loop=self.loop)
+ f.set_result(msg)
+ futures.append(f)
+
+ self.loop._proactor.recv_into.side_effect = futures
+ self.loop._run_once()
+ self.assertFalse(tr._paused)
+ self.assertTrue(tr.is_reading())
+ self.loop._run_once()
+ self.protocol.buffer_updated.assert_called_with(10)
+ self.loop._run_once()
+ self.protocol.buffer_updated.assert_called_with(20)
+
+ tr.pause_reading()
+ tr.pause_reading()
+ self.assertTrue(tr._paused)
+ self.assertFalse(tr.is_reading())
+ for i in range(10):
+ self.loop._run_once()
+ self.protocol.buffer_updated.assert_called_with(20)
+
+ tr.resume_reading()
+ tr.resume_reading()
+ self.assertFalse(tr._paused)
+ self.assertTrue(tr.is_reading())
+ self.loop._run_once()
+ self.protocol.buffer_updated.assert_called_with(30)
+ self.loop._run_once()
+ self.protocol.buffer_updated.assert_called_with(40)
+ tr.close()
+
+ self.assertFalse(tr.is_reading())
+
+
class BaseProactorEventLoopTests(test_utils.TestCase):
def setUp(self):
diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py
index 89c3d5e16c..219ab0eb5b 100644
--- a/Lib/test/test_asyncio/test_selector_events.py
+++ b/Lib/test/test_asyncio/test_selector_events.py
@@ -926,6 +926,34 @@ class SelectorSocketTransportTests(test_utils.TestCase):
self.assertFalse(tr.is_reading())
self.loop.assert_no_reader(7)
+ def test_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)
+
+ def test_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)
+
def test_read_ready(self):
transport = self.socket_transport()
@@ -1229,6 +1257,149 @@ class SelectorSocketTransportTests(test_utils.TestCase):
remove_writer.assert_called_with(self.sock_fd)
+class SelectorSocketTransportBufferedProtocolTests(test_utils.TestCase):
+
+ def setUp(self):
+ super().setUp()
+ self.loop = self.new_test_loop()
+
+ self.protocol = test_utils.make_test_protocol(asyncio.BufferedProtocol)
+ self.buf = mock.Mock()
+ self.protocol.get_buffer.side_effect = lambda: self.buf
+
+ self.sock = mock.Mock(socket.socket)
+ self.sock_fd = self.sock.fileno.return_value = 7
+
+ def socket_transport(self, waiter=None):
+ transport = _SelectorSocketTransport(self.loop, self.sock,
+ self.protocol, waiter=waiter)
+ self.addCleanup(close_transport, transport)
+ return transport
+
+ def test_ctor(self):
+ waiter = asyncio.Future(loop=self.loop)
+ 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)
+
+ def test_get_buffer_error(self):
+ transport = self.socket_transport()
+ transport._fatal_error = mock.Mock()
+
+ self.loop.call_exception_handler = mock.Mock()
+ self.protocol.get_buffer.side_effect = LookupError()
+
+ transport._read_ready()
+
+ self.assertTrue(transport._fatal_error.called)
+ self.assertTrue(self.protocol.get_buffer.called)
+ self.assertFalse(self.protocol.buffer_updated.called)
+
+ def test_buffer_updated_error(self):
+ transport = self.socket_transport()
+ transport._fatal_error = mock.Mock()
+
+ self.loop.call_exception_handler = mock.Mock()
+ self.protocol.buffer_updated.side_effect = LookupError()
+
+ self.sock.recv_into.return_value = 10
+ transport._read_ready()
+
+ self.assertTrue(transport._fatal_error.called)
+ self.assertTrue(self.protocol.get_buffer.called)
+ self.assertTrue(self.protocol.buffer_updated.called)
+
+ def test_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_into.return_value = 0
+ transport._read_ready()
+
+ self.protocol.eof_received.assert_called_with()
+ self.assertTrue(transport._fatal_error.called)
+
+ def test_read_ready(self):
+ transport = self.socket_transport()
+
+ self.sock.recv_into.return_value = 10
+ transport._read_ready()
+
+ self.protocol.get_buffer.assert_called_with()
+ self.protocol.buffer_updated.assert_called_with(10)
+
+ def test_read_ready_eof(self):
+ transport = self.socket_transport()
+ transport.close = mock.Mock()
+
+ self.sock.recv_into.return_value = 0
+ transport._read_ready()
+
+ self.protocol.eof_received.assert_called_with()
+ transport.close.assert_called_with()
+
+ def test_read_ready_eof_keep_open(self):
+ transport = self.socket_transport()
+ transport.close = mock.Mock()
+
+ self.sock.recv_into.return_value = 0
+ 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')
+ def test_read_ready_tryagain(self, m_exc):
+ self.sock.recv_into.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')
+ def test_read_ready_tryagain_interrupted(self, m_exc):
+ self.sock.recv_into.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')
+ def test_read_ready_conn_reset(self, m_exc):
+ err = self.sock.recv_into.side_effect = ConnectionResetError()
+
+ transport = self.socket_transport()
+ transport._force_close = mock.Mock()
+ with test_utils.disable_logger():
+ transport._read_ready()
+ transport._force_close.assert_called_with(err)
+
+ @mock.patch('logging.exception')
+ def test_read_ready_err(self, m_exc):
+ err = self.sock.recv_into.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')
+
+
class SelectorDatagramTransportTests(test_utils.TestCase):
def setUp(self):
View Lorry Controller Status