test_selector_events.py 文件源码

def test_write_ready_send_retry(self):
        transport = self._make_one()
        transport._buffer = list_to_buffer([b'data'])

        self.sslsock.send.side_effect = ssl.SSLWantWriteError
        transport._write_ready()
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)

        self.sslsock.send.side_effect = BlockingIOError()
        transport._write_ready()
        self.assertEqual(list_to_buffer([b'data']), transport._buffer)