package io.aiven.klaw.service;

import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_102;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_104;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_105;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_106;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_107;
import static io.aiven.klaw.error.KlawErrorMessages.ENV_CLUSTER_TNT_ERR_108;
import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_KAFKA_CONNECT_ENVS;
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.assertj.core.api.Assertions.assertThatExceptionOfType;
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.config.ManageDatabase;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.EnvID;
import io.aiven.klaw.dao.EnvTag;
import io.aiven.klaw.dao.KwClusters;
import io.aiven.klaw.dao.KwProperties;
import io.aiven.klaw.dao.KwRolesPermissions;
import io.aiven.klaw.dao.KwTenants;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawBadRequestException;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.error.KlawValidationException;
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.KwTenantModel;
import io.aiven.klaw.model.enums.ApiResultStatus;
import io.aiven.klaw.model.enums.ClusterStatus;
import io.aiven.klaw.model.enums.EntityType;
import io.aiven.klaw.model.enums.KafkaClustersType;
import io.aiven.klaw.model.enums.KafkaFlavors;
import io.aiven.klaw.model.enums.KafkaSupportedProtocol;
import io.aiven.klaw.model.enums.MetadataOperationType;
import io.aiven.klaw.model.enums.PermissionType;
import io.aiven.klaw.model.enums.RolesType;
import io.aiven.klaw.model.requests.EnvModel;
import io.aiven.klaw.model.requests.KwClustersModel;
import io.aiven.klaw.model.response.ClusterInfo;
import io.aiven.klaw.model.response.EnvIdInfo;
import io.aiven.klaw.model.response.EnvModelResponse;
import io.aiven.klaw.model.response.EnvParams;
import io.aiven.klaw.model.response.EnvUpdatedStatus;
import io.aiven.klaw.model.response.KwClustersModelResponse;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Stream;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.MethodOrderer;
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.security.core.userdetails.UserDetails;
import org.springframework.security.test.context.support.WithMockUser;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.test.util.ReflectionTestUtils;

