/*
 * Copyright 2017-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.r2dbc.spi.test;

import io.r2dbc.spi.Blob;
import io.r2dbc.spi.Clob;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.Parameters;
import io.r2dbc.spi.ReadableMetadata;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.Statement;
import io.r2dbc.spi.ValidationDepth;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.support.AbstractLobCreatingPreparedStatementCallback;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobCreator;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.NoSuchElementException;
import java.util.concurrent.Callable;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.junit.jupiter.api.Assertions.assertThrows;

/**
 * R2DBC TCK implementation to verify a driver. The TCK creates and removes tables after each test using JUnit lifecycle hooks (see {@link #createTable()} and {@link #dropTable()}).
 * <p>Functionality that is not supported by a driver should exclude the individual tests by using {@link Disabled @Disabled}.
 * Implementors need to provide the following methods:
 *
 * <ul>
 *     <li>{@link #getConnectionFactory()}: Provide the {@link ConnectionFactory} under test.</li>
 *     <li>{@link #getCreateTableWithAutogeneratedKey()}: Return the {@code CREATE TABLE} statement to create a table.</li>
 *     <li>{@link #getPlaceholder(int)}: Return an parameter placeholder for the {@code n}th argument. Arguments start at zero.</li>
 *     <li>{@link #getIdentifier(int)}: Return an identifier for the {@code n}th argument. The placeholder was obtained prior to this call via {@link #getPlaceholder(int)}. Arguments start at zero
 *     <li>{@link #getJdbcOperations}: Return a handle to {@link JdbcOperations} configured to the same data source as {@link #getConnectionFactory()}. Used for test fixture setup/teardown.</li>
 * </ul>
 *
 * <p>The following customization hooks can be overridden to adapt the TCK to a specific database:
 * <ul>
 *    <li>{@link #expand(TestStatement, Object...)}: Return the SQL statement for a specific {@link TestStatement}.</li>
 *    <li>{@link #blobType()}: Returns the SQL type for a {@link Blob} column.</li>
 *    <li>{@link #clobType()}: Returns the SQL type for a {@link Clob} column.</li>
 * </ul>
 *
 * @param <T> type of parameter placeholder identifiers, see {@link #getIdentifier(int)}.
 */
public interface TestKit<T> {

    /**
     * Customization hook to extract {@link Result#getRowsUpdated()}.
     *
     * @param result the result object
     * @return mono emitting the update row count.
     */
    default Mono<Long> extractRowsUpdated(Result result) {
        return Mono.from(result.getRowsUpdated());
    }

    /**
     * Customization hook to extract the {@code value} column from {@link Result}.
     *
     * @param result the result object
     * @return mono containing a collection of result values
     */
    default Mono<Collection<Integer>> extractColumns(Result result) {
        return Flux.from(result
                .map((row, rowMetadata) -> extractColumn(row, Integer.class)))
            .collect(Collectors.toSet());
    }

    /**
     * Customization hook to extract the {@code value} column from {@link Row}.
     *
     * @param row the row
     * @return the result value.
     */
    @Nullable
    default Object extractColumn(Row row) {
        return row.get("test_value");
    }

    /**
     * Customization hook to extract the {@code value} column from {@link Row}.
     *
     * @param row  the row
     * @param type column value type
     * @param <V>  column value type
     * @return the result value.
     */
    @Nullable
    default <V> V extractColumn(Row row, Class<V> type) {
        return row.get("test_value", type);
    }

    /**
     * Obtain and expand SQL text for a {@link TestStatement}. SQL retrieval can be customized by using {@link #doGetSql(TestStatement)}.
     *
     * @param statement the test statement
     * @param args      parameters for {@link String#format} replacement.
     * @return the expanded SQL query.
     */
    default String expand(TestStatement statement, Object... args) {
        return String.format(doGetSql(statement), args);
    }

