test_selector_events.py 文件源码

def test_read_ready_recv_write(self):
        self.loop.remove_reader = mock.Mock()
        self.loop.add_writer = mock.Mock()
        self.sslsock.recv.side_effect = ssl.SSLWantWriteError
        transport = self._make_one()
        transport._read_ready()
        self.assertFalse(self.protocol.data_received.called)
        self.assertTrue(transport._read_wants_write)

        self.loop.remove_reader.assert_called_with(transport._sock_fd)
        self.loop.add_writer.assert_called_with(
            transport._sock_fd, transport._write_ready)