test_signal.py 文件源码

def check_wakeup(self, test_body):
        # use a subprocess to have only one thread and to not change signal
        # handling of the parent process
        code = """if 1:
        import fcntl
        import os
        import signal

        def handler(signum, frame):
            pass

        {}

        signal.signal(signal.SIGALRM, handler)
        read, write = os.pipe()
        flags = fcntl.fcntl(write, fcntl.F_GETFL, 0)
        flags = flags | os.O_NONBLOCK
        fcntl.fcntl(write, fcntl.F_SETFL, flags)
        signal.set_wakeup_fd(write)

        test()

        os.close(read)
        os.close(write)
        """.format(test_body)

        assert_python_ok('-c', code)