test: add RestQueryTranslationTest

ksql-engine/src/test/java/io/confluent/ksql/integration/IntegrationTestHarness.java

@@ -477,26 +477,6 @@ public void waitForTopicsToBePresent(final String... topicNames) throws Exceptio
         "topics not all present after 30 seconds. topics: " + Arrays.toString(topicNames));
   }
 
-  /**
-   * Wait for topics with names {@code topicNames} to not exist in Kafka.
-   *
-   * @param topicNames the names of the topics to await absence for.
-   */
-  public void waitForTopicsToBeAbsent(final String... topicNames) throws Exception {
-    TestUtils.waitForCondition(
-        () -> {
-          try {
-            final KafkaTopicClient topicClient = serviceContext.get().getTopicClient();
-            return Arrays.stream(topicNames)
-                .noneMatch(topicClient::isTopicExists);
-          } catch (Exception e) {
-            throw new RuntimeException("could not get subjects");
-          }
-        },
-        30_000,
-        "topics not all absent after 30 seconds. topics: " + Arrays.toString(topicNames));
-  }
-
   /**
    * Wait for a subject with name {@code subjectName} to exist in Schema Registry.
    *

ksql-functional-tests/pom.xml

@@ -74,6 +74,21 @@
       <artifactId>kafka-streams-test-utils</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.confluent.ksql</groupId>
+      <artifactId>ksql-rest-app</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.confluent.ksql</groupId>
+      <artifactId>ksql-rest-app</artifactId>
+      <version>${project.version}</version>
+      <type>test-jar</type>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>io.confluent.ksql</groupId>
       <artifactId>ksql-test-util</artifactId>

ksql-functional-tests/src/main/java/io/confluent/ksql/engine/FakeInsertValuesExecutor.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
+import java.util.Optional;
 import org.apache.kafka.clients.producer.ProducerRecord;
 
 public final class FakeInsertValuesExecutor  {
@@ -59,7 +60,7 @@ void sendRecord(final ProducerRecord<byte[], byte[]> record) {
                   fakeKafkaService.getTopic(record.topic()),
                   new String(record.key(), StandardCharsets.UTF_8),
                   value,
-                  record.timestamp(),
+                  Optional.of(record.timestamp()),
                   null
               ),
               null)

ksql-functional-tests/src/main/java/io/confluent/ksql/test/TestFrameworkException.java

@@ -23,4 +23,8 @@ public class TestFrameworkException extends RuntimeException {
   public TestFrameworkException(final String msg) {
     super(msg);
   }
+
+  public TestFrameworkException(final String msg, final Throwable cause) {
+    super(msg, cause);
+  }
 }

ksql-functional-tests/src/main/java/io/confluent/ksql/test/loader/JsonTestLoader.java

@@ -0,0 +1,155 @@
+/*
+ * 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.test.loader;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.confluent.ksql.test.TestFrameworkException;
+import io.confluent.ksql.test.tools.Test;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Load JSON tests from a directory structure
+ */
+public final class JsonTestLoader<T extends Test> implements TestLoader<T> {
+
+  // Pass a single test or multiple tests separated by commas to the test framework.
+  // Example:
+  //   mvn test -pl ksql-engine -Dtest=QueryTranslationTest -Dksql.test.files=test1.json
+  //   mvn test -pl ksql-engine -Dtest=QueryTranslationTest -Dksql.test.files=test1.json,test2,json
+  private static final String KSQL_TEST_FILES = "ksql.test.files";
+
+  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+  private final Path testDir;
+  private final Class<? extends TestFile<T>> testFileType;
+
+  public static <T extends Test> JsonTestLoader<T> of(
+      final Path testDir,
+      final Class<? extends TestFile<T>> testFileType
+  ) {
+    return new JsonTestLoader<>(
+        testDir,
+        testFileType
+    );
+  }
+
+  private JsonTestLoader(
+      final Path testDir,
+      final Class<? extends TestFile<T>> testFileType
+  ) {
+    this.testDir = Objects.requireNonNull(testDir, "testDir");
+    this.testFileType = Objects.requireNonNull(testFileType, "testFileType");
+  }
+
+  public Stream<T> load() {
+    final List<String> whiteList = getWhiteList();
+    final List<Path> testPaths = whiteList.isEmpty()
+        ? loadTestPathsFromDirectory()
+        : getTestPathsFromWhiteList(whiteList);
+
+    final List<T> testCases = testPaths
+        .stream()
+        .flatMap(testPath -> buildTests(testPath, testFileType))
+        .collect(Collectors.toList());
+
+    throwOnDuplicateNames(testCases);
+
+    return testCases.stream();
+  }
+
+  private List<Path> getTestPathsFromWhiteList(final List<String> whiteList) {
+    return whiteList.stream()
+        .map(name -> testDir.resolve(name.trim()))
+        .collect(Collectors.toList());
+  }
+
+  private List<Path> loadTestPathsFromDirectory() {
+    final InputStream s = JsonTestLoader.class.getClassLoader()
+        .getResourceAsStream(testDir.toString());
+
+    if (s == null) {
+      throw new TestFrameworkException("Test directory not found: " + testDir);
+    }
+
+    try (BufferedReader reader = new BufferedReader(new InputStreamReader(s, UTF_8))) {
+      final List<Path> tests = new ArrayList<>();
+
+      String test;
+      while ((test = reader.readLine()) != null) {
+        if (test.endsWith(".json")) {
+          tests.add(testDir.resolve(test));
+        }
+      }
+      return tests;
+    } catch (final IOException e) {
+      throw new TestFrameworkException("Failed to read test dir: " + testDir, e);
+    }
+  }
+
+  private static List<String> getWhiteList() {
+    final String ksqlTestFiles = System.getProperty(KSQL_TEST_FILES, "").trim();
+    if (ksqlTestFiles.isEmpty()) {
+      return Collections.emptyList();
+    }
+
+    return Arrays.asList(ksqlTestFiles.split(","));
+  }
+
+  private static <TFT extends TestFile<T>, T extends Test> Stream<T> buildTests(
+      final Path testPath,
+      final Class<TFT> testFileType
+  ) {
+    try (InputStream stream = JsonTestLoader.class
+        .getClassLoader()
+        .getResourceAsStream(testPath.toString())
+    ) {
+      final TFT testFile = OBJECT_MAPPER.readValue(stream, testFileType);
+      return testFile.buildTests(testPath);
+    } catch (final Exception e) {
+      throw new RuntimeException("Unable to load test at path " + testPath, e);
+    }
+  }
+
+  private static void throwOnDuplicateNames(final List<? extends Test> testCases) {
+    final String duplicates = testCases.stream()
+        .collect(Collectors.groupingBy(Test::getName))
+        .entrySet()
+        .stream()
+        .filter(e -> e.getValue().size() > 1)
+        .map(e -> "test name: '" + e.getKey()
+            + "' found in files: " + e.getValue().stream().map(Test::getTestFile)
+            .collect(Collectors.joining(",")))
+        .collect(Collectors.joining(System.lineSeparator()));
+
+    if (!duplicates.isEmpty()) {
+      throw new IllegalStateException("There are tests with duplicate names: "
+          + System.lineSeparator() + duplicates);
+    }
+  }
+}

