test: migrate gRPC method invocation to testcontainers

sdk-tests/pom.xml

@@ -270,6 +270,8 @@
                     <include name="io/dapr/it/methodinvoke/http/MethodInvokeService*.class"/>
                     <include name="io/dapr/it/methodinvoke/http/MethodInvokeController*.class"/>
                     <include name="io/dapr/it/methodinvoke/http/Person*.class"/>
+                    <include name="io/dapr/it/methodinvoke/grpc/MethodInvokeIT*.class"/>
+                    <include name="io/dapr/it/methodinvoke/grpc/MethodInvokeService*.class"/>
                   </fileset>
                 </delete>
               </target>

sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeIT.java → sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/MethodInvokeIT.java

@@ -11,22 +11,29 @@
 limitations under the License.
 */
 
-package io.dapr.it.methodinvoke.grpc;
+package io.dapr.it.testcontainers.methodinvoke.grpc;
 
 import io.dapr.client.DaprClient;
 import io.dapr.client.resiliency.ResiliencyOptions;
-import io.dapr.it.AppRun;
 import io.dapr.it.MethodInvokeServiceGrpc;
-import io.dapr.it.containers.BaseContainerIT;
-import io.dapr.testcontainers.DaprContainer;
+import io.dapr.it.testcontainers.methodinvoke.grpc.app.MethodInvokeService;
+import io.dapr.it.testcontainers.support.DaprTestEnvironment;
 import io.dapr.testcontainers.DaprProtocol;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
-import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.time.Duration;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import static io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequest;
 import static io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest;
