package io.aiven.klaw.clusterapi.services;

import static io.aiven.klaw.clusterapi.services.SchemaService.SCHEMA_VALUE_URI;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.when;
import static org.springframework.test.web.client.match.MockRestRequestMatchers.requestTo;
import static org.springframework.test.web.client.response.MockRestResponseCreators.*;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.aiven.klaw.clusterapi.constants.TestConstants;
import io.aiven.klaw.clusterapi.models.ApiResponse;
import io.aiven.klaw.clusterapi.models.ClusterTopicRequest;
import io.aiven.klaw.clusterapi.models.SchemaCompatibilityCheckResponse;
import io.aiven.klaw.clusterapi.models.SchemaInfoOfTopic;
import io.aiven.klaw.clusterapi.models.SchemasInfoOfClusterResponse;
import io.aiven.klaw.clusterapi.models.enums.*;
import io.aiven.klaw.clusterapi.utils.ClusterApiUtils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;
import org.assertj.core.api.Assertions;
import org.assertj.core.util.Lists;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.client.RestClientTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.test.web.client.MockRestServiceServer;
import org.springframework.web.client.RestTemplate;

@RestClientTest(SchemaService.class)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
class SchemaServiceTest {
  public static final String TOPIC_COMPATIBILITY_URI_TEMPLATE =
      "/compatibility/subjects/{topic_name}-value/versions/latest";

  public static final String TOPIC_GET_VERSIONS_URI_TEMPLATE =
      "/subjects/{topic_name}-value/versions";

  public static final String SCHEMA_SUBJECTS_URL =
      "http://" + TestConstants.ENVIRONMENT + "/subjects";

  @Autowired SchemaService schemaService;
  RestTemplate restTemplate;
  @Autowired ObjectMapper objectMapper;
  private ObjectMapper mapper = new ObjectMapper();
  private MockRestServiceServer mockRestServiceServer;
  @MockBean private ClusterApiUtils getAdminClient;

  private static ClusterTopicRequest deleteTopicRequest(String topicName) {
    return ClusterTopicRequest.builder()
        .clusterName("DEV2")
        .topicName(topicName)
        .env("bootStrapServersSsl")
        .protocol(KafkaSupportedProtocol.SSL)
        .partitions(1)
        .replicationFactor(Short.parseShort("1"))
        .aclsNativeType(AclsNativeType.NATIVE)
        .deleteAssociatedSchema(true)
        .schemaClusterIdentification("DEV3")
        .schemaEnv("schemaservers")
        .schemaEnvProtocol(KafkaSupportedProtocol.SSL)
        .build();
  }

  @BeforeEach
  public void setUp() {
    restTemplate = new RestTemplate();
    schemaService = new SchemaService(getAdminClient);
    mockRestServiceServer = MockRestServiceServer.bindTo(restTemplate).build();
  }

  @Test
  @Order(1)
  public void getSchema_returnMapWithSubjectCompatibility() throws JsonProcessingException {
    // getSchemaVersions
    String getSchemaVersionsUrl = "env/subjects/topic-value/versions";
    getSchemaVersions(getSchemaVersionsUrl);

    // getSchemaCompatibility
    getSchemaCompatibilitySubject(true);

    // getSchema
    getSchema();

    Map<Integer, Map<String, Object>> schemaResponse =
        schemaService.getSchema("env", KafkaSupportedProtocol.PLAINTEXT, "CLID1", "topic");

    assertThat(schemaResponse).isNotEmpty();
    assertThat(schemaResponse.get(1).get("compatibility")).isEqualTo("BACKWARD");
  }

  @Test
  @Order(2)
  public void getSchema_returnMapWithGlobalCompatibility() throws JsonProcessingException {
    // getSchemaVersions
    String getSchemaVersionsUrl = "env/subjects/topic-value/versions";
    getSchemaVersions(getSchemaVersionsUrl);

    // getSchemaCompatibility
    getSchemaCompatibilitySubject(false);
    getSchemaCompatibilityGlobal();
    // getSchema
    getSchema();

    Map<Integer, Map<String, Object>> schemaResponse =
        schemaService.getSchema("env", KafkaSupportedProtocol.PLAINTEXT, "CLID1", "topic");

    assertThat(schemaResponse).isNotEmpty();
    assertThat(schemaResponse.get(1).get("compatibility")).isEqualTo("FORWARD");
  }

