// Path: src/main/java/uk/sky/cqlmigrate/exception/CannotAcquireLockException.java
// public class CannotAcquireLockException extends LockException {
// 
//     public CannotAcquireLockException(String message) {
//         super(message);
//     }
// 
//     public CannotAcquireLockException(String message, Throwable cause) {
//         super(message, cause);
//     }
// }


import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
import com.datastax.oss.driver.api.core.servererrors.SyntaxError;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.*;
import uk.sky.cqlmigrate.exception.CannotAcquireLockException;
import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Instant;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import static com.datastax.oss.driver.api.core.cql.SimpleStatement.newInstance;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonList;
import static org.assertj.core.api.Assertions.*;
import static org.junit.Assert.fail;
        //when
        MIGRATOR.migrate(CASSANDRA_HOSTS, LOCAL_DC, binaryPort, username, password, TEST_KEYSPACE, cqlPaths);

        //then
        assertThat(session.getMetadata().getKeyspace(TEST_KEYSPACE)).isNotEmpty();
    }

    @Test(timeout = 5000)
    public void shouldThrowCannotAcquireLockExceptionIfLockCannotBeAcquiredAfterTimeout() throws Exception {
        //given
        CqlMigrator migrator = new CqlMigratorImpl(CqlMigratorConfig.builder()
                .withLockConfig(CassandraLockConfig.builder().withPollingInterval(ofMillis(50)).withTimeout(ofMillis(300)).build())
                .withReadConsistencyLevel(ConsistencyLevel.ALL)
                .withWriteConsistencyLevel(ConsistencyLevel.ALL)
                .build(), new SessionContextFactory());

        String client = UUID.randomUUID().toString();
        session.execute(newInstance("INSERT INTO cqlmigrate.locks (name, client) VALUES (?, ?)", LOCK_NAME, client));

        Collection<Path> cqlPaths = singletonList(getResourcePath("cql_bootstrap"));

        //when
        Future<?> future = executorService.submit(() -> migrator.migrate(
                CASSANDRA_HOSTS, LOCAL_DC, binaryPort, username, password, TEST_KEYSPACE, cqlPaths));

        Thread.sleep(310);
        Throwable throwable = catchThrowable(future::get);

        //then
        assertThat(throwable).isInstanceOf(ExecutionException.class);
// Path: src/main/java/uk/sky/cqlmigrate/exception/CannotAcquireLockException.java
// public class CannotAcquireLockException extends LockException {
// 
//     public CannotAcquireLockException(String message) {
//         super(message);
//     }
// 
//     public CannotAcquireLockException(String message, Throwable cause) {
//         super(message, cause);
//     }
// }


// Path: src/test/java/uk/sky/cqlmigrate/CqlMigratorImplTest.java
import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.SimpleStatement;
import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata;
import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata;
import com.datastax.oss.driver.api.core.servererrors.SyntaxError;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.thrift.transport.TTransportException;
import org.cassandraunit.utils.EmbeddedCassandraServerHelper;
import org.junit.*;
import uk.sky.cqlmigrate.exception.CannotAcquireLockException;
import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Instant;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import static com.datastax.oss.driver.api.core.cql.SimpleStatement.newInstance;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonList;
import static org.assertj.core.api.Assertions.*;
import static org.junit.Assert.fail;

        //when
        MIGRATOR.migrate(CASSANDRA_HOSTS, LOCAL_DC, binaryPort, username, password, TEST_KEYSPACE, cqlPaths);

        //then
        assertThat(session.getMetadata().getKeyspace(TEST_KEYSPACE)).isNotEmpty();
    }

    @Test(timeout = 5000)
    public void shouldThrowCannotAcquireLockExceptionIfLockCannotBeAcquiredAfterTimeout() throws Exception {
        //given
        CqlMigrator migrator = new CqlMigratorImpl(CqlMigratorConfig.builder()
                .withLockConfig(CassandraLockConfig.builder().withPollingInterval(ofMillis(50)).withTimeout(ofMillis(300)).build())
                .withReadConsistencyLevel(ConsistencyLevel.ALL)
                .withWriteConsistencyLevel(ConsistencyLevel.ALL)
                .build(), new SessionContextFactory());

        String client = UUID.randomUUID().toString();
        session.execute(newInstance("INSERT INTO cqlmigrate.locks (name, client) VALUES (?, ?)", LOCK_NAME, client));

        Collection<Path> cqlPaths = singletonList(getResourcePath("cql_bootstrap"));

        //when
        Future<?> future = executorService.submit(() -> migrator.migrate(
                CASSANDRA_HOSTS, LOCAL_DC, binaryPort, username, password, TEST_KEYSPACE, cqlPaths));

        Thread.sleep(310);
        Throwable throwable = catchThrowable(future::get);

        //then
        assertThat(throwable).isInstanceOf(ExecutionException.class);
        assertThat(throwable.getCause()).isInstanceOf(CannotAcquireLockException.class);