Skip to content
Next Next commit
Add explicit tests for writelines
Includes two failing tests that demonstrate a bug where if the data passed to writelines can't be sent in one shot, the remaining buffer isn't written since there is no registered write handler
  • Loading branch information
alisaifee committed Apr 12, 2023
commit 273e1e74a9656b3e45cbc67b284457e730b38ea2
40 changes: 40 additions & 0 deletions Lib/test/test_asyncio/test_selector_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -747,6 +747,46 @@ def test_write_sendmsg_no_data(self):
self.assertFalse(self.sock.sendmsg.called)
self.assertEqual(list_to_buffer([b'data']), transport._buffer)

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)

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(sendmsg=False)
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(sendmsg=False)
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')
Expand Down