Skip to content

gh-85451: Add StreamReaderBufferedProtocol #21446

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tontinton wants to merge 6 commits into from
Closed

gh-85451: Add StreamReaderBufferedProtocol #21446

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9a65cfe
bpo-41279: Add StreamReaderBufferedProtocol (GH-21446)
tontinton Jul 22, 2020
6db642c
bpo-41279: Add support for BufferedProtocol in _UnixReadPipeTransport…
tontinton Jul 11, 2020
71d6a90
bpo-41279: Add support for BufferedProtocol in _ProactorReadPipeTrans…
tontinton Jul 14, 2020
cdc14d8
bpo-41279: Support resetting a protocol in _ProactorReadPipeTransport…
tontinton Jul 20, 2020
9e81f1a
bpo-41279: Add BufferedProtocol to proactor transport tests (GH-21446)
tontinton Jul 19, 2020
3dc4d18
bpo-41279: Add BufferedProtocol to unix read pipe transport tests (GH…
tontinton Jul 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
bpo-41279: Add BufferedProtocol to proactor transport tests (GH-21446)
  • Loading branch information
@tontinton
tontinton committed Jul 8, 2022
commit 9e81f1a660c6609b8c6494fc533d8b327894fd24
289 changes: 194 additions & 95 deletions Lib/test/test_asyncio/test_proactor_events.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,53 +30,27 @@ def close_transport(transport):
transport._sock = None


class ProactorSocketTransportTests(test_utils.TestCase):
class ProactorSocketTransportTestsBase:

def setUp(self):
super().setUp()
def make_test_protocol(self):
raise NotImplementedError

def setup_globals(self):
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.Protocol)
self.protocol = None
self.sock = mock.Mock(socket.socket)
self.buffer_size = 65536
self.make_test_protocol()

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 = self.loop.create_future()
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, bytearray(self.buffer_size))

def test_loop_reading(self):
tr = self.socket_transport()
tr._loop_reading()
self.loop._proactor.recv_into.assert_called_with(self.sock, bytearray(self.buffer_size))
self.assertFalse(self.protocol.data_received.called)
self.assertFalse(self.protocol.eof_received.called)

def test_loop_reading_data(self):
buf = b'data'
res = self.loop.create_future()
res.set_result(len(buf))

tr = self.socket_transport()
tr._read_fut = res
tr._data[:len(buf)] = buf
tr._loop_reading(res)
called_buf = bytearray(self.buffer_size)
called_buf[:len(buf)] = buf
self.loop._proactor.recv_into.assert_called_with(self.sock, called_buf)
self.protocol.data_received.assert_called_with(bytearray(buf))

@unittest.skipIf(sys.flags.optimize, "Assertions are disabled in optimized mode")
def test_loop_reading_no_data(self):
res = self.loop.create_future()
Expand All @@ -88,12 +62,12 @@ def test_loop_reading_no_data(self):
tr.close = mock.Mock()
tr._read_fut = res
tr._loop_reading(res)
self.assertFalse(self.loop._proactor.recv_into.called)
self.assertFalse(self.loop._proactor.recv.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()
err = self.loop._proactor.recv.side_effect = ConnectionAbortedError()

tr = self.socket_transport()
tr._fatal_error = mock.Mock()
Expand All @@ -103,7 +77,7 @@ def test_loop_reading_aborted(self):
'Fatal read error on pipe transport')

def test_loop_reading_aborted_closing(self):
self.loop._proactor.recv_into.side_effect = ConnectionAbortedError()
self.loop._proactor.recv.side_effect = ConnectionAbortedError()

tr = self.socket_transport()
tr._closing = True
Expand All @@ -112,15 +86,15 @@ def test_loop_reading_aborted_closing(self):
self.assertFalse(tr._fatal_error.called)

def test_loop_reading_aborted_is_fatal(self):
self.loop._proactor.recv_into.side_effect = ConnectionAbortedError()
self.loop._proactor.recv.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()
err = self.loop._proactor.recv.side_effect = ConnectionResetError()

tr = self.socket_transport()
tr._closing = False
Expand All @@ -131,7 +105,7 @@ def test_loop_reading_conn_reset_lost(self):
tr._force_close.assert_called_with(err)

def test_loop_reading_exception(self):
err = self.loop._proactor.recv_into.side_effect = (OSError())
err = self.loop._proactor.recv.side_effect = (OSError())

