test_asyncorereactor.py 文件源码

def test_socket_error_on_write(self, *args):
        c = self.make_connection()

        # make the OptionsMessage write fail
        c.socket.send.side_effect = socket_error(errno.EIO, "bad stuff!")
        c.handle_write()

        # make sure it errored correctly
        self.assertTrue(c.is_defunct)
        self.assertIsInstance(c.last_error, socket_error)
        self.assertTrue(c.connected_event.is_set())