package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.TOPICS_VLD_ERR_121;
import static io.aiven.klaw.helpers.KwConstants.KLAW_OPTIONAL_PERMISSION_NEW_TOPIC_CREATION_KEY;
import static org.assertj.core.api.Assertions.assertThat;
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 io.aiven.klaw.UtilMethods;
import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.CRUDResponse;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.KwClusters;
import io.aiven.klaw.dao.MessageSchema;
import io.aiven.klaw.dao.Topic;
import io.aiven.klaw.dao.TopicRequest;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.error.KlawNotAuthorizedException;
import io.aiven.klaw.helpers.KwConstants;
import io.aiven.klaw.helpers.db.rdbms.HandleDbRequestsJdbc;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.KwTenantConfigModel;
import io.aiven.klaw.model.TopicConfigEntry;
import io.aiven.klaw.model.TopicInfo;
import io.aiven.klaw.model.enums.AclPatternType;
import io.aiven.klaw.model.enums.AclType;
import io.aiven.klaw.model.enums.ApiResultStatus;
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.TopicRequestModel;
import io.aiven.klaw.model.response.TopicDetailsPerEnv;
import io.aiven.klaw.model.response.TopicRequestsResponseModel;
import io.aiven.klaw.model.response.TopicTeamResponse;
import java.sql.SQLDataException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Stream;
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.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
import org.springframework.dao.DataIntegrityViolationException;
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 TopicControllerServiceTest {

  public static final String TOPIC_1 = "topic1";
  public static final String EXPECTED_SUCCESS_RESPONSE =
      "Topic Status: success, TopicSchemaStatus: success";
  @Mock private ClusterApiService clusterApiService;

  @Mock private UserDetails userDetails;

  @Mock private UserInfo userInfo;

  @Mock private ManageDatabase manageDatabase;

  @Mock private HandleDbRequestsJdbc handleDbRequests;

  @Mock CommonUtilsService commonUtilsService;

  @Mock private MailUtils mailService;

  private TopicControllerService topicControllerService;

  @Mock RolesPermissionsControllerService rolesPermissionsControllerService;

  @Mock Map<Integer, KwTenantConfigModel> tenantConfig;

  @Mock KwTenantConfigModel tenantConfigModel;

  @Captor ArgumentCaptor<TopicRequest> topicRequestCaptor;
  private Env env;

  private UtilMethods utilMethods;

  @BeforeEach
  public void setUp() throws Exception {
    this.topicControllerService = new TopicControllerService(clusterApiService, mailService);
    utilMethods = new UtilMethods();
    this.env = new Env();
    env.setId("1");
    env.setName("DEV");
    ReflectionTestUtils.setField(topicControllerService, "manageDatabase", manageDatabase);
    ReflectionTestUtils.setField(topicControllerService, "mailService", mailService);
    ReflectionTestUtils.setField(topicControllerService, "commonUtilsService", commonUtilsService);
    ReflectionTestUtils.setField(
        topicControllerService,
        "rolesPermissionsControllerService",
        rolesPermissionsControllerService);

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

  @Test
  @Order(1)
  public void createTopicsSuccessAdvancedTopicConfigs()
      throws KlawException, KlawNotAuthorizedException {
    Map<String, String> resultMap = new HashMap<>();
    resultMap.put("result", ApiResultStatus.SUCCESS.value);

    when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
    when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
    when(tenantConfigModel.getBaseSyncEnvironment()).thenReturn("1");
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(handleDbRequests.requestForTopic(any())).thenReturn(resultMap);
    when(commonUtilsService.getEnvProperty(anyInt(), anyString())).thenReturn("1");

    ApiResponse apiResponse =
        topicControllerService.createTopicsCreateRequest(getTopicWithAdvancedConfigs());
    assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @Order(2)
  public void createTopicsSuccessDefaultValues() throws KlawException, KlawNotAuthorizedException {
    Map<String, String> resultMap = new HashMap<>();
    resultMap.put("result", ApiResultStatus.SUCCESS.value);

    when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
    when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
    when(tenantConfigModel.getBaseSyncEnvironment()).thenReturn("1");
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(handleDbRequests.requestForTopic(any())).thenReturn(resultMap);
    when(commonUtilsService.getEnvProperty(anyInt(), anyString())).thenReturn("1");

    ApiResponse apiResponse =
        topicControllerService.createTopicsCreateRequest(getTopicWithDefaultConfigs());
    assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  // invalid partitions
  @Test
  @Order(3)
  public void createTopicsFailureInvalidPartitions()
      throws KlawException, KlawNotAuthorizedException {
    Map<String, String> resultMap = new HashMap<>();
    resultMap.put("result", "failure");

    when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
    when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
    when(tenantConfigModel.getBaseSyncEnvironment()).thenReturn("1");
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(handleDbRequests.requestForTopic(any())).thenReturn(resultMap);
    when(commonUtilsService.getEnvProperty(anyInt(), anyString())).thenReturn("1");

    ApiResponse apiResponse = topicControllerService.createTopicsCreateRequest(getFailureTopic1());
    assertThat(apiResponse.getMessage()).contains("failure");
  }

  @Test
  @Order(4)
  public void createTopicsFailureInvalidClusterTenantIds()
      throws KlawException, KlawNotAuthorizedException {

    when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
    when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
    when(tenantConfigModel.getBaseSyncEnvironment()).thenReturn("1");
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvListsIncorrect1());
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(commonUtilsService.getEnvProperty(anyInt(), anyString())).thenReturn("1");
    Map<String, String> res = new HashMap<>();
    res.put("result", ApiResultStatus.FAILURE.value);
    when(handleDbRequests.requestForTopic(any())).thenReturn(res);

    ApiResponse apiResponse = topicControllerService.createTopicsCreateRequest(getFailureTopic1());
    assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.FAILURE.value);
  }

  @Test
  @Order(5)
  public void createTopicsFailureInvalidTopicDescriptionLength()
      throws KlawException, KlawNotAuthorizedException {

    when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
    when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
    when(tenantConfigModel.getBaseSyncEnvironment()).thenReturn("1");
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvListsIncorrect1());
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(commonUtilsService.getEnvProperty(anyInt(), anyString())).thenReturn("1");

    SQLDataException causeException =
        new SQLDataException(
            "Seed size for the object exceeds the\n"
                + "column size in the database. Value too long for column...");
    DataIntegrityViolationException exception =
        new DataIntegrityViolationException("Error message", causeException);
    when(handleDbRequests.requestForTopic(any())).thenThrow(exception);

    ApiResponse apiResponse =
        topicControllerService.createTopicsCreateRequest(getFailureTopicWithLongDescription());
    assertThat(apiResponse.getMessage())
        .isEqualTo("Failure. Topic description exceeds allowed length.");
  }

  @Test
  @Order(6)
  public void createTopicDeleteRequestFailureTopicAlreadyExists() {
    String topicName = "testtopic1";
    String envId = "1";
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(getListTopicRequests());
    when(handleDbRequests.existsTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(true);
    try {
      ApiResponse apiResponse =
          topicControllerService.createTopicDeleteRequest(topicName, envId, false);
      assertThat(apiResponse.getMessage())
          .isEqualTo("Failure. A delete topic request already exists.");
    } catch (KlawException | KlawNotAuthorizedException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(7)
  public void createTopicDeleteRequestFailureNotOwnerTeamOfTopic() {
    String topicName = "testtopic1";
    String envId = "1";
    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    when(handleDbRequests.existsTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(false);
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    try {
      ApiResponse apiResponse =
          topicControllerService.createTopicDeleteRequest(topicName, envId, false);
      assertThat(apiResponse.getMessage())
          .isEqualTo(
              "Failure. Sorry, you cannot delete this topic, as you are not part of this team.");
    } catch (KlawException | KlawNotAuthorizedException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(8)
  public void createTopicDeleteRequestFailureTopicWithSubscriptions() {
    String topicName = "testtopic1";
    String envId = "1";
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(1);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    when(handleDbRequests.getSyncAcls(anyString(), anyString(), anyInt()))
        .thenReturn(utilMethods.getAcls());
    try {
      ApiResponse apiResponse =
          topicControllerService.createTopicDeleteRequest(topicName, envId, false);
      assertThat(apiResponse.getMessage())
          .isEqualTo(
              "Failure. There are existing subscriptions for topic. Please get them deleted before.");
    } catch (KlawException | KlawNotAuthorizedException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(9)
  public void createTopicDeleteRequestFailureTopicNotInCluster() {
    String topicName = "testtopic1";
    String envId = "2";
    stubUserInfo();
    when(userInfo.getTeamId()).thenReturn(1);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    when(manageDatabase.getTeamsAndAllowedEnvs(anyInt(), anyInt()))
        .thenReturn(Collections.singletonList("1"));
    when(commonUtilsService.getTeamId(anyString())).thenReturn(1);
    try {
      ApiResponse apiResponse =
          topicControllerService.createTopicDeleteRequest(topicName, envId, false);
      assertThat(apiResponse.getMessage())
          .isEqualTo("Failure. Topic not found on cluster: " + topicName);
    } catch (KlawException | KlawNotAuthorizedException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(10)
  public void createTopicDeleteRequestSuccessDefaultValues() {
    String topicName = "testtopic1";
    String envId = "1";
    stubUserInfo();
    when(commonUtilsService.getTeamId(anyString())).thenReturn(1);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    when(handleDbRequests.getSyncAcls(anyString(), anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    Map<String, String> deleteReqResult = new HashMap<>();
    deleteReqResult.put("result", ApiResultStatus.SUCCESS.value);
    when(handleDbRequests.requestForTopic(any())).thenReturn(deleteReqResult);
    try {
      ApiResponse apiResponse =
          topicControllerService.createTopicDeleteRequest(topicName, envId, false);
      assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    } catch (KlawException e) {
      throw new RuntimeException(e);
    } catch (KlawNotAuthorizedException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(11)
  public void createClaimTopicRequestFailureRequestAlreadyExists() {
    String topicName = "testtopic1";
    String envId = "1";
    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(getListTopicRequests());
    when(handleDbRequests.existsTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(true);
    try {
      ApiResponse apiResponse = topicControllerService.createClaimTopicRequest(topicName, envId);
      assertThat(apiResponse.getMessage())
          .isEqualTo("Failure. A request already exists for this topic.");
    } catch (KlawException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(12)
  public void createClaimTopicRequestSuccessDefaultValues() {
    String topicName = "testtopic1";
    String envId = "1";
    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getTopicRequests(anyString(), anyString(), anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    List<UserInfo> userList = utilMethods.getUserInfoList("testuser", "");
    userList.get(0).setTeamId(1);
    when(handleDbRequests.getAllUsersInfo(anyInt())).thenReturn(userList);
    Map<String, String> claimReqResult = new HashMap<>();
    claimReqResult.put("result", ApiResultStatus.SUCCESS.value);
    when(handleDbRequests.requestForTopic(any())).thenReturn(claimReqResult);
    try {
      ApiResponse apiResponse = topicControllerService.createClaimTopicRequest(topicName, envId);
      assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    } catch (KlawException e) {
      throw new RuntimeException(e);
    }
  }

  @Test
  @Order(13)
  public void getTopicRequests() {
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(handleDbRequests.getAllTopicRequests(
            anyString(), anyString(), eq(null), eq(null), eq(null), eq(false), anyInt()))
        .thenReturn(getListTopicRequests());
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("INFRATEAM");

    List<TopicRequestsResponseModel> listTopicRqs =
        topicControllerService.getTopicRequests(
            "1",
            "",
            null,
            "all",
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            false);
    assertThat(listTopicRqs).hasSize(2);
  }

  @Test
  @Order(14)
  public void getTopicRequestsClaimType() {
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<TopicRequest> topicRequests = getListTopicRequests();
    topicRequests.get(0).setRequestOperationType(RequestOperationType.CLAIM.value);
    when(handleDbRequests.getAllTopicRequests(
            anyString(), anyString(), eq(null), eq(null), eq(null), eq(false), anyInt()))
        .thenReturn(topicRequests);
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("INFRATEAM");
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopics());

    List<TopicRequestsResponseModel> listTopicRqs =
        topicControllerService.getTopicRequests(
            "1",
            "",
            null,
            "all",
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            false);
    assertThat(listTopicRqs).hasSize(2);
  }

  @Test
  @Order(15)
  public void getTopicRequestsOtherType() {
    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<TopicRequest> topicRequests = getListTopicRequests();
    topicRequests.get(0).setRequestOperationType(RequestOperationType.CLAIM.value);
    when(handleDbRequests.getAllTopicRequests(
            anyString(), anyString(), eq(null), eq(null), eq(null), eq(false), anyInt()))
        .thenReturn(topicRequests);
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("INFRATEAM");
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(utilMethods.getTopics());

    List<TopicRequestsResponseModel> listTopicRqs =
        topicControllerService.getTopicRequests(
            "1",
            "",
            null,
            "created",
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            false);
    assertThat(listTopicRqs).hasSize(2);
  }

  @Test
  @Order(16)
  public void getTopicTeamOnlyFailureNoTopicsFound() {
    String topicName = "testtopic";
    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    TopicTeamResponse topicTeamMap =
        topicControllerService.getTopicTeamOnly(topicName, AclPatternType.PREFIXED);
    assertThat(topicTeamMap.getError()).contains("There are no topics found with this prefix.");
  }

  @Test
  @Order(17)
  public void getTopicTeamOnlyFailurePrefixInMultipleTeams() {
    String topicName = "testtopic";
    List<Topic> topicList = utilMethods.getTopics();
    Topic topic = new Topic();
    topic.setTopicname("testtopic2");
    topic.setEnvironment("1");
    topicList.add(topic);

    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getAllTopics(anyInt())).thenReturn(topicList);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    TopicTeamResponse topicTeamMap =
        topicControllerService.getTopicTeamOnly(topicName, AclPatternType.PREFIXED);
    assertThat(topicTeamMap.getError())
        .contains("There are at least two topics with same prefix owned by different teams.");
  }

  @Test
  @Order(18)
  public void getTopicTeamOnlySuccessPatternPrefixed() {
    String topicName = "testtopic";
    String teamName = "TestTeam";
    stubUserInfo();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(handleDbRequests.getAllTopics(anyInt())).thenReturn(utilMethods.getTopics());
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn(teamName);
    TopicTeamResponse topicTeamMap =
        topicControllerService.getTopicTeamOnly(topicName, AclPatternType.PREFIXED);
    assertThat(topicTeamMap.getTeam()).isEqualTo(teamName);
  }

  @Test
  @Order(19)
  public void getTopicTeamOnlySuccessPatternLiteral() {
    String topicName = "testtopic";
    String teamName = "TestTeam";
    stubUserInfo();
    List<Topic> topicList = utilMethods.getTopics();
    when(commonUtilsService.getTenantId(anyString())).thenReturn(101);
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt())).thenReturn(topicList);
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn(teamName);
    TopicTeamResponse topicTeamMap =
        topicControllerService.getTopicTeamOnly(topicName, AclPatternType.LITERAL);
    assertThat(topicTeamMap.getTeam()).isEqualTo(teamName);
  }

  @Test
  @Order(20)
  public void getCreatedTopicRequests() {
    List<TopicRequest> listTopicReqs = new ArrayList<>();
    listTopicReqs.add(getCorrectTopicDao());
    listTopicReqs.add(getCorrectTopicDao());

    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(handleDbRequests.getCreatedTopicRequests(
            anyString(),
            anyString(),
            anyBoolean(),
            anyInt(),
            eq(null),
            eq(null),
            eq(null),
            eq(null)))
        .thenReturn(listTopicReqs);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("INFTATEAM");

    List<TopicRequestsResponseModel> topicList =
        topicControllerService.getTopicRequestsForApprover(
            "1",
            "",
            "all",
            null,
            null,
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME);

    assertThat(topicList).hasSize(2);
  }

  @Test
  @Order(21)
  public void getCreatedTopicRequestsWithMoreElements() {
    List<TopicRequest> listTopicReqs = new ArrayList<>();
    listTopicReqs.add(getTopicRequest(TOPIC_1));
    listTopicReqs.add(getTopicRequest("topic2"));
    listTopicReqs.add(getTopicRequest("topic3"));
    listTopicReqs.add(getTopicRequest("topic4"));
    listTopicReqs.add(getTopicRequest("topic5"));

    stubUserInfo();
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());
    when(handleDbRequests.getCreatedTopicRequests(
            anyString(),
            anyString(),
            anyBoolean(),
            anyInt(),
            eq(null),
            eq(null),
            eq(null),
            eq(null)))
        .thenReturn(listTopicReqs);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("INFTATEAM");

    List<TopicRequestsResponseModel> topicList =
        topicControllerService.getTopicRequestsForApprover(
            "1",
            "",
            "all",
            null,
            null,
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME);

    assertThat(topicList).hasSize(5);
    assertThat(topicList.get(0).getTopicpartitions()).isEqualTo(2);
    assertThat(topicList.get(0).getTopicname()).isEqualTo(TOPIC_1);
    assertThat(topicList.get(1).getTopicname()).isEqualTo("topic2");
  }

  @Test
  @Order(22)
  public void deleteTopicRequests() throws KlawException {
    when(handleDbRequests.deleteTopicRequest(anyInt(), anyString(), anyInt()))
        .thenReturn(ApiResultStatus.SUCCESS.value);
    when(mailService.getUserName(any())).thenReturn("uiuser1");
    ApiResponse resultResp = topicControllerService.deleteTopicRequests("1001");
    assertThat(resultResp.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @Order(23)
  public void approveTopicRequests() throws KlawException {
    String topicName = TOPIC_1;
    int topicId = 1001;
    TopicRequest topicRequest = getTopicRequest(topicName);
    ApiResponse apiResponse = ApiResponse.SUCCESS;

    stubUserInfo();
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_TOPICS))
        .thenReturn(false);
    when(handleDbRequests.getTopicRequestsForTopic(anyInt(), anyInt())).thenReturn(topicRequest);
    when(handleDbRequests.updateTopicRequest(any(), anyString()))
        .thenReturn(
            CRUDResponse.<Topic>builder().resultStatus(ApiResultStatus.SUCCESS.value).build());
    when(clusterApiService.approveTopicRequests(
            anyString(),
            anyString(),
            anyInt(),
            anyString(),
            anyString(),
            any(),
            anyInt(),
            anyBoolean()))
        .thenReturn(new ResponseEntity<>(apiResponse, HttpStatus.OK));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));

    ApiResponse apiResponse1 = topicControllerService.approveTopicRequests(topicId + "");
    assertThat(apiResponse1.getMessage()).isEqualTo("Topic Status: success");
  }

  @Test
  @Order(24)
  public void approveTopicClaimRequests() throws KlawException {
    String topicName = TOPIC_1;
    int topicId = 1001;
    TopicRequest topicRequest = getTopicRequest(topicName);
    topicRequest.setRequestOperationType(RequestOperationType.CLAIM.value);
    ApiResponse apiResponse = ApiResponse.SUCCESS;

    stubUserInfo();
    when(handleDbRequests.getTopicRequestsForTopic(anyInt(), anyInt())).thenReturn(topicRequest);
    when(handleDbRequests.updateTopicRequest(any(), anyString()))
        .thenReturn(
            CRUDResponse.<Topic>builder().resultStatus(ApiResultStatus.SUCCESS.value).build());
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_TOPICS))
        .thenReturn(false);
    when(clusterApiService.approveTopicRequests(
            anyString(),
            anyString(),
            anyInt(),
            anyString(),
            anyString(),
            any(),
            anyInt(),
            anyBoolean()))
        .thenReturn(new ResponseEntity<>(apiResponse, HttpStatus.OK));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(handleDbRequests.addToSynctopics(any()))
        .thenReturn(
            CRUDResponse.<Topic>builder().resultStatus(ApiResultStatus.SUCCESS.value).build());
    when(handleDbRequests.updateTopicRequestStatus(any(), anyString()))
        .thenReturn(ApiResultStatus.SUCCESS.value);

    ApiResponse apiResponse1 = topicControllerService.approveTopicRequests(topicId + "");
    assertThat(apiResponse1.getMessage()).isEqualTo("Topic Status: success");
  }

  @Test
  @Order(25)
  public void approveTopicUpdateRequests() throws KlawException {
    String topicName = TOPIC_1;
    int topicId = 1001;
    TopicRequest topicRequest = getTopicRequest(topicName);
    topicRequest.setRequestOperationType(RequestOperationType.UPDATE.value);
    ApiResponse apiResponse = ApiResponse.SUCCESS;

    stubUserInfo();
    when(handleDbRequests.getTopicRequestsForTopic(anyInt(), anyInt())).thenReturn(topicRequest);
    when(handleDbRequests.updateTopicRequest(any(), anyString()))
        .thenReturn(
            CRUDResponse.<Topic>builder().resultStatus(ApiResultStatus.SUCCESS.value).build());
    when(commonUtilsService.getTopicsForTopicName(anyString(), anyInt()))
        .thenReturn(List.of(getTopic(topicName)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_TOPICS))
        .thenReturn(false);
    when(clusterApiService.approveTopicRequests(
            anyString(),
            anyString(),
            anyInt(),
            anyString(),
            anyString(),
            any(),
            anyInt(),
            anyBoolean()))
        .thenReturn(new ResponseEntity<>(apiResponse, HttpStatus.OK));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getTeamNameFromTeamId(anyInt(), anyInt())).thenReturn("INFRATEAM");
    when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(utilMethods.getEnvLists());

    ApiResponse apiResponse1 = topicControllerService.approveTopicRequests(topicId + "");
    assertThat(apiResponse1.getMessage()).isEqualTo("Topic Status: success");
  }

  @Test
  @Order(26)
  public void approveTopicRequestsFailureResponseFromCluster() throws KlawException {
    String topicName = TOPIC_1;
    int topicId = 1001;
    TopicRequest topicRequest = getTopicRequest(topicName);
    ApiResponse apiResponse = ApiResponse.FAILURE;

    stubUserInfo();
    when(handleDbRequests.getTopicRequestsForTopic(anyInt(), anyInt())).thenReturn(topicRequest);
    when(handleDbRequests.updateTopicRequest(any(), anyString()))
        .thenReturn(
            CRUDResponse.<Topic>builder().resultStatus(ApiResultStatus.SUCCESS.value).build());
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.APPROVE_TOPICS))
        .thenReturn(false);
    when(clusterApiService.approveTopicRequests(
            anyString(),
            anyString(),
            anyInt(),
            anyString(),
            anyString(),
            any(),
            anyInt(),
            anyBoolean()))
        .thenReturn(new ResponseEntity<>(apiResponse, HttpStatus.OK));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));

    ApiResponse apiResponse1 = topicControllerService.approveTopicRequests("" + topicId);
    assertThat(apiResponse1.getMessage()).isEqualTo("Topic Status: failure");
  }

  @Test
  @Order(27)
