content: package io.aiven.klaw.clusterapi.controller;
  
  import static org.hamcrest.Matchers.containsString;
  import static org.hamcrest.Matchers.hasSize;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.Mockito.when;
  import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
  import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
  import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
  
  import com.fasterxml.jackson.databind.ObjectMapper;
  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.enums.AclType;
  import io.aiven.klaw.clusterapi.models.enums.AclsNativeType;
  import io.aiven.klaw.clusterapi.models.enums.ApiResultStatus;
  import io.aiven.klaw.clusterapi.models.enums.ClusterStatus;
  import io.aiven.klaw.clusterapi.models.enums.KafkaSupportedProtocol;
  import io.aiven.klaw.clusterapi.services.AivenApiService;
  import io.aiven.klaw.clusterapi.services.ApacheKafkaAclService;
  import io.aiven.klaw.clusterapi.services.ApacheKafkaTopicService;
  import io.aiven.klaw.clusterapi.services.ConfluentCloudApiService;
  import io.aiven.klaw.clusterapi.services.UtilComponentsService;
  import java.nio.charset.StandardCharsets;
  import java.util.HashMap;
  import java.util.Map;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.extension.ExtendWith;
  import org.springframework.boot.test.mock.mockito.MockBean;
  import org.springframework.http.MediaType;
  import org.springframework.test.context.junit.jupiter.SpringExtension;
  import org.springframework.test.web.servlet.MockMvc;
  import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
  import org.springframework.test.web.servlet.setup.MockMvcBuilders;
  
  @ExtendWith(SpringExtension.class)
  public class ClusterApiControllerTest {
  
    @MockBean private UtilComponentsService utilComponentsService;
    @MockBean private ApacheKafkaAclService apacheKafkaAclService;
    @MockBean private ApacheKafkaTopicService apacheKafkaTopicService;
    @MockBean private AivenApiService aivenApiService;
  
    @MockBean private ConfluentCloudApiService confluentCloudApiService;
  
    private MockMvc mvc;
  
    private UtilMethods utilMethods;
  
    @BeforeEach
    public void setUp() throws Exception {
      utilMethods = new UtilMethods();
      ClusterApiController clusterApiController =
          new ClusterApiController(
              utilComponentsService,
              apacheKafkaAclService,
              apacheKafkaTopicService,
              aivenApiService,
              confluentCloudApiService);
      mvc = MockMvcBuilders.standaloneSetup(clusterApiController).dispatchOptions(true).build();
    }
  
    @Test
    public void getApiStatus() throws Exception {
      mvc.perform(get("/topics/getApiStatus"))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().json("\"ONLINE\""));
    }
  
    @Test
    public void getStatus() throws Exception {
      String clusterName = "testCluster";
      String clusterType = "sampleType";
      String bootstrapServers = "localhost:9092";
  
      when(utilComponentsService.getStatus(
              bootstrapServers,
              KafkaSupportedProtocol.PLAINTEXT,
              clusterName,
              clusterType,
              "Apache Kafka"))
          .thenReturn(ClusterStatus.ONLINE);
  
      String urlTemplate =
          String.join(
              "/",
              "/topics",
              "getStatus",
              bootstrapServers,
              KafkaSupportedProtocol.PLAINTEXT.getValue(),
              clusterName,
              clusterType,
              "kafkaFlavor",
              "Apache Kafka");
      mvc.perform(
              MockMvcRequestBuilders.get(urlTemplate)
                  .contentType(MediaType.APPLICATION_JSON)
                  .accept(MediaType.APPLICATION_JSON))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().json("\"ONLINE\""));
    }
  
    @Test
    public void getTopics() throws Exception {
      String clusterName = "testCluster";
      String bootstrapServers = "localhost:9092";
  
      when(apacheKafkaTopicService.loadTopics(
              bootstrapServers, KafkaSupportedProtocol.PLAINTEXT, clusterName, false))
          .thenReturn(utilMethods.getTopics());
  
      String urlTemplate =
          String.join(
              "/",
              "/topics",
              "getTopics",
              bootstrapServers,
              KafkaSupportedProtocol.PLAINTEXT.getValue(),
              clusterName,
              "topicsNativeType",
              AclsNativeType.NATIVE.value,
              "resetCache",
              "false");
      mvc.perform(get(urlTemplate))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(jsonPath("$.topicConfigSet", hasSize(1)));
    }
  
    @Test
    public void getAcls() throws Exception {
      String clusterName = "testCluster";
      String bootstrapServers = "localhost:9092";
      String aclsNativeType = AclsNativeType.NATIVE.name();
      String projectName = "projectName";
      String serviceName = "serviceName";
  
      when(apacheKafkaAclService.loadAcls(
              bootstrapServers, KafkaSupportedProtocol.PLAINTEXT, clusterName))
          .thenReturn(utilMethods.getAcls());
  
      String urlTemplate =
          String.join(
              "/",
              "/topics",
              "getAcls",
              bootstrapServers,
              aclsNativeType,
              KafkaSupportedProtocol.PLAINTEXT.getValue(),
              clusterName,
              projectName,
              serviceName);
      mvc.perform(get(urlTemplate))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(jsonPath("$", hasSize(2)));
    }
  
    @Test
    public void createTopics() throws Exception {
      String jsonReq = new ObjectMapper().writer().writeValueAsString(utilMethods.getTopicRequest());
      ApiResponse apiResponse = ApiResponse.builder().message(ApiResultStatus.SUCCESS.value).build();
  
      when(apacheKafkaTopicService.createTopic(any(ClusterTopicRequest.class)))
          .thenReturn(apiResponse);
  
      mvc.perform(
              post("/topics/createTopics")
                  .content(jsonReq)
                  .contentType(MediaType.APPLICATION_JSON)
                  .characterEncoding(StandardCharsets.UTF_8))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().string(containsString(ApiResultStatus.SUCCESS.value)));
    }
  
    @Test
    public void createTopicsConfluentCloud() throws Exception {
      ClusterTopicRequest topicReq = utilMethods.getConfluentCloudTopicRequest();
      String jsonReq = new ObjectMapper().writer().writeValueAsString(topicReq);
      ApiResponse apiResponse = ApiResponse.builder().message(ApiResultStatus.SUCCESS.value).build();
  
      when(confluentCloudApiService.createTopic(any(ClusterTopicRequest.class)))
          .thenReturn(apiResponse);
  
      mvc.perform(
              post("/topics/createTopics")
                  .content(jsonReq)
                  .contentType(MediaType.APPLICATION_JSON)
                  .characterEncoding(StandardCharsets.UTF_8))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().string(containsString(ApiResultStatus.SUCCESS.value)));
    }
  
    @Test
    public void createAclsProducer() throws Exception {
      ClusterAclRequest clusterAclRequest = utilMethods.getAclRequest(AclType.PRODUCER.value);
      String jsonReq = new ObjectMapper().writer().writeValueAsString(clusterAclRequest);
  
      when(apacheKafkaAclService.updateProducerAcl(any(ClusterAclRequest.class)))
          .thenReturn(ApiResultStatus.SUCCESS.value);
  
      mvc.perform(
              post("/topics/createAcls")
                  .content(jsonReq)
                  .contentType(MediaType.APPLICATION_JSON)
                  .characterEncoding(StandardCharsets.UTF_8))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().string(containsString(ApiResultStatus.SUCCESS.value)));
    }
  
    @Test
    public void createAclsProducerConfluentCloud() throws Exception {
      ClusterAclRequest clusterAclRequest = utilMethods.getConfluentCloudProducerAclRequest();
      String jsonReq = new ObjectMapper().writer().writeValueAsString(clusterAclRequest);
      Map<String, String> aclResponse = new HashMap<>();
      aclResponse.put("result", ApiResultStatus.SUCCESS.value);
  
      when(confluentCloudApiService.createAcls(any(ClusterAclRequest.class))).thenReturn(aclResponse);
  
      mvc.perform(
              post("/topics/createAcls")
                  .content(jsonReq)
                  .contentType(MediaType.APPLICATION_JSON)
                  .characterEncoding(StandardCharsets.UTF_8))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().string(containsString(ApiResultStatus.SUCCESS.value)));
    }
  
    @Test
    public void createAclsConsumer() throws Exception {
      ClusterAclRequest clusterAclRequest = utilMethods.getAclRequest(AclType.CONSUMER.value);
      String jsonReq = new ObjectMapper().writer().writeValueAsString(clusterAclRequest);
  
      when(apacheKafkaAclService.updateConsumerAcl(any(ClusterAclRequest.class)))
          .thenReturn("success1");
  
      mvc.perform(
              post("/topics/createAcls")
                  .content(jsonReq)
                  .contentType(MediaType.APPLICATION_JSON)
                  .characterEncoding(StandardCharsets.UTF_8))
          .andExpect(status().isOk())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(content().string(containsString(ApiResultStatus.SUCCESS.value)));
    }
  
    @Test
    public void createAclsConsumerFail() throws Exception {
      ClusterAclRequest clusterAclRequest = utilMethods.getAclRequest(AclType.CONSUMER.value);
      String jsonReq = new ObjectMapper().writer().writeValueAsString(clusterAclRequest);
  
      when(apacheKafkaAclService.updateConsumerAcl(any(ClusterAclRequest.class)))
          .thenThrow(new RuntimeException("Error creating acls"));
  
      mvc.perform(post("/topics/createAcls").content(jsonReq).contentType(MediaType.APPLICATION_JSON))
          .andExpect(status().is5xxServerError())
          .andExpect(content().contentType(MediaType.APPLICATION_JSON))
          .andExpect(jsonPath("$.message", containsString("Error creating acls")));
    }
  }
  