package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.SYNC_102;
import static io.aiven.klaw.error.KlawErrorMessages.SYNC_103;
import static io.aiven.klaw.service.SchemaRegistrySyncControllerService.IN_SYNC;
import static io.aiven.klaw.service.SchemaRegistrySyncControllerService.NOT_IN_SYNC;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
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 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.MessageSchema;
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.SyncSchemaUpdates;
import io.aiven.klaw.model.cluster.SchemasInfoOfClusterResponse;
import io.aiven.klaw.model.enums.PermissionType;
import io.aiven.klaw.model.response.SchemaDetailsResponse;
import io.aiven.klaw.model.response.SchemaSubjectInfoResponse;
import io.aiven.klaw.model.response.SyncSchemasList;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Stream;
import org.apache.commons.collections4.CollectionUtils;
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.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.Mock;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.test.util.ReflectionTestUtils;

@ExtendWith(SpringExtension.class)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class SchemaRegistrySyncControllerServiceTest {
  public static final String TEAM = "Octopus";
  @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;

  private SchemaRegistrySyncControllerService schemaRegistrySyncControllerService;
  private UtilMethods utilMethods;

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

    schemaRegistrySyncControllerService =
        new SchemaRegistrySyncControllerService(clusterApiService, mailService);
    ReflectionTestUtils.setField(
        schemaRegistrySyncControllerService, "manageDatabase", manageDatabase);
    ReflectionTestUtils.setField(schemaRegistrySyncControllerService, "mailService", mailService);
    ReflectionTestUtils.setField(
        schemaRegistrySyncControllerService, "commonUtilsService", commonUtilsService);

    when(manageDatabase.getHandleDbRequests()).thenReturn(handleDbRequests);
    when(commonUtilsService.getPrincipal()).thenReturn(userDetails);
    when(commonUtilsService.isNotAuthorizedUser(any(), any(PermissionType.class))).thenReturn(true);
  }

  @Test
  @Order(1)
  public void getSchemasOfEnvironmentNotInSyncSourceCluster() throws Exception {
    stubUserInfo();

    String source = "cluster";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    List<Topic> topics = utilMethods.generateTopics(14);
    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(10))).thenReturn("Team1");
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);

    when(clusterApiService.getSchemasFromCluster(anyString(), any(), anyString(), anyInt()))
        .thenReturn(utilMethods.getSchemasInfoOfEnv());
    when(handleDbRequests.getSyncTopics(eq("1"), eq(null), eq(101))).thenReturn(topics);

    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(0).getRemarks())
        .isEqualTo(NOT_IN_SYNC);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(1).getRemarks())
        .isEqualTo(NOT_IN_SYNC);
  }

  @Test
  @Order(2)
  public void getSchemasOfEnvironmentInSyncAndNotInSyncSourceCluster() throws Exception {
    stubUserInfo();

    String source = "cluster";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    List<Topic> topics = utilMethods.generateTopics(14);
    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(10))).thenReturn("Team1");
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);

    when(clusterApiService.getSchemasFromCluster(anyString(), any(), anyString(), anyInt()))
        .thenReturn(utilMethods.getSchemasInfoOfEnv());
    when(handleDbRequests.getSyncTopics(eq("1"), eq(null), eq(101))).thenReturn(topics);

    Map<String, Set<String>> topicSchemaVersionsInDb = utilMethods.getTopicSchemaVersionsInDb();
    when(handleDbRequests.getTopicAndVersionsForEnvAndTenantId(anyString(), anyInt()))
        .thenReturn(topicSchemaVersionsInDb);

    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(0).getRemarks())
        .isEqualTo(IN_SYNC);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(1).getRemarks())
        .isEqualTo(IN_SYNC);

    // All schemas don't exist
    topicSchemaVersionsInDb.put("Topic0", Set.of("1")); // remove a version from db
    topicSchemaVersionsInDb.put("Topic1", Set.of("1", "2")); // remove a version from db
    when(handleDbRequests.getTopicAndVersionsForEnvAndTenantId(anyString(), anyInt()))
        .thenReturn(topicSchemaVersionsInDb);
    schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(0).getRemarks())
        .isEqualTo(NOT_IN_SYNC);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(1).getRemarks())
        .isEqualTo(NOT_IN_SYNC);

    // Only one schema differs, set showAllTopics to false
    topicSchemaVersionsInDb.put("Topic0", Set.of("1")); // remove a version from db, so not-in-sync
    topicSchemaVersionsInDb.put("Topic1", Set.of("1", "2", "3")); // make it in sync with db
    when(handleDbRequests.getTopicAndVersionsForEnvAndTenantId(anyString(), anyInt()))
        .thenReturn(topicSchemaVersionsInDb);
    schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", false, source, 0);

    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(1);
  }

  @Test
  @Order(3)
  public void getSchemasOfEnvironmentSourceMetadata() throws Exception {
    stubUserInfo();

    String source = "metadata";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    List<Topic> topics = utilMethods.generateTopics(14);

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_BACK_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(10))).thenReturn("Team1");

    when(manageDatabase.getTopicsForTenant(anyInt())).thenReturn(topics);
    Map<String, Set<String>> topicSchemaVersionsInDb = utilMethods.getTopicSchemaVersionsInDb();
    when(handleDbRequests.getTopicAndVersionsForEnvAndTenantId(anyString(), anyInt()))
        .thenReturn(topicSchemaVersionsInDb);

    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList())
        .extracting(SchemaSubjectInfoResponse::getTopic)
        .containsExactlyInAnyOrder("Topic0", "Topic1");
  }

  @Test
  @Order(4)
  public void getSchemasOfEnvironmentNoSchemasSourceMetadata() throws Exception {
    stubUserInfo();

    String source = "metadata";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    List<Topic> topics = utilMethods.generateTopics(14);

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_BACK_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(10))).thenReturn("Team1");

    when(manageDatabase.getTopicsForTenant(anyInt())).thenReturn(topics);
    when(handleDbRequests.getTopicAndVersionsForEnvAndTenantId(anyString(), anyInt()))
        .thenReturn(new HashMap<>());

    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(0);
  }

  @Test
  @Order(5)
  public void updateDbFromCluster() throws Exception {
    stubUserInfo();
    String topicName = "2ndTopic";

    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    SyncSchemaUpdates syncSchemaUpdates = new SyncSchemaUpdates();
    syncSchemaUpdates.setSourceKafkaEnvSelected("1");
    syncSchemaUpdates.setTopicList(List.of("2ndTopic"));
    syncSchemaUpdates.setTypeOfSync("SYNC_SCHEMAS");

    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);

    when(clusterApiService.getAvroSchema(anyString(), any(), anyString(), anyString(), anyInt()))
        .thenReturn(utilMethods.createSchemaList());
    List<Topic> topicList = utilMethods.getTopics();
    topicList.get(0).setTopicname(topicName);
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt())).thenReturn(topicList);

    ApiResponse apiResponse =
        schemaRegistrySyncControllerService.updateSyncSchemas(syncSchemaUpdates);
    assertThat(apiResponse.isSuccess()).isTrue();
    assertThat(apiResponse.getMessage())
        .isEqualTo(
            "Topics/Schemas "
                + syncSchemaUpdates.getTopicList()
                + "\nSchemas removed "
                + CollectionUtils.emptyIfNull(syncSchemaUpdates.getTopicListForRemoval()));
  }

  @Test
  @Order(6)
  public void updateClusterFromDb() throws Exception {
    stubUserInfo();
    String topicName = "2ndTopic";

    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    SyncSchemaUpdates syncSchemaUpdates = new SyncSchemaUpdates();
    syncSchemaUpdates.setSourceKafkaEnvSelected("1");
    syncSchemaUpdates.setTargetKafkaEnvSelected("2");
    syncSchemaUpdates.setTopicList(List.of(topicName));
    syncSchemaUpdates.setTypeOfSync("SYNC_BACK_SCHEMAS");
    syncSchemaUpdates.setTopicsSelectionType("SELECTED_TOPICS");

    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_BACK_SCHEMAS))
        .thenReturn(false);
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);

    when(clusterApiService.getAvroSchema(anyString(), any(), anyString(), anyString(), anyInt()))
        .thenReturn(utilMethods.createSchemaList());
    ApiResponse apiResponseDelete = ApiResponse.builder().success(true).build();
    ResponseEntity<ApiResponse> responseEntity =
        new ResponseEntity<>(apiResponseDelete, HttpStatus.OK);

    when(clusterApiService.deleteSchema(anyString(), anyString(), anyInt()))
        .thenReturn(responseEntity);
    Map<String, Object> schemaCreationResponseMap = new HashMap<>();
    schemaCreationResponseMap.put("schemaRegistered", Boolean.TRUE);
    schemaCreationResponseMap.put("id", 1234);
    schemaCreationResponseMap.put("compatibility", "BACKWARD");
    schemaCreationResponseMap.put("version", 1);
    ApiResponse apiResponseCreate =
        ApiResponse.builder().success(true).data(schemaCreationResponseMap).build();
    ResponseEntity<ApiResponse> responseEntityCreate =
        new ResponseEntity<>(apiResponseCreate, HttpStatus.OK);

    when(clusterApiService.postSchema(any(), anyString(), anyString(), anyInt()))
        .thenReturn(responseEntityCreate);

    List<MessageSchema> schemaList = utilMethods.getMSchemas();
    schemaList.get(0).setTopicname(topicName);
    when(handleDbRequests.getSchemaForTenantAndEnvAndTopic(anyInt(), anyString(), anyString()))
        .thenReturn(schemaList);

    ApiResponse apiResponse =
        schemaRegistrySyncControllerService.updateSyncSchemas(syncSchemaUpdates);
    assertThat(apiResponse.isSuccess()).isTrue();
    assertThat(((List<String>) apiResponse.getData()))
        .contains("Topics/Schemas result")
        .contains("Schemas deleted for 2ndTopic")
        .contains("Schemas registered on cluster for 2ndTopic Version 1");
  }

  @Test
  @Order(7)
  public void getSchemaOfTopicFromSourceCluster() throws Exception {
    int schemaVersion = 1;
    String topicName = "2ndTopic";
    String kafkaEnvId = "1";
    String source = "cluster";

    stubUserInfo();

    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);
    when(clusterApiService.getAvroSchema(anyString(), any(), anyString(), anyString(), anyInt()))
        .thenReturn(utilMethods.createSchemaList());

    SchemaDetailsResponse schemaDetailsResponse =
        schemaRegistrySyncControllerService.getSchemaOfTopicFromSource(
            source, topicName, schemaVersion, kafkaEnvId);
    assertThat(schemaDetailsResponse.getSchemaContent()).contains("klaw.avro"); // namespace
  }

  @Test
  @Order(8)
  public void getSchemaOfTopicFromSourceMetadata() throws Exception {
    int schemaVersion = 1;
    String topicName = "2ndTopic";
    String kafkaEnvId = "1";
    String source = "metadata";

    stubUserInfo();

    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_BACK_SCHEMAS))
        .thenReturn(false);
    MessageSchema schema = utilMethods.getMSchemas().get(0);
    schema.setTopicname(topicName);
    schema.setSchemafull("\"namespace : klaw.avro\"");
    when(handleDbRequests.getFirstSchemaForTenantAndEnvAndTopicAndVersion(
            anyInt(), anyString(), anyString(), anyString()))
        .thenReturn(Optional.of(schema));

    SchemaDetailsResponse schemaDetailsResponse =
        schemaRegistrySyncControllerService.getSchemaOfTopicFromSource(
            source, topicName, schemaVersion, kafkaEnvId);
    assertThat(schemaDetailsResponse.getSchemaContent()).contains("klaw.avro"); // namespace
  }

  @ParameterizedTest
  @MethodSource("generateSchemasToDeleteData")
  @Order(9)
  public void deleteOrphanedSchemaMetaDataFromDb(List<String> schemasToBeDeleted) throws Exception {
    stubUserInfo();
    String topicName = "2ndTopic";

    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    SyncSchemaUpdates syncSchemaUpdates = new SyncSchemaUpdates();
    syncSchemaUpdates.setSourceKafkaEnvSelected("1");
    syncSchemaUpdates.setTopicList(List.of("2ndTopic"));
    syncSchemaUpdates.setTypeOfSync("SYNC_SCHEMAS");
    syncSchemaUpdates.setTopicListForRemoval(schemasToBeDeleted);

    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);

    when(clusterApiService.getAvroSchema(anyString(), any(), anyString(), anyString(), anyInt()))
        .thenReturn(utilMethods.createSchemaList());
    List<Topic> topicList = utilMethods.getTopics();
    topicList.get(0).setTopicname(topicName);
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt())).thenReturn(topicList);

    ApiResponse apiResponse =
        schemaRegistrySyncControllerService.updateSyncSchemas(syncSchemaUpdates);
    assertThat(apiResponse.isSuccess()).isTrue();
    assertThat(apiResponse.getMessage())
        .isEqualTo(
            "Topics/Schemas "
                + syncSchemaUpdates.getTopicList()
                + "\nSchemas removed "
                + CollectionUtils.emptyIfNull(syncSchemaUpdates.getTopicListForRemoval()));

    verify(handleDbRequests, times(schemasToBeDeleted.size()))
        .deleteSchema(eq(101), anyString(), eq("1"));
  }

  @Test
  @Order(10)
  public void getSchemaOfTopicFromSourceMetadataWithSchemasToBeDeleted() throws Exception {
    stubUserInfo();

    String source = "cluster";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");

    List<Topic> topics = utilMethods.generateTopics(1);
    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(10))).thenReturn("Team1");
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);

    when(clusterApiService.getSchemasFromCluster(anyString(), any(), anyString(), anyInt()))
        .thenReturn(utilMethods.getSchemasInfoOfEnv());
    when(handleDbRequests.getSyncTopics(eq("1"), eq(null), eq(101))).thenReturn(topics);
    when(handleDbRequests.getSchemaForTenantAndEnvAndTopic(eq(101), eq("1"), anyString()))
        .thenReturn(schemaMetaData());
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(3))).thenReturn(TEAM);

    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(0).getRemarks())
        .isEqualTo(NOT_IN_SYNC);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(1).getRemarks())
        .isEqualTo(SYNC_103);
    assertThat(
            schemasInfoOfClusterResponse
                .getSchemaSubjectInfoResponseList()
                .get(1)
                .getPossibleTeams())
        .contains(TEAM, SYNC_102);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().get(1).getTeamname())
        .isEqualTo(TEAM);
  }

  @Test
  @Order(11)
  public void getSchemasOfEnvironmentFromMetadataWithTopicDeletedAndSchemaStillAvailable()
      throws Exception {
    stubUserInfo();

    String source = "cluster";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");
    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());

    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(3))).thenReturn("Team1");
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);
    when(manageDatabase.getTopicsForTenant(anyInt())).thenReturn(new ArrayList<>());
    Map<String, Set<String>> topicSchemaVersionsInDb = utilMethods.getTopicSchemaVersionsInDb();

    when(manageDatabase
            .getHandleDbRequests()
            .getSchemaForTenantAndEnvAndTopic(eq(101), eq("1"), anyString()))
        .thenReturn(schemaMetaData());
    when(clusterApiService.getSchemasFromCluster(anyString(), any(), anyString(), anyInt()))
        .thenReturn(utilMethods.getSchemasInfoOfEnv());
    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList())
        .extracting(SchemaSubjectInfoResponse::getTopic)
        .containsExactlyInAnyOrder("Topic0", "Topic1");
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList())
        .extracting(SchemaSubjectInfoResponse::getRemarks)
        .containsExactlyInAnyOrder("ORPHANED", "ORPHANED");
    assertThat(
            schemasInfoOfClusterResponse
                .getSchemaSubjectInfoResponseList()
                .get(0)
                .getPossibleTeams())
        .contains("REMOVE FROM KLAW");

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

  @Test
  @Order(12)
  public void getSchemasOfEnvironmentFromMetadataWithSchemaDeletedFromCluster() throws Exception {
    stubUserInfo();

    String source = "cluster";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");
    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());
    List<Topic> topics = utilMethods.generateTopics(14);

    SchemasInfoOfClusterResponse clusterResp = new SchemasInfoOfClusterResponse();
    clusterResp.setSchemaInfoOfTopicList(new ArrayList<>());
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(3))).thenReturn("Team1");
    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getTeamNameFromTeamId(eq(101), eq(10))).thenReturn("Team1");
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);
    when(manageDatabase.getTopicsForTenant(anyInt())).thenReturn(topics);
    Map<String, Set<String>> topicSchemaVersionsInDb = utilMethods.getTopicSchemaVersionsInDb();
    when(handleDbRequests.getTopicAndVersionsForEnvAndTenantId(anyString(), anyInt()))
        .thenReturn(topicSchemaVersionsInDb);
    when(clusterApiService.getAvroSchema(anyString(), any(), anyString(), anyString(), anyInt()))
        .thenReturn(new TreeMap<>());
    when(manageDatabase
            .getHandleDbRequests()
            .getTeamIdFromSchemaTopicNameAndEnvAndTenantId(anyString(), eq("1"), eq(101)))
        .thenReturn(schemaMetaData().get(0));
    when(clusterApiService.getSchemasFromCluster(anyString(), any(), anyString(), anyInt()))
        .thenReturn(clusterResp);
    SyncSchemasList schemasInfoOfClusterResponse =
        schemaRegistrySyncControllerService.getSchemasOfEnvironment(
            "1", "1", "", "", true, source, 0);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList())
        .extracting(SchemaSubjectInfoResponse::getTopic)
        .containsExactlyInAnyOrder("Topic0", "Topic1");
    assertThat(schemasInfoOfClusterResponse.getSchemaSubjectInfoResponseList())
        .extracting(SchemaSubjectInfoResponse::getRemarks)
        .containsExactlyInAnyOrder("NOT_ON_CLUSTER", "NOT_ON_CLUSTER");
    assertThat(
            schemasInfoOfClusterResponse
                .getSchemaSubjectInfoResponseList()
                .get(0)
                .getPossibleTeams())
        .contains("REMOVE FROM KLAW");
    verify(clusterApiService, times(1))
        .getSchemasFromCluster(anyString(), any(), anyString(), anyInt());
  }

  @Test
  @Order(13)
  public void getSchemasOfEnvironmentFromMetadataExceptonContactingCluster() throws Exception {
    stubUserInfo();

    String source = "cluster";
    Env env = utilMethods.getEnvLists().get(0);
    env.setAssociatedEnv(new EnvTag("1", "SCH"));
    env.setType("kafka");
    Map<Integer, KwClusters> kwClustersMap = new HashMap<>();
    kwClustersMap.put(1, utilMethods.getKwClusters());
    List<Topic> topics = utilMethods.generateTopics(14);

    SchemasInfoOfClusterResponse clusterResp = new SchemasInfoOfClusterResponse();
    clusterResp.setSchemaInfoOfTopicList(new ArrayList<>());
    when(handleDbRequests.getEnvDetails(anyString(), anyInt())).thenReturn(env);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_SCHEMAS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(manageDatabase.getClusters(any(), anyInt())).thenReturn(kwClustersMap);
    assertThatThrownBy(
            () -> {
              schemaRegistrySyncControllerService.getSchemasOfEnvironment(
                  "1", "1", "", "", true, source, 0);
            })
        .isInstanceOf(KlawException.class);
  }

  private List<MessageSchema> schemaMetaData() {
    List<MessageSchema> schemas = new ArrayList<>();

    MessageSchema schema = new MessageSchema();
    schema.setSchemaId(1);
    schema.setTenantId(101);
    schema.setTopicname("Topic1");
    schema.setTeamId(3);
    schemas.add(schema);

    return schemas;
  }

  private static Stream<Arguments> generateSchemasToDeleteData() {
    return Stream.of(
        Arguments.of(Arrays.asList("schema1", "schema2")),
        Arguments.of(Arrays.asList("schema1")),
        Arguments.of(Arrays.asList("schema1", "schema2", "schema2")),
        Arguments.of(CollectionUtils.emptyCollection()));
  }

  private void stubUserInfo() {
    when(handleDbRequests.getUsersInfo(anyString())).thenReturn(userInfo);
    when(userInfo.getTeamId()).thenReturn(101);
    when(userInfo.getRole()).thenReturn("USER");
    when(mailService.getUserName(any())).thenReturn("kwusera");
  }
}
