package io.aiven.klaw.service;

import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_TOPIC_ENVS;
import static io.aiven.klaw.helpers.KwConstants.REQUEST_TOPICS_OF_ENVS;
import static org.assertj.core.api.Assertions.assertThat;
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 com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.aiven.klaw.UtilMethods;
import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.*;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.helpers.db.rdbms.HandleDbRequestsJdbc;
import io.aiven.klaw.model.KwTenantConfigModel;
import io.aiven.klaw.model.TopicOverviewInfo;
import io.aiven.klaw.model.enums.AclGroupBy;
import io.aiven.klaw.model.enums.AclType;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.KafkaFlavors;
import io.aiven.klaw.model.enums.PromotionStatusType;
import io.aiven.klaw.model.enums.RequestOperationType;
import io.aiven.klaw.model.enums.RequestStatus;
import io.aiven.klaw.model.response.AclOverviewInfo;
import io.aiven.klaw.model.response.TopicOverview;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.TreeMap;
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.CsvSource;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
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 TopicOverviewServiceTest {

  public static final String TESTTOPIC = "testtopic";
  public static final String TEAM_1 = "Team-1";
  public static final String TEAM_ID = "kwusera";
  public static final int TEAMID = 10;
  private UtilMethods utilMethods;
  @Mock private UserDetails userDetails;
  @Mock private HandleDbRequestsJdbc handleDbRequests;
  @Mock private ManageDatabase manageDatabase;
  @Mock private CommonUtilsService commonUtilsService;
  @Mock private MailUtils mailService;
  @Mock private UserInfo userInfo;

  @Mock private ClusterApiService clusterApiService;

  @Mock private Map<Integer, KwClusters> kwClustersHashMap;
  @Mock private KwClusters kwClusters;

  private TopicOverviewService topicOverviewService;

  private ObjectMapper mapper = new ObjectMapper();

  @BeforeEach
  public void setUp() throws Exception {
    utilMethods = new UtilMethods();
    this.topicOverviewService = new TopicOverviewService(mailService);

    Env env = new Env();
    env.setName("DEV");
    env.setId("1");
    ReflectionTestUtils.setField(topicOverviewService, "manageDatabase", manageDatabase);
    ReflectionTestUtils.setField(topicOverviewService, "commonUtilsService", commonUtilsService);
    ReflectionTestUtils.setField(topicOverviewService, "clusterApiService", clusterApiService);
    when(manageDatabase.getHandleDbRequests()).thenReturn(handleDbRequests);
    loginMock();
  }

  private void loginMock() {
    Authentication authentication = Mockito.mock(Authentication.class);
    SecurityContext securityContext = Mockito.mock(SecurityContext.class);
    when(securityContext.getAuthentication()).thenReturn(authentication);
    when(authentication.getPrincipal()).thenReturn(userDetails);
    SecurityContextHolder.setContext(securityContext);
  }

  @Test
  @Order(1)
  public void getAclsSyncFalse1() throws KlawException {
    String env1 = "1";
    stubUserInfo();

    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getKwPropertyValue(anyString(), anyInt())).thenReturn("true");
    when(manageDatabase.getEnv(anyInt(), anyInt()))
        .thenReturn(Optional.of(utilMethods.getEnvLists().get(0)));
    when(handleDbRequests.getAllTeamsOfUsers(anyString(), anyInt()))
        .thenReturn(utilMethods.getTeams());
    when(handleDbRequests.getSyncAcls(anyString(), anyString(), anyInt()))
        .thenReturn(getAclsSOT(TESTTOPIC));

    List<Topic> topics = utilMethods.getTopics(TESTTOPIC);
    topics
        .get(0)
        .setJsonParams("{\"advancedTopicConfiguration\":{\"retention.ms\":\"404800000\"}}");
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt())).thenReturn(topics);
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);
    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS))).thenReturn("1");
    mockTenantConfig();
    TopicOverviewInfo topicOverviewInfo =
        topicOverviewService
            .getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE)
            .getTopicInfoList()
            .get(0);
    List<AclOverviewInfo> aclList =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE).getAclInfoList();

    assertThat(aclList).hasSize(1);

    assertThat(topicOverviewInfo.getAdvancedTopicConfiguration())
        .containsEntry("retention.ms", "404800000");
    assertThat(aclList.get(0).getTopicname()).isEqualTo(TESTTOPIC);
    assertThat(aclList.get(0).getConsumergroup()).isEqualTo("mygrp1");
    assertThat(aclList.get(0).getAcl_ip()).isEqualTo("2.1.2.1");
  }

  @Test
  @Order(2)
  public void getAclsSyncFalse2() {
    String topicNameSearch = "testnewtopic1";

    stubUserInfo();
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getKwPropertyValue(anyString(), anyInt())).thenReturn("true");
    when(manageDatabase.getEnv(anyInt(), anyInt()))
        .thenReturn(Optional.of(utilMethods.getEnvLists().get(0)));
    when(handleDbRequests.getAllTeamsOfUsers(anyString(), anyInt()))
        .thenReturn(utilMethods.getTeams());
    when(handleDbRequests.getSyncAcls(anyString(), anyString(), anyInt()))
        .thenReturn(getAclsSOT(topicNameSearch));
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopics(topicNameSearch));
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS))).thenReturn("1");
    mockTenantConfig();

    List<AclOverviewInfo> aclList =
        topicOverviewService
            .getTopicOverview(topicNameSearch, "1", AclGroupBy.NONE)
            .getAclInfoList();

    assertThat(aclList).hasSize(1);

    assertThat(aclList.get(0).getTopicname()).isEqualTo(topicNameSearch);
    assertThat(aclList.get(0).getConsumergroup()).isEqualTo("mygrp1");
    assertThat(aclList.get(0).getAcl_ip()).isEqualTo("2.1.2.1");
  }

  @Test
  @Order(3)
  public void givenATopicWithOnlyOneKafkaEnv_ReturnNoPromotion() throws Exception {
    stubUserInfo();
    stubKafkaPromotion(TESTTOPIC, 1);
    stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.KAFKA, 15);
    when(commonUtilsService.getTopicsForTopicName(TESTTOPIC, 101))
        .thenReturn(List.of(createTopic(TESTTOPIC)));
    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS))).thenReturn("1");
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1");

    TopicOverview returnedValue =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(returnedValue.getTopicPromotionDetails()).isNotNull();
    assertThat(returnedValue.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
  }

  @Test
  @Order(4)
  public void getGivenATopicWithManyKafkaEnv_ReturnNextInPromotion() throws Exception {
    stubUserInfo();
    stubKafkaPromotion(TESTTOPIC, 15);
    stubEnvironment("1", "4");
    stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.KAFKA, 15);
    when(commonUtilsService.getTopicsForTopicName(TESTTOPIC, 101))
        .thenReturn(List.of(createTopic(TESTTOPIC)));
    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS)))
        .thenReturn("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15");
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS)))
        .thenReturn("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15");

    TopicOverview returnedValue =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(returnedValue.getTopicPromotionDetails()).isNotNull();
    assertThat(returnedValue.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.SUCCESS);
    assertThat(returnedValue.getTopicPromotionDetails().getSourceEnv()).isEqualTo("1");
    assertThat(returnedValue.getTopicPromotionDetails().getTargetEnv()).isEqualTo("test-2");
  }

  @Test
  @Order(5)
  public void givenATopicThatDoesntExist_ReturnNoPromotion() throws Exception {
    stubUserInfo();
    stubKafkaPromotion(TESTTOPIC, 1);
    stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.KAFKA, 15);

    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1");

    TopicOverview returnedValue =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(returnedValue.getTopicPromotionDetails()).isNull();
    assertThat(returnedValue.isTopicExists()).isFalse();
  }

  @Test
  @Order(6)
  public void getAclsSyncFalseMaskedData() {
    String env1 = "1";
    when(handleDbRequests.getUsersInfo(anyString())).thenReturn(userInfo);
    when(userInfo.getTeamId()).thenReturn(110);
    when(mailService.getUserName(any())).thenReturn(TEAM_ID);
    when(commonUtilsService.getTeamId(eq(TEAM_ID))).thenReturn(110);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getKwPropertyValue(anyString(), anyInt())).thenReturn("true");
    when(manageDatabase.getEnv(anyInt(), anyInt()))
        .thenReturn(Optional.of(utilMethods.getEnvLists().get(0)));
    when(handleDbRequests.getAllTeamsOfUsers(anyString(), anyInt()))
        .thenReturn(utilMethods.getTeams());
    when(handleDbRequests.getSyncAcls(anyString(), anyString(), anyInt()))
        .thenReturn(getAclsSOT(TESTTOPIC));
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopics(TESTTOPIC));
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS))).thenReturn("1");
    mockTenantConfig();
    List<AclOverviewInfo> aclList =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE).getAclInfoList();

    assertThat(aclList).hasSize(1);

    assertThat(aclList.get(0).getTopicname()).isEqualTo(TESTTOPIC);
    assertThat(aclList.get(0).getConsumergroup()).isEqualTo("mygrp1");
    // MASKED DATA
    assertThat(aclList.get(0).getAcl_ip()).isEqualTo("Not Authorized to see this.");
    assertThat(aclList.get(0).getAcl_ssl()).isEqualTo("Not Authorized to see this.");
  }

  @ParameterizedTest
  @Order(7)
  @CsvSource({"TEAM", "NONE"})
  public void givenARequestWithAnOrderEnsureItIsCorrectlyUsed(AclGroupBy groupBy) throws Exception {
    stubUserInfo();
    stubKafkaPromotion(TESTTOPIC, 1);
    stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.KAFKA, 15);

    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1");
    when(commonUtilsService.getTopicsForTopicName(eq(TESTTOPIC), eq(101)))
        .thenReturn(List.of(createTopic(TESTTOPIC)));
    when(handleDbRequests.getSyncAcls(anyString(), eq(TESTTOPIC), eq(101)))
        .thenReturn(createAcls(20));
    createListOfEnvs(KafkaClustersType.KAFKA, 3);

    when(manageDatabase.getTeamNameFromTeamId(eq(101), anyInt()))
        .thenReturn("Octopus")
        .thenReturn("Town")
        .thenReturn("Alias")
        .thenReturn("Octopus")
        .thenReturn("Town")
        .thenReturn("Alias")
        .thenReturn("Octopus")
        .thenReturn("Town")
        .thenReturn("Alias");

    when(manageDatabase.getClusters(eq(KafkaClustersType.KAFKA), eq(101)))
        .thenReturn(getKwClusterMap());
    TopicOverview returnedValue = topicOverviewService.getTopicOverview(TESTTOPIC, "1", groupBy);

    if (AclGroupBy.TEAM.equals(groupBy)) {
      String previousTeam = returnedValue.getAclInfoList().get(0).getTeamname();
      for (AclOverviewInfo info : returnedValue.getAclInfoList()) {
        assertThat(previousTeam).isLessThanOrEqualTo(info.getTeamname());
      }
    } else if (AclGroupBy.NONE.equals(groupBy)) {

    } else {
      throw new UnsupportedOperationException(
          "This is an unsupported Operation and should not occur.");
    }
  }

  @Test
  @Order(8)
  public void getTopicOverview() {
    mockTenantConfig();
    stubUserInfo();
    stubEnvironment("1", "4");
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Arrays.asList("1", "2", "3")));
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 3));

    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    when(commonUtilsService.isCreateNewSchemaAllowed(eq("4"), eq(101))).thenReturn(true);

    TopicOverview topicOverview =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(3);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(topicOverview.getTopicInfoList().get(0).isTopicDeletable())
        .isTrue(); // topic can be deleted
    assertThat(topicOverview.isCreateSchemaAllowed()).isTrue();
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 2));

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "2", AclGroupBy.NONE);
    stubEnvironment("1", "5");
    when(commonUtilsService.isCreateNewSchemaAllowed(eq("5"), eq(101))).thenReturn(false);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(2);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.SUCCESS);

    when(handleDbRequests.getSyncAcls(anyString(), anyString(), anyInt()))
        .thenReturn(getAclsSOT(TESTTOPIC));
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(2);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(topicOverview.getTopicInfoList().get(0).isTopicDeletable())
        .isFalse(); // topic can't be deleted

    assertThat(topicOverview.getTopicInfoList().get(0).isHasACL()).isTrue(); // topic hasAcl

    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenACLRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenTopicRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasSchema()).isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequestOnAnyEnv())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.isCreateSchemaAllowed()).isFalse();
  }

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

    mockTenantConfig();
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Arrays.asList("1", "2", "3")));
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 3));

    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    TopicOverview topicOverview =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(3);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(topicOverview.getTopicInfoList().get(0).isTopicDeletable())
        .isTrue(); // topic can be deleted

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 2));

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "2", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(2);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.SUCCESS);

    when(handleDbRequests.existsTopicRequest(
            eq(TESTTOPIC), eq(RequestStatus.CREATED.value), eq("1"), eq(101)))
        .thenReturn(true);
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);

    assertThat(topicOverview.getTopicInfoList().get(0).isHasACL()).isFalse(); // topic hasAcl

    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenACLRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenTopicRequest())
        .isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequest()).isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequestOnAnyEnv()).isTrue();
    assertThat(topicOverview.getTopicInfoList().get(0).isHasSchema()).isFalse(); // topic hasAcl
  }

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

    mockTenantConfig();
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Arrays.asList("1", "2", "3")));
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 3));

    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    when(manageDatabase.getAssociatedSchemaEnvIdFromTopicId(eq("1"), eq(101)))
        .thenReturn(Optional.of("3"));

    TopicOverview topicOverview =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(3);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(topicOverview.getTopicInfoList().get(0).isTopicDeletable())
        .isTrue(); // topic can be deleted

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 2));

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "2", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(2);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.SUCCESS);

    when(handleDbRequests.existsSchemaRequest(
            eq(TESTTOPIC), eq(RequestStatus.CREATED.value), eq("3"), eq(101)))
        .thenReturn(true);
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);

    assertThat(topicOverview.getTopicInfoList().get(0).isHasACL()).isFalse(); // topic hasAcl

    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenSchemaRequest())
        .isTrue(); // topic hasAcl

    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenACLRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenTopicRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequest()).isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequestOnAnyEnv())
        .isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasSchema()).isFalse(); // topic hasAcl
  }

  @Test
  @Order(11)
  public void getTopicOverviewWithNoAcls_OpenTopicACLSchemaRequests() {

    mockTenantConfig();
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Arrays.asList("1", "2", "3")));
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 3));

    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    TopicOverview topicOverview =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(3);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(topicOverview.getTopicInfoList().get(0).isTopicDeletable())
        .isTrue(); // topic can be deleted

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 2));

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "2", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(2);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.SUCCESS);

    when(handleDbRequests.existsSchemaRequest(
            eq(TESTTOPIC), eq(RequestStatus.CREATED.value), eq("1"), eq(101)))
        .thenReturn(true);
    when(handleDbRequests.existsAclRequest(
            eq(TESTTOPIC), eq(RequestStatus.CREATED.value), eq("1"), eq(101)))
        .thenReturn(true);
    when(handleDbRequests.existsTopicRequest(
            eq(TESTTOPIC), eq(RequestStatus.CREATED.value), eq("1"), eq(101)))
        .thenReturn(true);
    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);

    assertThat(topicOverview.getTopicInfoList().get(0).isHasACL()).isFalse(); // topic hasAcl

    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenACLRequest())
        .isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenTopicRequest())
        .isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequest()).isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequestOnAnyEnv())
        .isTrue(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasSchema()).isFalse(); // topic hasAcl
  }

  @Test
  @Order(12)
  public void getTopicOverviewWithNoAcls_hasSchema() {

    mockTenantConfig();
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Arrays.asList("1", "2", "3")));
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 3));

    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    TopicOverview topicOverview =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(3);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(topicOverview.getTopicInfoList().get(0).isTopicDeletable())
        .isTrue(); // topic can be deleted

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 2));

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "2", AclGroupBy.NONE);
    assertThat(topicOverview.getAvailableEnvironments().size()).isEqualTo(2);
    assertThat(topicOverview.getTopicInfoList().size()).isEqualTo(1);
    assertThat(topicOverview.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.SUCCESS);

    when(commonUtilsService.existsSchemaForTopic(eq(TESTTOPIC), eq("1"), eq(101))).thenReturn(true);

    when(manageDatabase.getClusters(any(KafkaClustersType.class), anyInt()))
        .thenReturn(kwClustersHashMap);
    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    topicOverview = topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);

    assertThat(topicOverview.getTopicInfoList().get(0).isHasACL()).isFalse(); // topic hasAcl

    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenACLRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenTopicRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequest())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasOpenRequestOnAnyEnv())
        .isFalse(); // topic hasAcl
    assertThat(topicOverview.getTopicInfoList().get(0).isHasSchema()).isTrue(); // topic hasAcl
  }

  @Test
  @Order(13)
  public void givenATopicWithPromoteRequestAlreadyOpen_ReturnREQUEST_OPEN() throws Exception {
    stubUserInfo();
    stubKafkaPromotion(TESTTOPIC, 15);
    stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.KAFKA, 15);
    when(commonUtilsService.getTopicsForTopicName(TESTTOPIC, 101))
        .thenReturn(List.of(createTopic(TESTTOPIC)));
    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS)))
        .thenReturn("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15");
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS)))
        .thenReturn("1,2,3,4,5,6,7,8,9,10,11,12,13,14,15");
    when(handleDbRequests.existsTopicRequest(
            eq(TESTTOPIC),
            eq(RequestStatus.CREATED.value),
            eq(RequestOperationType.PROMOTE.value),
            eq("2"),
            eq(101)))
        .thenReturn(true);
    TopicOverview returnedValue =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    assertThat(returnedValue.getTopicPromotionDetails()).isNotNull();
    assertThat(returnedValue.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.REQUEST_OPEN);
    assertThat(returnedValue.getTopicPromotionDetails().getSourceEnv()).isEqualTo("1");
    assertThat(returnedValue.getTopicPromotionDetails().getTargetEnv()).isEqualTo("test-2");
  }

  @Test
  @Order(14)
  public void givenATopicWithNohigherEnv_DoNotCheckForOpenPromoteRequest() throws Exception {

    stubUserInfo();
    stubKafkaPromotion(TESTTOPIC, 1);
    stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.KAFKA, 15);
    when(commonUtilsService.getTopicsForTopicName(TESTTOPIC, 101))
        .thenReturn(List.of(createTopic(TESTTOPIC)));
    when(commonUtilsService.getEnvProperty(eq(101), eq(REQUEST_TOPICS_OF_ENVS))).thenReturn("1");
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1");

    TopicOverview returnedValue =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
    verify(handleDbRequests, times(0))
        .existsTopicRequest(
            eq(TESTTOPIC),
            eq(RequestStatus.CREATED.value),
            eq(RequestOperationType.PROMOTE.value),
            eq("2"),
            eq(101));
    assertThat(returnedValue.getTopicPromotionDetails()).isNotNull();
    assertThat(returnedValue.getTopicPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
  }

  @Test
  @Order(15)
  public void getTopicOverviewWithNoAcls_OpenClaimRequest() {

    mockTenantConfig();
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Arrays.asList("1", "2", "3")));
    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");

    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopicInMultipleEnvs("testtopic", TEAMID, 3));

    when(kwClustersHashMap.get(anyInt())).thenReturn(kwClusters);

    createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, 5);

    when(manageDatabase.getAssociatedSchemaEnvIdFromTopicId(eq("1"), eq(101)))
        .thenReturn(Optional.of("3"));

    TopicOverview topicOverview =
        topicOverviewService.getTopicOverview(TESTTOPIC, "1", AclGroupBy.NONE);
