Introduce sdk-test module to provide testing utilities.

This module provides an easy to use opinionated test toolkit to get started with testing your service with a Restate container. The container is automatically configured and deployed together with the services on the local JVM, respecting the lifecycle of JUnit 5 tests. We also provide easy to use parameter injection to send requests to services. See CounterTest for an example.

Added new allowed licenses due to the testcontainers transitive dependencies.

config/allowed-licenses.json

@@ -8,6 +8,15 @@
     },
     {
       "moduleLicense": "The 3-Clause BSD License"
+    },
+    {
+      "moduleLicense": "The 2-Clause BSD License"
+    },
+    {
+      "moduleLicense": "Eclipse Public License - v 2.0"
+    },
+    {
+      "moduleLicense": "Eclipse Public License - v 1.0"
     }
   ]
 }

sdk-test/build.gradle.kts

@@ -0,0 +1,48 @@
+import com.google.protobuf.gradle.id
+
+plugins {
+  `java-library`
+  `maven-publish`
+}
+
+dependencies {
+  api(project(":sdk-core"))
+  api(testingLibs.junit.api)
+  api(testingLibs.testcontainers.core)
+
+  implementation(project(":admin-client"))
+  implementation(project(":sdk-http-vertx"))
+  implementation(coreLibs.log4j.api)
+  implementation(platform(vertxLibs.vertx.bom))
+  implementation(vertxLibs.vertx.core)
+  implementation(coreLibs.grpc.netty.shaded)
+
+  testCompileOnly(coreLibs.javax.annotation.api)
+  testImplementation(project(":sdk-java-blocking"))
+  testImplementation(testingLibs.assertj)
+  testImplementation(testingLibs.junit.jupiter)
+  testImplementation(coreLibs.grpc.stub)
+  testImplementation(coreLibs.grpc.protobuf)
+  testImplementation(coreLibs.log4j.core)
+}
+
+publishing {
+  publications {
+    register<MavenPublication>("maven") {
+      groupId = "dev.restate.sdk"
+      artifactId = "sdk-test"
+
+      from(components["java"])
+    }
+  }
+}
+
+// Protobuf codegen for tests
+
+protobuf {
+  plugins {
+    id("grpc") { artifact = "io.grpc:protoc-gen-grpc-java:${coreLibs.versions.grpc.get()}" }
+  }
+
+  generateProtoTasks { ofSourceSet("test").forEach { it.plugins { id("grpc") } } }
+}

sdk-test/src/main/java/dev/restate/sdk/testing/BaseRestateRunner.java

@@ -0,0 +1,97 @@
+package dev.restate.sdk.testing;
+
+import dev.restate.admin.client.ApiClient;
+import io.grpc.ManagedChannel;
+import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
+import java.net.URL;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.extension.*;
+import org.junit.jupiter.api.extension.ExtensionContext.Namespace;
+import org.junit.jupiter.api.extension.ExtensionContext.Store;
+
+abstract class BaseRestateRunner implements ParameterResolver {
+
+  static final Namespace NAMESPACE = Namespace.create(BaseRestateRunner.class);
+  private static final String MANAGED_CHANNEL_KEY = "ManagedChannelKey";
+  static final String DEPLOYER_KEY = "Deployer";
+
+  @Override
+  public boolean supportsParameter(
+      ParameterContext parameterContext, ExtensionContext extensionContext)
+      throws ParameterResolutionException {
+    return (parameterContext.isAnnotated(RestateAdminClient.class)
+            && ApiClient.class.isAssignableFrom(parameterContext.getParameter().getType()))
+        || (parameterContext.isAnnotated(RestateGrpcChannel.class)
+            && ManagedChannel.class.isAssignableFrom(parameterContext.getParameter().getType()))
+        || (parameterContext.isAnnotated(RestateURL.class)
+                && String.class.isAssignableFrom(parameterContext.getParameter().getType())
+            || URL.class.isAssignableFrom(parameterContext.getParameter().getType()));
+  }
+
+  @Override
+  public Object resolveParameter(
+      ParameterContext parameterContext, ExtensionContext extensionContext)
+      throws ParameterResolutionException {
+    if (parameterContext.isAnnotated(RestateAdminClient.class)) {
+      return getDeployer(extensionContext).getAdminClient();
+    } else if (parameterContext.isAnnotated(RestateGrpcChannel.class)) {
+      return resolveChannel(extensionContext);
+    } else if (parameterContext.isAnnotated(RestateURL.class)) {
+      URL url = getDeployer(extensionContext).getIngressUrl();
+      if (parameterContext.getParameter().getType().equals(String.class)) {
+        return url.toString();
+      }
+      return url;
+    }
+    throw new ParameterResolutionException("The parameter is not supported");
+  }
+
+  private ManagedChannel resolveChannel(ExtensionContext extensionContext) {
+    return extensionContext
+        .getStore(NAMESPACE)
+        .getOrComputeIfAbsent(
+            MANAGED_CHANNEL_KEY,
+            k -> {
+              URL url = getDeployer(extensionContext).getIngressUrl();
+
+              ManagedChannel channel =
+                  NettyChannelBuilder.forAddress(url.getHost(), url.getPort())
+                      .disableServiceConfigLookUp()
+                      .usePlaintext()
+                      .build();
+
+              return new ManagedChannelResource(channel);
+            },
+            ManagedChannelResource.class)
+        .channel;
+  }
+
+  private ManualRestateRunner getDeployer(ExtensionContext extensionContext) {
+    return (ManualRestateRunner) extensionContext.getStore(NAMESPACE).get(DEPLOYER_KEY);
+  }
+
+  // AutoCloseable wrapper around ManagedChannelResource
+  private static class ManagedChannelResource implements Store.CloseableResource {
+
+    private final ManagedChannel channel;
+
+    private ManagedChannelResource(ManagedChannel channel) {
+      this.channel = channel;
+    }
+
+    @Override
+    public void close() throws Exception {
+      // Shutdown channel
+      channel.shutdown();
+      if (channel.awaitTermination(5, TimeUnit.SECONDS)) {
+        return;
+      }
+
+      // Force shutdown now
+      channel.shutdownNow();
+      if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
+        throw new IllegalStateException("Cannot terminate ManagedChannel on time");
+      }
+    }
+  }
+}

