content: package io.aiven.klaw.service;
  
  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.when;
  
  import io.aiven.klaw.UtilMethods;
  import io.aiven.klaw.config.ManageDatabase;
  import io.aiven.klaw.dao.Env;
  import io.aiven.klaw.dao.KwClusters;
  import io.aiven.klaw.dao.KwKafkaConnector;
  import io.aiven.klaw.dao.UserInfo;
  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.cluster.ConnectorState;
  import io.aiven.klaw.model.cluster.ConnectorsStatus;
  import io.aiven.klaw.model.enums.ApiResultStatus;
  import io.aiven.klaw.model.enums.KafkaClustersType;
  import io.aiven.klaw.model.enums.KafkaSupportedProtocol;
  import io.aiven.klaw.model.enums.PermissionType;
  import io.aiven.klaw.model.response.KafkaConnectorModelResponse;
  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 org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.MethodOrderer;
  import org.junit.jupiter.api.Order;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.TestMethodOrder;
  import org.junit.jupiter.api.extension.ExtendWith;
  import org.mockito.Mock;
  import org.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 KafkaConnectSyncControllerServiceTest {
    public static final int TENANT_ID = 1;
    public static final String USERNAME = "kwusera";
    public static final String REMOVE_FROM_KLAW = "REMOVE FROM KLAW";
    public static final String IN_SYNC = "IN_SYNC";
    public static final String DELETED = "DELETED";
    public static final String ADDED = "ADDED";
    public static final String TEAM_NAME_1 = "Octopus";
    @Mock private ClusterApiService clusterApiService;
  
    @Mock private UserDetails userDetails;
  
    @Mock private UserInfo userInfo;
  
    @Mock private Map<Integer, KwClusters> clustersHashMap;
  
    @Mock private KwClusters kwClusters;
  
    @Mock private ManageDatabase manageDatabase;
  
    @Mock private HandleDbRequestsJdbc handleDbRequests;
  
    @Mock CommonUtilsService commonUtilsService;
  
    @Mock private MailUtils mailService;
    private KafkaConnectSyncControllerService kafkaConnectSyncControllerService;
  
    @Mock Map<Integer, KwTenantConfigModel> tenantConfig;
  
    @Mock KwTenantConfigModel tenantConfigModel;
  
    private UtilMethods utilMethods;
    private Env env = new Env();
    private Env test = new Env();
  
    @BeforeEach
    public void setUp() throws Exception {
      this.kafkaConnectSyncControllerService = new KafkaConnectSyncControllerService();
      utilMethods = new UtilMethods();
      environmentSetUp();
  
      ReflectionTestUtils.setField(
          kafkaConnectSyncControllerService, "manageDatabase", manageDatabase);
      ReflectionTestUtils.setField(kafkaConnectSyncControllerService, "mailService", mailService);
      ReflectionTestUtils.setField(
          kafkaConnectSyncControllerService, "commonUtilsService", commonUtilsService);
      ReflectionTestUtils.setField(
          kafkaConnectSyncControllerService, "clusterApiService", clusterApiService);
  
      when(manageDatabase.getHandleDbRequests()).thenReturn(handleDbRequests);
      when(commonUtilsService.getPrincipal()).thenReturn(userDetails);
      when(commonUtilsService.isNotAuthorizedUser(any(), any(PermissionType.class))).thenReturn(true);
    }
  
    private void environmentSetUp() {
      env.setId("1");
      env.setName("DEV");
      env.setClusterId(1);
  
      test.setId("2");
      test.setName("TST");
      test.setClusterId(2);
      when(manageDatabase.getKafkaEnvList(anyInt())).thenReturn(List.of(env, test));
    }
  
    private void stubUserInfo() {
      when(handleDbRequests.getUsersInfo(anyString())).thenReturn(userInfo);
      when(userInfo.getTeamId()).thenReturn(101);
      when(mailService.getUserName(userDetails)).thenReturn(USERNAME);
      when(commonUtilsService.getTenantId(eq(USERNAME))).thenReturn(TENANT_ID);
    }
  
    @Test
    @Order(1)
    public void updateSyncConnectorsSuccess() throws KlawException {
      stubUserInfo();
      when(manageDatabase.getTenantConfig()).thenReturn(tenantConfig);
      when(tenantConfig.get(anyInt())).thenReturn(tenantConfigModel);
      when(tenantConfigModel.getBaseSyncEnvironment()).thenReturn("1");
      when(commonUtilsService.isNotAuthorizedUser(userDetails, PermissionType.SYNC_CONNECTORS))
          .thenReturn(false);
      when(commonUtilsService.getEnvsFromUserId(anyString()))
          .thenReturn(new HashSet<>(Collections.singletonList("1")));
      when(manageDatabase.getKafkaConnectEnvList(anyInt())).thenReturn(List.of(env));
      when(manageDatabase.getClusters(eq(KafkaClustersType.KAFKA_CONNECT), anyInt()))
          .thenReturn(clustersHashMap);
      when(clustersHashMap.get(any())).thenReturn(kwClusters);
      when(handleDbRequests.addToSyncConnectors(any())).thenReturn(ApiResultStatus.SUCCESS.value);
  
      ApiResponse result =
          kafkaConnectSyncControllerService.updateSyncConnectors(
              utilMethods.getSyncConnectorUpdates());
      assertThat(result.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    @Order(2)
    public void getSync_WithDeleted_Connectors() throws KlawException {
      stubUserInfo();
      when(manageDatabase.getKafkaConnectEnvList(anyInt())).thenReturn(List.of(env, test));
      when(manageDatabase.getTeamNameFromTeamId(eq(1), eq(10))).thenReturn(TEAM_NAME_1);
      when(manageDatabase.getClusters(eq(KafkaClustersType.KAFKA_CONNECT), eq(1)))
          .thenReturn(getConnectlusters(2));
      when(manageDatabase.getHandleDbRequests().getSyncConnectors(eq("2"), eq(null), eq(1)))
          .thenReturn(getConnectorsFromDb(5));
      when(clusterApiService.getAllKafkaConnectors(
              anyString(),
              eq(KafkaSupportedProtocol.PLAINTEXT.getValue()),
              anyString(),
              eq(1),
              anyBoolean()))
          .thenReturn(getConnectorStatus(3));
      List<KafkaConnectorModelResponse> result =
          kafkaConnectSyncControllerService.getSyncConnectors("2", "1", "1", null, false);
  
      assertThat(result).hasSize(5);
      int added = 0, deleted = 0, synced = 0;
      for (KafkaConnectorModelResponse entry : result) {
  
        if (entry.getRemarks().equalsIgnoreCase(ADDED)) {
          added++;
        } else if (entry.getRemarks().equalsIgnoreCase(DELETED)) {
          deleted++;
          assertThat(entry.getPossibleTeams()).contains(REMOVE_FROM_KLAW);
        } else if (entry.getRemarks().equalsIgnoreCase(IN_SYNC)) {
          synced++;
        }
      }
      assertThat(added).isEqualTo(0);
      assertThat(deleted).isEqualTo(2);
      assertThat(synced).isEqualTo(3);
    }
  
    @Test
    @Order(2)
    public void getSync_WithAddedOnly_Connectors() throws KlawException {
      stubUserInfo();
      when(manageDatabase.getKafkaConnectEnvList(anyInt())).thenReturn(List.of(env, test));
      when(manageDatabase.getTeamNameFromTeamId(eq(1), eq(10))).thenReturn(TEAM_NAME_1);
      when(manageDatabase.getClusters(eq(KafkaClustersType.KAFKA_CONNECT), eq(1)))
          .thenReturn(getConnectlusters(2));
      when(manageDatabase.getHandleDbRequests().getSyncConnectors(eq("2"), eq(null), eq(1)))
          .thenReturn(getConnectorsFromDb(0));
      when(clusterApiService.getAllKafkaConnectors(
              anyString(),
              eq(KafkaSupportedProtocol.PLAINTEXT.getValue()),
              anyString(),
              eq(1),
              anyBoolean()))
          .thenReturn(getConnectorStatus(5));
      List<KafkaConnectorModelResponse> result =
          kafkaConnectSyncControllerService.getSyncConnectors("2", "1", "1", null, false);
  
      assertThat(result).hasSize(5);
      int added = 0, deleted = 0, synced = 0;
      for (KafkaConnectorModelResponse entry : result) {
  
        if (entry.getRemarks().equalsIgnoreCase(ADDED)) {
          added++;
        } else if (entry.getRemarks().equalsIgnoreCase(DELETED)) {
          deleted++;
          assertThat(entry.getPossibleTeams()).contains(REMOVE_FROM_KLAW);
        } else if (entry.getRemarks().equalsIgnoreCase(IN_SYNC)) {
          synced++;
        }
      }
      assertThat(added).isEqualTo(5);
      assertThat(deleted).isEqualTo(0);
      assertThat(synced).isEqualTo(0);
    }
  
    @Test
    @Order(2)
    public void getSync_WithDeletedOnly_Connectors() throws KlawException {
      stubUserInfo();
      when(manageDatabase.getKafkaConnectEnvList(anyInt())).thenReturn(List.of(env, test));
      when(manageDatabase.getTeamNameFromTeamId(eq(1), eq(10))).thenReturn(TEAM_NAME_1);
      when(manageDatabase.getClusters(eq(KafkaClustersType.KAFKA_CONNECT), eq(1)))
          .thenReturn(getConnectlusters(2));
      when(manageDatabase.getHandleDbRequests().getSyncConnectors(eq("2"), eq(null), eq(1)))
          .thenReturn(getConnectorsFromDb(5));
      when(clusterApiService.getAllKafkaConnectors(
              anyString(),
              eq(KafkaSupportedProtocol.PLAINTEXT.getValue()),
              anyString(),
              eq(1),
              anyBoolean()))
          .thenReturn(getConnectorStatus(0));
      List<KafkaConnectorModelResponse> result =
          kafkaConnectSyncControllerService.getSyncConnectors("2", "1", "1", null, false);
  
      assertThat(result).hasSize(5);
      int added = 0, deleted = 0, synced = 0;
      for (KafkaConnectorModelResponse entry : result) {
  
        if (entry.getRemarks().equalsIgnoreCase(ADDED)) {
          added++;
        } else if (entry.getRemarks().equalsIgnoreCase(DELETED)) {
          deleted++;
          assertThat(entry.getPossibleTeams()).contains(REMOVE_FROM_KLAW);
        } else if (entry.getRemarks().equalsIgnoreCase(IN_SYNC)) {
          synced++;
        }
      }
      assertThat(added).isEqualTo(0);
      assertThat(deleted).isEqualTo(5);
      assertThat(synced).isEqualTo(0);
    }
  
    private List<KwKafkaConnector> getConnectorsFromDb(int number) {
      List<KwKafkaConnector> list = new ArrayList<>();
      for (int i = 0; i < number; i++) {
        KwKafkaConnector conn = new KwKafkaConnector();
        conn.setConnectorId(i);
        conn.setConnectorName("Connector" + i);
        conn.setEnvironment("1");
        conn.setTenantId(1);
        conn.setTeamId(10);
        list.add(conn);
      }
      return list;
    }
  
    private ConnectorsStatus getConnectorStatus(int number) {
      ConnectorsStatus status = new ConnectorsStatus();
      status.setConnectorStateList(getConnectorStateList(number));
      return status;
    }
  
    private List<ConnectorState> getConnectorStateList(int number) {
      List<ConnectorState> list = new ArrayList<>();
  
      for (int i = 0; i < number; i++) {
  
        ConnectorState state = new ConnectorState();
        state.setConnectorName("Connector" + i);
        state.setFailedTasks(0);
        state.setRunningTasks(1);
        state.setConnectorStatus("RUNNING");
        list.add(state);
      }
      return list;
    }
  
    private Map<Integer, KwClusters> getConnectlusters(int number) {
      Map<Integer, KwClusters> retval = new HashMap<>();
      for (int i = 1; i <= number; i++) {
        KwClusters clusters = new KwClusters();
        clusters.setProtocol(KafkaSupportedProtocol.PLAINTEXT);
        clusters.setClusterId(i);
        clusters.setClusterType(KafkaClustersType.KAFKA_CONNECT.value);
        clusters.setClusterName(i + "");
        clusters.setBootstrapServers("localhost:9092");
        retval.put(i, clusters);
      }
  
      return retval;
    }
  }
  