    /**
     * Customization hook to obtain a SQL text for a {@link TestStatement}.
     *
     * @param statement the test statement
     * @return the SQL query.
     */
    default String doGetSql(TestStatement statement) {
        return statement.getSql();
    }

    /**
     * Customization hook: Returns the {@code INSERT INTO} statement for a table which created
     * by {@link #getCreateTableWithAutogeneratedKey}, only contains the
     * data for second column {@code value} which should be {@literal 100}.
     * <p>
     * Example:
     * <pre class="code">
     *     INSERT INTO test VALUES(100);
     * // or
     *     INSERT INTO test VALUES(DEFAULT,100);
     * </pre>
     *
     * @return the {@code INSERT INTO} statement
     * @deprecated since 0.8.3, use {@link #doGetSql(TestStatement)} as customization hook instead using {@link TestStatement#INSERT_VALUE_AUTOGENERATED_KEY}.
     */
    @Deprecated
    default String getInsertIntoWithAutogeneratedKey() {
        return expand(TestStatement.INSERT_VALUE_AUTOGENERATED_KEY);
    }

    /**
     * Returns the {@code CREATE TABLE} statement for a table named {@code test}
     * with two columns: First one uses auto-generated keys, second one is named {@code value} of type {@code INT}.
     * <p>
     * Example:
     * <pre class="code">
     *     CREATE TABLE test ( id SERIAL,  value INTEGER );
     *  // or
     *     CREATE TABLE test ( id INTEGER IDENTITY,  value INTEGER );
     *     </pre>
     *
     * @return the {@code CREATE TABLE} statement
     * @deprecated since 0.8.3, use {@link #doGetSql(TestStatement)} as customization hook instead using {@link TestStatement#CREATE_TABLE_AUTOGENERATED_KEY}.
     */
    @Deprecated
    default String getCreateTableWithAutogeneratedKey() {
        return expand(TestStatement.CREATE_TABLE_AUTOGENERATED_KEY);
    }

    /**
     * Customization hook: Returns the SQL type for a {@link Blob} column.
     *
     * @return the SQL type for a {@link Blob} column.
     */
    default String blobType() {
        return "BLOB";
    }

    /**
     * Customization hook: Returns the SQL type for a {@link Clob} column.
     *
     * @return the SQL type for a {@link Clob} column.
     */
    default String clobType() {
        return "CLOB";
    }

    @BeforeEach
    default void createTable() {
        getJdbcOperations().execute(expand(TestStatement.CREATE_TABLE));
        getJdbcOperations().execute(expand(TestStatement.CREATE_TABLE_TWO_COLUMNS));
        getJdbcOperations().execute(expand(TestStatement.CREATE_BLOB_TABLE, blobType()));
        getJdbcOperations().execute(expand(TestStatement.CREATE_CLOB_TABLE, clobType()));
    }

    @AfterEach
    default void dropTable() {
        getJdbcOperations().execute(expand(TestStatement.DROP_TABLE));
        getJdbcOperations().execute(expand(TestStatement.DROP_TABLE_TWO_COLUMNS));
        getJdbcOperations().execute(expand(TestStatement.DROP_BLOB_TABLE));
        getJdbcOperations().execute(expand(TestStatement.DROP_CLOB_TABLE));
    }

    /**
     * Returns a {@link ConnectionFactory} for the connected database.
     *
     * @return a {@link ConnectionFactory} for the connected database
     */
    ConnectionFactory getConnectionFactory();

    /**
     * Returns the database-specific placeholder for a given substitution.
     *
     * @param index the zero-index number of the substitution
     * @return the database-specific placeholder for a given substitution
     */
    String getPlaceholder(int index);

    /**
     * Returns the bind identifier for a given substitution.
     *
     * @param index the zero-index number of the substitution
     * @return the bind identifier for a given substitution
     */
    T getIdentifier(int index);

    /**
     * Returns a {@link JdbcOperations} for the connected database.
     *
     * @return a {@link JdbcOperations} for the connected database
     */
    JdbcOperations getJdbcOperations();

