package io.aiven.klaw.clusterapi.services;

import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.mock;

import io.aiven.klaw.clusterapi.constants.TestConstants;
import io.aiven.klaw.clusterapi.models.ApiResponse;
import io.aiven.klaw.clusterapi.models.ClusterTopicRequest;
import io.aiven.klaw.clusterapi.models.LoadTopicsResponse;
import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
import io.aiven.klaw.clusterapi.utils.AdminClientProperties;
import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.errors.TopicExistsException;
import org.assertj.core.api.AbstractThrowableAssert;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;

@ExtendWith(MockitoExtension.class)
class ApacheKafkaTopicServiceTest {

  @Mock private ClusterApiUtils clusterApiUtils;
  @Mock private SchemaService schemaService;

  private ApacheKafkaTopicService apacheKafkaTopicService;

  @Mock private KafkaSupportedProtocol protocol;
  @Mock private AdminClient adminClient;

  @Mock private AdminClientProperties adminClientProperties;
  @Mock private ListTopicsResult listTopicsResult;
  @Mock private DescribeTopicsResult describeTopicsResult;
  @Mock private TopicDescription topicDescription;
  @Mock private TopicPartitionInfo topicPartitionInfo;
  @Mock private CreateTopicsResult createTopicsResult;
  @Mock private DeleteTopicsResult deleteTopicsResult;
  @Mock private KafkaFuture<Void> kafkaFuture;

  private static Stream<Exception> exceptionProviderForCreateTopic() {
    return Stream.of(
        new RuntimeException(),
        new NumberFormatException(),
        new KafkaException(),
        new InterruptedException(),
        new ExecutionException(new RuntimeException()));
  }

  private static Stream<Exception> exceptionProviderForDeleteTopic() {
    return Stream.of(
        new RuntimeException(""),
        new KafkaException(""),
        new InterruptedException(""),
        new ExecutionException(new RuntimeException("")));
  }

  @BeforeEach
  void setUp() {
    apacheKafkaTopicService = new ApacheKafkaTopicService(clusterApiUtils, schemaService);
  }

  @Test
  void loadTopicsClientNull() throws Exception {
    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(null);

    AbstractThrowableAssert<?, ? extends Throwable> exception =
        assertThatThrownBy(
            () ->
                apacheKafkaTopicService.loadTopics(
                    TestConstants.ENVIRONMENT,
                    protocol,
                    TestConstants.CLUSTER_IDENTIFICATION,
                    false));

    exception.isInstanceOf(Exception.class);
    exception.hasMessage("Cannot connect to cluster.");
  }

  @Test
  void loadTopics() throws Exception {
    Map<String, TopicDescription> topicDescriptionsPerAdminClient =
        new HashMap<>(
            Map.of(
                "_confluent", topicDescription,
                "__connect", topicDescription,
                "_schemas", topicDescription,
                "topic", topicDescription));

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.listTopics(any(ListTopicsOptions.class))).thenReturn(listTopicsResult);
    Mockito.when(listTopicsResult.names()).thenReturn(KafkaFuture.completedFuture(Set.of("name")));
    Mockito.when(adminClient.describeTopics(any(Collection.class)))
        .thenReturn(describeTopicsResult);
    Mockito.when(describeTopicsResult.allTopicNames())
        .thenReturn(KafkaFuture.completedFuture(topicDescriptionsPerAdminClient));
    Mockito.when(topicDescription.partitions()).thenReturn(List.of(topicPartitionInfo));

    LoadTopicsResponse topicConfigs =
        apacheKafkaTopicService.loadTopics(
            TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION, false);

