test_signals.py 文件源码

def test_catch_signals():
    print = lambda *args: None
    orig = signal.getsignal(signal.SIGILL)
    print(orig)
    with catch_signals([signal.SIGILL]) as queue:
        # Raise it a few times, to exercise signal coalescing, both at the
        # call_soon level and at the SignalQueue level
        signal_raise(signal.SIGILL)
        signal_raise(signal.SIGILL)
        await _core.wait_all_tasks_blocked()
        signal_raise(signal.SIGILL)
        await _core.wait_all_tasks_blocked()
        async for batch in queue:  # pragma: no branch
            assert batch == {signal.SIGILL}
            break
        signal_raise(signal.SIGILL)
        async for batch in queue:  # pragma: no branch
            assert batch == {signal.SIGILL}
            break
    with pytest.raises(RuntimeError):
        await queue.__anext__()
    assert signal.getsignal(signal.SIGILL) is orig