@@ -36,37 +43,43 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-public class MethodInvokeIT extends BaseContainerIT {
+@Testcontainers
+@Tag("testcontainers")
+@Execution(ExecutionMode.SAME_THREAD)
+public class MethodInvokeIT {
 
     private static final String APP_NAME = "methodinvoke-grpc-it";
     private static final int NUM_MESSAGES = 10;
     private static final int TIMEOUT_MS = 100;
     private static final ResiliencyOptions RESILIENCY_OPTIONS = new ResiliencyOptions()
         .setTimeout(Duration.ofMillis(TIMEOUT_MS));
 
-    private static DaprContainer dapr;
-    private static AppRun app;
-
-    @BeforeAll
-    public static void init() throws Exception {
-        var pair = startAppAndAttach(
-            APP_NAME,
-            MethodInvokeService.class,
-            AppRun.AppProtocol.GRPC,
-            appPort -> {
-                DaprContainer d = daprBuilder(APP_NAME)
-                    .withAppPort(appPort)
-                    .withAppChannelAddress("host.testcontainers.internal")
-                    .withAppProtocol(DaprProtocol.GRPC);
-                return d;
-            });
-        dapr = pair.dapr();
-        app = pair.app();
+    @Container
+    private static final DaprTestEnvironment ENVIRONMENT =
+        new DaprTestEnvironment(APP_NAME, MethodInvokeService.class, DaprProtocol.GRPC);
+
+    @BeforeEach
+    public void reset() throws InterruptedException {
+        ManagedChannel channel = ManagedChannelBuilder.forTarget(ENVIRONMENT.appAddress()).usePlaintext().build();
+
+        try {
+            var stub = MethodInvokeServiceGrpc.newBlockingStub(channel).withDeadlineAfter(10, TimeUnit.SECONDS);
+            var request = GetMessagesRequest.getDefaultInstance();
+
+            for (int id : stub.getMessages(request).getMessagesMap().keySet()) {
+                stub.deleteMessage(DeleteMessageRequest.newBuilder().setId(id).build());
+            }
+
+            assertTrue(stub.getMessages(request).getMessagesMap().isEmpty());
+        } finally {
+            channel.shutdownNow();
+            assertTrue(channel.awaitTermination(5, TimeUnit.SECONDS), "reset channel did not terminate");
+        }
     }
 
     @Test
     public void testInvoke() throws Exception {
-        try (DaprClient client = newDaprClient(dapr)) {
+        try (DaprClient client = ENVIRONMENT.clientBuilder().build()) {
             client.waitForSidecar(10000).block();
 
             MethodInvokeServiceGrpc.MethodInvokeServiceBlockingStub stub = createGrpcStub(client);
@@ -97,7 +110,7 @@ public void testInvoke() throws Exception {
 
     @Test
     public void testInvokeTimeout() throws Exception {
-        try (DaprClient client = newDaprClientBuilder(dapr).withResiliencyOptions(RESILIENCY_OPTIONS).build()) {
+        try (DaprClient client = ENVIRONMENT.clientBuilder().withResiliencyOptions(RESILIENCY_OPTIONS).build()) {
             client.waitForSidecar(10000).block();
 
             MethodInvokeServiceGrpc.MethodInvokeServiceBlockingStub stub = createGrpcStub(client);
@@ -114,7 +127,7 @@ public void testInvokeTimeout() throws Exception {
 
     @Test
     public void testInvokeException() throws Exception {
-        try (DaprClient client = newDaprClient(dapr)) {
+        try (DaprClient client = ENVIRONMENT.clientBuilder().build()) {
             client.waitForSidecar(10000).block();
 
             MethodInvokeServiceGrpc.MethodInvokeServiceBlockingStub stub = createGrpcStub(client);

sdk-tests/src/test/java/io/dapr/it/methodinvoke/grpc/MethodInvokeService.java → sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/grpc/app/MethodInvokeService.java

@@ -11,7 +11,7 @@
 limitations under the License.
 */
 
-package io.dapr.it.methodinvoke.grpc;
+package io.dapr.it.testcontainers.methodinvoke.grpc.app;
 
 import io.dapr.grpc.GrpcHealthCheckService;
 import io.dapr.it.MethodInvokeServiceGrpc;
@@ -83,7 +83,7 @@ private void awaitTermination() throws InterruptedException {
      * {@inheritDoc}
      */
     public void postMessage(io.dapr.it.MethodInvokeServiceProtos.PostMessageRequest request,
-                            io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.PostMessageResponse> responseObserver) {
+        io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.PostMessageResponse> responseObserver) {
       this.messages.put(request.getId(), request.getMessage());
 
       io.dapr.it.MethodInvokeServiceProtos.PostMessageResponse.Builder responseBuilder =
@@ -96,7 +96,7 @@ public void postMessage(io.dapr.it.MethodInvokeServiceProtos.PostMessageRequest
      * {@inheritDoc}
      */
     public void deleteMessage(io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequest request,
-                              io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.DeleteMessageResponse> responseObserver) {
+        io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.DeleteMessageResponse> responseObserver) {
       this.messages.remove(request.getId());
 
       io.dapr.it.MethodInvokeServiceProtos.DeleteMessageResponse.Builder responseBuilder =
@@ -109,7 +109,7 @@ public void deleteMessage(io.dapr.it.MethodInvokeServiceProtos.DeleteMessageRequ
      * {@inheritDoc}
      */
     public void getMessages(io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest request,
-                            io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.GetMessagesResponse> responseObserver) {
+        io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.GetMessagesResponse> responseObserver) {
       GetMessagesResponse res = GetMessagesResponse.newBuilder().putAllMessages(this.messages).build();
 
       io.dapr.it.MethodInvokeServiceProtos.GetMessagesResponse.Builder responseBuilder
@@ -122,7 +122,7 @@ public void getMessages(io.dapr.it.MethodInvokeServiceProtos.GetMessagesRequest
      * {@inheritDoc}
      */
     public void sleep(io.dapr.it.MethodInvokeServiceProtos.SleepRequest request,
-                      io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.SleepResponse> responseObserver) {
+        io.grpc.stub.StreamObserver<io.dapr.it.MethodInvokeServiceProtos.SleepResponse> responseObserver) {
       SleepResponse res = this.sleep(request);
 
       io.dapr.it.MethodInvokeServiceProtos.SleepResponse.Builder responseBuilder =

sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/MethodInvokeIT.java

@@ -65,6 +65,7 @@ public void reset() throws Exception {
     public void testInvoke() throws Exception {
         try (DaprClient client = ENVIRONMENT.clientBuilder().build()) {
             client.waitForSidecar(10000).block();
+
             for (int i = 0; i < NUM_MESSAGES; i++) {
                 String message = String.format("This is message #%d", i);
                 client.invokeMethod(APP_NAME, "messages", message.getBytes(), HttpExtension.POST).block();
@@ -90,6 +91,7 @@ public void testInvoke() throws Exception {
     public void testInvokeWithObjects() throws Exception {
         try (DaprClient client = ENVIRONMENT.clientBuilder().build()) {
             client.waitForSidecar(10000).block();
+
             for (int i = 0; i < NUM_MESSAGES; i++) {
                 Person person = new Person();
                 person.setName(String.format("Name %d", i));

sdk-tests/src/test/java/io/dapr/it/testcontainers/methodinvoke/http/app/MethodInvokeController.java

@@ -80,6 +80,7 @@ public void postPerson(@RequestBody Person person){
     @PutMapping(path = "/persons/{personId}")
     public void putPerson(@PathVariable("personId") Integer personId, @RequestBody Person person){
         final Optional<Person> auxPerson = persons.stream().filter(person1 -> person1.getId() == personId).findFirst();
+
         if(auxPerson.isPresent()){
             auxPerson.get().setName(person.getName());
             auxPerson.get().setLastName(person.getLastName());
@@ -90,6 +91,7 @@ public void putPerson(@PathVariable("personId") Integer personId, @RequestBody P
     @DeleteMapping(path = "/persons/{personId}")
     public void deletePerson(@PathVariable("personId") Integer personId){
         final Optional<Person> auxPerson = persons.stream().filter(person1 -> person1.getId() == personId).findFirst();
+
         if(auxPerson.isPresent()) {
             persons.remove(auxPerson.get());
         }
@@ -105,6 +107,7 @@ public void sleep(@RequestBody int seconds) throws InterruptedException {
         if (seconds < 0) {
             throw new IllegalArgumentException("Sleep time cannot be negative.");
         }
+
         Thread.sleep(seconds * 1000);
     }
 

sdk-tests/src/test/java/io/dapr/it/testcontainers/support/DaprTestEnvironment.java

@@ -17,11 +17,13 @@
 import io.dapr.testcontainers.DaprContainer;
 import io.dapr.testcontainers.DaprContainerConstants;
 import io.dapr.testcontainers.DaprPlacementContainer;
+import io.dapr.testcontainers.DaprProtocol;
 import io.dapr.testcontainers.DaprSchedulerContainer;
 import org.testcontainers.containers.BindMode;
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitStrategy;
 import org.testcontainers.lifecycle.Startable;
 import reactor.util.retry.Retry;
 
@@ -43,6 +45,7 @@ public final class DaprTestEnvironment implements Startable {
   private static final String APP_RUNTIME = "/it-runtime";
   private final String identity;
   private final String mainClass;
+  private final DaprProtocol protocol;
   private final Network network;
   private final OwnedResources resources;
   private final DaprPlacementContainer placement;
@@ -53,12 +56,31 @@ public final class DaprTestEnvironment implements Startable {
 
   /** Defines an isolated topology; no container is started during construction. */
   public DaprTestEnvironment(String appId, Class<?> mainClass) {
-    this(appId, mainClass.getName(), Duration.ofMinutes(2));
+    this(appId, mainClass, DaprProtocol.HTTP);
+  }
+
+  /**
+   * Defines an isolated topology with the specified application callback protocol.
+   * @param appId Dapr application ID.
+   * @param mainClass Application entry point.
+   * @param protocol Application callback protocol.
+   */
+  public DaprTestEnvironment(String appId, Class<?> mainClass, DaprProtocol protocol) {
+    this(appId, mainClass.getName(), protocol, Duration.ofMinutes(2));
   }
 
   DaprTestEnvironment(String appId, String mainClass, Duration appReadinessTimeout) {
+    this(appId, mainClass, DaprProtocol.HTTP, appReadinessTimeout);
+  }
+
+  DaprTestEnvironment(String appId, String mainClass, DaprProtocol protocol, Duration appReadinessTimeout) {
+    if (protocol != DaprProtocol.HTTP && protocol != DaprProtocol.GRPC) {
+      throw new IllegalArgumentException("Unsupported application protocol: " + protocol);
+    }
+
     identity = appId + "-" + UUID.randomUUID();
     this.mainClass = mainClass;
+    this.protocol = protocol;
     Map<String, String> labels = Map.of("io.dapr.it.environment", identity);
     network = Network.builder().createNetworkCmdModifier(cmd -> cmd.withLabels(labels)).build();
     resources = new OwnedResources(network);
@@ -68,11 +90,13 @@ public DaprTestEnvironment(String appId, Class<?> mainClass) {
     scheduler = new DaprSchedulerContainer(DaprContainerConstants.DAPR_SCHEDULER_IMAGE_TAG)
         .withNetwork(network).withNetworkAliases("scheduler").withLabels(labels)
         .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2)));
+    WaitStrategy appWait = protocol == DaprProtocol.GRPC
+        ? new GrpcCallbackWaitStrategy(APP_PORT).withStartupTimeout(appReadinessTimeout)
+        : Wait.forHttp("/health").forPort(APP_PORT).forStatusCode(200).withStartupTimeout(appReadinessTimeout);
     app = new GenericContainer<>(JDK_17_TEMURIN_JAMMY)
         .withNetwork(network).withNetworkAliases("app").withLabels(labels)
         .withExposedPorts(APP_PORT)
-        .waitingFor(Wait.forHttp("/health").forPort(APP_PORT).forStatusCode(200)
-            .withStartupTimeout(appReadinessTimeout))
+        .waitingFor(appWait)
         .withEnv("DAPR_HTTP_ENDPOINT", Endpoints.http("dapr", 3500).toString())
         .withEnv("DAPR_GRPC_ENDPOINT", "dapr:50001")
         .withEnv("DAPR_SIDECAR_IP", "dapr")
@@ -81,7 +105,7 @@ public DaprTestEnvironment(String appId, Class<?> mainClass) {
     dapr = new DaprContainer(DaprContainerConstants.DAPR_RUNTIME_IMAGE_TAG)
         .withNetwork(network).withNetworkAliases("dapr").withLabels(labels)
         .withAccessToHost(false)
-        .withAppName(appId).withAppPort(APP_PORT).withAppChannelAddress("app")
+        .withAppName(appId).withAppPort(APP_PORT).withAppChannelAddress("app").withAppProtocol(protocol)
         .withPlacementContainer(placement).withSchedulerContainer(scheduler)
         .waitingFor(Wait.forHttp("/v1.0/healthz/outbound").forPort(3500)
             .forStatusCodeMatching(status -> status >= 200 && status <= 399)
@@ -95,9 +119,11 @@ public DaprTestEnvironment(String appId, Class<?> mainClass) {
   private void attachLogs(GenericContainer<?> container, String alias) {
     container.withLogConsumer(frame -> {
       String text = frame.getUtf8String();
+
       if ("app".equals(alias)) {
         appOutput.append(text);
       }
+
       if (!text.isEmpty()) {
         System.out.print("[" + identity + " " + alias + "] " + text);
       }
@@ -110,13 +136,15 @@ private void configureApplication() {
     List<String> classpath = new ArrayList<>();
     classpath.add(APP_RUNTIME + "/test-classes");
     classpath.add(APP_RUNTIME + "/classes");
+
     try (var files = Files.walk(lib)) {
       classpath.addAll(files.filter(path -> path.toString().endsWith(".jar"))
           .sorted().map(path -> APP_RUNTIME + "/lib/" + lib.relativize(path).toString().replace('\\', '/'))
           .toList());
     } catch (IOException e) {
       throw new IllegalStateException("Cannot read staged IT dependencies", e);
     }
+
     app.withFileSystemBind(runtime.toString(), APP_RUNTIME, BindMode.READ_ONLY);
     app.withCommand("java", "-cp", String.join(":", classpath), mainClass, Integer.toString(APP_PORT));
   }
@@ -130,25 +158,29 @@ public void start() {
       System.out.printf("[%s] scheduler id=%s%n", identity, scheduler.getContainerId());
       configureApplication();
       resources.start(app);
-      System.out.printf("[%s] app id=%s endpoint=%s%n", identity, app.getContainerId(), appEndpoint());
+      System.out.printf("[%s] app id=%s image=%s protocol=%s address=%s%n", identity, app.getContainerId(),
+          JDK_17_TEMURIN_JAMMY, protocol, appAddress());
       resources.start(dapr);
       System.out.printf("[%s] dapr id=%s endpoint=%s grpc=%s:%d%n", identity, dapr.getContainerId(),
           Endpoints.http(dapr.getHost(), dapr.getHttpPort()), dapr.getHost(), dapr.getGrpcPort());
+
       try (DaprClient client = clientBuilder().build()) {
         client.waitForSidecar(30_000).block(Duration.ofSeconds(35));
         client.getMetadata().retryWhen(Retry.fixedDelay(30, Duration.ofSeconds(1))).block(Duration.ofSeconds(35));
       } catch (Exception e) {
         throw new IllegalStateException("Sidecar client readiness failed", e);
       }
-      System.out.printf("[%s] app=%s (%s), dapr=%s (%s), placement=%s, scheduler=%s%n", identity,
-          appEndpoint(), app.getContainerId(), Endpoints.http(dapr.getHost(), dapr.getHttpPort()),
+
+      System.out.printf("[%s] app=%s (%s, %s), dapr=%s (%s), placement=%s, scheduler=%s%n", identity,
+          appAddress(), protocol, app.getContainerId(), Endpoints.http(dapr.getHost(), dapr.getHttpPort()),
           dapr.getContainerId(), placement.getContainerId(), scheduler.getContainerId());
     } catch (RuntimeException | Error failure) {
       try {
         resources.stop();
       } catch (RuntimeException | Error cleanup) {
         failure.addSuppressed(cleanup);
       }
+
       throw failure;
     }
   }
@@ -158,11 +190,23 @@ public void stop() {
     resources.stop();
   }
 
-  /** Returns the app address reachable by the test JVM. */
+  /** Returns the HTTP application endpoint reachable by the test JVM. */
   public URI appEndpoint() {
+    if (protocol != DaprProtocol.HTTP) {
+      throw new IllegalStateException("Application is not HTTP: " + protocol);
+    }
+
     return Endpoints.http(app.getHost(), app.getMappedPort(APP_PORT));
   }
 
+  /**
+   * Returns the mapped host and port as an IPv6-safe authority.
+   * @return Application authority reachable by the test JVM.
+   */
+  public String appAddress() {
+    return Endpoints.http(app.getHost(), app.getMappedPort(APP_PORT)).getRawAuthority();
+  }
+
   /** Returns a client builder configured for the mapped sidecar ports. */
   public DaprClientBuilder clientBuilder() {
     return Endpoints.client(dapr.getHost(), dapr.getHttpPort(), dapr.getGrpcPort());
@@ -177,6 +221,10 @@ String appOutput() {
     return appOutput.toString();
   }
 
+  DaprProtocol appProtocol() {
+    return dapr.getAppProtocol();
+  }
+
   String appContainerId() {
     return app.getContainerId();
   }