package io.aiven.klaw.clusterapi.services;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.when;
import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
import static org.springframework.test.web.client.response.MockRestResponseCreators.withBadRequest;
import static org.springframework.test.web.client.response.MockRestResponseCreators.withRawStatus;
import static org.springframework.test.web.client.response.MockRestResponseCreators.withSuccess;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.aiven.klaw.clusterapi.UtilMethods;
import io.aiven.klaw.clusterapi.models.ApiResponse;
import io.aiven.klaw.clusterapi.models.ClusterConnectorRequest;
import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
import io.aiven.klaw.clusterapi.models.enums.KafkaClustersType;
import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
import io.aiven.klaw.clusterapi.models.error.RestErrorResponse;
import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
import java.util.Collections;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.client.RestClientTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.test.web.client.MockRestServiceServer;
import org.springframework.web.client.RestTemplate;

@RestClientTest(KafkaConnectService.class)
class KafkaConnectServiceTest {

  public static final String THIS_IS_A_MISCONFIGURED_CONNECTOR =
      "This is a misconfigured connector";
  @Autowired KafkaConnectService kafkaConnectService;

  RestTemplate restTemplate;
  @Autowired ObjectMapper objectMapper;
  private MockRestServiceServer mockRestServiceServer;
  @MockBean private ClusterApiUtils getAdminClient;

  private UtilMethods utilMethods;

  @BeforeEach
  public void setUp() {
    utilMethods = new UtilMethods();
    restTemplate = new RestTemplate();
    kafkaConnectService = new KafkaConnectService(getAdminClient);
    mockRestServiceServer = MockRestServiceServer.bindTo(restTemplate).build();
  }