    Assertions.assertThat(topicConfigs.getTopicConfigSet().size()).isEqualTo(1);
  }

  @Test
  void createTopicClientNull() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(null);

    AbstractThrowableAssert<?, ? extends Throwable> exception =
        assertThatThrownBy(() -> apacheKafkaTopicService.createTopic(clusterTopicRequest));

    exception.isInstanceOf(Exception.class);
    exception.hasMessage("Cannot connect to cluster.");
  }

  @Test
  void createTopic() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.MULTIPLE_PARTITIONS)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.createTopics(anyCollection())).thenReturn(createTopicsResult);
    Mockito.when(createTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, KafkaFuture.completedFuture(null)));

    ApiResponse response = apacheKafkaTopicService.createTopic(clusterTopicRequest);

    Assertions.assertThat(response.isSuccess()).isTrue();
    Assertions.assertThat(response.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  void createTopicTopicExistsException() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.MULTIPLE_PARTITIONS)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .build();
    Map<String, TopicDescription> topicDescriptionsPerAdminClient =
        new HashMap<>(Map.of(TestConstants.TOPIC_NAME, topicDescription));
    Exception expected = new ExecutionException(new TopicExistsException(TestConstants.TOPIC_NAME));

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.createTopics(anyCollection())).thenReturn(createTopicsResult);
    Mockito.when(createTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, kafkaFuture));
    Mockito.when(kafkaFuture.get(10L, TimeUnit.SECONDS)).thenThrow(expected);
    Mockito.when(adminClient.describeTopics(any(Collection.class)))
        .thenReturn(describeTopicsResult);
    Mockito.when(describeTopicsResult.all())
        .thenReturn(KafkaFuture.completedFuture(topicDescriptionsPerAdminClient));
    Mockito.when(topicDescription.partitions()).thenReturn(List.of(topicPartitionInfo));

    ApiResponse response = apacheKafkaTopicService.createTopic(clusterTopicRequest);

    Assertions.assertThat(response.isSuccess()).isFalse();
    Assertions.assertThat(response.getMessage()).isEqualTo(expected.getMessage());
  }

  @Test
  void createTopicTopicExistsWithSameConfig() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.SINGLE_PARTITION)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .build();
    Map<String, TopicDescription> topicDescriptionsPerAdminClient =
        new HashMap<>(Map.of(TestConstants.TOPIC_NAME, topicDescription));
    Exception expected = new ExecutionException(new TopicExistsException(TestConstants.TOPIC_NAME));

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.createTopics(anyCollection())).thenReturn(createTopicsResult);
    Mockito.when(createTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, kafkaFuture));
    Mockito.when(kafkaFuture.get(10L, TimeUnit.SECONDS)).thenThrow(expected);
    Mockito.when(adminClient.describeTopics(any(Collection.class)))
        .thenReturn(describeTopicsResult);
    Mockito.when(describeTopicsResult.all())
        .thenReturn(KafkaFuture.completedFuture(topicDescriptionsPerAdminClient));
    Mockito.when(topicDescription.partitions()).thenReturn(List.of(topicPartitionInfo));
    Mockito.when(topicPartitionInfo.replicas()).thenReturn(List.of(mock(Node.class)));

    ApiResponse response = apacheKafkaTopicService.createTopic(clusterTopicRequest);

    Assertions.assertThat(response.isSuccess()).isTrue();
    Assertions.assertThat(response.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @ParameterizedTest
  @MethodSource("exceptionProviderForCreateTopic")
  void createTopicException(Exception expected) throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.MULTIPLE_PARTITIONS)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.createTopics(anyCollection())).thenReturn(createTopicsResult);
    Mockito.when(createTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, kafkaFuture));
    Mockito.when(kafkaFuture.get(10L, TimeUnit.SECONDS)).thenThrow(expected);

    AbstractThrowableAssert<?, ? extends Throwable> exception =
        assertThatThrownBy(() -> apacheKafkaTopicService.createTopic(clusterTopicRequest));
    exception.isEqualTo(expected);
  }

  @Test
  void updateTopicClientNull() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(null);

    AbstractThrowableAssert<?, ? extends Throwable> exception =
        assertThatThrownBy(() -> apacheKafkaTopicService.updateTopic(clusterTopicRequest));

    exception.isInstanceOf(Exception.class);
    exception.hasMessage("Cannot connect to cluster.");
  }

  @Test
  void updateTopic() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.MULTIPLE_PARTITIONS)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .build();
    Map<String, TopicDescription> topicDescriptionsPerAdminClient =
        new HashMap<>(Map.of(TestConstants.TOPIC_NAME, topicDescription));

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.describeTopics(any(Collection.class)))
        .thenReturn(describeTopicsResult);
    Mockito.when(describeTopicsResult.all())
        .thenReturn(KafkaFuture.completedFuture(topicDescriptionsPerAdminClient));
    Mockito.when(topicDescription.partitions()).thenReturn(List.of(topicPartitionInfo));
    Mockito.when(adminClient.createPartitions(anyMap())).thenReturn(null);

    ApiResponse response = apacheKafkaTopicService.updateTopic(clusterTopicRequest);

    Assertions.assertThat(response.isSuccess()).isTrue();
    Assertions.assertThat(response.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  void deleteTopicClientNull() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(null);

    AbstractThrowableAssert<?, ? extends Throwable> exception =
        assertThatThrownBy(() -> apacheKafkaTopicService.deleteTopic(clusterTopicRequest));

    exception.isInstanceOf(Exception.class);
    exception.hasMessage("Cannot connect to cluster.");
  }

  @Test
  void deleteTopic() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.SINGLE_PARTITION)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .deleteAssociatedSchema(true)
            .build();
    ApiResponse schemaApiResponse =
        ApiResponse.builder().message(ApiResultStatus.SUCCESS.value).build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.deleteTopics(any(Collection.class))).thenReturn(deleteTopicsResult);
    Mockito.when(deleteTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, KafkaFuture.completedFuture(null)));
    Mockito.when(schemaService.deleteSchema(clusterTopicRequest)).thenReturn(schemaApiResponse);

    ApiResponse response = apacheKafkaTopicService.deleteTopic(clusterTopicRequest);

    Assertions.assertThat(response.isSuccess()).isTrue();
    Assertions.assertThat(response.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  void deleteTopicUnknownTopicOrPartitionException() throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.SINGLE_PARTITION)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .deleteAssociatedSchema(true)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.deleteTopics(any(Collection.class))).thenReturn(deleteTopicsResult);
    Mockito.when(deleteTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, kafkaFuture));
    Mockito.when(kafkaFuture.get(10L, TimeUnit.SECONDS))
        .thenThrow(new InterruptedException("UnknownTopicOrPartition"));

    ApiResponse response = apacheKafkaTopicService.deleteTopic(clusterTopicRequest);

    Assertions.assertThat(response.isSuccess()).isTrue();
    Assertions.assertThat(response.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @ParameterizedTest
  @MethodSource("exceptionProviderForDeleteTopic")
  void deleteTopicException(Exception expected) throws Exception {
    ClusterTopicRequest clusterTopicRequest =
        ClusterTopicRequest.builder()
            .env(TestConstants.ENVIRONMENT)
            .clusterName(TestConstants.CLUSTER_IDENTIFICATION)
            .protocol(protocol)
            .topicName(TestConstants.TOPIC_NAME)
            .partitions(TestConstants.SINGLE_PARTITION)
            .replicationFactor(TestConstants.REPLICATION_FACTOR)
            .advancedTopicConfiguration(TestConstants.ADVANCED_TOPIC_CONFIGURATION)
            .deleteAssociatedSchema(true)
            .build();

    Mockito.when(
            clusterApiUtils.getAdminClient(
                TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION))
        .thenReturn(adminClient);
    Mockito.when(clusterApiUtils.getAdminClientProperties()).thenReturn(adminClientProperties);
    Mockito.when(adminClientProperties.getTopicsTimeoutSecs()).thenReturn(10L);
    Mockito.when(adminClient.deleteTopics(any(Collection.class))).thenReturn(deleteTopicsResult);
    Mockito.when(deleteTopicsResult.values())
        .thenReturn(Map.of(TestConstants.TOPIC_NAME, kafkaFuture));
    Mockito.when(kafkaFuture.get(10L, TimeUnit.SECONDS)).thenThrow(expected);

    AbstractThrowableAssert<?, ? extends Throwable> exception =
        assertThatThrownBy(() -> apacheKafkaTopicService.deleteTopic(clusterTopicRequest));
    exception.isEqualTo(expected);
  }
}