ksql-functional-tests/src/main/java/io/confluent/ksql/test/loader/TestFile.java

@@ -0,0 +1,25 @@
+/*
+ * 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.test.loader;
+
+import io.confluent.ksql.test.tools.Test;
+import java.nio.file.Path;
+import java.util.stream.Stream;
+
+public interface TestFile<TestTypeT extends Test> {
+
+  Stream<TestTypeT> buildTests(Path testPath);
+}

ksql-functional-tests/src/main/java/io/confluent/ksql/test/loader/TestLoader.java

@@ -0,0 +1,29 @@
+/*
+ * 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.test.loader;
+
+import io.confluent.ksql.test.tools.Test;
+import java.util.stream.Stream;
+
+/**
+ * Loader of tests.
+ *
+ * @param <T> the type of test.
+ */
+public interface TestLoader<T extends Test> {
+
+  Stream<T> load();
+}

ksql-functional-tests/src/main/java/io/confluent/ksql/test/model/ExpectedExceptionNode.java

@@ -24,8 +24,9 @@
 import io.confluent.ksql.util.KsqlExceptionMatcher;
 import io.confluent.ksql.util.KsqlStatementException;
 import java.util.Optional;
+import org.hamcrest.Matcher;
 
-final class ExpectedExceptionNode {
+public final class ExpectedExceptionNode {
 
   private final Optional<String> type;
   private final Optional<String> message;
@@ -42,7 +43,7 @@ final class ExpectedExceptionNode {
     }
   }
 
-  public KsqlExpectedException build(final String lastStatement) {
+  public Matcher<Throwable> build(final String lastStatement) {
     final KsqlExpectedException expectedException = KsqlExpectedException.none();
 
     type
@@ -58,7 +59,7 @@ public KsqlExpectedException build(final String lastStatement) {
         });
 
     message.ifPresent(expectedException::expectMessage);
-    return expectedException;
+    return expectedException.build();
   }
 
   @SuppressWarnings("unchecked")
@@ -73,5 +74,4 @@ private static Class<? extends Throwable> parseThrowable(final String className)
       throw new InvalidFieldException("expectedException.type", "Type was not found", e);
     }
   }
-
 }

ksql-functional-tests/src/main/java/io/confluent/ksql/test/model/RecordNode.java

@@ -36,7 +36,7 @@ public class RecordNode {
   private final String topicName;
   private final String key;
   private final JsonNode value;
-  private final long timestamp;
+  private final Optional<Long> timestamp;
   private final Optional<WindowData> window;
 
   RecordNode(
@@ -49,7 +49,7 @@ public class RecordNode {
     this.topicName = topicName == null ? "" : topicName;
     this.key = key == null ? "" : key;
     this.value = requireNonNull(value, "value");
-    this.timestamp = timestamp == null ? 0L : timestamp;
+    this.timestamp = Optional.ofNullable(timestamp);
     this.window = Optional.ofNullable(window);
 
     if (this.topicName.isEmpty()) {