    @Test
    default void autoCommitByDefault() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.just(connection.isAutoCommit()),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(true).as("new connections are in auto-commit mode")
            .verifyComplete();
    }

    @Test
    default void changeAutoCommitCommitsTransaction() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection ->
                    Flux.from(connection.setAutoCommit(false))
                        .thenMany(connection.beginTransaction())
                        .thenMany(connection.createStatement(expand(TestStatement.INSERT_VALUE200)).execute())
                        .flatMap(Result::getRowsUpdated)
                        .thenMany(connection.setAutoCommit(true))
                        .thenMany(connection.createStatement(expand(TestStatement.SELECT_VALUE)).execute())
                        .flatMap(it -> it.map((row, metadata) -> extractColumn(row))),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(200).as("autoCommit(true) committed the transaction. Expecting a value to be present")
            .verifyComplete();
    }

    @Test
    default void sameAutoCommitLeavesTransactionUnchanged() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection ->
                    Flux.from(connection.setAutoCommit(false))
                        .thenMany(connection.beginTransaction())
                        .thenMany(connection.createStatement(expand(TestStatement.INSERT_VALUE200)).execute())
                        .flatMap(Result::getRowsUpdated)
                        .thenMany(connection.setAutoCommit(false))
                        .thenMany(connection.rollbackTransaction())
                        .thenMany(connection.createStatement(expand(TestStatement.SELECT_VALUE)).execute())
                        .flatMap(it -> it.map((row, metadata) -> extractColumn(row))),
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void batch() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createBatch()
                        .add(expand(TestStatement.INSERT_VALUE200))
                        .add(expand(TestStatement.SELECT_VALUE))
                        .execute())

                    .flatMap(Result::getRowsUpdated),
                Connection::close)
            .then()
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void bindFails() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {

                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    assertThrows(IllegalArgumentException.class, () -> statement.bind(0, null), "bind(0, null) should fail");
                    assertThrows(IndexOutOfBoundsException.class, () -> statement.bind(99, ""), "bind(nonexistent-index, null) should fail");
                    assertThrows(IllegalArgumentException.class, () -> bind(statement, getIdentifier(0), null), "bind(identifier, null) should fail");
                    assertThrows(IllegalArgumentException.class, () -> bind(statement, getIdentifier(0), Class.class), "bind(identifier, Class.class) should fail");
                    assertThrows(NoSuchElementException.class, () -> statement.bind("unknown-placeholder", ""), "bind(unknown-placeholder, \"\") should fail");
                    return Mono.empty();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void bindValueAsParameter() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection
                        .createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    bind(statement, getIdentifier(0), Parameters.in(100));
                    return Flux.from(statement.execute())
                        .flatMap(this::extractRowsUpdated);
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(1).as("rows inserted")
            .verifyComplete();

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                        .execute())
                    .flatMap(this::extractColumns),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(collectionOf(100)).as("test_value from select")
            .verifyComplete();
    }

    @Test
    default void bindNullAsParameter() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection
                        .createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    bind(statement, getIdentifier(0), Parameters.in(Integer.class));
                    return Flux.from(statement.execute())
                        .flatMap(this::extractRowsUpdated);
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(1).as("rows inserted")
            .verifyComplete();
    }

    @Test
    default void bindNull() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection
                        .createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    bindNull(statement, getIdentifier(0), Integer.class);
                    return Flux.from(statement.execute())
                        .flatMap(this::extractRowsUpdated);
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(1).as("rows inserted")
            .verifyComplete();
    }

    @Test
    default void bindNullFails() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {

                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    assertThrows(IllegalArgumentException.class, () -> statement.bindNull(null, String.class), "bindNull(null, …) should fail");
                    assertThrows(IllegalArgumentException.class, () -> bind(statement, getIdentifier(0), null), "bindNull(identifier, null) should fail");
                    return Mono.empty();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void blobInsert() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_BLOB_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    bind(statement, getIdentifier(0), Blob.from(Mono.just(StandardCharsets.UTF_8.encode("test-value"))));
                    return Flux.from(statement.execute())
                        .flatMap(this::extractRowsUpdated);
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(1).as("rows inserted")
            .verifyComplete();
    }

    @Test
    default void blobSelect() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_BLOB_VALUE_PLACEHOLDER, "?"), new AbstractLobCreatingPreparedStatementCallback(new DefaultLobHandler()) {

            @Override
            protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException {
                lobCreator.setBlobAsBytes(ps, 1, "test-value".getBytes(StandardCharsets.UTF_8));
            }

        });

        // BLOB as ByteBuffer
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_BLOB_VALUE))
                        .execute())
                    .flatMap(result -> result
                        .map((row, rowMetadata) -> extractColumn(row)))
                    .cast(ByteBuffer.class)
                    .map(buffer -> {
                        byte[] bytes = new byte[buffer.remaining()];
                        buffer.get(bytes);
                        return bytes;
                    }),
                Connection::close)
            .as(StepVerifier::create)
            .expectNextMatches(actual -> {
                ByteBuffer expected = ByteBuffer.wrap("test-value".getBytes(StandardCharsets.UTF_8));
                return Arrays.equals(expected.array(), actual);
            })
            .verifyComplete();

        // BLOB as Blob
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_BLOB_VALUE))
                        .execute())
                    .flatMap(result -> Flux.usingWhen(result
                            .map((row, rowMetadata) -> extractColumn(row, Blob.class)),
                        blob -> Flux.from(blob.stream()).reduce(ByteBuffer::put),
                        Blob::discard)),
                Connection::close)
            .as(StepVerifier::create)
            .expectNextMatches(actual -> {
                ByteBuffer expected = StandardCharsets.UTF_8.encode("test-value");
                return actual.compareTo(expected) == 0;
            })
            .verifyComplete();
    }

    @Test
    default void clobInsert() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_CLOB_VALUE_PLACEHOLDER, getPlaceholder(0)));
                    bind(statement, getIdentifier(0), Clob.from(Mono.just("test-value")));
                    return Flux.from(statement.execute())
                        .flatMap(Result::getRowsUpdated);
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(1).as("rows inserted")
            .verifyComplete();
    }

    @Test
    default void clobSelect() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_CLOB_VALUE_PLACEHOLDER, "?"), new AbstractLobCreatingPreparedStatementCallback(new DefaultLobHandler()) {

            @Override
            protected void setValues(PreparedStatement ps, LobCreator lobCreator) throws SQLException {
                lobCreator.setClobAsString(ps, 1, "test-value");
            }

        });

        // CLOB defaults to String
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_CLOB_VALUE))
                        .execute())
                    .flatMap(result -> result
                        .map((row, rowMetadata) -> extractColumn(row))),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext("test-value").as("test_value from select")
            .verifyComplete();

        // CLOB consume as Clob
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_CLOB_VALUE))
                        .execute())
                    .flatMap(result -> Flux.usingWhen(result
                            .map((row, rowMetadata) -> extractColumn(row, Clob.class)),
                        clob -> Flux.from(clob.stream())
                            .reduce(new StringBuilder(), StringBuilder::append)
                            .map(StringBuilder::toString),
                        Clob::discard)),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext("test-value").as("test_value from select")
            .verifyComplete();
    }

    @Test
    default void columnMetadata() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_TWO_COLUMNS));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_VALUE_TWO_COLUMNS))
                        .execute())
                    .flatMap(result -> {
                        return result.map((row, rowMetadata) -> {
                            return Arrays.asList(rowMetadata.contains("test_value"), rowMetadata.contains("TEST_VALUE"),
                                captureException(() -> rowMetadata.getColumnMetadata(-1)),
                                captureException(() -> rowMetadata.getColumnMetadata(100)),
                                captureException(() -> rowMetadata.getColumnMetadata("unknown")));
                        });
                    })
                    .flatMapIterable(Function.identity()),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(true).as("rowMetadata.contains(value)")
            .expectNext(true).as("rowMetadata.contains(VALUE)")
            .expectNextMatches(IndexOutOfBoundsException.class::isInstance).as("getColumnMetadata(-1) throws IndexOutOfBoundsException")
            .expectNextMatches(IndexOutOfBoundsException.class::isInstance).as("getColumnMetadata(100) throws IndexOutOfBoundsException")
            .expectNextMatches(NoSuchElementException.class::isInstance).as("getColumnMetadata(unknown) throws NoSuchElementException")
            .verifyComplete();
    }

    @Test
    default void rowMetadata() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_TWO_COLUMNS));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_VALUE_ALIASED_COLUMNS))
                        .execute())
                    .flatMap(result -> result.map((row, rowMetadata) -> rowMetadata.getColumnMetadatas().stream().map(ReadableMetadata::getName).collect(Collectors.toList())))
                    .flatMapIterable(Function.identity()),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext("b").as("First column label: b")
            .expectNext("c").as("First column label: c")
            .expectNext("a").as("First column label: a")
            .verifyComplete();
    }

    @Test
    default void row() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_TWO_COLUMNS));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_VALUE_ALIASED_COLUMNS))
                        .execute())
                    .flatMap(result -> result.map(readable -> Arrays.asList(captureException(() -> readable.get(-1)),
                        captureException(() -> readable.get(100)),
                        captureException(() -> readable.get("unknown")))))
                    .flatMapIterable(Function.identity()),
                Connection::close)
            .as(StepVerifier::create)
            .expectNextMatches(IndexOutOfBoundsException.class::isInstance).as("get(-1) throws IndexOutOfBoundsException")
            .expectNextMatches(IndexOutOfBoundsException.class::isInstance).as("get(100) throws IndexOutOfBoundsException")
            .expectNextMatches(NoSuchElementException.class::isInstance).as("get(unknown) throws NoSuchElementException")
            .verifyComplete();
    }

    @Test
    default void compoundStatement() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_VALUE_BATCH))
                        .execute())
                    .flatMap(this::extractColumns),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(collectionOf(100)).as("test_value from first select")
            .expectNext(collectionOf(100)).as("test_value from second select")
            .verifyComplete();
    }

    @Test
    default void createStatementFails() {

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    assertThrows(IllegalArgumentException.class, () -> connection.createStatement(null));
                    return Mono.empty();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void duplicateColumnNames() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_TWO_COLUMNS));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection

                        .createStatement(expand(TestStatement.SELECT_VALUE_TWO_COLUMNS))
                        .execute())

                    .flatMap(result -> result
                        .map((row, rowMetadata) -> Arrays.asList(row.get("test_value"), row.get("TEST_VALUE"))))
                    .flatMapIterable(Function.identity()),
                Connection::close)

            .as(StepVerifier::create)
            .expectNext(100).as("test_value from col1")
            .expectNext(100).as("test_value from col1 (upper case)")
            .verifyComplete();
    }

    @Test
    default void prepareStatement() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));

                    IntStream.range(0, 10)
                        .forEach(i -> {
                            bind(statement, getIdentifier(0), i);

                            if (i != 9) {
                                statement.add();
                            }
                        });

                    return Flux.from(statement
                            .execute())
                        .flatMap(this::extractRowsUpdated);
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(10).as("values from insertions")
            .verifyComplete();
    }

    @Test
    default void prepareStatementWithTrailingAddShouldFail() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));

                    bind(statement, getIdentifier(0), 0).add();

                    return Flux.from(statement
                            .execute())
                        .flatMap(this::extractRowsUpdated).then();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyError();
    }

    @Test
    default void prepareStatementWithIncompleteBatchFails() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_TWO_VALUES_PLACEHOLDER, getPlaceholder(0), getPlaceholder(1)));

                    bind(statement, getIdentifier(0), 0);

                    assertThrows(IllegalStateException.class, statement::add);
                    return Mono.empty();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void prepareStatementWithIncompleteBindingFails() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_TWO_VALUES_PLACEHOLDER, getPlaceholder(0), getPlaceholder(1)));

                    bind(statement, getIdentifier(0), 0);

                    assertThrows(IllegalStateException.class, statement::execute);
                    return Mono.empty();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void returnGeneratedValues() {

        getJdbcOperations().execute(expand(TestStatement.DROP_TABLE));
        getJdbcOperations().execute(getCreateTableWithAutogeneratedKey());

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(getInsertIntoWithAutogeneratedKey());

                    statement.returnGeneratedValues();

                    return Flux.from(statement
                            .execute())
                        .flatMap(it -> it.map((row, rowMetadata) -> row.get(0)));
                },
                Connection::close)
            .as(StepVerifier::create)
            .expectNextCount(1)
            .verifyComplete();
    }

    @Test
    default void returnGeneratedValuesFails() {

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> {
                    Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE100));

                    assertThrows(IllegalArgumentException.class, () -> statement.returnGeneratedValues((String[]) null));
                    return Mono.empty();
                },
                Connection::close)
            .as(StepVerifier::create)
            .verifyComplete();
    }

    @Test
    default void savePoint() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Mono.from(connection

                        .beginTransaction())
                    .<Object>thenMany(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(Flux.defer(() -> {
                            Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                            bind(statement, getIdentifier(0), 200);
                            return statement.execute();
                        })
                        .flatMap(this::extractRowsUpdated))
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(connection.createSavepoint("test_savepoint"))
                    .concatWith(Flux.defer(() -> {
                            Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                            bind(statement, getIdentifier(0), 300);
                            return statement.execute();
                        })
                        .flatMap(this::extractRowsUpdated))
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(connection.rollbackTransactionToSavepoint("test_savepoint"))
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns)),

                Connection::close)
            .as(StepVerifier::create)
            .expectNext(collectionOf(100)).as("test_value from select")
            .expectNext(1L).as("rows inserted")
            .expectNext(collectionOf(100, 200)).as("values from select")
            .expectNext(1L).as("rows inserted")
            .expectNext(collectionOf(100, 200, 300)).as("values from select")
            .expectNext(collectionOf(100, 200)).as("values from select")
            .verifyComplete();
    }

    @Test
    default void savePointStartsTransaction() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Mono.from(connection
                        .createSavepoint("test_savepoint"))
                    .then(Mono.fromSupplier(() -> connection.isAutoCommit())),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(false).as("createSavepoint starts a transaction")
            .verifyComplete();
    }

    @Test
    default void segmentInsertEmitsUpdateCount() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection
                        .createStatement(expand(TestStatement.INSERT_VALUE100))
                        .execute())
                    .flatMap(result -> result.flatMap(segment -> Mono.just(((Result.UpdateCount) segment).value()))),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(1L).as("insert of a single row")
            .verifyComplete();
    }

    @Test
    default void segmentInsertWithFilterCompletesWithoutOnNext() {
        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection
                        .createStatement(expand(TestStatement.INSERT_VALUE100))
                        .execute())
                    .flatMap(result -> result
                        .filter(it -> false)
                        .flatMap(segment -> Mono.just(false))),
                Connection::close)
            .as(StepVerifier::create)
            .as("filter(it -> false) should complete without data signals")
            .verifyComplete();
    }

    @Test
    default void segmentSelectWithFilterCompletesWithoutOnNext() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection
                        .createStatement(expand(TestStatement.SELECT_VALUE))
                        .execute())
                    .flatMap(result -> result
                        .filter(it -> false)
                        .flatMap(segment -> Mono.just(false))),
                Connection::close)
            .as(StepVerifier::create)
            .as("filter(it -> false) should complete without data signals")
            .verifyComplete();
    }

    @Test
    default void segmentSelectWithEmitsRow() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Flux.from(connection
                        .createStatement(expand(TestStatement.SELECT_VALUE))
                        .execute())
                    .flatMap(result -> result
                        .filter(Result.RowSegment.class::isInstance)
                        .flatMap(segment -> Mono.just(extractColumn(((Result.RowSegment) segment).row())))),
                Connection::close)
            .as(StepVerifier::create)
            .expectNext(100).as("test_value from select")
            .verifyComplete();
    }

    @Test
    default void transactionCommit() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Mono.from(connection

                        .beginTransaction())
                    .<Object>thenMany(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(Flux.defer(() -> {
                            Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                            bind(statement, getIdentifier(0), 200);
                            return statement.execute();
                        })
                        .flatMap(this::extractRowsUpdated))
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(connection.commitTransaction())
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns)),

                Connection::close)
            .as(StepVerifier::create)
            .expectNext(collectionOf(100)).as("test_value from select")
            .expectNext(1L).as("rows inserted")
            .expectNext(collectionOf(100, 200)).as("values from select")
            .expectNext(collectionOf(100, 200)).as("values from select")
            .verifyComplete();
    }

    @Test
    default void transactionRollback() {
        getJdbcOperations().execute(expand(TestStatement.INSERT_VALUE100));

        Flux.usingWhen(getConnectionFactory().create(),
                connection -> Mono.from(connection

                        .beginTransaction())
                    .<Object>thenMany(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(Flux.defer(() -> {
                            Statement statement = connection.createStatement(expand(TestStatement.INSERT_VALUE_PLACEHOLDER, getPlaceholder(0)));
                            bind(statement, getIdentifier(0), 200);
                            return statement.execute();
                        })
                        .flatMap(this::extractRowsUpdated))
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns))

                    .concatWith(connection.rollbackTransaction())
                    .concatWith(Flux.from(connection.createStatement(expand(TestStatement.SELECT_VALUE))
                            .execute())
                        .flatMap(this::extractColumns)),

                Connection::close)
            .as(StepVerifier::create)
            .expectNext(collectionOf(100)).as("test_value from select")
            .expectNext(1L).as("rows inserted")
            .expectNext(collectionOf(100, 200)).as("values from select")
            .expectNext(collectionOf(100)).as("test_value from select")
            .verifyComplete();
    }

    @Test
    default void validate() {
        Mono.from(getConnectionFactory().create())
            .flatMapMany(connection -> Flux.concat(connection.validate(ValidationDepth.LOCAL),
                connection.validate(ValidationDepth.REMOTE),
                connection.close(),
                connection.validate(ValidationDepth.LOCAL),
                connection.validate(ValidationDepth.REMOTE)))
            .as(StepVerifier::create)
            .expectNext(true).as("successful local validation")
            .expectNext(true).as("successful remote validation")
            .expectNext(false).as("failed local validation after close")
            .expectNext(false).as("failed remote validation after close")
            .verifyComplete();
    }

    static Statement bind(Statement statement, Object identifier, Object value) {
        Assert.requireNonNull(identifier, "Identifier must not be null");
        if (identifier instanceof String) {
            return statement.bind((String) identifier, value);
        }
        if (identifier instanceof Integer) {
            return statement.bind((Integer) identifier, value);
        }
        throw new IllegalArgumentException(String.format("Identifier %s must be a String or Integer. Was: %s", identifier, identifier.getClass().getName()));
    }

    static Statement bindNull(Statement statement, Object identifier, Class<?> type) {
        Assert.requireNonNull(identifier, "Identifier must not be null");
        if (identifier instanceof String) {
            return statement.bindNull((String) identifier, type);
        }
        if (identifier instanceof Integer) {
            return statement.bindNull((Integer) identifier, type);
        }
        throw new IllegalArgumentException(String.format("Identifier %s must be a String or Integer. Was: %s", identifier, identifier.getClass().getName()));
    }

    /**
     * Returns an unordered {@code Collection} containing 0 or more {@code values}.
     *
     * @param <T>    Class of objects in {@code values}
     * @param values 0 or more values
     * @return {@code Collection} containing {@code values}
     */
    @SafeVarargs
    @SuppressWarnings("varargs")
    static <T> Collection<T> collectionOf(T... values) {
        return new HashSet<>(Arrays.asList(values));
    }

    /**
     * Returns an {@link Exception} from a {@link Callable}.
     *
     * @param throwingCallable the callable code block
     * @return the thrown exception
     * @throws IllegalStateException if {@code throwingCallable} did not throw an exception.
     */
    static Exception captureException(Callable<?> throwingCallable) {

        try {
            throwingCallable.call();
        } catch (Exception e) {
            return e;
        }

        throw new IllegalStateException("Callable did not throw an exception.");
    }

    /**
     * Enumeration of TCK statements. Can be customized by overriding {@link #expand(TestStatement, Object...)}.
     */
    enum TestStatement {

        //-------------------------------------------------------------------------
        // Methods dealing with a single-column table.
        //-------------------------------------------------------------------------

        INSERT_VALUE_PLACEHOLDER("INSERT INTO test VALUES(%s)"),
        INSERT_VALUE100("INSERT INTO test VALUES(100)"),
        INSERT_VALUE200("INSERT INTO test VALUES(200)"),
        INSERT_TWO_VALUES_PLACEHOLDER("INSERT INTO test VALUES(%s,%s)"),
        SELECT_VALUE("SELECT test_value FROM test"),
        CREATE_TABLE("CREATE TABLE test ( test_value INTEGER )"),
        DROP_TABLE("DROP TABLE test"),
        SELECT_VALUE_BATCH("SELECT test_value FROM test; SELECT test_value FROM test"),

        INSERT_VALUE_AUTOGENERATED_KEY("INSERT INTO test VALUES(100)"),
        CREATE_TABLE_AUTOGENERATED_KEY("CREATE TABLE test ( id INTEGER IDENTITY,  test_value INTEGER )"),

        //-------------------------------------------------------------------------
        // Methods dealing with a single-column BLOB table.
        //-------------------------------------------------------------------------

        INSERT_BLOB_VALUE_PLACEHOLDER("INSERT INTO blob_test VALUES (%s)"),
        CREATE_BLOB_TABLE("CREATE TABLE blob_test ( test_value %s )"),
        DROP_BLOB_TABLE("DROP TABLE blob_test"),
        SELECT_BLOB_VALUE("SELECT test_value FROM blob_test"),

        //-------------------------------------------------------------------------
        // Methods dealing with a single-column CLOB table.
        //-------------------------------------------------------------------------

        INSERT_CLOB_VALUE_PLACEHOLDER("INSERT INTO clob_test VALUES (%s)"),
        SELECT_CLOB_VALUE("SELECT test_value FROM clob_test"),
        CREATE_CLOB_TABLE("CREATE TABLE clob_test ( test_value %s )"),
        DROP_CLOB_TABLE("DROP TABLE clob_test"),

        //-------------------------------------------------------------------------
        // Methods dealing with a two-column table.
        //-------------------------------------------------------------------------

        INSERT_TWO_COLUMNS("INSERT INTO test_two_column VALUES (100, 'hello')"),
        SELECT_VALUE_TWO_COLUMNS("SELECT col1 AS test_value, col2 AS test_value FROM test_two_column"),
        SELECT_VALUE_ALIASED_COLUMNS("SELECT col1 AS b, col1 AS c, col1 AS a FROM test_two_column"),
        CREATE_TABLE_TWO_COLUMNS("CREATE TABLE test_two_column ( col1 INTEGER, col2 VARCHAR(100) )"),
        DROP_TABLE_TWO_COLUMNS("DROP TABLE test_two_column");

        private final String sql;

        TestStatement(String sql) {
            this.sql = sql;
        }

        public String getSql() {
            return this.sql;
        }

    }

}
