rlx_port_test.py 文件源码

def test_socket_busy_on_accept(self, monkeypatch):
        accept_responses = [
            lambda: MockUtils.raise_(socket.error(errno.ENOBUFS, "fatal error message")),
            lambda: MockUtils.raise_(socket.error(errno.EAGAIN, "busy, try accept again")),
            lambda: MockUtils.raise_(socket.error(errno.EPERM, "rejected, but try accept again"))
        ]
        if sys.platform != 'win32':
            monkeypatch.setattr(os, 'fork', lambda: 0)
        monkeypatch.setattr(socket, 'socket', lambda af, st: self.socket)
        self.socket.accept = lambda: accept_responses.pop()()
        try:
            RLXPort.listen(('localhost', 7000))
            assert False
        except Exception as e:
            traceback.format_exc()
            assert str(e) == '[Errno %d] fatal error message' % errno.ENOBUFS