package io.aiven.klaw.service;

import static io.aiven.klaw.service.MailUtils.MailType.SCHEMA_APPROVED_NOTIFY_SUBSCRIBERS;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.aiven.klaw.UtilMethods;
import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.EnvTag;
import io.aiven.klaw.dao.KwClusters;
import io.aiven.klaw.dao.SchemaRequest;
import io.aiven.klaw.dao.Topic;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.helpers.db.rdbms.HandleDbRequestsJdbc;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.enums.ApiResultStatus;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.PermissionType;
import io.aiven.klaw.model.enums.RequestOperationType;
import io.aiven.klaw.model.enums.RequestStatus;
import io.aiven.klaw.model.requests.SchemaPromotion;
import io.aiven.klaw.model.requests.SchemaRequestModel;
import io.aiven.klaw.model.response.SchemaRequestsResponseModel;
import java.sql.Timestamp;
import java.util.*;
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.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.security.test.context.support.WithMockUser;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.test.util.ReflectionTestUtils;

@ExtendWith(SpringExtension.class)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class SchemaRegistryControllerServiceTest {

  public static final String TESTTOPIC = "topic-1";
  public static final String VALIDATION_FAILURE_MSG =
      ApiResultStatus.FAILURE.value + "  Schema is not compatible.";
  public static final String VALIDATION_SUCCESS_MSG =
      ApiResultStatus.SUCCESS.value + " Schema is compatible.";
  @Mock private UserDetails userDetails;

  @Mock private HandleDbRequestsJdbc handleDbRequests;

  @Mock private MailUtils mailService;

  @Mock private ManageDatabase manageDatabase;

  @Mock private UserInfo userInfo;

  @Mock private ClusterApiService clusterApiService;

  @Mock CommonUtilsService commonUtilsService;

  @Mock RolesPermissionsControllerService rolesPermissionsControllerService;

  private SchemaRegistryControllerService schemaRegistryControllerService;

  private ObjectMapper mapper = new ObjectMapper();

  private Env env;
  private UtilMethods utilMethods;

  @Captor private ArgumentCaptor<SchemaRequest> schemaRequestCaptor;

  @BeforeEach
  public void setUp() throws Exception {
    this.env = new Env();
    env.setId("1");
    env.setName("DEV");

    schemaRegistryControllerService =
        new SchemaRegistryControllerService(clusterApiService, mailService);
    ReflectionTestUtils.setField(schemaRegistryControllerService, "manageDatabase", manageDatabase);
    ReflectionTestUtils.setField(schemaRegistryControllerService, "mailService", mailService);
    ReflectionTestUtils.setField(
        schemaRegistryControllerService, "commonUtilsService", commonUtilsService);
    ReflectionTestUtils.setField(
        schemaRegistryControllerService,
        "rolesPermissionsControllerService",
        rolesPermissionsControllerService);

    when(manageDatabase.getHandleDbRequests()).thenReturn(handleDbRequests);
    Boolean validateOnSave = true;
    ReflectionTestUtils.setField(
        schemaRegistryControllerService, "validateCompatiblityOnSave", validateOnSave);
    utilMethods = new UtilMethods();
    when(commonUtilsService.getPrincipal()).thenReturn(userDetails);
    when(commonUtilsService.isNotAuthorizedUser(any(), any(PermissionType.class))).thenReturn(true);
  }

  @Test
  @Order(1)
  public void getSchemaRequests() {
    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getAllSchemaRequests(
            anyBoolean(),
            anyString(),
            anyInt(),
            eq(null),
            eq(null),
            eq(null),
            eq("all"),
            eq(null),
            eq(true),
            eq(false)))
        .thenReturn(getSchemasReqs());
    when(rolesPermissionsControllerService.getApproverRoles(anyString(), anyInt()))
        .thenReturn(Set.of(""));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(handleDbRequests.getAllUsersInfoForTeam(anyInt(), anyInt())).thenReturn(List.of(userInfo));
    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(this.env);
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("teamname");
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.APPROVE_ALL_REQUESTS_TEAMS))
        .thenReturn(false);

    List<SchemaRequestsResponseModel> listReqs =
        schemaRegistryControllerService.getSchemaRequests(
            "1",
            "",
            "all",
            null,
            true,
            null,
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            false);
    assertThat(listReqs).hasSize(2);
  }

  @Test
  @Order(2)
  public void deleteSchemaRequestsSuccess() throws KlawException {
    int schemaReqId = 1001;

    stubUserInfo();
    when(handleDbRequests.deleteSchemaRequest(anyInt(), anyString(), anyInt()))
        .thenReturn(ApiResultStatus.SUCCESS.value);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_DELETE_SCHEMAS))
        .thenReturn(false);
    ApiResponse resultResp = schemaRegistryControllerService.deleteSchemaRequests("" + schemaReqId);
    assertThat(resultResp.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @Order(3)
  public void deleteSchemaRequestsFailure() {
    int schemaReqId = 1001;

    stubUserInfo();
    when(handleDbRequests.deleteSchemaRequest(anyInt(), anyString(), anyInt()))
        .thenThrow(new RuntimeException("Error from Schema upload"));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_DELETE_SCHEMAS))
        .thenReturn(false);

    KlawException ex =
        assertThrows(
            KlawException.class,
            () -> schemaRegistryControllerService.deleteSchemaRequests("" + schemaReqId));
    assertThat(ex.getMessage()).contains("Error from Schema upload");
  }

  @Test
  @Order(4)
  public void execSchemaRequestsSuccess() throws KlawException {
    int schemaReqId = 1001;
    Map<String, Object> registerSchemaCustomResponse = new HashMap<>();
    registerSchemaCustomResponse.put("schemaRegistered", true);
    registerSchemaCustomResponse.put("version", 1);
    registerSchemaCustomResponse.put("id", 1);
    registerSchemaCustomResponse.put("compatibility", "BACKWARD");

    ApiResponse apiResponse =
        ApiResponse.builder()
            .message("Schema registered id\": 215")
            .data(registerSchemaCustomResponse)
            .build();

    ResponseEntity<ApiResponse> response = new ResponseEntity<>(apiResponse, HttpStatus.OK);
    SchemaRequest schemaRequest = new SchemaRequest();
    schemaRequest.setSchemafull("schema..");
    schemaRequest.setRequestor("kwuserb");
    schemaRequest.setEnvironment("1");
    schemaRequest.setTopicname("topic");
    schemaRequest.setTeamId(1001);

    stubUserInfo();
    when(handleDbRequests.getSchemaRequest(anyInt(), anyInt())).thenReturn(schemaRequest);
    when(clusterApiService.postSchema(any(), anyString(), anyString(), anyInt()))
        .thenReturn(response);
    when(handleDbRequests.updateSchemaRequest(any(), anyString()))
        .thenReturn(ApiResultStatus.SUCCESS.value);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_SCHEMAS))
        .thenReturn(false);

    ApiResponse resultResp = schemaRegistryControllerService.execSchemaRequests("" + schemaReqId);
    assertThat(resultResp.isSuccess()).isTrue();
  }

  @Test
  @Order(5)
  public void execSchemaRequestsFailure1() throws KlawException {
    int schemaReqId = 1001;

    ApiResponse apiResponse = ApiResponse.notOk("Schema not registered");
    ResponseEntity<ApiResponse> response = new ResponseEntity<>(apiResponse, HttpStatus.OK);
    SchemaRequest schemaRequest = new SchemaRequest();
    schemaRequest.setSchemafull("schema..");
    schemaRequest.setRequestor("kwuserb");
    schemaRequest.setEnvironment("1");
    schemaRequest.setTopicname("topic");

    stubUserInfo();
    when(handleDbRequests.getSchemaRequest(anyInt(), anyInt())).thenReturn(schemaRequest);
    when(clusterApiService.postSchema(any(), anyString(), anyString(), anyInt()))
        .thenReturn(response);
    when(handleDbRequests.updateSchemaRequest(any(), anyString()))
        .thenReturn(ApiResultStatus.SUCCESS.value);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_SCHEMAS))
        .thenReturn(false);

    ApiResponse resultResp = schemaRegistryControllerService.execSchemaRequests("" + schemaReqId);
    assertThat(resultResp.getMessage()).contains("Schema not registered");
  }

  @Test()
  @Order(6)
  public void execSchemaRequestsFailure2() {
    int schemaReqId = 1001;

    ApiResponse apiResponse = ApiResponse.ok("Schema registered id\": 215");
    ResponseEntity<ApiResponse> response = new ResponseEntity<>(apiResponse, HttpStatus.OK);

    SchemaRequest schemaRequest = new SchemaRequest();
    schemaRequest.setSchemafull("schema..");
    schemaRequest.setRequestor("kwuserb");
    schemaRequest.setEnvironment("1");
    schemaRequest.setTopicname("topic");

    stubUserInfo();
    when(handleDbRequests.getSchemaRequest(anyInt(), anyInt())).thenReturn(schemaRequest);
    try {
      when(clusterApiService.postSchema(any(), anyString(), anyString(), anyInt()))
          .thenReturn(response);
    } catch (KlawException e) {
      throw new RuntimeException(e);
    }
    when(handleDbRequests.updateSchemaRequest(any(), anyString()))
        .thenThrow(new RuntimeException("Error in registering"));
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_SCHEMAS))
        .thenReturn(false);

    try {
      schemaRegistryControllerService.execSchemaRequests("" + schemaReqId);
    } catch (KlawException e) {
      assertThat(e.getMessage()).contains("Error in registering");
    }
  }

  @Test
  @Order(7)
  public void uploadSchemaSuccess() throws KlawException {
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    mockGetEnvironment();
    when(manageDatabase.getEnv(eq(101), eq(1))).thenReturn(Optional.of(createEnv(1)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);
    when(handleDbRequests.requestForSchema(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(topic));

    ApiResponse resultResp =
        schemaRegistryControllerService.uploadSchema(schemaRequest, RequestOperationType.CREATE);
    assertThat(resultResp.isSuccess()).isTrue();
  }

  @Test
  @Order(8)
  public void uploadSchemaFailure() throws KlawException {
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();

    stubUserInfo();
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);
    when(handleDbRequests.requestForSchema(any()))
        .thenThrow(new RuntimeException("Error from schema upload"));
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(topic));
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));

    try {
      schemaRegistryControllerService.uploadSchema(schemaRequest, RequestOperationType.CREATE);
    } catch (KlawException e) {
      assertThat(e.getMessage()).contains("Error from schema upload");
    }
  }

  @Test
  @Order(9)
  public void promoteSchemaNotAuthorized() throws Exception {
    // Make user unauthorized
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(true);
    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "1"));
    assertThat(returnedValue.getMessage()).isEqualTo(ApiResultStatus.NOT_AUTHORIZED.value);
  }

  @Test
  @WithMockUser(
      username = "kwusera",
      authorities = {"ADMIN", "USER"})
  @Order(10)
  public void promoteSchemaCanNotFindSourceEnvironmentSchema() throws Exception {
    //    mockGetEnvironment();
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(createTopic()));
    when(commonUtilsService.getTeamId(any())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);
    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "1"));
    assertThat(returnedValue.getMessage())
        .isEqualTo("Unable to find or access the source Schema Registry");
  }

  @Test
  @Order(12)
  public void promoteSchemaSuccess() throws Exception {
    mockGetEnvironment();
    mockSchema();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    mockSchemaCreation();
    when(manageDatabase.getEnv(eq(101), eq(8))).thenReturn(Optional.of(createEnv(8)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);

    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "1"));
    assertThat(returnedValue.getMessage())
        .isNotEqualTo("Unable to find or access the source Schema Registry");
    assertThat(returnedValue.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @Order(13)
  public void promoteSchemaEnsureCorrectSchemaSelectedV1Success() throws Exception {
    mockGetEnvironment();
    mockSchema();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    mockSchemaCreation();

    when(manageDatabase.getEnv(eq(101), eq(8))).thenReturn(Optional.of(createEnv(8)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);

    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "1"));
    assertThat(returnedValue.getMessage())
        .isNotEqualTo("Unable to find or access the source Schema Registry");
    assertThat(returnedValue.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    verify(handleDbRequests, times(1)).requestForSchema(schemaRequestCaptor.capture());
    assertThat(schemaRequestCaptor.getValue().getSchemaversion()).isEqualTo("1");
    JsonNode json = mapper.readTree(schemaRequestCaptor.getValue().getSchemafull());
    assertThat(json.get("name").asText()).isEqualTo("klawTestAvroV1");
  }

  @Test
  @Order(14)
  public void promoteSchemaEnsureCorrectSchemaSelectedV2Success() throws Exception {
    mockGetEnvironment();
    mockSchema();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    mockSchemaCreation();
    when(manageDatabase.getEnv(eq(101), eq(8))).thenReturn(Optional.of(createEnv(8)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);

    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "2"));
    assertThat(returnedValue.getMessage())
        .isNotEqualTo("Unable to find or access the source Schema Registry");
    assertThat(returnedValue.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    verify(handleDbRequests, times(1)).requestForSchema(schemaRequestCaptor.capture());
    assertThat(schemaRequestCaptor.getValue().getSchemaversion()).isEqualTo("2");
    JsonNode json = mapper.readTree(schemaRequestCaptor.getValue().getSchemafull());
    assertThat(json.get("name").asText()).isEqualTo("klawTestAvroV2");
  }

  @Test
  @Order(15)
  public void promoteSchemaEnsureCorrectSchemaSelectedV3Success() throws Exception {
    mockGetEnvironment();
    mockSchema();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    mockSchemaCreation();
    when(manageDatabase.getEnv(eq(101), eq(8))).thenReturn(Optional.of(createEnv(8)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);

    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "3"));
    assertThat(returnedValue.getMessage())
        .isNotEqualTo("Unable to find or access the source Schema Registry");
    assertThat(returnedValue.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);

    verify(handleDbRequests, times(1)).requestForSchema(schemaRequestCaptor.capture());
    assertThat(schemaRequestCaptor.getValue().getSchemaversion()).isEqualTo("3");
    JsonNode json = mapper.readTree(schemaRequestCaptor.getValue().getSchemafull());
    assertThat(json.get("name").asText()).isEqualTo("klawTestAvroV3");
  }

  @Test
  @Order(16)
  public void promoteSchemaEnsureCorrectSchemaSelectedV4Success() throws Exception {
    mockGetEnvironment();
    mockSchema();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    mockSchemaCreation();
    when(manageDatabase.getEnv(eq(101), eq(8))).thenReturn(Optional.of(createEnv(8)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);

    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "4"));
    assertThat(returnedValue.getMessage())
        .isNotEqualTo("Unable to find or access the source Schema Registry");
    assertThat(returnedValue.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    verify(handleDbRequests, times(1)).requestForSchema(schemaRequestCaptor.capture());
    assertThat(schemaRequestCaptor.getValue().getSchemaversion()).isEqualTo("4");
    JsonNode json = mapper.readTree(schemaRequestCaptor.getValue().getSchemafull());
    assertThat(json.get("name").asText()).isEqualTo("klawTestAvroV4");
  }

  @Test
  @Order(17)
  public void promoteSchemaWithInCompatibleSchemaReturnFailure() throws Exception {
    mockGetEnvironment();
    mockSchema();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(false));
    mockSchemaCreation();
    when(manageDatabase.getEnv(eq(101), eq(8))).thenReturn(Optional.of(createEnv(8)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);

    ApiResponse returnedValue =
        schemaRegistryControllerService.promoteSchema(buildPromoteSchemaRequest(false, "4"));
    assertThat(returnedValue.getMessage())
        .isNotEqualTo("Unable to find or access the source Schema Registry");
    assertThat(returnedValue.getMessage()).isEqualTo(VALIDATION_FAILURE_MSG);
    verify(clusterApiService, times(1))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(18)
  public void uploadSchemaIncompatibleSchemaError() throws KlawException {
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(false));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);
    mockGetEnvironment();
    when(manageDatabase.getEnv(eq(101), eq(1))).thenReturn(Optional.of(createEnv(1)));

    ApiResponse resultResp =
        schemaRegistryControllerService.uploadSchema(schemaRequest, RequestOperationType.CREATE);
    assertThat(resultResp.getMessage()).isEqualTo(VALIDATION_FAILURE_MSG);
    verify(clusterApiService, times(1))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(19)
  public void validateSchemaReturnSchemaError() throws KlawException {
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(false));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);

    ApiResponse resultResp = schemaRegistryControllerService.validateSchema(schemaRequest);
    assertThat(resultResp.getMessage()).isEqualTo(VALIDATION_FAILURE_MSG);
    verify(clusterApiService, times(1))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(19)
  public void validateSchemaReturnSchemaCompatible() throws KlawException {
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);

    ApiResponse resultResp = schemaRegistryControllerService.validateSchema(schemaRequest);
    assertThat(resultResp.getMessage()).isEqualTo(VALIDATION_SUCCESS_MSG);
    verify(clusterApiService, times(1))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(19)
  public void validateSchemaThrowError() throws KlawException {
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    String exceptionMsg = "Unable to contact the schema registry";
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenThrow(new KlawException(exceptionMsg));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);

    KlawException ex =
        assertThrows(
            KlawException.class,
            () -> schemaRegistryControllerService.validateSchema(schemaRequest));
    assertThat(ex.getMessage()).isEqualTo(exceptionMsg);
    verify(clusterApiService, times(1))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(20)
  public void validateSchema_NoValidationOnSaveIgnored_ReturnSuccess() throws KlawException {
    Object validateOnSave = false;
    ReflectionTestUtils.setField(
        schemaRegistryControllerService, "validateCompatiblityOnSave", validateOnSave);
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    String exceptionMsg = "Unable to contact the schema registry";
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));

    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);

    ApiResponse resultResp = schemaRegistryControllerService.validateSchema(schemaRequest);
    assertThat(resultResp.getMessage()).isEqualTo(VALIDATION_SUCCESS_MSG);

    verify(clusterApiService, times(1))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(21)
  public void uploadSchema_NoValidationOnSave() throws KlawException {
    Object validateOnSave = false;
    ReflectionTestUtils.setField(
        schemaRegistryControllerService, "validateCompatiblityOnSave", validateOnSave);
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);
    when(handleDbRequests.requestForSchema(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(topic));
    mockGetEnvironment();
    when(manageDatabase.getEnv(eq(101), eq(1))).thenReturn(Optional.of(createEnv(1)));

    ApiResponse resultResp =
        schemaRegistryControllerService.uploadSchema(schemaRequest, RequestOperationType.CREATE);
    assertThat(resultResp.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);

    verify(clusterApiService, times(0))
        .validateSchema(anyString(), anyString(), anyString(), anyInt(), any());
  }

  @Test
  @Order(22)
  public void uploadSchema_ValidationPropertyNotSet() throws KlawException {

    ReflectionTestUtils.setField(
        schemaRegistryControllerService, "validateCompatiblityOnSave", null);
    SchemaRequestModel schemaRequest = createDefaultSchemaRequestModel();
    Topic topic = createTopic();
    when(clusterApiService.validateSchema(anyString(), anyString(), anyString(), anyInt(), any()))
        .thenReturn(buildValidationResponse(true));
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.REQUEST_CREATE_SCHEMAS))
        .thenReturn(false);
    mockGetEnvironment();
    when(manageDatabase.getEnv(eq(101), eq(1))).thenReturn(Optional.of(createEnv(1)));

    NullPointerException ex =
        assertThrows(
            NullPointerException.class,
            () ->
                schemaRegistryControllerService.uploadSchema(
                    schemaRequest, RequestOperationType.CREATE));
    assertThat(ex.getMessage())
        .contains("Cannot invoke \"java.lang.Boolean.booleanValue()\"")
        .contains("validateCompatiblityOnSave\" is null");
  }

  @Test
  public void getListofSchemaRequestsIn_NEWEST_FIRST_ORDER() {
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(handleDbRequests.getAllSchemaRequests(
            anyBoolean(),
            anyString(),
            anyInt(),
            eq(null),
            eq(null),
            eq(null),
            eq(null),
            eq(null),
            eq(true),
            eq(false)))
        .thenReturn(getSchemasReqs(40));
    when(rolesPermissionsControllerService.getApproverRoles(anyString(), anyInt()))
        .thenReturn(Set.of(""));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(handleDbRequests.getAllUsersInfoForTeam(anyInt(), anyInt())).thenReturn(List.of(userInfo));
    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(this.env);
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("teamname");
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.APPROVE_ALL_REQUESTS_TEAMS))
        .thenReturn(false);

    List<SchemaRequestsResponseModel> ordered_response =
        schemaRegistryControllerService.getSchemaRequests(
            "1",
            "1",
            null,
            null,
            true,
            null,
            null,
            null,
            io.aiven.klaw.model.enums.Order.DESC_REQUESTED_TIME,
            false);

    assertThat(ordered_response).hasSize(10);
    Timestamp origReqTime = ordered_response.get(0).getRequesttime();
    for (SchemaRequestsResponseModel req : ordered_response) {

      // assert That each new Request time is older than or equal to the previous request
      assertThat(origReqTime.compareTo(req.getRequesttime()) >= 0).isTrue();
      origReqTime = req.getRequesttime();
    }
  }

  @Test
  public void getListofSchemaRequestsIn_OLDEST_FIRST_ORDER() {
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(101);
    when(handleDbRequests.getAllSchemaRequests(
            anyBoolean(),
            anyString(),
            anyInt(),
            eq(null),
            eq(null),
            eq(null),
            eq(null),
            eq(null),
            eq(true),