  private void getSchemaCompatibilitySubject(boolean compatibilitySet)
      throws JsonProcessingException {
    String getSchemaCompatibilityUrl = "env/config/topic-value";
    when(getAdminClient.getRequestDetails(
            eq(getSchemaCompatibilityUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(getSchemaCompatibilityUrl, restTemplate));
    if (compatibilitySet) {
      this.mockRestServiceServer
          .expect(requestTo("/" + getSchemaCompatibilityUrl))
          .andRespond(
              withSuccess(
                  objectMapper.writeValueAsString(
                      Collections.singletonMap("compatibilityLevel", "BACKWARD")),
                  MediaType.APPLICATION_JSON));
    } else {
      this.mockRestServiceServer
          .expect(requestTo("/" + getSchemaCompatibilityUrl))
          .andRespond(withServerError());
    }
  }

  private void getSchemaCompatibilityGlobal() throws JsonProcessingException {
    String getSchemaCompatibilityUrl = "env/config";
    when(getAdminClient.getRequestDetails(
            eq(getSchemaCompatibilityUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(getSchemaCompatibilityUrl, restTemplate));
    this.mockRestServiceServer
        .expect(requestTo("/" + getSchemaCompatibilityUrl))
        .andRespond(
            withSuccess(
                objectMapper.writeValueAsString(
                    Collections.singletonMap("compatibilityLevel", "FORWARD")),
                MediaType.APPLICATION_JSON));
  }

  private void getSchema() throws JsonProcessingException {
    String getSchemaUrl = "env/subjects/topic-value/versions/1";
    when(getAdminClient.getRequestDetails(eq(getSchemaUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(getSchemaUrl, restTemplate));

    when(getAdminClient.getRequestDetails(eq(getSchemaUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(getSchemaUrl, restTemplate));
    this.mockRestServiceServer
        .expect(requestTo("/" + getSchemaUrl))
        .andRespond(
            withSuccess(
                objectMapper.writeValueAsString(Collections.singletonMap("foo", "bar")),
                MediaType.APPLICATION_JSON));
  }

  private void getSchemaVersions(String getSchemaVersionsUrl) throws JsonProcessingException {
    when(getAdminClient.getRequestDetails(
            eq(getSchemaVersionsUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(getSchemaVersionsUrl, restTemplate));
    this.mockRestServiceServer
        .expect(requestTo("/" + getSchemaVersionsUrl))
        .andRespond(
            withSuccess(
                objectMapper.writeValueAsString(Lists.list(1)), MediaType.APPLICATION_JSON));
  }

  @Test
  @Order(3)
  public void deleteSchema() {
    ClusterTopicRequest clusterTopicRequest = deleteTopicRequest("testtopic");

    String deleteSchemaUrl = "schemaservers/subjects/testtopic-value";
    when(getAdminClient.getRequestDetails(eq(deleteSchemaUrl), eq(KafkaSupportedProtocol.SSL)))
        .thenReturn(Pair.of(deleteSchemaUrl, restTemplate));

    this.mockRestServiceServer.expect(requestTo("/" + deleteSchemaUrl)).andRespond(withSuccess());

    ApiResponse apiResponse = schemaService.deleteSchema(clusterTopicRequest);
    assertThat(apiResponse.getMessage()).isEqualTo(ApiResultStatus.SUCCESS.value);
  }

  @Test
  @Order(4)
  public void checkSchemaCompatibility_ReturnSuccess() throws JsonProcessingException {
    String topicName = "Octopus";
    String dev = "Dev";
    mockVersionsofSchema(topicName, dev, false);
    String validationUrl =
        dev + TOPIC_COMPATIBILITY_URI_TEMPLATE.replace("{topic_name}", topicName);
    when(getAdminClient.getRequestDetails(eq(validationUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(validationUrl, restTemplate));

    when(getAdminClient.createHeaders(eq("18"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());
    SchemaCompatibilityCheckResponse check = new SchemaCompatibilityCheckResponse();
    check.setCompatible(true);
    this.mockRestServiceServer
        .expect(requestTo("/" + validationUrl))
        .andRespond(withSuccess(mapper.writeValueAsString(check), MediaType.APPLICATION_JSON));

    ApiResponse apiResponse =
        schemaService.checkSchemaCompatibility(
            "schema: {}", topicName, KafkaSupportedProtocol.PLAINTEXT, dev, "18", SchemaType.AVRO);
    assertThat(apiResponse.getMessage()).startsWith(ApiResultStatus.SUCCESS.value);

    mockRestServiceServer.verify();
  }

  @Test
  @Order(5)
  public void checkSchemaCompatibility_IsNotCompatibleReturnFailure()
      throws JsonProcessingException {
    String topicName = "Octopus";
    String dev = "Dev";

    mockVersionsofSchema(topicName, dev, false);

    String validationUrl =
        dev + TOPIC_COMPATIBILITY_URI_TEMPLATE.replace("{topic_name}", topicName);
    when(getAdminClient.getRequestDetails(eq(validationUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(validationUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("18"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());
    SchemaCompatibilityCheckResponse check = new SchemaCompatibilityCheckResponse();
    check.setCompatible(false);
    this.mockRestServiceServer
        .expect(requestTo("/" + validationUrl))
        .andRespond(withSuccess(mapper.writeValueAsString(check), MediaType.APPLICATION_JSON));

    ApiResponse apiResponse =
        schemaService.checkSchemaCompatibility(
            "schema: {}", topicName, KafkaSupportedProtocol.PLAINTEXT, dev, "18", SchemaType.AVRO);
    assertThat(apiResponse.getMessage()).startsWith(ApiResultStatus.FAILURE.value);

    mockRestServiceServer.verify();
  }

  @Test
  @Order(6)
  public void checkSchemaCompatibility_NoExistingSchema() throws JsonProcessingException {
    String topicName = "Octopus";
    String dev = "Dev";
    String validationUrl =
        dev + TOPIC_COMPATIBILITY_URI_TEMPLATE.replace("{topic_name}", topicName);
    mockVersionsofSchema(topicName, dev, true);
    when(getAdminClient.getRequestDetails(eq(validationUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(validationUrl, restTemplate));

    ApiResponse apiResponse =
        schemaService.checkSchemaCompatibility(
            "schema: {}", topicName, KafkaSupportedProtocol.PLAINTEXT, dev, "18", SchemaType.AVRO);
    assertThat(apiResponse.isSuccess()).isTrue();

    mockRestServiceServer.verify();
  }

  private void mockVersionsofSchema(String topicName, String dev, boolean isNotFound)
      throws JsonProcessingException {
    String versionUrl = dev + TOPIC_GET_VERSIONS_URI_TEMPLATE.replace("{topic_name}", topicName);
    when(getAdminClient.getRequestDetails(eq(versionUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(versionUrl, restTemplate));
    if (isNotFound) {
      this.mockRestServiceServer
          .expect(requestTo("/" + versionUrl))
          .andRespond(withResourceNotFound());
    } else {
      this.mockRestServiceServer
          .expect(requestTo("/" + versionUrl))
          .andRespond(
              withSuccess(
                  mapper.writeValueAsString(isNotFound ? new ArrayList<>() : List.of(1)),
                  MediaType.APPLICATION_JSON));
    }
  }

  @Test
  @Order(7)
  public void checkSchemaCompatibility_UnExpectedExceptionFailure() throws JsonProcessingException {
    String topicName = "Octopus";
    String dev = "Dev";

    mockVersionsofSchema(topicName, dev, false);
    // creating a different url so that the reqDetails is not created properly and causes an
    // exception
    String validationUrl =
        dev + TOPIC_COMPATIBILITY_URI_TEMPLATE.replace("{topic_name}", "notTheCorrectTopicName");
    when(getAdminClient.getRequestDetails(eq(validationUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(validationUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("18"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());
    SchemaCompatibilityCheckResponse check = new SchemaCompatibilityCheckResponse();
    check.setCompatible(true);

    ApiResponse apiResponse =
        schemaService.checkSchemaCompatibility(
            "schema: {}", topicName, KafkaSupportedProtocol.PLAINTEXT, dev, "18", SchemaType.AVRO);
    assertThat(apiResponse.getMessage())
        .isEqualTo(ApiResultStatus.FAILURE.value + " Unable to validate Schema Compatibility.");
  }

  @Test
  @Order(8)
  public void checkSchemaCompatibility_InvalidAvroSchema() throws JsonProcessingException {
    String topicName = "Octopus";
    String dev = "Dev";

    mockVersionsofSchema(topicName, dev, false);
    // creating a different url so that the reqDetails is not created properly and causes an
    // exception
    String validationUrl =
        dev + TOPIC_COMPATIBILITY_URI_TEMPLATE.replace("{topic_name}", topicName);
    when(getAdminClient.getRequestDetails(eq(validationUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(validationUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("18"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());
    this.mockRestServiceServer
        .expect(requestTo("/" + validationUrl))
        .andRespond(withRawStatus(422));

    ApiResponse apiResponse =
        schemaService.checkSchemaCompatibility(
            "schema: {}", topicName, KafkaSupportedProtocol.PLAINTEXT, dev, "18", SchemaType.AVRO);
    assertThat(apiResponse.getMessage())
        .isEqualTo(
            ApiResultStatus.FAILURE.value
                + " Invalid Schema. Unable to validate Schema Compatibility.");
  }

  @Test
  @Order(9)
  public void getSchemasOfCluster() throws JsonProcessingException {
    String dev = "Dev";
    String subjectsUrl = dev + "/subjects";
    String topic1 = "test1", topic2 = "test2";
    Set<Integer> topic1Versions = Set.of(1, 2);
    Set<Integer> topic2Versions = Set.of(1, 2, 3);

    when(getAdminClient.getRequestDetails(eq(subjectsUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(subjectsUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("19"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());
    this.mockRestServiceServer
        .expect(requestTo("/" + subjectsUrl))
        .andRespond(
            withSuccess(
                mapper.writeValueAsString(
                    List.of(topic1 + SCHEMA_VALUE_URI, topic2 + SCHEMA_VALUE_URI)),
                MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/" + subjectsUrl + "/" + topic1 + SCHEMA_VALUE_URI + "/versions"))
        .andRespond(
            withSuccess(mapper.writeValueAsString(topic1Versions), MediaType.APPLICATION_JSON));
    this.mockRestServiceServer
        .expect(requestTo("/" + subjectsUrl + "/" + topic2 + SCHEMA_VALUE_URI + "/versions"))
        .andRespond(
            withSuccess(mapper.writeValueAsString(topic2Versions), MediaType.APPLICATION_JSON));
    when(getAdminClient.getRequestDetails(
            eq(subjectsUrl + "/" + topic1 + SCHEMA_VALUE_URI + "/versions"),
            eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(
            Pair.of(subjectsUrl + "/" + topic1 + SCHEMA_VALUE_URI + "/versions", restTemplate));
    when(getAdminClient.getRequestDetails(
            eq(subjectsUrl + "/" + topic2 + SCHEMA_VALUE_URI + "/versions"),
            eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(
            Pair.of(subjectsUrl + "/" + topic2 + SCHEMA_VALUE_URI + "/versions", restTemplate));

    SchemasInfoOfClusterResponse schemasInfoOfClusterResponse =
        schemaService.loadAllSchemasInfoFromCluster(
            dev, KafkaSupportedProtocol.PLAINTEXT, "19", false, SchemaCacheUpdateType.NONE, null);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList())
        .extracting(SchemaInfoOfTopic::getTopic)
        .containsExactlyInAnyOrder(topic1, topic2);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList())
        .extracting(SchemaInfoOfTopic::getSchemaVersions)
        .containsExactlyInAnyOrder(topic1Versions, topic2Versions);
  }

  @Test
  @Order(10)
  public void getSchemasOfClusterCreateNewSchema() throws JsonProcessingException {
    String dev = "Dev";
    String subjectsUrl = dev + "/subjects";
    String topic1 = "test1", topic2 = "test2", topic3 = "test3";
    Set<Integer> topic1Versions = Set.of(1, 2);
    Set<Integer> topic2Versions = Set.of(1, 2, 3);
    Set<Integer> topic3Versions = Set.of(1);

    when(getAdminClient.getRequestDetails(eq(subjectsUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(subjectsUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("19"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());

    this.mockRestServiceServer
        .expect(requestTo("/" + subjectsUrl + "/" + topic3 + SCHEMA_VALUE_URI + "/versions"))
        .andRespond(
            withSuccess(mapper.writeValueAsString(topic3Versions), MediaType.APPLICATION_JSON));
    when(getAdminClient.getRequestDetails(
            eq(subjectsUrl + "/" + topic3 + SCHEMA_VALUE_URI + "/versions"),
            eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(
            Pair.of(subjectsUrl + "/" + topic3 + SCHEMA_VALUE_URI + "/versions", restTemplate));

    SchemasInfoOfClusterResponse schemasInfoOfClusterResponse =
        schemaService.loadAllSchemasInfoFromCluster(
            dev,
            KafkaSupportedProtocol.PLAINTEXT,
            "19",
            true,
            SchemaCacheUpdateType.CREATE,
            "test3");
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList().size()).isEqualTo(3);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList())
        .extracting(SchemaInfoOfTopic::getTopic)
        .containsExactlyInAnyOrder(topic1, topic2, topic3);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList())
        .extracting(SchemaInfoOfTopic::getSchemaVersions)
        .containsExactlyInAnyOrder(topic1Versions, topic2Versions, topic3Versions);
  }

  @Test
  @Order(11)
  public void getSchemasOfClusterDeleteSchema() throws JsonProcessingException {
    String dev = "Dev";
    String subjectsUrl = dev + "/subjects";
    String topic1 = "test1", topic2 = "test2", topic3 = "test3";
    Set<Integer> topic1Versions = Set.of(1, 2);
    Set<Integer> topic3Versions = Set.of(1);

    when(getAdminClient.getRequestDetails(eq(subjectsUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(subjectsUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("19"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());

    SchemasInfoOfClusterResponse schemasInfoOfClusterResponse =
        schemaService.loadAllSchemasInfoFromCluster(
            dev,
            KafkaSupportedProtocol.PLAINTEXT,
            "19",
            true,
            SchemaCacheUpdateType.DELETE,
            topic2);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList().size()).isEqualTo(2);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList())
        .extracting(SchemaInfoOfTopic::getTopic)
        .containsExactlyInAnyOrder(topic1, topic3);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList())
        .extracting(SchemaInfoOfTopic::getSchemaVersions)
        .containsExactlyInAnyOrder(topic1Versions, topic3Versions);
  }

  @Test
  @Order(12)
  public void getSchemasOfClusterNoSchemas() throws JsonProcessingException {
    String dev = "Dev";
    String subjectsUrl = dev + "/subjects";

    when(getAdminClient.getRequestDetails(eq(subjectsUrl), eq(KafkaSupportedProtocol.PLAINTEXT)))
        .thenReturn(Pair.of(subjectsUrl, restTemplate));
    when(getAdminClient.createHeaders(eq("20"), eq(KafkaClustersType.SCHEMA_REGISTRY)))
        .thenReturn(new HttpHeaders());
    this.mockRestServiceServer
        .expect(requestTo("/" + subjectsUrl))
        .andRespond(withSuccess(mapper.writeValueAsString(List.of()), MediaType.APPLICATION_JSON));

    SchemasInfoOfClusterResponse schemasInfoOfClusterResponse =
        schemaService.loadAllSchemasInfoFromCluster(
            dev, KafkaSupportedProtocol.PLAINTEXT, "20", false, SchemaCacheUpdateType.NONE, null);
    assertThat(schemasInfoOfClusterResponse.getSchemaInfoOfTopicList().size()).isEqualTo(0);
  }

  @Test
  @Order(13)
  public void getSchemaRegistryStatusClusterStatusOnline() throws JsonProcessingException {
    KafkaSupportedProtocol protocol = KafkaSupportedProtocol.PLAINTEXT;

    when(getAdminClient.getRequestDetails(anyString(), eq(protocol)))
        .thenReturn(Pair.of(SCHEMA_SUBJECTS_URL, restTemplate));
    this.mockRestServiceServer
        .expect(requestTo(SCHEMA_SUBJECTS_URL))
        .andRespond(
            withSuccess(objectMapper.writeValueAsString(Object.class), MediaType.APPLICATION_JSON));

    ClusterStatus actual =
        schemaService.getSchemaRegistryStatus(
            TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION);
    ClusterStatus expected = ClusterStatus.ONLINE;

    Assertions.assertThat(actual).isEqualTo(expected);
  }

  @Test
  @Order(14)
  public void getSchemaRegistryStatusClusterStatusOffline() {
    KafkaSupportedProtocol protocol = KafkaSupportedProtocol.PLAINTEXT;

    when(getAdminClient.getRequestDetails(anyString(), eq(protocol)))
        .thenReturn(Pair.of(SCHEMA_SUBJECTS_URL, restTemplate));
    this.mockRestServiceServer
        .expect(requestTo(SCHEMA_SUBJECTS_URL))
        .andRespond(withUnauthorizedRequest());

    ClusterStatus actual =
        schemaService.getSchemaRegistryStatus(
            TestConstants.ENVIRONMENT, protocol, TestConstants.CLUSTER_IDENTIFICATION);
    ClusterStatus expected = ClusterStatus.OFFLINE;

    Assertions.assertThat(actual).isEqualTo(expected);
  }
}
