content: package io.aiven.klaw.clusterapi.services;
  
  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.anyString;
  import static org.mockito.ArgumentMatchers.eq;
  import static org.mockito.Mockito.when;
  
  import io.aiven.klaw.clusterapi.UtilMethods;
  import io.aiven.klaw.clusterapi.models.ApiResponse;
  import io.aiven.klaw.clusterapi.models.ClusterAclRequest;
  import io.aiven.klaw.clusterapi.models.ClusterTopicRequest;
  import io.aiven.klaw.clusterapi.models.LoadTopicsResponse;
  import io.aiven.klaw.clusterapi.models.confluentcloud.ListAclsResponse;
  import io.aiven.klaw.clusterapi.models.confluentcloud.ListTopicsResponse;
  import io.aiven.klaw.clusterapi.models.confluentcloud.TopicCreateRequest;
  import io.aiven.klaw.clusterapi.models.enums.*;
  import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Set;
  import org.apache.commons.lang3.tuple.Pair;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.extension.ExtendWith;
  import org.mockito.Mock;
  import org.springframework.core.ParameterizedTypeReference;
  import org.springframework.core.env.Environment;
  import org.springframework.http.HttpHeaders;
  import org.springframework.http.HttpMethod;
  import org.springframework.http.HttpStatus;
  import org.springframework.http.ResponseEntity;
  import org.springframework.test.context.junit.jupiter.SpringExtension;
  import org.springframework.web.client.RestClientException;
  import org.springframework.web.client.RestTemplate;
  
  @ExtendWith(SpringExtension.class)
  public class ConfluentCloudApiServiceTest {
    public static final String KAFKA_V_3_CLUSTERS_CLUSTER_ID_TOPICS =
        "/kafka/v3/clusters/cluster-id/topics";
    public static final String KAFKA_V_3_CLUSTERS_CLUSTER_ID_ACLS =
        "/kafka/v3/clusters/cluster-id/acls";
    public static final String CLUSTER_ID = "CC_DEV1";
    ConfluentCloudApiService confluentCloudApiService;
    @Mock RestTemplate restTemplate;
    @Mock ClusterApiUtils clusterApiUtils;
    @Mock private Environment env;
    private UtilMethods utilMethods;
  
    @BeforeEach
    public void setUp() {
      confluentCloudApiService = new ConfluentCloudApiService(env, clusterApiUtils);
      utilMethods = new UtilMethods();
    }
  
    @Test
    public void listTopics() throws Exception {
      ListTopicsResponse listTopicsResponse = utilMethods.getConfluentCloudListTopicsResponse();
      ResponseEntity<ListTopicsResponse> listTopicsResponseResponseEntity =
          new ResponseEntity<>(listTopicsResponse, HttpStatus.OK);
      stubTopics();
      when(restTemplate.exchange(
              anyString(),
              eq(HttpMethod.GET),
              any(),
              (ParameterizedTypeReference<ListTopicsResponse>) any()))
          .thenReturn(listTopicsResponseResponseEntity);
      LoadTopicsResponse listTopicsSet =
          confluentCloudApiService.listTopics(
              "localhost:443", KafkaSupportedProtocol.SSL, CLUSTER_ID);
  
      assertThat(listTopicsSet.getTopicConfigSet()).hasSize(2); // two topics
      assertThat(listTopicsSet.getTopicConfigSet().stream().toList().get(0).getTopicName())
          .isNotNull();
      assertThat(listTopicsSet.getTopicConfigSet().stream().toList().get(0).getPartitions())
          .isNotNull();
      assertThat(listTopicsSet.getTopicConfigSet().stream().toList().get(0).getReplicationFactor())
          .isNotNull();
    }
  
    @Test
    public void listTopicsConnectivityFailure() {
      stubTopics();
      when(restTemplate.exchange(
              anyString(),
              eq(HttpMethod.GET),
              any(),
              (ParameterizedTypeReference<ListTopicsResponse>) any()))
          .thenThrow(new RestClientException("Cannot connect to confluent cluster"));
      assertThatThrownBy(
              () -> {
                confluentCloudApiService.listTopics(
                    "localhost:443", KafkaSupportedProtocol.SSL, CLUSTER_ID);
              })
          .hasMessage("Error in listing topics : " + "Cannot connect to confluent cluster");
    }
  
    @Test
    public void listAcls() throws Exception {
      ListAclsResponse listAclsResponse = utilMethods.getConfluentCloudListAclsResponse();
      ResponseEntity<ListAclsResponse> listAclsResponseResponseEntity =
          new ResponseEntity<>(listAclsResponse, HttpStatus.OK);
  
      stubAcls();
      when(restTemplate.exchange(
              anyString(),
              eq(HttpMethod.GET),
              any(),
              (ParameterizedTypeReference<ListAclsResponse>) any()))
          .thenReturn(listAclsResponseResponseEntity);
  
      Set<Map<String, String>> aclsSet =
          confluentCloudApiService.listAcls("localhost:443", KafkaSupportedProtocol.SSL, CLUSTER_ID);
  
      assertThat(aclsSet).hasSize(2); // two acls
      assertThat(aclsSet.stream().toList().get(0))
          .hasSize(6); // operation, resourceType, resourceName, principle, host, permissionType
      assertThat(aclsSet.stream().toList().get(0))
          .containsKeys(
              "operation", "resourceType", "resourceName", "principle", "host", "permissionType");
      assertThat(aclsSet.stream().toList().get(0)).containsEntry("resourceName", "testtopic");
      assertThat(aclsSet.stream().toList().get(0)).containsEntry("permissionType", "ALLOW");
      assertThat(aclsSet.stream().toList().get(0)).containsEntry("resourceType", "TOPIC");
      assertThat(aclsSet.stream().toList().get(0)).containsEntry("host", "12.12.43.123");
    }
  
    @Test
    public void listAclsConnectivityFailure() {
      stubAcls();
      when(restTemplate.exchange(
              anyString(),
              eq(HttpMethod.GET),
              any(),
              (ParameterizedTypeReference<ListAclsResponse>) any()))
          .thenThrow(new RestClientException("Cannot connect to confluent cluster"));
  
      assertThatThrownBy(
              () -> {
                confluentCloudApiService.listAcls(
                    "localhost:443", KafkaSupportedProtocol.SSL, CLUSTER_ID);
              })
          .hasMessage("Error in listing acls : " + "Cannot connect to confluent cluster");
    }
  
    @Test
    public void createAclsProducer() throws Exception {
      stubAcls();
      ClusterAclRequest clusterAclRequest = getClusterAclRequest(AclType.PRODUCER.value);
      Map<String, String> createAclsResponse = confluentCloudApiService.createAcls(clusterAclRequest);
  
      when(restTemplate.postForEntity(anyString(), any(), any())).thenReturn(null);
      assertThat(createAclsResponse.get("result")).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    public void createAclsConsumer() throws Exception {
      stubAcls();
      ClusterAclRequest clusterAclRequest = utilMethods.getConfluentCloudConsumerAclRequest();
      Map<String, String> createAclsResponse = confluentCloudApiService.createAcls(clusterAclRequest);
  
      when(restTemplate.postForEntity(anyString(), any(), any())).thenReturn(null);
      assertThat(createAclsResponse.get("result")).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    public void updateAclMapConsumer() {
      ClusterAclRequest clusterAclRequest = utilMethods.getConfluentCloudConsumerAclRequest();
      Map<String, String> aclMap = new HashMap<>();
      Map<String, String> resultMap =
          confluentCloudApiService.updateAclMap(clusterAclRequest, aclMap);
      assertThat(resultMap.get("pattern_type")).isEqualTo("LITERAL");
      assertThat(resultMap.get("principal")).isEqualTo("User:CN=host");
      assertThat(resultMap.get("host")).isEqualTo("*");
      assertThat(resultMap.get("permission")).isEqualTo("ALLOW");
    }
  
    @Test
    public void updateAclMapProducer() {
      ClusterAclRequest clusterAclRequest = getClusterAclRequest(AclType.PRODUCER.value);
      Map<String, String> aclMap = new HashMap<>();
      Map<String, String> resultMap =
          confluentCloudApiService.updateAclMap(clusterAclRequest, aclMap);
      assertThat(resultMap.get("pattern_type")).isEqualTo("LITERAL");
      assertThat(resultMap.get("principal")).isEqualTo("User:*");
      assertThat(resultMap.get("host")).isEqualTo("11.12.33.122");
      assertThat(resultMap.get("permission")).isEqualTo("ALLOW");
    }
  
    @Test
    public void updateAclMapProducerPrefixed() {
      ClusterAclRequest clusterAclRequest = utilMethods.getConfluentCloudProducerPrefixedAclRequest();
      Map<String, String> aclMap = new HashMap<>();
      Map<String, String> resultMap =
          confluentCloudApiService.updateAclMap(clusterAclRequest, aclMap);
      assertThat(resultMap.get("pattern_type")).isEqualTo("PREFIXED");
      assertThat(resultMap.get("principal")).isEqualTo("User:*");
      assertThat(resultMap.get("host")).isEqualTo("11.12.33.122");
      assertThat(resultMap.get("permission")).isEqualTo("ALLOW");
    }
  
    @Test
    public void deleteAcls_AclTypeProducer() throws Exception {
      ClusterAclRequest clusterAclRequest = getClusterAclRequest(AclType.PRODUCER.value);
      stubAcls();
      when(restTemplate.exchange(
              anyString(),
              eq(HttpMethod.DELETE),
              any(),
              (ParameterizedTypeReference<ListAclsResponse>) any()))
          .thenReturn(null);
  
      String response = confluentCloudApiService.deleteAcls(clusterAclRequest);
      assertThat(response).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    public void deleteAcls_AclTypeConsumer() throws Exception {
      ClusterAclRequest clusterAclRequest = getClusterAclRequest(AclType.CONSUMER.value);
      stubAcls();
      when(restTemplate.exchange(
              anyString(),
              eq(HttpMethod.DELETE),
              any(),
              (ParameterizedTypeReference<ListAclsResponse>) any()))
          .thenReturn(null);
  
      String response = confluentCloudApiService.deleteAcls(clusterAclRequest);
      assertThat(response).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    public void createTopic() throws Exception {
      stubTopics();
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
      when(restTemplate.postForEntity(anyString(), any(), any())).thenReturn(null);
  
      ApiResponse apiResponse = confluentCloudApiService.createTopic(clusterTopicRequest);
      assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    public void createTopic_AlredyExistsFailure() throws Exception {
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
  
      stubTopics();
      when(restTemplate.postForEntity(anyString(), any(), any()))
          .thenThrow(new RestClientException("Topic already exists."));
  
      ApiResponse apiResponse = confluentCloudApiService.createTopic(clusterTopicRequest);
  
      assertThat(apiResponse.getMessage()).isEqualTo("Topic already exists.");
      assertThat(apiResponse.isSuccess()).isFalse();
    }
  
    @Test
    public void createTopic_Failure() {
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
      Exception exception = new RestClientException("Error occurred.");
  
      stubTopics();
      when(restTemplate.postForEntity(anyString(), any(), any())).thenThrow(exception);
  
      assertThatThrownBy(() -> confluentCloudApiService.createTopic(clusterTopicRequest))
          .isEqualTo(exception);
    }
  
    @Test
    public void deleteTopic() throws Exception {
      stubTopics();
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
      when(restTemplate.postForEntity(anyString(), any(), any())).thenReturn(null);
  
      ApiResponse apiResponse = confluentCloudApiService.deleteTopic(clusterTopicRequest);
      assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
    }
  
    @Test
    public void deleteTopic_AlredyExistsFailure() throws Exception {
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
  
      stubTopics();
      when(restTemplate.exchange(anyString(), eq(HttpMethod.DELETE), any(), eq(String.class)))
          .thenThrow(new RestClientException("This server does not host this topic"));
  
      ApiResponse apiResponse = confluentCloudApiService.deleteTopic(clusterTopicRequest);
  
      assertThat(apiResponse.getMessage()).isEqualTo("This server does not host this topic");
      assertThat(apiResponse.isSuccess()).isFalse();
    }
  
    @Test
    public void deleteTopic_Failure() {
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
      Exception exception = new RestClientException("Error occurred.");
  
      stubTopics();
      when(restTemplate.exchange(anyString(), eq(HttpMethod.DELETE), any(), eq(String.class)))
          .thenThrow(exception);
  
      assertThatThrownBy(() -> confluentCloudApiService.deleteTopic(clusterTopicRequest))
          .isEqualTo(exception);
    }
  
    @Test
    public void getTopicCreateObj() {
      ClusterTopicRequest clusterTopicRequest = utilMethods.getTopicRequest();
      TopicCreateRequest topicCreateRequest =
          confluentCloudApiService.getTopicCreateObj(clusterTopicRequest);
      assertThat(topicCreateRequest.getTopic_name()).isEqualTo(clusterTopicRequest.getTopicName());
      assertThat(topicCreateRequest.getPartitions_count())
          .isEqualTo(clusterTopicRequest.getPartitions());
      assertThat(topicCreateRequest.getReplication_factor())
          .isEqualTo(clusterTopicRequest.getReplicationFactor());
      assertThat(topicCreateRequest.getConfigs())
          .hasSize(clusterTopicRequest.getAdvancedTopicConfiguration().size());
    }
  
    @Test
    public void getQueryParamsHost() {
      ClusterAclRequest clusterAclRequest = getClusterAclRequest(AclType.PRODUCER.value);
      String resourceType = "TOPIC";
      String operation = "WRITE";
      String response =
          confluentCloudApiService.updateQueryParams(
              clusterAclRequest,
              resourceType,
              clusterAclRequest.getTopicName(),
              operation,
              "https://host");
      assertThat(response)
          .isEqualTo(
              "https://host?resource_type="
                  + resourceType
                  + "&resource_name="
                  + clusterAclRequest.getTopicName()
                  + "&pattern_type=LITERAL&principal=User:*&host="
                  + clusterAclRequest.getAclIp()
                  + "&operation="
                  + operation
                  + "&permission=ALLOW");
    }
  
    @Test
    public void getQueryParamsPrincipal() {
      ClusterAclRequest clusterAclRequest = utilMethods.getConfluentCloudConsumerAclRequest();
      String resourceType = "CONSUMER";
      String operation = "READ";
      String response =
          confluentCloudApiService.updateQueryParams(
              clusterAclRequest,
              resourceType,
              clusterAclRequest.getConsumerGroup(),
              operation,
              "https://host");
      assertThat(response)
          .isEqualTo(
              "https://host?resource_type="
                  + resourceType
                  + "&resource_name="
                  + clusterAclRequest.getConsumerGroup()
                  + "&pattern_type=LITERAL&principal=User:"
                  + clusterAclRequest.getAclSsl()
                  + "&host=*&operation="
                  + operation
                  + "&permission=ALLOW");
    }
  
    private void stubTopics() {
      when(clusterApiUtils.getRequestDetails(anyString(), eq(KafkaSupportedProtocol.SSL)))
          .thenReturn(Pair.of("", restTemplate));
      when(clusterApiUtils.createHeaders(anyString(), any())).thenReturn(new HttpHeaders());
      when(env.getProperty(any())).thenReturn(KAFKA_V_3_CLUSTERS_CLUSTER_ID_TOPICS);
    }
  
    private void stubAcls() {
      when(clusterApiUtils.getRequestDetails(anyString(), eq(KafkaSupportedProtocol.SSL)))
          .thenReturn(Pair.of("", restTemplate));
      when(clusterApiUtils.createHeaders(anyString(), any())).thenReturn(new HttpHeaders());
      when(env.getProperty(any())).thenReturn(KAFKA_V_3_CLUSTERS_CLUSTER_ID_ACLS);
    }
  
    public ClusterAclRequest getClusterAclRequest(String aclType) {
      return ClusterAclRequest.builder()
          .env("localhost")
          .topicName("testtopic")
          .protocol(KafkaSupportedProtocol.PLAINTEXT)
          .consumerGroup("congroup1")
          .clusterName("clusterName")
          .aclType(aclType)
          .aclIp("11.12.33.122")
          .aclSsl(null)
          .requestOperationType(RequestOperationType.CREATE)
          .aclNativeType(AclsNativeType.CONFLUENT_CLOUD.name())
          .aclIpPrincipleType(AclIPPrincipleType.IP_ADDRESS.name())
          .transactionalId("transactionalId")
          .build();
    }
  }
  