content: package io.aiven.klaw.service;
  
  import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_TOPIC_ENVS;
  import static org.assertj.core.api.Assertions.assertThat;
  import static org.mockito.ArgumentMatchers.any;
  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.Acl;
  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.helpers.db.rdbms.HandleDbRequestsJdbc;
  import io.aiven.klaw.model.KwTenantConfigModel;
  import io.aiven.klaw.model.enums.AclType;
  import io.aiven.klaw.model.enums.KafkaClustersType;
  import io.aiven.klaw.model.enums.KafkaSupportedProtocol;
  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.enums.SchemaType;
  import io.aiven.klaw.model.response.SchemaOverview;
  import java.util.ArrayList;
  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.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 SchemaOverviewServiceTest {
  
    public static final String TESTTOPIC = "testtopic";
    public static final String TEAM_1 = "Team-1";
    public static final String TEAM_ID = "kwusera";
    public static final String SCHEMA_TYPE = "schemaType";
    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;
  
    private SchemaOverviewService schemaOverviewService;
  
    @BeforeEach
    public void setUp() throws Exception {
      utilMethods = new UtilMethods();
      this.schemaOverviewService = new SchemaOverviewService(mailService);
  
      ReflectionTestUtils.setField(schemaOverviewService, "manageDatabase", manageDatabase);
      ReflectionTestUtils.setField(schemaOverviewService, "commonUtilsService", commonUtilsService);
      ReflectionTestUtils.setField(schemaOverviewService, "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 givenASchemaWithOnlyOneSchemaEnv_ReturnNoPromotion() throws Exception {
      stubUserInfo();
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 1);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3");
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaPromotionDetails().getStatus())
          .isEqualTo(PromotionStatusType.NO_PROMOTION);
    }
  
    @Test
    @Order(2)
    public void getGivenASchemaWithManySchemaEnv_ReturnNextInPromotion() throws Exception {
      // Schema Envs are 3,4
      //      kafka envs are 1,2
      //      Expecting kafka env ids to be returned now instead of schema env ids
      stubUserInfo();
      stubEnvironments();
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3,4");
      when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1"), createTopic(TESTTOPIC, "2")));
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaPromotionDetails().getStatus())
          .isEqualTo(PromotionStatusType.SUCCESS);
      assertThat(returnedValue.isCreateSchemaAllowed()).isTrue();
      assertThat(returnedValue.getSchemaPromotionDetails().getSourceEnv()).isEqualTo("1");
      assertThat(returnedValue.getSchemaPromotionDetails().getTargetEnv()).isEqualTo("test-2");
    }
  
    @Test
    @Order(3)
    public void givenASchemaThatDoesntExist_ReturnNoPromotion() throws Exception {
      stubUserInfo();
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("1");
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 3, "1");
      assertThat(returnedValue.getSchemaPromotionDetails()).isNull();
      assertThat(returnedValue.isSchemaExists()).isFalse();
    }
  
    @Test
    @Order(4)
    public void
        getGivenASchemaWithManySchemaEnv_WithRequestorFromDifferentTeam_DoNotReturnNextInPromotion()
            throws Exception {
      stubUserInfo();
  
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3,4");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(8);
      when(handleDbRequests.getTopics(eq(TESTTOPIC), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNull();
    }
  
    @Test
    @Order(5)
    public void givenARequestForSchemaWithoutAnInCorrectDBEntry_CallAPIAndUpdateSchema()
        throws Exception {
      stubUserInfo();
  
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3,4");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(8);
      when(handleDbRequests.getTopics(eq(TESTTOPIC), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      when(manageDatabase
              .getHandleDbRequests()
              .getSchemaForTenantAndEnvAndTopic(eq(101), eq("3"), eq(TESTTOPIC)))
          .thenReturn(createIncompleteMessages(2));
  
      when(clusterApiService.getAvroSchema(
              any(), eq(KafkaSupportedProtocol.PLAINTEXT), any(), eq(TESTTOPIC), eq(101)))
          .thenReturn(getAvroSchemas(2));
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      verify(manageDatabase.getHandleDbRequests(), times(1)).insertIntoMessageSchemaSOT(any());
      assertThat(returnedValue.getSchemaPromotionDetails()).isNull();
    }
  
    @Test
    @Order(6)
    public void givenARequestForSchemaWithoutACorrectDBEntry_CallAPIAndDontUpdateSchema()
        throws Exception {
      stubUserInfo();
  
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3,4");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(8);
      when(handleDbRequests.getTopics(eq(TESTTOPIC), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      when(manageDatabase
              .getHandleDbRequests()
              .getSchemaForTenantAndEnvAndTopic(eq(101), eq("3"), eq(TESTTOPIC)))
          .thenReturn(createMessages(2));
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      verify(manageDatabase.getHandleDbRequests(), times(0)).insertIntoMessageSchemaSOT(any());
      assertThat(returnedValue.getSchemaPromotionDetails()).isNull();
    }
  
    @Test
    @Order(7)
    public void givenARequestForSchemaWithoutACorrectDBEntry_CallAPIButDontUpdateSchema()
        throws Exception {
      stubUserInfo();
  
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3,4");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(8);
      when(handleDbRequests.getTopics(eq(TESTTOPIC), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      when(manageDatabase
              .getHandleDbRequests()
              .getSchemaForTenantAndEnvAndTopic(eq(101), eq("3"), eq(TESTTOPIC)))
          .thenReturn(createDBMessages(2));
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      verify(manageDatabase.getHandleDbRequests(), times(0)).insertIntoMessageSchemaSOT(any());
      assertThat(returnedValue.getSchemaPromotionDetails()).isNull();
    }
  
    @Test
    @Order(8)
    public void givenASchemaWithAPromotionRequestAlreadyOpen_ReturnREQUESTOPEN() throws Exception {
      // Schema Envs are 3,4
      //      kafka envs are 1,2
      //      Expecting kafka env ids to be returned now instead of schema env ids
      stubUserInfo();
  
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3,4");
      when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1"), createTopic(TESTTOPIC, "2")));
      when(handleDbRequests.existsSchemaRequest(
              eq(TESTTOPIC),
              eq(RequestStatus.CREATED.value),
              eq(RequestOperationType.PROMOTE.value),
              eq("4"),
              eq(101)))
          .thenReturn(true);
      when(manageDatabase.getAssociatedSchemaEnvIdFromTopicId(eq("2"), eq(101)))
          .thenReturn(Optional.of("4"));
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaPromotionDetails().getStatus())
          .isEqualTo(PromotionStatusType.REQUEST_OPEN);
  
      assertThat(returnedValue.getSchemaPromotionDetails().getSourceEnv()).isEqualTo("1");
      assertThat(returnedValue.getSchemaPromotionDetails().getTargetEnv()).isEqualTo("test-2");
    }
  
    @Test
    @Order(9)
    public void givenASchemaWithNoTopicInHigherEnv_DoNotCheckForOpenRequestInHigherEnv()
        throws Exception {
      // Schema Envs are 3,4
      //      kafka envs are 1,2
      //      Expecting kafka env ids to be returned now instead of schema env ids
      stubUserInfo();
  
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 5);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3");
      when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1");
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1"), createTopic(TESTTOPIC, "2")));
      when(handleDbRequests.existsSchemaRequest(
              eq(TESTTOPIC),
              eq(RequestStatus.CREATED.value),
              eq(RequestOperationType.PROMOTE.value),
              eq("4"),
              eq(101)))
          .thenReturn(true);
      when(manageDatabase.getAssociatedSchemaEnvIdFromTopicId(eq("2"), eq(101)))
          .thenReturn(Optional.of("4"));
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "1");
  
      verify(handleDbRequests, times(0))
          .existsSchemaRequest(
              eq(TESTTOPIC),
              eq(RequestStatus.CREATED.value),
              eq(RequestOperationType.CREATE.value),
              eq("4"),
              eq(101));
      verify(manageDatabase, times(0)).getAssociatedSchemaEnvIdFromTopicId(eq("2"), eq(101));
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaPromotionDetails().getStatus())
          .isEqualTo(PromotionStatusType.NO_PROMOTION);
    }
  
    @Test
    @Order(10)
    public void givenASchemaInHighestEnv_ReturnNoPromotion() throws Exception {
      stubUserInfo();
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 2);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3");
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 1, "2");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaPromotionDetails().getStatus())
          .isEqualTo(PromotionStatusType.NO_PROMOTION);
      assertThat(returnedValue.isCreateSchemaAllowed()).isFalse();
    }
  
    @Test
    @Order(11)
    public void givenASchemaWithNoSchemaTypeDefaultToAvro() throws Exception {
      stubUserInfo();
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 2);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3");
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 2, "2");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaDetailsPerEnv().getSchemaType()).isEqualTo(SchemaType.AVRO);
      assertThat(returnedValue.isCreateSchemaAllowed()).isFalse();
    }
  
    @Test
    @Order(11)
    public void givenASchemaWithNullSchemaTypeDefaultToAvro() throws Exception {
      stubUserInfo();
      when(commonUtilsService.getTeamId(anyString())).thenReturn(10);
      when(handleDbRequests.getAllTopicsByTopicNameAndTeamIdAndTenantId(
              eq(TESTTOPIC), eq(10), eq(101)))
          .thenReturn(List.of(createTopic(TESTTOPIC, "1")));
      stubSchemaPromotionInfo(TESTTOPIC, KafkaClustersType.SCHEMA_REGISTRY, 2);
  
      when(commonUtilsService.getSchemaPromotionEnvsFromKafkaEnvs(eq(101))).thenReturn("3");
  
      TreeMap<Integer, Map<String, Object>> schemaList = utilMethods.createSchemaList();
      schemaList.get(2).put(SCHEMA_TYPE, null);
      schemaList.get(1).put(SCHEMA_TYPE, null);
      // override default schemas returned
      when(clusterApiService.getAvroSchema(any(), any(), any(), eq(TESTTOPIC), eq(101)))
          .thenReturn(schemaList)
          .thenReturn(null);
  
      SchemaOverview returnedValue = schemaOverviewService.getSchemaOfTopic(TESTTOPIC, 2, "2");
  
      assertThat(returnedValue.getSchemaPromotionDetails()).isNotNull();
      assertThat(returnedValue.getSchemaDetailsPerEnv().getSchemaType()).isEqualTo(SchemaType.AVRO);
      assertThat(returnedValue.isCreateSchemaAllowed()).isFalse();
    }
  
    private TreeMap<Integer, Map<String, Object>> getAvroSchemas(int numOfEntries) {
      TreeMap<Integer, Map<String, Object>> messages = new TreeMap<>();
      for (int i = 1; i <= numOfEntries; i++) {
        Map<String, Object> schema = new HashMap<>();
        schema.put("compatibility", "BACKWARD");
        schema.put("id", String.valueOf(i));
        schema.put("schema", "my schema");
        messages.put(i, schema);
      }
  
      return messages;
    }
  
    private List<MessageSchema> createIncompleteMessages(int numOfEntries) {
  
      List<MessageSchema> messages = new ArrayList<>();
      for (int i = 1; i <= numOfEntries; i++) {
        MessageSchema msg = new MessageSchema();
        msg.setSchemaversion(String.valueOf(i));
        msg.setSchemaversion(String.valueOf(i));
        msg.setSchemafull(null);
        msg.setCompatibility(null);
        msg.setTeamId(103);
        msg.setTenantId(101);
        msg.setTopicname(TESTTOPIC);
        messages.add(msg);
      }
  
      return messages;
    }
  
    private List<MessageSchema> createMessages(int numOfEntries) {
  
      List<MessageSchema> messages = new ArrayList<>();
      for (int i = 1; i <= numOfEntries; i++) {
        MessageSchema msg = new MessageSchema();
        msg.setSchemaversion(String.valueOf(i));
        msg.setSchemafull("1");
        msg.setSchemaId(Integer.valueOf(i));
        msg.setCompatibility("NOT_SET");
        msg.setTeamId(103);
        msg.setTenantId(101);
        msg.setTopicname(TESTTOPIC);
        messages.add(msg);
      }
  
      return messages;
    }
  
    private List<MessageSchema> createDBMessages(int numOfEntries) {
  
      List<MessageSchema> messages = new ArrayList<>();
      for (int i = 1; i <= numOfEntries; i++) {
        MessageSchema msg = new MessageSchema();
        msg.setSchemaversion(String.valueOf(i) + ".0");
        msg.setSchemafull("1");
        msg.setSchemaId(Integer.valueOf(i));
        msg.setCompatibility("NOT_SET");
        msg.setTeamId(103);
        msg.setTenantId(101);
        msg.setTopicname(TESTTOPIC);
        messages.add(msg);
      }
  
      return messages;
    }
  
    private Topic createTopic(String topicName, String envName) {
      Topic t = new Topic();
      t.setTopicname(topicName);
      t.setTopicid(1);
      t.setEnvironment(envName);
      t.setTeamId(10);
      return t;
    }
  
    private List<Env> createListOfEnvs(KafkaClustersType clusterType, int numberOfEnvs) {
      ArrayList<Env> envs = new ArrayList<>();
      for (int i = 1; i <= numberOfEnvs; i++) {
        Env e = new Env();
        e.setEnvExists("true");
        e.setName("test-" + i);
        e.setId(String.valueOf(i));
        e.setTenantId(101);
        e.setClusterId(i);
        e.setType(clusterType.value);
        when(manageDatabase.getEnv(eq(101), eq(i))).thenReturn(Optional.of(e));
        envs.add(e);
      }
  
      return envs;
    }
  
    private KwClusters createCluster(KafkaClustersType clusterType) {
      KwClusters cluster = new KwClusters();
      cluster.setClusterId(1);
      cluster.setClusterType(clusterType.value);
      cluster.setBootstrapServers("server:8081");
      cluster.setTenantId(101);
      return cluster;
    }
  
    private void stubSchemaPromotionInfo(
        String testtopic, KafkaClustersType clusterType, int numberOfEnvs) throws Exception {
      createListOfEnvs(KafkaClustersType.SCHEMA_REGISTRY, numberOfEnvs);
  
      when(commonUtilsService.getTenantId(any())).thenReturn(101);
      when(manageDatabase.getClusters(eq(KafkaClustersType.SCHEMA_REGISTRY), eq(101)))
          .thenReturn(createClusterMap(numberOfEnvs));
      when(clusterApiService.getAvroSchema(any(), any(), any(), eq(testtopic), eq(101)))
          .thenReturn(utilMethods.createSchemaList())
          .thenReturn(null);
  
      stubEnvironments();
    }
  
    private void stubEnvironments() {
      Env kafkaEnv1 = new Env();
      kafkaEnv1.setName("DEV");
      kafkaEnv1.setId("1");
      kafkaEnv1.setClusterId(1);
      EnvTag envTag = new EnvTag();
      envTag.setId("3");
      envTag.setName("DEV");
      kafkaEnv1.setAssociatedEnv(envTag);
  
      Env kafkaEnv2 = new Env();
      kafkaEnv2.setName("TST");
      kafkaEnv2.setId("2");
      kafkaEnv2.setClusterId(2);
      EnvTag envTag2 = new EnvTag();
      envTag2.setId("4");
      envTag2.setName("TST");
      kafkaEnv2.setAssociatedEnv(envTag2);
  
      Env schemaEnv1 = new Env();
      schemaEnv1.setId("3");
      schemaEnv1.setName("DEV");
      schemaEnv1.setClusterId(1);
      schemaEnv1.setAssociatedEnv(new EnvTag("1", "DEV"));
  
      Env schemaEnv2 = new Env();
      schemaEnv2.setId("4");
      schemaEnv2.setName("TST");
      schemaEnv2.setClusterId(1);
      schemaEnv2.setAssociatedEnv(new EnvTag("2", "TST"));
      KwTenantConfigModel model = new KwTenantConfigModel();
      model.setRequestTopicsEnvironmentsList(List.of("1"));
  
      when(handleDbRequests.getEnvDetails("1", 101)).thenReturn(kafkaEnv1);
      when(handleDbRequests.getEnvDetails("2", 101)).thenReturn(kafkaEnv2);
      when(handleDbRequests.getEnvDetails("3", 101)).thenReturn(schemaEnv1);
      when(handleDbRequests.getEnvDetails("4", 101)).thenReturn(schemaEnv2);
      when(commonUtilsService.isCreateNewSchemaAllowed(eq("3"), eq(101))).thenReturn(true);
      when(commonUtilsService.isCreateNewSchemaAllowed(eq("4"), eq(101))).thenReturn(false);
    }
  
    private void stubKafkaPromotion(String testtopic, int numberOfEnvs) throws Exception {
  
      when(commonUtilsService.getEnvsFromUserId(anyString()))
          .thenReturn(
              new HashSet<>() {
                {
                  add("1");
                  add("2");
                  add("3");
                  add("4");
                  add("5");
                  add("6");
                }
              });
  
      mockTenantConfig();
  
      when(manageDatabase.getTeamNameFromTeamId(101, 10)).thenReturn(TEAM_1);
      when(manageDatabase.getClusters(KafkaClustersType.SCHEMA_REGISTRY, 101))
          .thenReturn(createClusterMap(numberOfEnvs));
    }
  
    private Map<Integer, KwClusters> createClusterMap(int numberOfClusters) {
      Map<Integer, KwClusters> map = new HashMap<>();
  
      for (int i = 0; i < numberOfClusters; i++) {
        map.put(i + 1, createCluster(KafkaClustersType.SCHEMA_REGISTRY));
      }
      return map;
    }
  
    private void mockTenantConfig() {
      when(manageDatabase.getTenantConfig())
          .thenReturn(
              new HashMap<>() {
                {
                  put(101, createModel());
                }
              });
    }
  
    private KwTenantConfigModel createModel() {
      KwTenantConfigModel model = new KwTenantConfigModel();
      model.setBaseSyncEnvironment("1");
      model.setOrderOfTopicPromotionEnvsList(List.of("1", "2", "3", "4", "5", "6", "7"));
      return model;
    }
  
    private List<Acl> getAclsSOT(String topicName) {
      List<Acl> aclList = new ArrayList<>();
  
      Acl aclReq = new Acl();
      aclReq.setReq_no(1001);
      aclReq.setTopicname(topicName);
      aclReq.setTeamId(1);
      aclReq.setAclip("2.1.2.1");
      aclReq.setAclssl(null);
      aclReq.setEnvironment("1");
      aclReq.setConsumergroup("mygrp1");
      aclReq.setAclType(AclType.CONSUMER.value);
  
      aclList.add(aclReq);
      return aclList;
    }
  
    private void stubUserInfo() {
      when(handleDbRequests.getUsersInfo(anyString())).thenReturn(userInfo);
      when(userInfo.getTeamId()).thenReturn(10);
      when(mailService.getUserName(any())).thenReturn(TEAM_ID);
      when(commonUtilsService.getTeamId(eq(TEAM_ID))).thenReturn(10);
    }
  }
  