refactor: split out KsqlTarget helper methods

ksql-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlClientUtil.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.rest.client;
+
+import io.confluent.ksql.rest.Errors;
+import io.confluent.ksql.rest.entity.KsqlErrorMessage;
+import java.util.Optional;
+import java.util.function.Function;
+import javax.naming.AuthenticationException;
+import javax.ws.rs.core.Response;
+import org.eclipse.jetty.http.HttpStatus;
+import org.eclipse.jetty.http.HttpStatus.Code;
+
+public final class KsqlClientUtil {
+
+  private KsqlClientUtil() {
+  }
+
+  public static <T> RestResponse<T> toRestResponse(
+      final Response response,
+      final String path,
+      final Function<Response, T> mapper
+  ) {
+    final Code statusCode = HttpStatus.getCode(response.getStatus());
+    return statusCode == Code.OK
+        ? RestResponse.successful(statusCode, mapper.apply(response))
+        : createErrorResponse(path, response);
+  }
+
+  private static <T> RestResponse<T> createErrorResponse(
+      final String path,
+      final Response response
+  ) {
+    final Code statusCode = HttpStatus.getCode(response.getStatus());
+    final Optional<KsqlErrorMessage> errorMessage = tryReadErrorMessage(response);
+    if (errorMessage.isPresent()) {
+      return RestResponse.erroneous(statusCode, errorMessage.get());
+    }
+
+    if (statusCode == Code.NOT_FOUND) {
+      return RestResponse.erroneous(statusCode,
+          "Path not found. Path='" + path + "'. "
+              + "Check your ksql http url to make sure you are connecting to a ksql server."
+      );
+    }
+
+    if (statusCode == Code.UNAUTHORIZED) {
+      return RestResponse.erroneous(statusCode, unauthorizedErrorMsg());
+    }
+
+    if (statusCode == Code.FORBIDDEN) {
+      return RestResponse.erroneous(statusCode, forbiddenErrorMsg());
+    }
+
+    return RestResponse.erroneous(
+        statusCode,
+        "The server returned an unexpected error: "
+            + response.getStatusInfo().getReasonPhrase());
+  }
+
+  private static Optional<KsqlErrorMessage> tryReadErrorMessage(final Response response) {
+    try {
+      return Optional.ofNullable(response.readEntity(KsqlErrorMessage.class));
+    } catch (final Exception e) {
+      return Optional.empty();
+    }
+  }
+
+  private static KsqlErrorMessage unauthorizedErrorMsg() {
+    return new KsqlErrorMessage(
+        Errors.ERROR_CODE_UNAUTHORIZED,
+        new AuthenticationException(
+            "Could not authenticate successfully with the supplied credentials.")
+    );
+  }
+
+  private static KsqlErrorMessage forbiddenErrorMsg() {
+    return new KsqlErrorMessage(
+        Errors.ERROR_CODE_FORBIDDEN,
+        new AuthenticationException("You are forbidden from using this cluster.")
+    );
+  }
+}

ksql-rest-client/src/main/java/io/confluent/ksql/rest/client/KsqlTarget.java

@@ -18,19 +18,16 @@
 import static java.util.Objects.requireNonNull;
 
 import io.confluent.ksql.properties.LocalProperties;
-import io.confluent.ksql.rest.Errors;
 import io.confluent.ksql.rest.entity.CommandStatus;
 import io.confluent.ksql.rest.entity.CommandStatuses;
 import io.confluent.ksql.rest.entity.KsqlEntityList;
-import io.confluent.ksql.rest.entity.KsqlErrorMessage;
 import io.confluent.ksql.rest.entity.KsqlRequest;
 import io.confluent.ksql.rest.entity.ServerInfo;
 import java.io.InputStream;
 import java.net.SocketTimeoutException;
 import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
-import javax.naming.AuthenticationException;
 import javax.ws.rs.ProcessingException;
 import javax.ws.rs.client.Entity;
 import javax.ws.rs.client.WebTarget;
@@ -39,8 +36,6 @@
 import javax.ws.rs.core.MultivaluedHashMap;
 import javax.ws.rs.core.MultivaluedMap;
 import javax.ws.rs.core.Response;
-import org.eclipse.jetty.http.HttpStatus;
-import org.eclipse.jetty.http.HttpStatus.Code;
 import org.glassfish.jersey.client.ClientProperties;
 
 @SuppressWarnings("WeakerAccess") // Public API