tr = self.socket_transport()
tr._fatal_error = mock.Mock()
Expand Down Expand Up @@ -363,62 +337,6 @@ def test_write_eof_duplex_pipe(self):
tr.write_eof()
close_transport(tr)

def test_pause_resume_reading(self):
tr = self.socket_transport()
index = 0
msgs = [b'data1', b'data2', b'data3', b'data4', b'data5', b'']
reversed_msgs = list(reversed(msgs))

def recv_into(sock, data):
f = self.loop.create_future()
msg = reversed_msgs.pop()

result = f.result
def monkey():
data[:len(msg)] = msg
return result()
f.result = monkey

f.set_result(len(msg))
return f

self.loop._proactor.recv_into.side_effect = recv_into
self.loop._run_once()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())

for msg in msgs[:2]:
self.loop._run_once()
self.protocol.data_received.assert_called_with(bytearray(msg))

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.data_received.assert_called_with(bytearray(msgs[1]))

tr.resume_reading()
tr.resume_reading()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())

for msg in msgs[2:4]:
self.loop._run_once()
self.protocol.data_received.assert_called_with(bytearray(msg))

tr.pause_reading()
tr.resume_reading()
self.loop.call_exception_handler = mock.Mock()
self.loop._run_once()
self.loop.call_exception_handler.assert_not_called()
self.protocol.data_received.assert_called_with(bytearray(msgs[4]))
tr.close()

self.assertFalse(tr.is_reading())


def pause_writing_transport(self, high):
tr = self.socket_transport()
tr.set_write_buffer_limits(high=high)
Expand Down Expand Up @@ -498,6 +416,187 @@ def test_dont_pause_writing(self):
self.assertFalse(self.protocol.pause_writing.called)


class ProactorSocketTransportWithProtocolTests(test_utils.TestCase, ProactorSocketTransportTestsBase):

def make_test_protocol(self):
self.protocol = test_utils.make_test_protocol(asyncio.Protocol)

def setUp(self):
super().setUp()
self.setup_globals()

def test_ctor(self):
fut = self.loop.create_future()
tr = self.socket_transport(waiter=fut)
test_utils.run_briefly(self.loop)
self.assertIsNone(fut.result())
self.protocol.connection_made(tr)
self.proactor.recv.assert_called_with(self.sock, self.buffer_size)

def test_loop_reading(self):
tr = self.socket_transport()
tr._loop_reading()
self.loop._proactor.recv.assert_called_with(self.sock, self.buffer_size)
self.assertFalse(self.protocol.data_received.called)
self.assertFalse(self.protocol.eof_received.called)

def test_loop_reading_data(self):
buf = b'data'
res = self.loop.create_future()
res.set_result(buf)

tr = self.socket_transport()
tr._read_fut = res
tr._loop_reading(res)
self.loop._proactor.recv.assert_called_with(self.sock, self.buffer_size)
self.protocol.data_received.assert_called_with(buf)

def test_pause_resume_reading(self):
tr = self.socket_transport()
futures = []
msgs = [b'data1', b'data2', b'data3', b'data4', b'data5', b'']
for msg in msgs:
f = self.loop.create_future()
f.set_result(msg)
futures.append(f)

self.loop._proactor.recv.side_effect = futures
self.loop._run_once()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())

for msg in msgs[:2]:
self.loop._run_once()
self.protocol.data_received.assert_called_with(msg)

@gst gst Jul 7, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

before the _run_once() would not it be better to assert that the msg is not in the _mock_call_args_list of the mock object ?

and same below for the others data_received.assert_called_with following a _run_once().

@tontinton tontinton Jul 8, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't understand which mock object you are talking about, and I don't see any other test reading _mock_call_args_list.

Can you elaborate?

@gst gst Jul 8, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

something like:

for msg in msgs[:2]:
            self.assertFalse(unittest.mock.call(msg) in self.protocol.data_received.mock_calls)
            self.loop._run_once()
            self.protocol.data_received.assert_called_with(msg)

to ensure that each msg is effectively processed during the _run_once() call.

@tontinton tontinton Jul 8, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh I see, but protocol's buffer_updated (data_received is only in a regular stream object, not buffered) is not a mock, it's an actual function, so no mock_calls field.

@gst gst Jul 8, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah ok :/


