Function test_roundtrip_errors

python/pyarrow/tests/test_flight.py:2024–2059

Ensure that Flight errors propagate from server to client.

()

def test_roundtrip_errors():
    """Ensure that Flight errors propagate from server to client."""
    with ErrorFlightServer() as server, \
            FlightClient(('localhost', server.port)) as client:

        for arg, exc_type in ErrorFlightServer.error_cases().items():
            with pytest.raises(exc_type, match=".*foo.*"):
                list(client.do_action(flight.Action(arg, b"")))
        with pytest.raises(flight.FlightInternalError, match=".*foo.*"):
            list(client.list_flights())

        data = [pa.array([-10, -5, 0, 5, 10])]
        table = pa.Table.from_arrays(data, names=['a'])

        exceptions = {
            'internal': flight.FlightInternalError,
            'timedout': flight.FlightTimedOutError,
            'cancel': flight.FlightCancelledError,
            'unauthenticated': flight.FlightUnauthenticatedError,
            'unauthorized': flight.FlightUnauthorizedError,
        }

        for command, exception in exceptions.items():

            with pytest.raises(exception, match=".*foo.*"):
                writer, reader = client.do_put(
                    flight.FlightDescriptor.for_command(command),
                    table.schema)
                writer.write_table(table)
                writer.close()

            with pytest.raises(exception, match=".*foo.*"):
                writer, reader = client.do_put(
                    flight.FlightDescriptor.for_command(command),
                    table.schema)
                writer.close()


def test_do_put_independent_read_write():