rlx_port_test.py 文件源码

def test_fork_error(self, monkeypatch):
        def critical(*args):
            called.times += 1
            called.args = args

        logger = Mock()
        called = MockUtils.Placeholder()
        logger.critical = critical
        accept_responses = [
            lambda: MockUtils.raise_(socket.error(errno.EMFILE, "fatal error message")),
            lambda: (MockSocket.create(), ('some-address', 7000))
        ]
        monkeypatch.setattr('relaax.server.rlx_server.rlx_port.log', logger)
        if sys.platform != 'win32':
            monkeypatch.setattr(os, 'fork', lambda: MockUtils.raise_(OSError('can\'t fork')))
        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:
            if sys.platform != 'win32':
                assert called.args == ("Can't start child process ('some-address', 7000): can't fork",)
                assert called.times == 1
            assert str(e) == '[Errno %d] fatal error message' % errno.EMFILE