tr.pause_reading()
self.assertTrue(tr._paused)
self.assertFalse(tr.is_reading())
for _ in range(10):
self.loop._run_once()
self.protocol.data_received.assert_called_with(msgs[1])

tr.resume_reading()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())

for msg in msgs[2:4]:
self.loop._run_once()
self.protocol.data_received.assert_called_with(msg)

tr.pause_reading()
tr.resume_reading()
self.loop.call_exception_handler = mock.Mock()
self.loop._run_once()
self.loop.call_exception_handler.assert_not_called()
self.protocol.data_received.assert_called_with(msgs[4])
tr.close()

self.assertFalse(tr.is_reading())


class ProactorSocketTransportWithBufferedProtocolTests(test_utils.TestCase, ProactorSocketTransportTestsBase):

def make_test_protocol(self):
self.protocol = test_utils.make_test_buffered_protocol(
asyncio.BufferedProtocol, self.buffer_size)

# patch for some test in the base class to pass
# BufferedProtocol calls recv_into instead of recv and all the tests
# that assert if the function is called check if `recv` is called.
self.loop._proactor.recv = self.loop._proactor.recv_into

def setUp(self):
super().setUp()
self.setup_globals()

def assert_buffer_received(self, data):
self.assertTrue(self.protocol.buffer_updated.called)
self.assertEqual(self.protocol._last_called_buffer[:len(data)],
bytearray(data))

def test_ctor(self):
fut = self.loop.create_future()
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, bytearray(self.buffer_size))

def test_loop_reading(self):
tr = self.socket_transport()
tr._loop_reading()
self.loop._proactor.recv_into.assert_called_with(self.sock, bytearray(self.buffer_size))
self.assertFalse(self.protocol.buffer_updated.called)
self.assertFalse(self.protocol.eof_received.called)

def test_loop_reading_data(self):
buf = b'data'
res = self.loop.create_future()
res.set_result(len(buf))

tr = self.socket_transport()
tr._read_fut = res
tr._data[:len(buf)] = buf
tr._loop_reading(res)
called_buf = bytearray(self.buffer_size)
called_buf[:len(buf)] = buf
self.loop._proactor.recv_into.assert_called_with(self.sock, called_buf)
self.assert_buffer_received(buf)

def test_pause_resume_reading(self):
tr = self.socket_transport()
msgs = [b'data1', b'data2', b'data3', b'data4', b'data5', b'']
reversed_msgs = list(reversed(msgs))

def recv_into(sock, data):
msg = reversed_msgs.pop()

f = self.loop.create_future()
result = f.result
def monkey():
data[:len(msg)] = msg
return result()
f.result = monkey

f.set_result(len(msg))
return f

self.loop._proactor.recv_into.side_effect = recv_into
self.loop._run_once()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())

for msg in msgs[:2]:
self.loop._run_once()
self.assert_buffer_received(msg)

tr.pause_reading()
self.assertTrue(tr._paused)
self.assertFalse(tr.is_reading())
for _ in range(10):
self.loop._run_once()
self.assert_buffer_received(msgs[1])

tr.resume_reading()
self.assertFalse(tr._paused)
self.assertTrue(tr.is_reading())

for msg in msgs[2:4]:
self.loop._run_once()
self.assert_buffer_received(msg)

tr.pause_reading()
tr.resume_reading()
self.loop.call_exception_handler = mock.Mock()
self.loop._run_once()
self.loop.call_exception_handler.assert_not_called()
self.assert_buffer_received(msgs[4])
tr.close()

self.assertFalse(tr.is_reading())


class ProactorDatagramTransportTests(test_utils.TestCase):

def setUp(self):
Expand Down
19 changes: 19 additions & 0 deletions Lib/test/test_asyncio/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,25 @@ def make_test_protocol(base):
return type('TestProtocol', (base,) + base.__bases__, dct)()


def make_test_buffered_protocol(base, buffer_size):
protocol = make_test_protocol(base)
protocol._buffer = bytearray(buffer_size)
protocol._last_called_buffer = None

def get_buffer(*_, **__):
return protocol._buffer

def buffer_updated(nbytes):
protocol.buffer_updated.called = True
protocol._last_called_buffer = protocol._buffer[:nbytes]

protocol.get_buffer = get_buffer

protocol.buffer_updated = buffer_updated
protocol.buffer_updated.called = False
return protocol


class TestSelector(selectors.BaseSelector):

def __init__(self):
Expand Down