sdk-test/src/main/java/dev/restate/sdk/testing/ManualRestateRunner.java

@@ -0,0 +1,180 @@
+package dev.restate.sdk.testing;
+
+import dev.restate.admin.api.ServiceEndpointApi;
+import dev.restate.admin.client.ApiClient;
+import dev.restate.admin.client.ApiException;
+import dev.restate.admin.model.RegisterServiceEndpointRequest;
+import dev.restate.admin.model.RegisterServiceEndpointResponse;
+import dev.restate.admin.model.RegisterServiceResponse;
+import io.vertx.core.http.HttpServer;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.testcontainers.Testcontainers;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.utility.DockerImageName;
+
+/** Manual runner for Restate. We recommend using {@link RestateRunner} with JUnit 5. */
+public class ManualRestateRunner
+    implements AutoCloseable, ExtensionContext.Store.CloseableResource {
+
+  private static final Logger LOG = LogManager.getLogger(ManualRestateRunner.class);
+
+  private static final String RESTATE_RUNTIME = "runtime";
+  public static final int RESTATE_INGRESS_ENDPOINT_PORT = 8080;
+  public static final int RESTATE_ADMIN_ENDPOINT_PORT = 9070;
+
+  private final HttpServer server;
+  private final GenericContainer<?> runtimeContainer;
+
+  ManualRestateRunner(
+      HttpServer server,
+      String runtimeContainerImage,
+      Map<String, String> additionalEnv,
+      String configFile) {
+    this.server = server;
+    this.runtimeContainer = new GenericContainer<>(DockerImageName.parse(runtimeContainerImage));
+
+    // Configure runtimeContainer
+    this.runtimeContainer
+        // We expose these ports only to enable port checks
+        .withExposedPorts(RESTATE_INGRESS_ENDPOINT_PORT, RESTATE_ADMIN_ENDPOINT_PORT)
+        .withEnv(additionalEnv)
+        // These envs should not be overriden by additionalEnv
+        .withEnv("RESTATE_META__REST_ADDRESS", "0.0.0.0:" + RESTATE_ADMIN_ENDPOINT_PORT)
+        .withEnv(
+            "RESTATE_WORKER__INGRESS_GRPC__BIND_ADDRESS",
+            "0.0.0.0:" + RESTATE_INGRESS_ENDPOINT_PORT)
+        .withNetworkAliases(RESTATE_RUNTIME)
+        // Configure wait strategy on health paths
+        .setWaitStrategy(
+            new WaitAllStrategy()
+                .withStrategy(Wait.forHttp("/health").forPort(RESTATE_ADMIN_ENDPOINT_PORT))
+                .withStrategy(
+                    Wait.forHttp("/grpc.health.v1.Health/Check")
+                        .forPort(RESTATE_INGRESS_ENDPOINT_PORT)));
+
+    if (configFile != null) {
+      this.runtimeContainer.withCopyToContainer(Transferable.of(configFile), "/config.yaml");
+      this.runtimeContainer.withEnv("RESTATE_CONFIG", "/config.yaml");
+    }
+  }
+
+  /** Run restate, run the embedded service endpoint server, and register the services. */
+  public void run() {
+    // Start listening the local server
+    try {
+      server.listen(0).toCompletionStage().toCompletableFuture().get();
+    } catch (InterruptedException | ExecutionException e) {
+      throw new RuntimeException(e);
+    }
+
+    // Expose the server port
+    int serviceEndpointPort = server.actualPort();
+    LOG.debug("Started embedded service endpoint server on port {}", serviceEndpointPort);
+    Testcontainers.exposeHostPorts(serviceEndpointPort);
+
+    // Now create the runtime container and deploy it
+    this.runtimeContainer.start();
+    LOG.debug("Started Restate container");
+
+    // Register services now
+    ApiClient client = getAdminClient();
+    try {
+      RegisterServiceEndpointResponse response =
+          new ServiceEndpointApi(client)
+              .createServiceEndpoint(
+                  new RegisterServiceEndpointRequest()
+                      .uri("http://host.testcontainers.internal:" + serviceEndpointPort)
+                      .force(true));
+      LOG.debug(
+          "Registered services {}",
+          response.getServices().stream()
+              .map(RegisterServiceResponse::getName)
+              .collect(Collectors.toList()));
+    } catch (ApiException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Get restate ingress url to send HTTP/gRPC requests to services.
+   *
+   * @throws IllegalStateException if the restate container is not running.
+   */
+  public URL getRestateUrl() {
+    try {
+      return new URL(
+          "http",
+          runtimeContainer.getHost(),
+          runtimeContainer.getMappedPort(RESTATE_INGRESS_ENDPOINT_PORT),
+          "/");
+    } catch (MalformedURLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Get restate admin url to send HTTP requests to the admin API.
+   *
+   * @throws IllegalStateException if the restate container is not running.
+   */
+  public URL getAdminUrl() {
+    try {
+      return new URL(
+          "http",
+          runtimeContainer.getHost(),
+          runtimeContainer.getMappedPort(RESTATE_ADMIN_ENDPOINT_PORT),
+          "/");
+    } catch (MalformedURLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /** Get the restate container. */
+  public GenericContainer<?> getRestateContainer() {
+    return this.runtimeContainer;
+  }
+
+  /** Stop restate and the embedded service endpoint server. */
+  public void stop() {
+    this.close();
+  }
+
+  /** Like {@link #stop()}. */
+  @Override
+  public void close() {
+    runtimeContainer.stop();
+    LOG.debug("Stopped Restate container");
+    server.close().toCompletionStage().toCompletableFuture();
+    LOG.debug("Stopped Embedded Service endpoint server");
+  }
+
+  // -- Methods used by the JUnit5 extension
+
+  ApiClient getAdminClient() {
+    return new ApiClient()
+        .setHost(runtimeContainer.getHost())
+        .setPort(runtimeContainer.getMappedPort(RESTATE_ADMIN_ENDPOINT_PORT));
+  }
+
+  URL getIngressUrl() {
+    try {
+      return new URL(
+          "http",
+          runtimeContainer.getHost(),
+          runtimeContainer.getMappedPort(RESTATE_INGRESS_ENDPOINT_PORT),
+          "/");
+    } catch (MalformedURLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

sdk-test/src/main/java/dev/restate/sdk/testing/RestateAdminClient.java

@@ -0,0 +1,14 @@
+package dev.restate.sdk.testing;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Inject the Restate {@link dev.restate.admin.client.ApiClient}, useful to build admin clients,
+ * such as {@link dev.restate.admin.api.ServiceEndpointApi}.
+ */
+@Target(value = ElementType.PARAMETER)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface RestateAdminClient {}

sdk-test/src/main/java/dev/restate/sdk/testing/RestateGrpcChannel.java

@@ -0,0 +1,11 @@
+package dev.restate.sdk.testing;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/** Inject a {@link io.grpc.Channel} to interact with the deployed runtime. */
+@Target(value = ElementType.PARAMETER)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface RestateGrpcChannel {}