@@ -139,11 +134,7 @@ private <T> RestResponse<T> get(final String path, final Class<T> type) {
         .headers(headers())
         .get()
     ) {
-      final Code statusCode = HttpStatus.getCode(response.getStatus());
-      return statusCode == Code.OK
-          ? RestResponse.successful(statusCode, response.readEntity(type))
-          : createErrorResponse(path, response);
-
+      return KsqlClientUtil.toRestResponse(response, path, r -> r.readEntity(type));
     } catch (final Exception e) {
       throw new KsqlRestClientException("Error issuing GET to KSQL server. path:" + path, e);
     }
@@ -166,11 +157,7 @@ private <T> RestResponse<T> post(
           .headers(headers())
           .post(Entity.json(jsonEntity));
 
-      final Code statusCode = HttpStatus.getCode(response.getStatus());
-      return statusCode == Code.OK
-          ? RestResponse.successful(statusCode, mapper.apply(response))
-          : createErrorResponse(path, response);
-
+      return KsqlClientUtil.toRestResponse(response, path, mapper);
     } catch (final ProcessingException e) {
       if (shouldRetry(readTimeoutMs, e)) {
         return post(path, jsonEntity, calcReadTimeout(readTimeoutMs), closeResponse, mapper);
@@ -202,58 +189,4 @@ private static boolean shouldRetry(
   private static Optional<Integer> calcReadTimeout(final Optional<Integer> previousTimeoutMs) {
     return previousTimeoutMs.map(timeout -> Math.min(timeout * 2, MAX_TIMEOUT));
   }
-
-  private static <T> RestResponse<T> createErrorResponse(
-      final String path,
-      final Response response
-  ) {
-    final Code statusCode = HttpStatus.getCode(response.getStatus());
-    final Optional<KsqlErrorMessage> errorMessage = tryReadErrorMessage(response);
-    if (errorMessage.isPresent()) {
-      return RestResponse.erroneous(statusCode, errorMessage.get());
-    }
-
-    if (statusCode == Code.NOT_FOUND) {
-      return RestResponse.erroneous(statusCode,
-          "Path not found. Path='" + path + "'. "
-              + "Check your ksql http url to make sure you are connecting to a ksql server."
-      );
-    }
-
-    if (statusCode == Code.UNAUTHORIZED) {
-      return RestResponse.erroneous(statusCode, unauthorizedErrorMsg());
-    }
-
-    if (statusCode == Code.FORBIDDEN) {
-      return RestResponse.erroneous(statusCode, forbiddenErrorMsg());
-    }
-
-    return RestResponse.erroneous(
-        statusCode,
-        "The server returned an unexpected error: "
-            + response.getStatusInfo().getReasonPhrase());
-  }
-
-  private static Optional<KsqlErrorMessage> tryReadErrorMessage(final Response response) {
-    try {
-      return Optional.ofNullable(response.readEntity(KsqlErrorMessage.class));
-    } catch (final Exception e) {
-      return Optional.empty();
-    }
-  }
-
-  private static KsqlErrorMessage unauthorizedErrorMsg() {
-    return new KsqlErrorMessage(
-        Errors.ERROR_CODE_UNAUTHORIZED,
-        new AuthenticationException(
-            "Could not authenticate successfully with the supplied credentials.")
-    );
-  }
-
-  private static KsqlErrorMessage forbiddenErrorMsg() {
-    return new KsqlErrorMessage(
-        Errors.ERROR_CODE_FORBIDDEN,
-        new AuthenticationException("You are forbidden from using this cluster.")
-    );
-  }
 }

ksql-rest-client/src/test/java/io/confluent/ksql/rest/client/KsqlClientUtilTest.java

@@ -0,0 +1,160 @@
+/*
+ * Copyright 2019 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.rest.client;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.confluent.ksql.rest.entity.KsqlEntityList;
+import io.confluent.ksql.rest.entity.KsqlErrorMessage;
+import java.util.function.Function;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import org.eclipse.jetty.http.HttpStatus.Code;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KsqlClientUtilTest {
+
+  private static final String PATH = "/ksql";
+  private static final String ERROR_REASON = "something bad";
+
+  @Mock
+  private Response response;
+  @Mock
+  private Function<Response, KsqlEntityList> mapper;
+  @Mock
+  private KsqlEntityList entities;
+  @Mock
+  private KsqlErrorMessage errorMessage;
+  @Mock
+  private Response.StatusType statusInfo;
+
+  @Before
+  public void setUp() {
+    when(mapper.apply(response)).thenReturn(entities);
+  }
+
+  @Test
+  public void shouldCreateRestResponseFromSuccessfulResponse() {
+    // Given:
+    when(response.getStatus()).thenReturn(Status.OK.getStatusCode());
+
+    // When:
+    final RestResponse<KsqlEntityList> restResponse =
+        KsqlClientUtil.toRestResponse(response, PATH, mapper);
+
+    // Then:
+    assertThat("is successful", restResponse.isSuccessful());
+    assertThat(restResponse.getStatusCode(), is(Code.OK));
+    assertThat(restResponse.getResponse(), sameInstance(entities));
+  }
+
+  @Test
+  public void shouldCreateRestResponseFromUnsuccessfulResponseWithMessage() {
+    // Given:
+    when(response.getStatus()).thenReturn(Status.BAD_REQUEST.getStatusCode());
+    when(response.readEntity(KsqlErrorMessage.class)).thenReturn(errorMessage);
+
+    // When:
+    final RestResponse<KsqlEntityList> restResponse =
+        KsqlClientUtil.toRestResponse(response, PATH, mapper);
+
+    // Then:
+    assertThat("is erroneous", restResponse.isErroneous());
+    assertThat(restResponse.getStatusCode(), is(Code.BAD_REQUEST));
+    assertThat(restResponse.getErrorMessage(), is(errorMessage));
+    verify(mapper, never()).apply(any());
+  }
+
+  @Test
+  public void shouldCreateRestResponseFromNotFoundResponse() {
+    // Given:
+    when(response.getStatus()).thenReturn(Status.NOT_FOUND.getStatusCode());
+
+    // When:
+    final RestResponse<KsqlEntityList> restResponse =
+        KsqlClientUtil.toRestResponse(response, PATH, mapper);
+
+    // Then:
+    assertThat("is erroneous", restResponse.isErroneous());
+    assertThat(restResponse.getStatusCode(), is(Code.NOT_FOUND));
+    assertThat(restResponse.getErrorMessage().getMessage(),
+        containsString("Check your ksql http url to make sure you are connecting to a ksql server"));
+  }
+
+  @Test
+  public void shouldCreateRestResponseFromUnauthorizedResponse() {
+    // Given:
+    when(response.getStatus()).thenReturn(Status.UNAUTHORIZED.getStatusCode());
+
+    // When:
+    final RestResponse<KsqlEntityList> restResponse =
+        KsqlClientUtil.toRestResponse(response, PATH, mapper);
+
+    // Then:
+    assertThat("is erroneous", restResponse.isErroneous());
+    assertThat(restResponse.getStatusCode(), is(Code.UNAUTHORIZED));
+    assertThat(restResponse.getErrorMessage().getMessage(),
+        containsString("Could not authenticate successfully with the supplied credential"));
+  }
+
+  @Test
+  public void shouldCreateRestResponseFromForbiddenResponse() {
+    // Given:
+    when(response.getStatus()).thenReturn(Status.FORBIDDEN.getStatusCode());
+
+    // When:
+    final RestResponse<KsqlEntityList> restResponse =
+        KsqlClientUtil.toRestResponse(response, PATH, mapper);
+
+    // Then:
+    assertThat("is erroneous", restResponse.isErroneous());
+    assertThat(restResponse.getStatusCode(), is(Code.FORBIDDEN));
+    assertThat(restResponse.getErrorMessage().getMessage(),
+        containsString("You are forbidden from using this cluster"));
+  }
+
+  @Test
+  public void shouldCreateRestResponseFromUnknownResponse() {
+    // Given:
+    when(response.getStatus()).thenReturn(Status.INTERNAL_SERVER_ERROR.getStatusCode());
+    when(response.getStatusInfo()).thenReturn(statusInfo);
+    when(statusInfo.getReasonPhrase()).thenReturn(ERROR_REASON);
+
+    // When:
+    final RestResponse<KsqlEntityList> restResponse =
+        KsqlClientUtil.toRestResponse(response, PATH, mapper);
+
+    // Then:
+    assertThat("is erroneous", restResponse.isErroneous());
+    assertThat(restResponse.getStatusCode(), is(Code.INTERNAL_SERVER_ERROR));
+    assertThat(restResponse.getErrorMessage().getMessage(),
+        containsString("The server returned an unexpected error"));
+    assertThat(restResponse.getErrorMessage().getMessage(),
+        containsString(ERROR_REASON));
+  }
+}