test_selector_events.py 文件源码

def test_read_ready_recv_retry(self):
        self.sslsock.recv.side_effect = ssl.SSLWantReadError
        transport = self._make_one()
        transport._read_ready()
        self.assertTrue(self.sslsock.recv.called)
        self.assertFalse(self.protocol.data_received.called)

        self.sslsock.recv.side_effect = BlockingIOError
        transport._read_ready()
        self.assertFalse(self.protocol.data_received.called)

        self.sslsock.recv.side_effect = InterruptedError
        transport._read_ready()
        self.assertFalse(self.protocol.data_received.called)