mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-17 20:28:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

gh-103462: Ensure SelectorSocketTransport.writelines registers a writer when data is still pending (#103463)

Browse source
This commit is contained in:
Ali-Akber Saifee 2023-04-12 21:46:52 -07:00 • committed by GitHub
parent 9e677406ee
commit 19d2639d1e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 49 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

42
Lib/test/test_asyncio/test_selector_events.py
View file

@ -747,6 +747,48 @@ class SelectorSocketTransportTests(test_utils.TestCase):
self.assertFalse(self.sock.sendmsg.called)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)
@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
def test_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')
def test_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)
def test_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)
def test_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)
@unittest.skipUnless(selector_events._HAS_SENDMSG, 'no sendmsg')
def test_write_sendmsg_full(self):
data = memoryview(b'data')