content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.kafka.connect.runtime.rest;
  
  import org.apache.kafka.connect.runtime.rest.entities.ErrorMessage;
  import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
  
  import com.fasterxml.jackson.annotation.JsonCreator;
  import com.fasterxml.jackson.annotation.JsonProperty;
  import com.fasterxml.jackson.core.type.TypeReference;
  import com.fasterxml.jackson.databind.ObjectMapper;
  
  import org.eclipse.jetty.client.ContentResponse;
  import org.eclipse.jetty.client.HttpClient;
  import org.eclipse.jetty.client.Request;
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.extension.ExtendWith;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.Arguments;
  import org.junit.jupiter.params.provider.MethodSource;
  import org.mockito.Mock;
  import org.mockito.junit.jupiter.MockitoExtension;
  import org.mockito.junit.jupiter.MockitoSettings;
  import org.mockito.quality.Strictness;
  
  import java.nio.charset.StandardCharsets;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.TimeoutException;
  import java.util.stream.Stream;
  
  import javax.crypto.SecretKey;
  
  import jakarta.ws.rs.core.Response;
  
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.Mockito.doReturn;
  import static org.mockito.Mockito.doThrow;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.spy;
  import static org.mockito.Mockito.when;
  
  @ExtendWith(MockitoExtension.class)
  @MockitoSettings(strictness = Strictness.STRICT_STUBS)
  public class RestClientTest {
  
      private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
      private static final String MOCK_URL = "http://localhost:1234/api/endpoint";
      private static final String TEST_METHOD = "GET";
      private static final TestDTO TEST_DTO = new TestDTO("requestBodyData");
      private static final TypeReference<TestDTO> TEST_TYPE = new TypeReference<>() { };
      private static final SecretKey MOCK_SECRET_KEY = getMockSecretKey();
      private static final String TEST_SIGNATURE_ALGORITHM = "HmacSHA1";
  
      @Mock
      private HttpClient httpClient;
  
      private static void assertIsInternalServerError(ConnectRestException e) {
          assertEquals(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), e.statusCode());
          assertEquals(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(), e.errorCode());
      }
  
      private static SecretKey getMockSecretKey() {
          SecretKey mockKey = mock(SecretKey.class);
          when(mockKey.getFormat()).thenReturn("RAW");
          when(mockKey.getEncoded()).thenReturn("SomeKey".getBytes(StandardCharsets.UTF_8));
          return mockKey;
      }
  
      private static <T> RestClient.HttpResponse<T> httpRequest(
              HttpClient httpClient,
              String url,
              String method,
              TypeReference<T> responseFormat,
              String requestSignatureAlgorithm
      ) {
          RestClient client = spy(new RestClient(null));
          doReturn(httpClient).when(client).httpClient(any());
          return client.httpRequest(
                  url,
                  method,
                  null,
                  TEST_DTO,
                  responseFormat,
                  MOCK_SECRET_KEY,
                  requestSignatureAlgorithm
          );
      }
  
      private static Stream<Arguments> requestExceptions() {
          return Stream.of(
                  Arguments.of(new InterruptedException()),
                  Arguments.of(new ExecutionException(null)),
                  Arguments.of(new TimeoutException())
          );
      }
  
      private static Request buildThrowingMockRequest(Throwable t) throws ExecutionException, InterruptedException, TimeoutException {
          Request req = mock(Request.class);
          when(req.headers(any())).thenReturn(req);
          when(req.send()).thenThrow(t);
          return req;
      }
  
      @ParameterizedTest
      @MethodSource("requestExceptions")
      public void testFailureDuringRequestCausesInternalServerError(Throwable requestException) throws Exception {
          Request request = buildThrowingMockRequest(requestException);
          when(httpClient.newRequest(anyString())).thenReturn(request);
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          assertIsInternalServerError(e);
          assertEquals(requestException, e.getCause());
      }
  
      @Test
      public void testNullUrl() {
          RestClient client = spy(new RestClient(null));
          assertThrows(NullPointerException.class, () -> client.httpRequest(
                  null,
                  TEST_METHOD,
                  null,
                  TEST_DTO,
                  TEST_TYPE,
                  MOCK_SECRET_KEY,
                  TEST_SIGNATURE_ALGORITHM
          ));
      }
  
      @Test
      public void testNullMethod() {
          RestClient client = spy(new RestClient(null));
          assertThrows(NullPointerException.class, () -> client.httpRequest(
                  MOCK_URL,
                  null,
                  null,
                  TEST_DTO,
                  TEST_TYPE,
                  MOCK_SECRET_KEY,
                  TEST_SIGNATURE_ALGORITHM
          ));
      }
  
      @Test
      public void testNullResponseType() {
          RestClient client = spy(new RestClient(null));
          assertThrows(NullPointerException.class, () -> client.httpRequest(
                  MOCK_URL,
                  TEST_METHOD,
                  null,
                  TEST_DTO,
                  null,
                  MOCK_SECRET_KEY,
                  TEST_SIGNATURE_ALGORITHM
          ));
      }
  
      @Test
      public void testSuccess() throws Exception {
          int statusCode = Response.Status.OK.getStatusCode();
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
          when(resp.getContentAsString()).thenReturn(toJsonString(TEST_DTO));
          setupHttpClient(statusCode, req, resp);
  
          RestClient.HttpResponse<TestDTO> httpResp = httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          );
          assertEquals(statusCode, httpResp.status());
          assertEquals(TEST_DTO, httpResp.body());
      }
  
      @Test
      public void testNoContent() throws Exception {
          int statusCode = Response.Status.NO_CONTENT.getStatusCode();
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
          setupHttpClient(statusCode, req, resp);
  
          RestClient.HttpResponse<TestDTO> httpResp = httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          );
          assertEquals(statusCode, httpResp.status());
          assertNull(httpResp.body());
      }
  
      @Test
      public void testStatusCodeAndErrorMessagePreserved() throws Exception {
          int statusCode = Response.Status.CONFLICT.getStatusCode();
          ErrorMessage errorMsg = new ErrorMessage(Response.Status.GONE.getStatusCode(), "Some Error Message");
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
          when(resp.getContentAsString()).thenReturn(toJsonString(errorMsg));
          setupHttpClient(statusCode, req, resp);
  
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          assertEquals(statusCode, e.statusCode());
          assertEquals(errorMsg.errorCode(), e.errorCode());
          assertEquals(errorMsg.message(), e.getMessage());
      }
  
      @Test
      public void testNonEmptyResponseWithVoidResponseType() throws Exception {
          int statusCode = Response.Status.OK.getStatusCode();
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
          when(resp.getContentAsString()).thenReturn(toJsonString(TEST_DTO));
          setupHttpClient(statusCode, req, resp);
  
          TypeReference<Void> voidResponse = new TypeReference<>() { };
          RestClient.HttpResponse<Void> httpResp = httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, voidResponse, TEST_SIGNATURE_ALGORITHM
          );
          assertEquals(statusCode, httpResp.status());
          assertNull(httpResp.body());
      }
  
      @Test
      public void testUnexpectedHttpResponseCausesInternalServerError() throws Exception {
          int statusCode = Response.Status.NOT_MODIFIED.getStatusCode();
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
  
          setupHttpClient(statusCode, req, resp);
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          assertIsInternalServerError(e);
      }
  
      @Test
      public void testRuntimeExceptionCausesInternalServerError() {
          when(httpClient.newRequest(anyString())).thenThrow(new RuntimeException());
  
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          assertIsInternalServerError(e);
      }
  
      @Test
      public void testRequestSignatureFailureCausesInternalServerError() {
          String invalidRequestSignatureAlgorithm = "Foo";
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, invalidRequestSignatureAlgorithm
          ));
          assertIsInternalServerError(e);
      }
  
      @Test
      public void testIOExceptionCausesInternalServerError() throws Exception {
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
          setupHttpClient(201, req, resp);
  
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          assertIsInternalServerError(e);
      }
  
      @Test
      public void testUseSslConfigsOnlyWhenNecessary() throws Exception {
          int statusCode = Response.Status.OK.getStatusCode();
          Request req = mock(Request.class);
          ContentResponse resp = mock(ContentResponse.class);
          when(resp.getContentAsString()).thenReturn(toJsonString(TEST_DTO));
          setupHttpClient(statusCode, req, resp);
          assertDoesNotThrow(() -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          String httpsUrl = "https://localhost:1234/api/endpoint";
          RestClient client = spy(new RestClient(null));
          assertThrows(RuntimeException.class, () -> client.httpRequest(
                  httpsUrl,
                  TEST_METHOD,
                  null,
                  TEST_DTO,
                  TEST_TYPE,
                  MOCK_SECRET_KEY,
                  TEST_SIGNATURE_ALGORITHM
          ));
      }
  
      @Test
      public void testHttpRequestInterrupted() throws ExecutionException, InterruptedException, TimeoutException {
          Request req = mock(Request.class);
          doThrow(new InterruptedException()).when(req).send();
          doReturn(req).when(req).headers(any());
          doReturn(req).when(httpClient).newRequest(anyString());
          ConnectRestException e = assertThrows(ConnectRestException.class, () -> httpRequest(
                  httpClient, MOCK_URL, TEST_METHOD, TEST_TYPE, TEST_SIGNATURE_ALGORITHM
          ));
          assertIsInternalServerError(e);
          assertInstanceOf(InterruptedException.class, e.getCause());
          assertTrue(Thread.interrupted());
      }
  
      private void setupHttpClient(int responseCode, Request req, ContentResponse resp) throws Exception {
          when(resp.getStatus()).thenReturn(responseCode);
          when(req.send()).thenReturn(resp);
          when(req.headers(any())).thenReturn(req);
          when(httpClient.newRequest(anyString())).thenReturn(req);
      }
  
      private String toJsonString(Object obj) {
          return assertDoesNotThrow(() -> OBJECT_MAPPER.writeValueAsString(obj));
      }
  
      private record TestDTO(String content) {
          @JsonCreator
          private TestDTO(@JsonProperty(value = "content") String content) {
              this.content = content;
          }
  
      }
  }
  