Function test_interrupt

python/pyarrow/tests/test_flight.py:2403–2444

()

@pytest.mark.threading
def test_interrupt():
    if threading.current_thread().ident != threading.main_thread().ident:
        pytest.skip("test only works from main Python thread")

    def signal_from_thread():
        time.sleep(0.5)
        signal.raise_signal(signal.SIGINT)

    exc_types = (KeyboardInterrupt, pa.ArrowCancelled)

    def test(read_all):
        try:
            try:
                t = threading.Thread(target=signal_from_thread)
                with pytest.raises(exc_types) as exc_info:
                    t.start()
                    read_all()
            finally:
                t.join()
        except KeyboardInterrupt:
            # In case KeyboardInterrupt didn't interrupt read_all
            # above, at least prevent it from stopping the test suite
            pytest.fail("KeyboardInterrupt didn't interrupt Flight read_all")
        # __context__ is sometimes None
        e = exc_info.value
        assert isinstance(e, (pa.ArrowCancelled, KeyboardInterrupt)) or \
            isinstance(e.__context__, (pa.ArrowCancelled, KeyboardInterrupt))

    with CancelFlightServer() as server, \
            FlightClient(("localhost", server.port)) as client:

        reader = client.do_get(flight.Ticket(b""))
        test(reader.read_all)

        descriptor = flight.FlightDescriptor.for_command(b"echo")
        writer, reader = client.do_exchange(descriptor)
        test(reader.read_all)
        try:
            writer.close()
        except (KeyboardInterrupt, flight.FlightCancelledError):
            # Silence the Cancelled/Interrupt exception
            pass


def test_never_sends_data():