  // TODO need to add proper return value
  @Test
  public void getConnectors_returnList() throws JsonProcessingException {
    when(getAdminClient.getRequestDetails(any(), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors", restTemplate));
    this.mockRestServiceServer
        .expect(requestTo("/env/connectors"))
        .andRespond(
            withSuccess(
                objectMapper.writeValueAsString(utilMethods.getConnectorsListMap()),
                MediaType.APPLICATION_JSON));
    assertThat(
            kafkaConnectService
                .getConnectors("env", KafkaSupportedProtocol.PLAINTEXT, "CLID1", true)
                .getConnectorStateList())
        .isNotEmpty();
  }

  // TODO need to add proper return value
  @Test
  public void getConnectorDetails_returnMap() throws JsonProcessingException {

    // Connector operations
    when(getAdminClient.getRequestDetails(
            eq("env/connectors/conn1"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors/conn1", restTemplate));

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(
            withSuccess(
                objectMapper.writeValueAsString(Collections.singletonMap("conn1", "conn2")),
                MediaType.APPLICATION_JSON));

    assertThat(
            kafkaConnectService.getConnectorDetails(
                "conn1", "env", KafkaSupportedProtocol.PLAINTEXT, "CLID1"))
        .isNotEmpty();
  }

  @Test
  public void createConnector_bad_request() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();
    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(
            withBadRequest()
                .contentType(MediaType.APPLICATION_JSON)
                .body(getRestErrorResponse(THIS_IS_A_MISCONFIGURED_CONNECTOR)));
    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);

    assertThat(connectorResponse.isSuccess()).isFalse();
    assertThat(connectorResponse.getMessage()).isEqualTo(THIS_IS_A_MISCONFIGURED_CONNECTOR);
  }

  private String getRestErrorResponse(String resp) throws JsonProcessingException {
    RestErrorResponse restErrorResponse = new RestErrorResponse();
    restErrorResponse.setErrorCode(400);
    restErrorResponse.setMessage(resp);
    return objectMapper.writeValueAsString(restErrorResponse);
  }

  @Test
  public void createConnector_success() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isTrue();
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  public void createConnector_fail() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(withRawStatus(404).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isFalse();
    assertThat(connectorResponse.getMessage()).isEqualTo("Unable to create Connector on Cluster.");
  }

  @Test
  public void updateConnector_success() {
    ClusterConnectorRequest connectorRequest = stubUpdateConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1/config"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.updateConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isTrue();
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  public void updateConnector_badRequest() throws JsonProcessingException {
    ClusterConnectorRequest connectorRequest = stubUpdateConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1/config"))
        .andRespond(
            withRawStatus(400)
                .contentType(MediaType.APPLICATION_JSON)
                .body(getRestErrorResponse(THIS_IS_A_MISCONFIGURED_CONNECTOR)));
    ApiResponse connectorResponse = kafkaConnectService.updateConnector(connectorRequest);

    assertThat(connectorResponse.isSuccess()).isFalse();
    assertThat(connectorResponse.getMessage()).isEqualTo(THIS_IS_A_MISCONFIGURED_CONNECTOR);
  }

  @Test
  public void deleteConnector_badRequest() throws JsonProcessingException {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(
            withRawStatus(400)
                .contentType(MediaType.APPLICATION_JSON)
                .body(getRestErrorResponse(THIS_IS_A_MISCONFIGURED_CONNECTOR)));
    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);

    assertThat(connectorResponse.isSuccess()).isFalse();
    assertThat(connectorResponse.getMessage()).isEqualTo(THIS_IS_A_MISCONFIGURED_CONNECTOR);
  }

  @Test
  public void deleteConnector_badRequest_undetermined_response() throws JsonProcessingException {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(
            withRawStatus(400)
                .contentType(MediaType.APPLICATION_JSON)
                .body(THIS_IS_A_MISCONFIGURED_CONNECTOR));
    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);

    assertThat(connectorResponse.isSuccess()).isFalse();
    assertThat(connectorResponse.getMessage()).isEqualTo("Unable To Delete Connector on Cluster.");
  }

  @Test
  public void deleteConnector_success() {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isTrue();
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  public void createConnectorMultiEnvs_success() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnectorMultiUrl();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors"))
        .andRespond(withRawStatus(404).contentType(MediaType.APPLICATION_JSON));

    this.mockRestServiceServer
        .expect(requestTo("/env1/connectors"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isTrue();
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    mockRestServiceServer.verify();
  }

  @Test
  public void deleteConnectorMultiEnvs_success() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnectorMultiUrl();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(withRawStatus(404).contentType(MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/env1/connectors/conn1"))
        .andRespond(withRawStatus(500).contentType(MediaType.APPLICATION_JSON));

    this.mockRestServiceServer
        .expect(requestTo("/env2/connectors/conn1"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));

    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    mockRestServiceServer.verify();
  }

  @Test
  public void updateConnector_WithMulti_url_success() {
    ClusterConnectorRequest connectorRequest = stubUpdateConnectorMultiUrl();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1/config"))
        .andRespond(withRawStatus(401).contentType(MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/env1/connectors/conn1/config"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));

    ApiResponse connectorResponse = kafkaConnectService.updateConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isTrue();
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  public void createConnectorMultiEnvs_failure() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnectorMultiUrl();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors"))
        .andRespond(withRawStatus(404).contentType(MediaType.APPLICATION_JSON));

    this.mockRestServiceServer
        .expect(requestTo("/env1/connectors"))
        .andRespond(withRawStatus(201).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.postNewConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isTrue();
    assertThat(connectorResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    mockRestServiceServer.verify();
  }

  @Test
  public void deleteConnectorMultiEnvs_failure() throws Exception {
    ClusterConnectorRequest connectorRequest = stubCreateOrDeleteConnectorMultiUrl();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1"))
        .andRespond(withRawStatus(404).contentType(MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/env1/connectors/conn1"))
        .andRespond(withRawStatus(500).contentType(MediaType.APPLICATION_JSON));

    this.mockRestServiceServer
        .expect(requestTo("/env2/connectors/conn1"))
        .andRespond(withRawStatus(401).contentType(MediaType.APPLICATION_JSON));
    ApiResponse connectorResponse = kafkaConnectService.deleteConnector(connectorRequest);
    assertThat(connectorResponse.getMessage()).isEqualTo("Unable To Delete Connector on Cluster.");
    mockRestServiceServer.verify();
  }

  @Test
  public void updateConnector_WithMulti_url_failure() {
    ClusterConnectorRequest connectorRequest = stubUpdateConnector();

    this.mockRestServiceServer
        .expect(requestTo("/env/connectors/conn1/config"))
        .andRespond(withRawStatus(401).contentType(MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/env1/connectors/conn1/config"))
        .andRespond(withRawStatus(401).contentType(MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/env2/connectors/conn1/config"))
        .andRespond(withRawStatus(401).contentType(MediaType.APPLICATION_JSON));

    ApiResponse connectorResponse = kafkaConnectService.updateConnector(connectorRequest);
    assertThat(connectorResponse.isSuccess()).isFalse();
    assertThat(connectorResponse.getMessage()).isEqualTo("Unable to update Connector on Cluster");
  }

  private ClusterConnectorRequest stubCreateOrDeleteConnector() {
    when(getAdminClient.getRequestDetails(any(), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors/conn1", restTemplate));
    when(getAdminClient.createHeaders(eq("1"), eq(KafkaClustersType.KAFKA_CONNECT)))
        .thenReturn(new HttpHeaders());
    ClusterConnectorRequest connectorRequest =
        ClusterConnectorRequest.builder()
            .connectorName("conn1")
            .clusterIdentification("1")
            .env("env")
            .protocol(KafkaSupportedProtocol.PLAINTEXT)
            .build();
    return connectorRequest;
  }

  private ClusterConnectorRequest stubCreateOrDeleteConnectorMultiUrl() {
    when(getAdminClient.getRequestDetails(
            eq("env/connectors"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq("env1/connectors"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env1/connectors", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq("env2/connectors"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env2/connectors", restTemplate));
    // Connector operations
    when(getAdminClient.getRequestDetails(
            eq("env/connectors/conn1"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors/conn1", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq("env1/connectors/conn1"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env1/connectors/conn1", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq("env2/connectors/conn1"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env2/connectors/conn1", restTemplate));
    when(getAdminClient.createHeaders(eq("1"), eq(KafkaClustersType.KAFKA_CONNECT)))
        .thenReturn(new HttpHeaders());
    ClusterConnectorRequest connectorRequest =
        ClusterConnectorRequest.builder()
            .connectorName("conn1")
            .clusterIdentification("1")
            .env("env,env1,env2")
            .protocol(KafkaSupportedProtocol.PLAINTEXT)
            .build();
    return connectorRequest;
  }

  private ClusterConnectorRequest stubUpdateConnector() {
    when(getAdminClient.getRequestDetails(
            eq("env/connectors/conn1/config"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors/conn1/config", restTemplate));
    when(getAdminClient.createHeaders(eq("1"), eq(KafkaClustersType.KAFKA_CONNECT)))
        .thenReturn(new HttpHeaders());
    ClusterConnectorRequest connectorRequest =
        ClusterConnectorRequest.builder()
            .connectorName("conn1")
            .clusterIdentification("1")
            .env("env")
            .protocol(KafkaSupportedProtocol.PLAINTEXT)
            .build();
    return connectorRequest;
  }

  private ClusterConnectorRequest stubUpdateConnectorMultiUrl() {
    when(getAdminClient.getRequestDetails(
            eq("env/connectors/conn1/config"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env/connectors/conn1/config", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq("env1/connectors/conn1/config"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env1/connectors/conn1/config", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq("env2/connectors/conn1/config"), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of("/env2/connectors/conn1/config", restTemplate));

    when(getAdminClient.createHeaders(eq("1"), eq(KafkaClustersType.KAFKA_CONNECT)))
        .thenReturn(new HttpHeaders());
    ClusterConnectorRequest connectorRequest =
        ClusterConnectorRequest.builder()
            .connectorName("conn1")
            .clusterIdentification("1")
            .env("env,env1,env2")
            .protocol(KafkaSupportedProtocol.PLAINTEXT)
            .build();
    return connectorRequest;
  }
}