@ExtendWith(SpringExtension.class)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class EnvsClustersTenantsControllerServiceTest {

  private EnvsClustersTenantsControllerService service;
  @Mock private MailUtils mailService;
  @Mock private UserDetails userDetails;

  @Mock private CommonUtilsService commonUtilsService;

  @Mock ManageDatabase manageDatabase;

  @Mock private ClusterApiService clusterApiService;

  @Mock private UsersTeamsControllerService usersTeamsControllerService;
  @Mock private HandleDbRequestsJdbc handleDbRequestsJdbc;
  @Mock private DefaultDataService defaultDataService;

  @Captor ArgumentCaptor<Env> envCapture;

  @BeforeEach
  public void setup() {
    service = new EnvsClustersTenantsControllerService();
    ReflectionTestUtils.setField(service, "mailService", mailService);
    ReflectionTestUtils.setField(service, "commonUtilsService", commonUtilsService);
    ReflectionTestUtils.setField(service, "manageDatabase", manageDatabase);
    ReflectionTestUtils.setField(service, "clusterApiService", clusterApiService);
    ReflectionTestUtils.setField(
        service, "usersTeamsControllerService", usersTeamsControllerService);
    ReflectionTestUtils.setField(service, "defaultDataService", defaultDataService);
    when(mailService.getUserName(userDetails)).thenReturn("testuser");
    when(handleDbRequestsJdbc.getUsersInfo(any())).thenReturn(buildUserInfo());
    when(manageDatabase.getHandleDbRequests()).thenReturn(handleDbRequestsJdbc);
    when(commonUtilsService.getTenantId(any())).thenReturn(101);
    when(commonUtilsService.getPrincipal()).thenReturn(userDetails);
    when(commonUtilsService.isNotAuthorizedUser(any(), any(PermissionType.class))).thenReturn(true);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewEnv() throws KlawException, KlawValidationException {
    EnvModel env = getTestEnvModel(null);
    Env SchemaEnv = generateKafkaEnv("9", "Schema");
    when(handleDbRequestsJdbc.addNewEnv(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(handleDbRequestsJdbc.getEnvDetails(anyString(), anyInt())).thenReturn(SchemaEnv);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);
    assertThat(response.getMessage()).contains("success");
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewEnvNameAlreadyInUse() throws KlawException, KlawValidationException {
    EnvModel env = getTestEnvModel(null);
    when(handleDbRequestsJdbc.getAllEnvs(anyInt()))
        .thenReturn(
            List.of(
                buildEnv("4", 101, "DEV", KafkaClustersType.KAFKA, 4),
                buildEnv("5", 101, "TST", KafkaClustersType.SCHEMA_REGISTRY, 5)));
    when(manageDatabase.getKafkaEnvList(anyInt()))
        .thenReturn(List.of(buildEnv("4", 101, "DEV", KafkaClustersType.KAFKA, 4)));
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);
    assertThat(response.getMessage())
        .contains("Failure. Please choose a different name. This environment name already exists.");
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewEnvWithAssociatedEnv() throws KlawException, KlawValidationException {
    EnvModel env = getTestSchemaEnvModel(new EnvTag("1", "Kafka"));
    Env kafkaEnv = generateKafkaEnv("1", "Kafka");
    when(handleDbRequestsJdbc.addNewEnv(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(handleDbRequestsJdbc.getEnvDetails(eq("1"), eq(101)))
        .thenReturn(kafkaEnv)
        .thenReturn(null);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);
    kafkaEnv.setAssociatedEnv(env.getAssociatedEnv());
    verify(handleDbRequestsJdbc, times(1)).addNewEnv(eq(kafkaEnv));
    // 1 for saving the schema env 1 for updating the kafka env
    verify(handleDbRequestsJdbc, times(2)).addNewEnv(any(Env.class));
    assertThat(response.getMessage()).contains("success");
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void updateEnvWithDifferentAssociatedEnv() throws KlawException, KlawValidationException {
    EnvModel env = getTestSchemaEnvModel(new EnvTag("1", "Kafka"));
    Env SchemaEnv = new Env();
    SchemaEnv.setId("9");
    SchemaEnv.setName("DEV_SCH");
    SchemaEnv.setType(KafkaClustersType.SCHEMA_REGISTRY.value);
    SchemaEnv.setTenantId(101);
    SchemaEnv.setAssociatedEnv(new EnvTag("2", "Kafka"));

    Env kafkaEnv = generateKafkaEnv("1", "Kafka");
    when(handleDbRequestsJdbc.addNewEnv(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(handleDbRequestsJdbc.getEnvDetails(eq("1"), eq(101)))
        .thenReturn(kafkaEnv)
        .thenReturn(SchemaEnv);
    when(handleDbRequestsJdbc.getEnvDetails(eq("2"), eq(101)))
        .thenReturn(generateKafkaEnv("2", "Kafka"));
    when(handleDbRequestsJdbc.getNextSeqIdAndUpdate(eq(EntityType.ENVIRONMENT.name()), eq(101)))
        .thenReturn(1);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);
    kafkaEnv.setAssociatedEnv(env.getAssociatedEnv());
    verify(handleDbRequestsJdbc, times(1)).addNewEnv(eq(kafkaEnv));
    // 1 for saving the schema env 1 for updating the kafka env 1 for updating previous kafka env
    verify(handleDbRequestsJdbc, times(3)).addNewEnv(any(Env.class));
    assertThat(response.getMessage()).contains("success");
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void updateEnvWithKafkaEnvAlreadyAssociatedWithAnotherEnv() {
    EnvModel env = getTestSchemaEnvModel(new EnvTag("1", "Kafka"));
    Env SchemaEnv = new Env();
    SchemaEnv.setId("9");
    SchemaEnv.setName("DEV_SCH");
    SchemaEnv.setType(KafkaClustersType.SCHEMA_REGISTRY.value);
    SchemaEnv.setTenantId(101);
    SchemaEnv.setAssociatedEnv(new EnvTag("2", "Kafka"));
    Env kafkaEnv = generateKafkaEnv("1", "Kafka");
    kafkaEnv.setAssociatedEnv(new EnvTag("2", "TST_SCH"));
    when(handleDbRequestsJdbc.getEnvDetails(eq("1"), eq(101))).thenReturn(kafkaEnv);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    assertThatExceptionOfType(KlawValidationException.class)
        .isThrownBy(
            () -> {
              service.addNewEnv(env);
            });
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addEnvWithoutAssociatedEnv() throws KlawValidationException, KlawException {
    EnvModel env = getTestSchemaEnvModel(null);
    Env env1 = generateKafkaEnv("1", "Kafka");
    env1.setType(KafkaClustersType.SCHEMA_REGISTRY.value);

    when(handleDbRequestsJdbc.getEnvDetails(eq("1"), eq(101))).thenReturn(env1).thenReturn(null);
    when(handleDbRequestsJdbc.addNewEnv(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);

    assertThat(response.getMessage()).contains("success");
    verify(handleDbRequestsJdbc, times(1)).addNewEnv(any());
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addEnvRemoveAssociatedEnv() throws KlawValidationException, KlawException {
    EnvModel env = getTestSchemaEnvModel(null);
    Env env1 = generateKafkaEnv("1", "Kafka");
    env1.setType(KafkaClustersType.SCHEMA_REGISTRY.value);
    env1.setAssociatedEnv(new EnvTag("2", "TST_SCH"));
    when(handleDbRequestsJdbc.getEnvDetails(eq("1"), eq(101))).thenReturn(env1);
    when(handleDbRequestsJdbc.getEnvDetails(eq("2"), eq(101)))
        .thenReturn(generateKafkaEnv("2", "Kafka"));
    when(handleDbRequestsJdbc.addNewEnv(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(handleDbRequestsJdbc.getNextSeqIdAndUpdate(eq(EntityType.ENVIRONMENT.name()), eq(101)))
        .thenReturn(1);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);

    assertThat(response.getMessage()).contains("success");
    // 1 time for the env we are saving and 1 time for removing an existing mapping of a kafka env.
    verify(handleDbRequestsJdbc, times(2)).addNewEnv(any(Env.class));
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addEnvRemoveAssociatedEnvIncorrectIdSupplied()
      throws KlawValidationException, KlawException {
    EnvModel env = getTestSchemaEnvModel(null);
    Env env1 = generateKafkaEnv("1", "Kafka");
    env1.setType(KafkaClustersType.SCHEMA_REGISTRY.value);
    env1.setAssociatedEnv(new EnvTag("2", "TST_SCH"));
    when(handleDbRequestsJdbc.getEnvDetails(eq("1"), eq(101))).thenReturn(null);
    when(handleDbRequestsJdbc.addNewEnv(any())).thenReturn(ApiResultStatus.SUCCESS.value);
    when(handleDbRequestsJdbc.getNextSeqIdAndUpdate(eq(EntityType.ENVIRONMENT.name()), eq(101)))
        .thenReturn(1);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);
    ApiResponse response = service.addNewEnv(env);

    assertThat(response.getMessage()).contains("success");
    // 1 time for the env we are saving and 1 time for removing an existing mapping of a kafka env.
    verify(handleDbRequestsJdbc, times(1)).addNewEnv(any(Env.class));
  }

  @ParameterizedTest(name = "actual={0} / search={1} / pageNo={2} / expectedNumberOfMatches={4}")
  @MethodSource
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void getEnvs(KafkaClustersType type, String searchBy, String pageNo, int expectedMatches)
      throws KlawValidationException, KlawException {

    when(commonUtilsService.getEnvProperty(eq(101), eq(ORDER_OF_TOPIC_ENVS))).thenReturn("1,2,3");
    // dependent on the type
    when(manageDatabase.getKafkaEnvList(eq(101)))
        .thenReturn(
            List.of(
                buildEnv("1", 101, "DEV", type, 1),
                buildEnv("2", 101, "TST", type, 2),
                buildEnv("3", 101, "PRD", type, 3)));
    when(manageDatabase.getSchemaRegEnvList(eq(101)))
        .thenReturn(
            List.of(
                buildEnv("1", 101, "DEV", type, 1),
                buildEnv("2", 101, "TST", type, 2),
                buildEnv("3", 101, "PRD", type, 3)));
    when(manageDatabase.getKafkaConnectEnvList(eq(101)))
        .thenReturn(
            List.of(
                buildEnv("1", 101, "DEV", type, 1),
                buildEnv("2", 101, "TST", type, 2),
                buildEnv("3", 101, "PRD", type, 3)));

    when(manageDatabase.getClusters(eq(type), eq(101))).thenReturn(buildClusters(type, 3));
    when(manageDatabase.getTenantMap())
        .thenReturn(
            new HashMap<>() {
              {
                put(101, "");
              }
            });
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    List<EnvModelResponse> response = service.getEnvsPaginated(type, "", pageNo, searchBy);

    assertThat(response).hasSize(expectedMatches);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteTenantUnauthorizedUser1() throws KlawException {
    ApiResponse response = service.deleteTenant();

    assertThat(response).isEqualTo(ApiResponse.NOT_AUTHORIZED);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteTenantUnauthorizedUser2() throws KlawException {
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.UPDATE_DELETE_MY_TENANT))
        .thenReturn(false);
    ApiResponse response = service.deleteTenant();

    assertThat(response).isEqualTo(ApiResponse.NOT_AUTHORIZED);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteTenantFailed() throws KlawException {
    int tenantId = 102;
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.UPDATE_DELETE_MY_TENANT))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(any())).thenReturn(tenantId);
    when(manageDatabase.getTenantMap())
        .thenReturn(
            new HashMap<>() {
              {
                put(tenantId, "first");
                put(103, "second");
              }
            });
    String message = "failed";
    when(handleDbRequestsJdbc.disableTenant(tenantId)).thenReturn(message);
    UserInfo info = new UserInfo();
    info.setTenantId(tenantId);
    info.setUsername("first user");
    when(handleDbRequestsJdbc.getAllUsersInfo(tenantId)).thenReturn(List.of(info));
    ApiResponse response = service.deleteTenant();

    assertThat(response.getMessage()).contains(message);
    assertThat(response.isSuccess()).isFalse();
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteTenant() throws KlawException {
    int tenantId = 102;
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.UPDATE_DELETE_MY_TENANT))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(any())).thenReturn(tenantId);
    when(manageDatabase.getTenantMap())
        .thenReturn(
            new HashMap<>() {
              {
                put(tenantId, "first");
                put(103, "second");
              }
            });
    String message = "success";
    when(handleDbRequestsJdbc.disableTenant(tenantId)).thenReturn(message);
    UserInfo info = new UserInfo();
    info.setTenantId(tenantId);
    info.setUsername("first user");
    when(handleDbRequestsJdbc.getAllUsersInfo(tenantId)).thenReturn(List.of(info));
    ApiResponse response = service.deleteTenant();

    assertThat(response.getMessage()).contains(message);
    assertThat(String.valueOf(response.getData())).contains("first");
    assertThat(response.isSuccess()).isTrue();
    verify(usersTeamsControllerService, times(1)).deleteUser("first user", false);
    verify(handleDbRequestsJdbc, times(1)).deleteAllUsers(tenantId);
    verify(handleDbRequestsJdbc, times(1)).deleteAllTeams(tenantId);
    verify(handleDbRequestsJdbc, times(1)).deleteAllEnvs(tenantId);
    verify(handleDbRequestsJdbc, times(1)).deleteAllClusters(tenantId);
    verify(handleDbRequestsJdbc, times(1)).deleteAllRolesPerms(tenantId);
    verify(handleDbRequestsJdbc, times(1)).deleteAllKwProps(tenantId);
    verify(handleDbRequestsJdbc, times(1)).deleteTxnData(tenantId);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addTenantIdExceedMax() throws KlawException {
    when(handleDbRequestsJdbc.getTenants()).thenReturn(List.of(new KwTenants(), new KwTenants()));
    ReflectionTestUtils.setField(service, "maxNumberOfTenantsCanBeCreated", 1);
    ApiResponse response = service.addTenantId(new KwTenantModel(), true);

    assertThat(response.getMessage()).contains(ENV_CLUSTER_TNT_ERR_108);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addTenantIdUnauthorized() throws KlawException {
    when(handleDbRequestsJdbc.getTenants()).thenReturn(List.of(new KwTenants(), new KwTenants()));
    ReflectionTestUtils.setField(service, "maxNumberOfTenantsCanBeCreated", 100);
    ApiResponse response = service.addTenantId(new KwTenantModel(), true);

    assertThat(response).isEqualTo(ApiResponse.NOT_AUTHORIZED);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addTenantIdInternal() throws KlawException {
    KwTenantModel kwTenantModel = new KwTenantModel();
    kwTenantModel.setTenantName("first tenant");
    KwTenants kwTenant1 = new KwTenants();
    kwTenant1.setTenantName("first tenant");
    kwTenant1.setTenantId(101);
    KwTenants kwTenant2 = new KwTenants();
    kwTenant2.setTenantName("second tenant");
    kwTenant2.setTenantId(102);
    when(handleDbRequestsJdbc.getTenants()).thenReturn(List.of(kwTenant1, kwTenant2));
    ReflectionTestUtils.setField(service, "maxNumberOfTenantsCanBeCreated", 100);
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_TENANT))
        .thenReturn(false);
    when(handleDbRequestsJdbc.addNewTenant(any())).thenReturn("add new tenant");
    ApiResponse response = service.addTenantId(kwTenantModel, false);

    assertThat(response.getMessage()).contains("add new tenant");
    assertThat(response.isSuccess()).isTrue();
    verify(commonUtilsService, times(1))
        .updateMetadata(
            anyInt(), eq(EntityType.TENANT), eq(MetadataOperationType.CREATE), eq(null));
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addTenantIdExternal() throws KlawException {
    KwTenantModel kwTenantModel = new KwTenantModel();
    kwTenantModel.setTenantName("first tenant");
    KwTenants kwTenant1 = new KwTenants();
    kwTenant1.setTenantName("first tenant");
    kwTenant1.setTenantId(101);
    KwTenants kwTenant2 = new KwTenants();
    kwTenant2.setTenantName("second tenant");
    kwTenant2.setTenantId(102);
    when(handleDbRequestsJdbc.getTenants()).thenReturn(List.of(kwTenant1, kwTenant2));
    ReflectionTestUtils.setField(service, "maxNumberOfTenantsCanBeCreated", 100);
    ReflectionTestUtils.setField(service, "kwInstallationType", "kwInstallationType");
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_TENANT))
        .thenReturn(false);
    when(handleDbRequestsJdbc.addNewTenant(any())).thenReturn("add new tenant");
    List kwProperties = List.of(new KwProperties());
    List kwRolesPermissions = List.of(new KwRolesPermissions());
    when(defaultDataService.createDefaultProperties(101, "")).thenReturn(kwProperties);
    when(defaultDataService.createDefaultRolesPermissions(101, false, "kwInstallationType"))
        .thenReturn(kwRolesPermissions);
    ApiResponse response = service.addTenantId(kwTenantModel, true);

    assertThat(response.getMessage()).contains("add new tenant");
    assertThat(response.isSuccess()).isTrue();
    verify(commonUtilsService, times(1))
        .updateMetadata(101, EntityType.TENANT, MetadataOperationType.CREATE, null);
    verify(handleDbRequestsJdbc, times(1)).insertDefaultKwProperties(kwProperties);
    verify(handleDbRequestsJdbc, times(1)).insertDefaultRolesPermissions(kwRolesPermissions);
    verify(commonUtilsService, times(1))
        .updateMetadata(101, EntityType.ROLES_PERMISSIONS, MetadataOperationType.CREATE, null);
    verify(commonUtilsService, times(1))
        .updateMetadata(101, EntityType.PROPERTIES, MetadataOperationType.CREATE, null);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void getUpdateEnvStatusUnknownEnv() {
    int tenantId = 101;
    when(manageDatabase.getAllEnvList(tenantId))
        .thenReturn(
            List.of(
                buildEnv("1", 101, "DEV", KafkaClustersType.KAFKA, 1),
                buildEnv("2", 101, "TST", KafkaClustersType.KAFKA, 2),
                buildEnv("3", 101, "PRD", KafkaClustersType.KAFKA, 3)));

    assertThatExceptionOfType(KlawBadRequestException.class)
        .isThrownBy(
            () -> {
              service.getUpdateEnvStatus("4");
            });
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void getUpdateEnvStatus() throws KlawBadRequestException {
    int tenantId = 101;
    Env env1 = buildEnv("1", 101, "PRD", KafkaClustersType.KAFKA, 1);
    Env env2 = buildEnv("2", 101, "PRD", KafkaClustersType.KAFKA, 2);
    Env env3 = buildEnv("3", 101, "PRD", KafkaClustersType.KAFKA, 3);
    when(manageDatabase.getAllEnvList(tenantId)).thenReturn(List.of(env1, env2, env3));
    Map<Integer, KwClusters> clusters = buildClusters(KafkaClustersType.KAFKA, 3);
    KwClusters kwCluster = clusters.get(3);
    when(manageDatabase.getClusters(KafkaClustersType.KAFKA, tenantId)).thenReturn(clusters);
    when(clusterApiService.getKafkaClusterStatus(
            null,
            KafkaSupportedProtocol.SSL,
            "33",
            "kafka",
            KafkaFlavors.APACHE_KAFKA.value,
            tenantId))
        .thenReturn(ClusterStatus.ONLINE);

    EnvUpdatedStatus envUpdatedStatus = service.getUpdateEnvStatus("3");
    assertThat(kwCluster.getClusterStatus()).isEqualTo(ClusterStatus.ONLINE);
    verify(handleDbRequestsJdbc, times(1)).addNewCluster(kwCluster);
    verify(manageDatabase, times(1)).addEnvToCache(tenantId, env3, false);
    assertThat(envUpdatedStatus.getResult()).isEqualTo(ApiResultStatus.SUCCESS.value);
    assertThat(envUpdatedStatus.getEnvStatus()).isEqualTo(ClusterStatus.ONLINE);
    assertThat(envUpdatedStatus.getResult()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteEnvironmentUnauthorized() throws KlawException {
    ApiResponse response = service.deleteEnvironment("envId", "envType");

    assertThat(response).isEqualTo(ApiResponse.NOT_AUTHORIZED);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteEnvironmentExistKafkaComponents() throws KlawException {
    int tenantId = 101;
    String envId = "20";
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.existsKafkaComponentsForEnv(envId, tenantId)).thenReturn(true);
    ApiResponse response = service.deleteEnvironment(envId, KafkaClustersType.KAFKA.value);

    assertThat(response.getMessage()).contains(ENV_CLUSTER_TNT_ERR_105);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteEnvironmentExistKafkaConnectComponents() throws KlawException {
    int tenantId = 101;
    String envId = "20";
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.existsConnectorComponentsForEnv(envId, tenantId)).thenReturn(true);
    ApiResponse response = service.deleteEnvironment(envId, KafkaClustersType.KAFKA_CONNECT.value);

    assertThat(response.getMessage()).contains(ENV_CLUSTER_TNT_ERR_106);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteEnvironmentExistSchemaRegistryComponents() throws KlawException {
    int tenantId = 101;
    String envId = "20";
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.existsSchemaComponentsForEnv(envId, tenantId)).thenReturn(true);
    ApiResponse response =
        service.deleteEnvironment(envId, KafkaClustersType.SCHEMA_REGISTRY.value);

    assertThat(response.getMessage()).contains(ENV_CLUSTER_TNT_ERR_107);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteEnvironmentWithAssociatedEnv() throws KlawException {
    int tenantId = 101;
    String envId = "20";
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.existsKafkaComponentsForEnv(envId, tenantId)).thenReturn(false);
    Env env = buildEnv(envId, tenantId, "Env Name", KafkaClustersType.KAFKA, 1);
    Env associatedEnv = buildEnv("21", tenantId, "Associated Env Name", KafkaClustersType.KAFKA, 2);
    EnvTag associatedEnvTag = new EnvTag();
    associatedEnvTag.setId(associatedEnv.getId());
    env.setAssociatedEnv(associatedEnvTag);
    when(handleDbRequestsJdbc.getEnvDetails(envId, tenantId)).thenReturn(env);
    when(handleDbRequestsJdbc.getEnvDetails(associatedEnv.getId(), tenantId))
        .thenReturn(associatedEnv);
    when(handleDbRequestsJdbc.deleteEnvironmentRequest(envId, tenantId))
        .thenReturn(ApiResultStatus.SUCCESS.value);
    ApiResponse response = service.deleteEnvironment(envId, KafkaClustersType.KAFKA.value);

    assertThat(response.getMessage()).contains(ApiResultStatus.SUCCESS.value);
    assertThat(response.isSuccess()).isTrue();
    verify(manageDatabase, times(1)).removeEnvFromCache(tenantId, 20, false);
    verify(handleDbRequestsJdbc, times(1)).addNewEnv(associatedEnv);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void deleteEnvironmentFailed() throws KlawException {
    int tenantId = 101;
    String envId = "20";
    when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.ADD_EDIT_DELETE_ENVS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.existsKafkaComponentsForEnv(envId, tenantId)).thenReturn(false);
    Env env = buildEnv(envId, tenantId, "Env Name", KafkaClustersType.KAFKA, 1);
    when(handleDbRequestsJdbc.getEnvDetails(envId, tenantId)).thenReturn(env);
    when(handleDbRequestsJdbc.deleteEnvironmentRequest(envId, tenantId))
        .thenReturn("Failed to delete");
    ApiResponse response = service.deleteEnvironment(envId, KafkaClustersType.KAFKA.value);

    assertThat(response.getMessage()).contains("Failed to delete");
    assertThat(response.isSuccess()).isFalse();
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewClusterUnauthorized() {
    ApiResponse response = service.addNewCluster(new KwClustersModel());

    assertThat(response).isEqualTo(ApiResponse.NOT_AUTHORIZED);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewCluster() {
    int tenantId = 101;
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.ADD_EDIT_DELETE_CLUSTERS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.addNewCluster(any(KwClusters.class)))
        .thenReturn(ApiResultStatus.SUCCESS.value);

    KwClustersModel model = new KwClustersModel();
    model.setClusterName("new cluster");
    model.setClusterType(KafkaClustersType.KAFKA);
    model.setKafkaFlavor(KafkaFlavors.APACHE_KAFKA);
    model.setClusterType(KafkaClustersType.KAFKA);
    ApiResponse response = service.addNewCluster(model);

    assertThat(response).isEqualTo(ApiResponse.SUCCESS);
    verify(commonUtilsService, times(1))
        .updateMetadata(tenantId, EntityType.CLUSTER, MetadataOperationType.CREATE, null);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewClusterFailed() {
    int tenantId = 101;
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.ADD_EDIT_DELETE_CLUSTERS))
        .thenReturn(false);

    when(handleDbRequestsJdbc.addNewCluster(any(KwClusters.class)))
        .thenReturn("failed to add cluster");

    KwClustersModel model = new KwClustersModel();
    model.setClusterName("new cluster");
    model.setClusterType(KafkaClustersType.KAFKA);
    model.setKafkaFlavor(KafkaFlavors.APACHE_KAFKA);
    model.setClusterType(KafkaClustersType.KAFKA);
    ApiResponse response = service.addNewCluster(model);

    assertThat(response).isEqualTo(ApiResponse.FAILURE);
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void addNewClusterNameExists() {
    int tenantId = 101;
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.ADD_EDIT_DELETE_CLUSTERS))
        .thenReturn(false);

    when(manageDatabase.getClusters(KafkaClustersType.ALL, tenantId))
        .thenReturn(buildClusters(KafkaClustersType.KAFKA, 3));

    KwClustersModel model = new KwClustersModel();
    model.setClusterName("3");
    model.setClusterType(KafkaClustersType.KAFKA);
    ApiResponse response = service.addNewCluster(model);

    assertThat(response.getMessage()).isEqualTo(ENV_CLUSTER_TNT_ERR_102);
    assertThat(response.isSuccess()).isFalse();
  }

  @Test
  @WithMockUser(
      username = "james",
      authorities = {"ADMIN", "USER"})
  void getSyncEnvsEmptySyncCluster() {
    when(mailService.getUserName(anyString())).thenReturn("user name");
    UserInfo userInfo = buildUserInfo();
    when(handleDbRequestsJdbc.getUsersInfo("user name")).thenReturn(userInfo);
    KwTenantConfigModel model = new KwTenantConfigModel();
    when(manageDatabase.getTenantConfig())
        .thenReturn(
            new HashMap<>() {
              {
