test_gipc.py 文件源码

def usecase_child_c(reader, syncwriter):
    with syncwriter:
        # Tell partner process that we are up and running!
        syncwriter.put("SYN")
        # Wait for confirmation.
        assert reader.get() == 'SYNACK'
    with reader:
        # Processes are synchronized. CHICKEN must be incoming within no time.
        with gevent.Timeout(SHORTTIME, False) as t:
            assert reader.get(timeout=t) == "CHICKEN"
        # Timeout is invalidated.
        # The write end becomes closed right now.
        with raises(EOFError):
            reader.get()
    sys.exit(5)