package io.aiven.klaw.service;

import static io.aiven.klaw.helpers.KwConstants.ORDER_OF_KAFKA_CONNECT_ENVS;
import static io.aiven.klaw.service.KafkaConnectControllerService.OBJECT_MAPPER;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import io.aiven.klaw.config.ManageDatabase;
import io.aiven.klaw.dao.Env;
import io.aiven.klaw.dao.KafkaConnectorRequest;
import io.aiven.klaw.dao.KwKafkaConnector;
import io.aiven.klaw.dao.Team;
import io.aiven.klaw.dao.UserInfo;
import io.aiven.klaw.error.KlawBadRequestException;
import io.aiven.klaw.error.KlawException;
import io.aiven.klaw.helpers.db.rdbms.HandleDbRequestsJdbc;
import io.aiven.klaw.model.ApiResponse;
import io.aiven.klaw.model.KwTenantConfigModel;
import io.aiven.klaw.model.enums.ApiResultStatus;
import io.aiven.klaw.model.enums.PermissionType;
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.requests.KafkaConnectorRequestModel;
import io.aiven.klaw.model.response.ConnectorOverview;
import io.aiven.klaw.model.response.KafkaConnectorRequestsResponseModel;
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 org.jasypt.util.text.BasicTextEncryptor;
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.ArgumentCaptor;
import org.mockito.Captor;
import org.mockito.Mock;
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 KafkaConnectControllerServiceTest {
  public static final String USERNAME = "kwusera";
  public static final String CONNECTOR_NAME = "conn1";
  public static final int TENANT_ID = 101;
  @Mock private UserDetails userDetails;

  @Mock private UserInfo userInfo;

  @Mock private ManageDatabase manageDatabase;

  @Mock private HandleDbRequestsJdbc handleDbRequests;

  @Mock CommonUtilsService commonUtilsService;

  @Mock private MailUtils mailService;

  private KafkaConnectControllerService kafkaConnectControllerService;

  @Mock RolesPermissionsControllerService rolesPermissionsControllerService;

  @Mock Map<Integer, KwTenantConfigModel> tenantConfig;

  @Mock KwTenantConfigModel tenantConfigModel;

  @Mock BasicTextEncryptor basicTextEncryptor;

  @Captor ArgumentCaptor<KafkaConnectorRequest> kafkaConnectorRequest;

  private Env env;

  @BeforeEach
  public void setUp() throws Exception {
    this.kafkaConnectControllerService = new KafkaConnectControllerService();
    this.env = new Env();
    env.setId("1");
    env.setName("DEV");
    ReflectionTestUtils.setField(kafkaConnectControllerService, "manageDatabase", manageDatabase);
    ReflectionTestUtils.setField(kafkaConnectControllerService, "mailService", mailService);
    ReflectionTestUtils.setField(
        kafkaConnectControllerService, "kafkaConnectorSensitiveFields", "password,username");
    ReflectionTestUtils.setField(
        kafkaConnectControllerService, "commonUtilsService", commonUtilsService);
    ReflectionTestUtils.setField(
        kafkaConnectControllerService,
        "rolesPermissionsControllerService",
        rolesPermissionsControllerService);

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

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

    stubUserInfo();
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.REQUEST_CREATE_CONNECTORS))
        .thenReturn(false);
    when(commonUtilsService.getTenantId(anyString())).thenReturn(TENANT_ID);
    when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
    when(tenantConfigModel.getBaseSyncKafkaConnectCluster()).thenReturn("1");
    when(commonUtilsService.getTeamId(anyString())).thenReturn(1);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
    when(commonUtilsService.getEnvProperty(anyInt(), anyString())).thenReturn("1");
    when(handleDbRequests.getConnectorsFromName(anyString(), anyInt()))
        .thenReturn(Collections.emptyList());
    when(handleDbRequests.requestForConnector(any())).thenReturn(resultMap);
    when(commonUtilsService.getJasyptEncryptor()).thenReturn(basicTextEncryptor);
    String encryptedText = "encryptedText";
    when(basicTextEncryptor.encrypt(any())).thenReturn(encryptedText);

    ApiResponse apiResponse =
        kafkaConnectControllerService.createConnectorRequest(getConnectRequestModel());
    assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @Order(2)
  public void createConnectorRequestInvalidJsonConfig() {
    KafkaConnectorRequestModel kafkaConnectorRequestModel = getConnectRequestModel();
    kafkaConnectorRequestModel.setConnectorConfig("plain string"); // Invalid json
    stubUserInfo();
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.REQUEST_CREATE_CONNECTORS))
        .thenReturn(false);
    when(handleDbRequests.requestForConnector(any()))
        .thenThrow(new RuntimeException("Unrecognized token"));

    try {
      kafkaConnectControllerService.createConnectorRequest(kafkaConnectorRequestModel);
    } catch (KlawException e) {
      assertThat(e.getMessage()).contains("Unrecognized token");
    }
  }

  @Test
  @Order(3)
  public void createConnectorRequestParameterTopicsDoesntExist() throws KlawException {
    KafkaConnectorRequestModel kafkaConnectorRequestModel = getConnectRequestModel();
    kafkaConnectorRequestModel.setConnectorConfig(getInvalidValidConnConfig());
    stubUserInfo();
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.REQUEST_CREATE_CONNECTORS))
        .thenReturn(false);

    ApiResponse apiResponse =
        kafkaConnectControllerService.createConnectorRequest(kafkaConnectorRequestModel);
    assertThat(apiResponse.getMessage())
        .isEqualTo("Failure. Invalid config. topics/topics.regex is not configured");
  }

  // both topics and topics regex are configured which is not correct
  @Test
  @Order(4)
  public void createConnectorRequestParameterTopicsTopicRegexExist() throws KlawException {
    KafkaConnectorRequestModel kafkaConnectorRequestModel = getConnectRequestModel();
    kafkaConnectorRequestModel.setConnectorConfig(getInvalidValidConnConfigTopicsTopicsRegex());
    stubUserInfo();
    when(commonUtilsService.isNotAuthorizedUser(
            userDetails, PermissionType.REQUEST_CREATE_CONNECTORS))
        .thenReturn(false);

    ApiResponse apiResponse =
        kafkaConnectControllerService.createConnectorRequest(kafkaConnectorRequestModel);
    assertThat(apiResponse.getMessage())
        .isEqualTo("Failure. Invalid config. topics and topics.regex both cannot be configured.");
  }

  @Test
  @Order(5)
  public void createClaimConnectorRequest() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("1");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(handleDbRequests.getConnectorsFromName(eq("ConnectorOne"), eq(TENANT_ID)))
        .thenReturn(List.of(getKwKafkaConnector()));
    when(commonUtilsService.getEnvsFromUserId(any())).thenReturn(envListIds);
    Map<String, String> res = new HashMap<>();
    res.put("result", "success");
    when(handleDbRequests.requestForConnector(any())).thenReturn(res);
    ApiResponse apiResponse =
        kafkaConnectControllerService.createClaimConnectorRequest("ConnectorOne", "1");

    verify(handleDbRequests, times(1)).requestForConnector(kafkaConnectorRequest.capture());
    assertThat(kafkaConnectorRequest.getValue().getApprovingTeamId()).isEqualTo("8");
    assertThat(kafkaConnectorRequest.getValue().getDescription()).isNull();
  }

  @Test
  @Order(6)
  public void createClaimConnectorRequestAlreadyExists() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(handleDbRequests.existsConnectorRequest(
            "ConnectorOne", "1", RequestStatus.CREATED.value, TENANT_ID))
        .thenReturn(true);
    ApiResponse apiResponse =
        kafkaConnectControllerService.createClaimConnectorRequest("ConnectorOne", "1");

    assertThat(apiResponse.getMessage())
        .isEqualTo("Failure. A request already exists for this connector.");
  }

  @Test
  @Order(7)
  public void getRequests_OrderBy_NEWEST_FIRST() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);

    when(handleDbRequests.getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(false)))
        .thenReturn(generateKafkaConnectorRequests(50));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<KafkaConnectorRequestsResponseModel> ordered_response =
        kafkaConnectControllerService.getConnectorRequests(
            "1",
            "1",
            RequestStatus.CREATED,
            null,
            null,
            io.aiven.klaw.model.enums.Order.DESC_REQUESTED_TIME,
            null,
            false);

    assertThat(ordered_response).hasSize(10);

    Timestamp origReqTime = ordered_response.get(0).getRequesttime();

    for (KafkaConnectorRequestsResponseModel req : ordered_response) {

      // assert That each new Request time is older than or equal to the previous request
      assertThat(origReqTime.compareTo(req.getRequesttime()) >= 0).isTrue();
      origReqTime = req.getRequesttime();
    }
  }

  @Test
  @Order(8)
  public void getRequests_OrderBy_OLDEST_FIRST() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);

    when(handleDbRequests.getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(false)))
        .thenReturn(generateKafkaConnectorRequests(50));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<KafkaConnectorRequestsResponseModel> ordered_response =
        kafkaConnectControllerService.getConnectorRequests(
            "1",
            "1",
            RequestStatus.CREATED,
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            null,
            false);

    assertThat(ordered_response).hasSize(10);

    Timestamp origReqTime = ordered_response.get(0).getRequesttime();

    for (KafkaConnectorRequestsResponseModel req : ordered_response) {

      // assert That each new Request time is newer than or equal to the previous request
      assertThat(origReqTime.compareTo(req.getRequesttime()) <= 0).isTrue();
      origReqTime = req.getRequesttime();
    }
  }

  @Test
  @Order(9)
  public void getRequests_IsOnlyMyRequests() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);

    when(handleDbRequests.getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(true)))
        .thenReturn(generateKafkaConnectorRequests(50));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<KafkaConnectorRequestsResponseModel> ordered_response =
        kafkaConnectControllerService.getConnectorRequests(
            "1",
            "1",
            RequestStatus.CREATED,
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            null,
            true);

    assertThat(ordered_response).hasSize(10);

    verify(handleDbRequests, times(1))
        .getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(true));
  }

  @Test
  @Order(10)
  public void getRequests_() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);

    when(handleDbRequests.getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(true)))
        .thenReturn(generateKafkaConnectorRequests(50));
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<KafkaConnectorRequestsResponseModel> ordered_response =
        kafkaConnectControllerService.getConnectorRequests(
            "1",
            "1",
            RequestStatus.CREATED,
            null,
            null,
            io.aiven.klaw.model.enums.Order.ASC_REQUESTED_TIME,
            null,
            true);

    assertThat(ordered_response).hasSize(10);

    verify(handleDbRequests, times(1))
        .getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(true));
  }

  @Test
  @Order(11)
  public void getClaimRequests_WhereConnectorIsDeleted() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    List<KafkaConnectorRequest> connectorRequests = generateKafkaConnectorRequests(9);
    connectorRequests.addAll(generateKafkaConnectorRequests(1, 7, RequestOperationType.CLAIM));
    when(handleDbRequests.getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(false)))
        .thenReturn(connectorRequests);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    List<KafkaConnectorRequestsResponseModel> ordered_response =
        kafkaConnectControllerService.getConnectorRequests(
            "1",
            "1",
            RequestStatus.CREATED,
            null,
            null,
            io.aiven.klaw.model.enums.Order.DESC_REQUESTED_TIME,
            null,
            false);

    assertThat(ordered_response).hasSize(10);

    Timestamp origReqTime = ordered_response.get(0).getRequesttime();

    for (KafkaConnectorRequestsResponseModel req : ordered_response) {
      if (req.getRequestOperationType().equals(RequestOperationType.CLAIM)) {
        assertThat(req.getRemarks())
            .isEqualTo("This Connector is not found in Klaw. Please contact your Administrator.");
      }
      // assert That each new Request time is older than or equal to the previous request
      assertThat(origReqTime.compareTo(req.getRequesttime()) >= 0).isTrue();
      origReqTime = req.getRequesttime();
    }
  }

  @Test
  @Order(12)
  public void getClaimRequests_WhereConnectorIsNotDeleted() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    List<KafkaConnectorRequest> connectorRequests = generateKafkaConnectorRequests(9);
    connectorRequests.addAll(generateKafkaConnectorRequests(1, 7, RequestOperationType.CLAIM));
    when(handleDbRequests.getAllConnectorRequests(
            anyString(),
            eq(null),
            eq(RequestStatus.CREATED),
            eq(null),
            eq(null),
            eq(TENANT_ID),
            eq(false)))
        .thenReturn(connectorRequests);
    when(commonUtilsService.getEnvsFromUserId(anyString()))
        .thenReturn(new HashSet<>(Collections.singletonList("1")));
    when(handleDbRequests.getConnectorsFromName(eq("Conn0"), eq(TENANT_ID)))
        .thenReturn(List.of(getKwKafkaConnector()));
    List<KafkaConnectorRequestsResponseModel> ordered_response =
        kafkaConnectControllerService.getConnectorRequests(
            "1",
            "1",
            RequestStatus.CREATED,
            null,
            null,
            io.aiven.klaw.model.enums.Order.DESC_REQUESTED_TIME,
            null,
            false);

    assertThat(ordered_response).hasSize(10);

    Timestamp origReqTime = ordered_response.get(0).getRequesttime();

    for (KafkaConnectorRequestsResponseModel req : ordered_response) {
      if (req.getRequestOperationType().equals(RequestOperationType.CLAIM)) {
        assertThat(req.getRemarks())
            .isNotEqualTo(
                "This Connector is not found in Klaw. Please contact your Administrator.");
      }
      // assert That each new Request time is older than or equal to the previous request
      assertThat(origReqTime.compareTo(req.getRequesttime()) >= 0).isTrue();
      origReqTime = req.getRequesttime();
    }
  }

  @Test
  @Order(13)
  public void getConnectorOverview_WithNoParams_returnsNull() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));

    ConnectorOverview response = kafkaConnectControllerService.getConnectorOverview(null, null);

    assertThat(response).isNull();
  }

  @Test
  @Order(14)
  public void getConnectorOverview_WithAllEnvs_noPromotionOrderSet() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));
    when(commonUtilsService.getEnvsFromUserId(eq(USERNAME))).thenReturn(Set.of("0", "1", "2", "3"));
    when(commonUtilsService.getEnvProperty(
            eq(TENANT_ID), eq("REQUEST_CONNECTORS_OF_KAFKA_CONNECT_ENVS")))
        .thenReturn("0,1,2,3");
    when(commonUtilsService.getEnvProperty(eq(TENANT_ID), eq(ORDER_OF_KAFKA_CONNECT_ENVS)))
        .thenReturn("");
    when(manageDatabase.getKafkaConnectEnvList(commonUtilsService.getTenantId(eq(USERNAME))))
        .thenReturn(generateEnvironments());
    when(manageDatabase
            .getHandleDbRequests()
            .getConnectorsFromName(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));
    ConnectorOverview response =
        kafkaConnectControllerService.getConnectorOverview(CONNECTOR_NAME, null);

    assertThat(response.getConnectorInfoList()).hasSize(3);
    assertThat(response.getPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(response.getAvailableEnvironments()).hasSize(3);
  }

  private List<Env> generateEnvironments() {
    Env dev = new Env();
    dev.setId("0");
    dev.setName("DEV");
    Env tst = new Env();
    tst.setId("1");
    tst.setName("TST");
    Env prd = new Env();
    prd.setId("2");
    prd.setName("PRD");
    return List.of(dev, tst, prd);
  }

  @Test
  @Order(15)
  public void getConnectorOverview_WithAllEnvs() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));
    when(commonUtilsService.getEnvsFromUserId(eq(USERNAME))).thenReturn(Set.of("0", "1", "2", "3"));
    when(commonUtilsService.getEnvProperty(
            eq(TENANT_ID), eq("REQUEST_CONNECTORS_OF_KAFKA_CONNECT_ENVS")))
        .thenReturn("0,1,2,3");
    when(commonUtilsService.getEnvProperty(eq(TENANT_ID), eq(ORDER_OF_KAFKA_CONNECT_ENVS)))
        .thenReturn("2,1,0");
    when(manageDatabase.getKafkaConnectEnvList(commonUtilsService.getTenantId(eq(USERNAME))))
        .thenReturn(generateEnvironments());
    when(manageDatabase
            .getHandleDbRequests()
            .getConnectorsFromName(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));
    ConnectorOverview response =
        kafkaConnectControllerService.getConnectorOverview(CONNECTOR_NAME, null);

    assertThat(response.getConnectorInfoList()).hasSize(3);
    assertThat(response.getPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(response.getAvailableEnvironments()).hasSize(3);
  }

  @Test
  @Order(15)
  public void getConnectorOverview_WithOneEnv() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));
    when(commonUtilsService.getEnvsFromUserId(eq(USERNAME))).thenReturn(Set.of("0", "1", "2", "3"));
    when(commonUtilsService.getEnvProperty(
            eq(TENANT_ID), eq("REQUEST_CONNECTORS_OF_KAFKA_CONNECT_ENVS")))
        .thenReturn("0,1,2,3");
    when(commonUtilsService.getEnvProperty(eq(TENANT_ID), eq(ORDER_OF_KAFKA_CONNECT_ENVS)))
        .thenReturn("2,1,0");
    when(manageDatabase.getKafkaConnectEnvList(commonUtilsService.getTenantId(eq(USERNAME))))
        .thenReturn(generateEnvironments());
    when(manageDatabase
            .getHandleDbRequests()
            .getConnectorsFromName(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(3));
    ConnectorOverview response =
        kafkaConnectControllerService.getConnectorOverview(CONNECTOR_NAME, "1");

    assertThat(response.getConnectorInfoList()).hasSize(1);
    assertThat(response.getConnectorInfoList().get(0).getConnectorId()).isEqualTo(1);
    assertThat(response.getPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(response.getAvailableEnvironments()).hasSize(3);
  }

  @Test
  @Order(15)
  public void getConnectorOverview_WithOneEnvAndPromotion() throws KlawException {
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(2));
    when(commonUtilsService.getEnvsFromUserId(eq(USERNAME))).thenReturn(Set.of("0", "1", "2", "3"));
    when(commonUtilsService.getEnvProperty(
            eq(TENANT_ID), eq("REQUEST_CONNECTORS_OF_KAFKA_CONNECT_ENVS")))
        .thenReturn("0,1,2,3");
    when(commonUtilsService.getEnvProperty(eq(TENANT_ID), eq(ORDER_OF_KAFKA_CONNECT_ENVS)))
        .thenReturn("0,1,2");
    when(manageDatabase.getKafkaConnectEnvList(commonUtilsService.getTenantId(eq(USERNAME))))
        .thenReturn(generateEnvironments());
    when(manageDatabase
            .getHandleDbRequests()
            .getConnectorsFromName(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(2));
    ConnectorOverview response =
        kafkaConnectControllerService.getConnectorOverview(CONNECTOR_NAME, "1");

    assertThat(response.getConnectorInfoList()).hasSize(1);
    assertThat(response.getPromotionDetails().getStatus()).isEqualTo(PromotionStatusType.SUCCESS);
    assertThat(response.getPromotionDetails().getTargetEnv()).isEqualTo("PRD");
    assertThat(response.getAvailableEnvironments()).hasSize(2);
  }

  @Test
  @Order(15)
  public void getConnectorOverview_WithOneEnvAndNoPromotionForBaseEnv() throws KlawException {
    // A promotion is available for the tst connector but we are checking for the dev one and that
    // has already been promoted to tst.
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(2));
    when(commonUtilsService.getEnvsFromUserId(eq(USERNAME))).thenReturn(Set.of("0", "1", "2", "3"));
    when(commonUtilsService.getEnvProperty(
            eq(TENANT_ID), eq("REQUEST_CONNECTORS_OF_KAFKA_CONNECT_ENVS")))
        .thenReturn("0,1,2,3");
    when(commonUtilsService.getEnvProperty(eq(TENANT_ID), eq(ORDER_OF_KAFKA_CONNECT_ENVS)))
        .thenReturn("0,1,2");
    when(manageDatabase.getKafkaConnectEnvList(commonUtilsService.getTenantId(eq(USERNAME))))
        .thenReturn(generateEnvironments());
    when(manageDatabase
            .getHandleDbRequests()
            .getConnectorsFromName(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(2));
    ConnectorOverview response =
        kafkaConnectControllerService.getConnectorOverview(CONNECTOR_NAME, "0");

    assertThat(response.getConnectorInfoList()).hasSize(1);
    assertThat(response.getPromotionDetails().getStatus())
        .isEqualTo(PromotionStatusType.NO_PROMOTION);
    assertThat(response.getAvailableEnvironments()).hasSize(2);
  }

  @Test
  @Order(16)
  public void updateJsonNodeTestEncryptAndDecrypt() throws JsonProcessingException {
    JsonNode jsonNode = OBJECT_MAPPER.readTree(getValidConnConfig().trim());
    when(commonUtilsService.getJasyptEncryptor()).thenReturn(basicTextEncryptor);
    String encryptedText = "encryptedText";
    when(basicTextEncryptor.encrypt(any())).thenReturn(encryptedText);
    kafkaConnectControllerService.updateJsonNode("encrypt", jsonNode);
    assertThat(jsonNode.get("connector.password").asText()).isEqualTo(encryptedText);
    assertThat(jsonNode.get("jdbc.username").asText()).isEqualTo(encryptedText);
    verify(basicTextEncryptor, times(2)).encrypt(any());

    String decryptedText = "decryptedText";
    when(basicTextEncryptor.decrypt(any())).thenReturn(decryptedText);
    kafkaConnectControllerService.updateJsonNode("decrypt", jsonNode);
    assertThat(jsonNode.get("connector.password").asText()).isEqualTo(decryptedText);
    assertThat(jsonNode.get("jdbc.username").asText()).isEqualTo(decryptedText);
    verify(basicTextEncryptor, times(2)).decrypt(any());

    jsonNode = OBJECT_MAPPER.readTree(getValidConnConfig().trim());
    when(basicTextEncryptor.decrypt(any()))
        .thenThrow(new RuntimeException("NotSupportedOperation"));
    kafkaConnectControllerService.updateJsonNode("decrypt", jsonNode);
    assertThat(jsonNode.get("connector.password").asText()).isEqualTo("testpwd");
  }

  @Test
  @Order(17)
  public void updateJsonNodeTestEncryptAndDecryptNoKeyMatches() throws JsonProcessingException {
    ReflectionTestUtils.setField(
        kafkaConnectControllerService, "kafkaConnectorSensitiveFields", "fielddoesnotexist");

    JsonNode jsonNode = OBJECT_MAPPER.readTree(getValidConnConfig().trim());
    when(commonUtilsService.getJasyptEncryptor()).thenReturn(basicTextEncryptor);
    String encryptedText = "encryptedText";
    when(basicTextEncryptor.encrypt(any())).thenReturn(encryptedText);
    kafkaConnectControllerService.updateJsonNode("encrypt", jsonNode);
    assertThat(jsonNode.get("connector.password").asText()).isEqualTo("testpwd");
    assertThat(jsonNode.get("jdbc.username").asText()).isEqualTo("testuser");

    String decryptedText = "decryptedText";
    when(basicTextEncryptor.decrypt(any())).thenReturn(decryptedText);
    kafkaConnectControllerService.updateJsonNode("decrypt", jsonNode);
    assertThat(jsonNode.get("connector.password").asText()).isEqualTo("testpwd");
    assertThat(jsonNode.get("jdbc.username").asText()).isEqualTo("testuser");

    verify(basicTextEncryptor, times(0)).encrypt(any());
    verify(basicTextEncryptor, times(0)).decrypt(any());
  }

  @Test
  @Order(18)
  public void getConnectorOverview_WithHighestEnvAndConnectorOwnerSet() throws KlawException {
    // A promotion is available for the tst connector but we are checking for the dev one and that
    // has already been promoted to tst.
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(2));
    when(commonUtilsService.getEnvsFromUserId(eq(USERNAME))).thenReturn(Set.of("0", "1", "2", "3"));
    when(commonUtilsService.getEnvProperty(
            eq(TENANT_ID), eq("REQUEST_CONNECTORS_OF_KAFKA_CONNECT_ENVS")))
        .thenReturn("0,1,2,3");
    when(commonUtilsService.getEnvProperty(eq(TENANT_ID), eq(ORDER_OF_KAFKA_CONNECT_ENVS)))
        .thenReturn("0,1,2");
    when(manageDatabase.getKafkaConnectEnvList(commonUtilsService.getTenantId(eq(USERNAME))))
        .thenReturn(generateEnvironments());
    when(manageDatabase
            .getHandleDbRequests()
            .getConnectorsFromName(eq(CONNECTOR_NAME), eq(TENANT_ID)))
        .thenReturn(generateKafkaConnectors(2));
    ConnectorOverview response =
        kafkaConnectControllerService.getConnectorOverview(CONNECTOR_NAME, "1");

    assertThat(response.getConnectorInfoList().get(0).isHighestEnv()).isTrue();
    assertThat(response.getConnectorInfoList().get(0).isConnectorOwner()).isTrue();
    assertThat(response.getPromotionDetails().getStatus()).isEqualTo(PromotionStatusType.SUCCESS);
    assertThat(response.getAvailableEnvironments()).hasSize(2);
  }

  @Test
  @Order(19)
  public void getConnectorOverview_WithRequestsOpen() throws KlawException {
    // A promotion is available for the tst connector but we are checking for the dev one and that
    // has already been promoted to tst.
    Set<String> envListIds = new HashSet<>();
    envListIds.add("DEV");
    stubUserInfo();
    when(commonUtilsService.getTenantId(any())).thenReturn(TENANT_ID);
    when(commonUtilsService.getTeamId(eq(USERNAME))).thenReturn(8);
    when(handleDbRequests.getConnectors(eq(CONNECTOR_NAME), eq(TENANT_ID)))
