test_protocol.py 文件源码

def test_non_retryable_error(self, m_select):
        m_select.side_effect = iter([
            ([self.sck], [], []),
        ])
        err = socket.error()
        err.errno = errno.E2BIG
        self.sck.recv.side_effect = err
        msg_gen = self.reader.new_messages(timeout=1)
        self.assertRaises(socket.error, next, msg_gen)