package io.aiven.klaw.clusterapi.services;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import io.aiven.klaw.clusterapi.UtilMethods;
import io.aiven.klaw.clusterapi.models.ApiResponse;
import io.aiven.klaw.clusterapi.models.ClusterSchemaRequest;
import io.aiven.klaw.clusterapi.models.RegisterSchemaCustomResponse;
import io.aiven.klaw.clusterapi.models.RegisterSchemaResponse;
import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.MethodOrderer;
import org.junit.jupiter.api.Order;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestMethodOrder;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.web.client.RestTemplate;

@ExtendWith(SpringExtension.class)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class SchemaServiceRegisterSchemaTest {

  public static final String CUSTOM_COMP = "CustomComp";
  public static final String FULL_SCHEMA = "Nonsense Schema";
  public static final String ENV = "unittest";
  public static final String TOPIC_NAME = "topic-1";
  public static final String CLUSTER_IDENTIFICATION = "1";
  public static final String REGISTRY_URL = "https:registryEtc";
  public static final String COMPATIBILITY_NODE_KEY = "compatibility";
  SchemaService schemaService;
  @Captor ArgumentCaptor<HttpEntity<Map<String, String>>> schemaCompatibility;
  @Mock private RestTemplate restTemplate;
  @Mock private ClusterApiUtils clusterApiUtil;
  private UtilMethods utilMethods;

  @BeforeEach
  public void setUp() {
    schemaService = new SchemaService(clusterApiUtil);
    utilMethods = new UtilMethods();
  }

  @Test
  @Order(1)
  public void givenSchemaWithForceRegisterEnabled_setAndResetSchemaCompatability() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.OK));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called twice once to change to null second to change it back.
    verify(restTemplate, times(2)).put(any(), schemaCompatibility.capture(), eq(String.class));
    // 1 call to get the current SchemaCompatibility
    verify(restTemplate, times(1))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
    // Sets it to None.
    assertThat(schemaCompatibility.getAllValues().get(0).getBody().get(COMPATIBILITY_NODE_KEY))
        .isEqualTo("NONE");
    // resets it to previous compatability
    assertThat(schemaCompatibility.getAllValues().get(1).getBody().get(COMPATIBILITY_NODE_KEY))
        .isEqualTo(CUSTOM_COMP);
  }

  @Test
  @Order(2)
  public void givenSchemaWithForceRegisterEnabled_404ReturnedAndResetSchemaCompatability() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.NOT_FOUND));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called twice once to change to null second to change it back.
    verify(restTemplate, times(2)).put(any(), schemaCompatibility.capture(), eq(String.class));
    // 1 call to get the current SchemaCompatibility
    verify(restTemplate, times(1))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
    // Sets it to None.
    assertThat(schemaCompatibility.getAllValues().get(0).getBody().get(COMPATIBILITY_NODE_KEY))
        .isEqualTo("NONE");
    // resets it to previous compatability
    assertThat(schemaCompatibility.getAllValues().get(1).getBody().get(COMPATIBILITY_NODE_KEY))
        .isEqualTo(CUSTOM_COMP);
  }

  @Test
  @Order(3)
  public void givenSchemaWithForceRegisterEnabled_ConflictReturnedAndResetSchemaCompatability() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.CONFLICT));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called twice once to change to null second to change it back.
    verify(restTemplate, times(2)).put(any(), schemaCompatibility.capture(), eq(String.class));
    // 1 call to get the current SchemaCompatibility
    verify(restTemplate, times(1))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
    // Sets it to None.
    assertThat(schemaCompatibility.getAllValues().get(0).getBody().get(COMPATIBILITY_NODE_KEY))
        .isEqualTo("NONE");
    // resets it to previous compatability
    assertThat(schemaCompatibility.getAllValues().get(1).getBody().get(COMPATIBILITY_NODE_KEY))
        .isEqualTo(CUSTOM_COMP);
  }

  @Test
  @Order(4)
  public void givenSchemaWithoutForceRegisterEnabled_setAndResetSchemaCompatability() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(false)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.OK));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called twice once to change to null second to change it back.
    verify(restTemplate, times(0)).put(any(), any(), eq(String.class));
    // 1 call to get the current SchemaCompatibility as isForce==false
    verify(restTemplate, times(1))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
  }

  @Test
  @Order(5)
  public void givenSchemaWithoutForceRegisterEnabled_404ReturnedAndResetSchemaCompatability() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(false)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.NOT_FOUND));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called twice once to change to null second to change it back.
    verify(restTemplate, times(0)).put(any(), any(), eq(String.class));
    // 1 call to get the current SchemaCompatibility as isForce==false
    verify(restTemplate, times(1))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
  }

  @Test
  @Order(6)
  public void givenSchemaWithoutForceRegisterEnabled_ConflictReturnedAndResetSchemaCompatability() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(false)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.CONFLICT));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called twice once to change to null second to change it back.
    verify(restTemplate, times(0)).put(any(), any(), eq(String.class));
    // 1 call to get the current SchemaCompatibility as isForce==false
    verify(restTemplate, times(1))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
  }

  @Test
  @Order(7)
  public void
      givenSchemaWithForceRegisterEnabled_GetSchemaCompatibilityException_applyGlobalCompatibility() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class)))
        .thenReturn(createErrorCompatibilityResponseEntity(HttpStatus.UNAUTHORIZED));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called by setting global compatibility.
    verify(restTemplate, times(1)).put(any(), schemaCompatibility.capture(), eq(String.class));
    // 1 call to get the current SchemaCompatibility
    verify(restTemplate, times(2))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
  }

  @Test
  @Order(8)
  public void
      givenSchemaWithForceRegisterEnabled_ExistingCompatibilityIsFailValidationAndGetGlobalCompatibility() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();
    // If null returns default schema should be reverted to.
    when(restTemplate.exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class)))
        .thenReturn(createCompatibilityResponseEntity(null));
    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.OK));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility called once to change to none.
    verify(restTemplate, times(1)).put(any(), schemaCompatibility.capture(), eq(String.class));
    // 2 calls to get the SchemaCompatibility subject, global
    verify(restTemplate, times(2))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
  }

  @Test
  @Order(9)
  public void givenSchemaWithForceRegisterEnabled_NotFoundFallBackToGlobalSchemaSettingReturned() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.OK));

    when(restTemplate.exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class)))
        .thenReturn(createErrorCompatibilityResponseEntity(HttpStatus.NOT_FOUND));

    schemaService.registerSchema(schemaReq);
    verify(restTemplate, times(1)).put(any(), schemaCompatibility.capture(), eq(String.class));
  }

  @Test
  @Order(10)
  public void givenSchemaWithForceRegisterEnabled_NotFoundGetGlobalCompatibility() {

    ClusterSchemaRequest schemaReq =
        ClusterSchemaRequest.builder()
            .forceRegister(true)
            .fullSchema(FULL_SCHEMA)
            .clusterIdentification(CLUSTER_IDENTIFICATION)
            .protocol(KafkaSupportedProtocol.SSL)
            .env(ENV)
            .topicName(TOPIC_NAME)
            .build();
    mockGetSchemaCompatibility();

    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));

    when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
        .thenReturn(createSchemaRegisterResponseEntity(HttpStatus.OK));
    when(restTemplate.exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class)))
        .thenReturn(createErrorCompatibilityResponseEntity(HttpStatus.NOT_FOUND))
        .thenReturn(createErrorCompatibilityResponseEntity(HttpStatus.NOT_FOUND));
    when(restTemplate.exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class)))
        .thenReturn(createErrorCompatibilityResponseEntity(HttpStatus.NOT_FOUND))
        .thenReturn(createErrorCompatibilityResponseEntity(HttpStatus.NOT_FOUND));
    schemaService.registerSchema(schemaReq);
    // change schema compatibility to NONE, once
    verify(restTemplate, times(1)).put(any(), any(), eq(String.class));

    // 2 calls to get the current SchemaCompatibility (subject, global) as isForce==false
    verify(restTemplate, times(2))
        .exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class));
  }

  @Test
  public void registerNewSchema() {
    int id = 1;
    ClusterSchemaRequest clusterSchemaRequest = utilMethods.getSchema();
    RegisterSchemaResponse registerSchemaResponse = new RegisterSchemaResponse();
    registerSchemaResponse.setId(id);
    ResponseEntity<Set<Integer>> response = new ResponseEntity<>(Set.of(id), HttpStatus.OK);

    when(clusterApiUtil.getRequestDetails(any(), any())).thenReturn(Pair.of("", restTemplate));
    when(clusterApiUtil.createHeaders(anyString(), any())).thenReturn(new HttpHeaders());
    when(restTemplate.postForEntity(anyString(), any(), eq(RegisterSchemaResponse.class)))
        .thenReturn(new ResponseEntity<>(registerSchemaResponse, HttpStatus.OK));
    when(restTemplate.exchange(
            anyString(),
            eq(HttpMethod.GET),
            any(),
            eq(new ParameterizedTypeReference<Set<Integer>>() {}),
            anyMap()))
        .thenReturn(response);

    ApiResponse resultResp = schemaService.registerSchema(clusterSchemaRequest);
    assertThat(resultResp.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    RegisterSchemaCustomResponse registerSchemaCustomResponse =
        (RegisterSchemaCustomResponse) resultResp.getData();
    assertThat(registerSchemaCustomResponse.isSchemaRegistered()).isTrue();
    assertThat(registerSchemaCustomResponse.getVersion()).isEqualTo(1);
    assertThat(registerSchemaCustomResponse.getId()).isEqualTo(id);
  }

  @Test
  public void registerSchemaFailure() {
    ClusterSchemaRequest clusterSchemaRequest = utilMethods.getSchema();

    when(clusterApiUtil.getRequestDetails(any(), any())).thenReturn(Pair.of("", restTemplate));
    when(clusterApiUtil.createHeaders(anyString(), any())).thenReturn(new HttpHeaders());
    when(restTemplate.postForEntity(anyString(), any(), eq(RegisterSchemaResponse.class)))
        .thenThrow(new RuntimeException("Unable to connect"));

    ApiResponse resultResp = schemaService.registerSchema(clusterSchemaRequest);
    assertThat(resultResp.getMessage())
        .contains("Failure in registering schema.")
        .contains("Unable to connect");
  }

  private ResponseEntity<Map<String, String>> createErrorCompatibilityResponseEntity(
      HttpStatus status) {
    return ResponseEntity.status(status).build();
  }

  private ResponseEntity<String> createSchemaRegisterResponseEntity(HttpStatus status) {

    return ResponseEntity.status(status).body("Completed API Call.");
  }

  private void mockGetSchemaCompatibility() {
    when(clusterApiUtil.getRequestDetails(any(), any()))
        .thenReturn(Pair.of(REGISTRY_URL, restTemplate));
    when(clusterApiUtil.createHeaders(any(), any())).thenReturn(new HttpHeaders());
    Map<String, String> params = new HashMap<>();

    when(restTemplate.exchange(
            eq(REGISTRY_URL),
            any(HttpMethod.class),
            any(HttpEntity.class),
            eq(new ParameterizedTypeReference<Map<String, String>>() {}),
            any(HashMap.class)))
        .thenReturn(createCompatibilityResponseEntity(CUSTOM_COMP));
  }

  private ResponseEntity<Map<String, String>> createCompatibilityResponseEntity(
      String compatiblityType) {
    Map<String, String> map = new HashMap<>();
    map.put("compatibilityLevel", compatiblityType);
    return ResponseEntity.ok(map);
  }
}
