How to unit test grpc-java server implementation functions?

Question

I have an implementation of GRPC-java server code, but I didn't find the example code to unit test the StreamObserver. Does anyone know the right way to unit test the function?

public class RpcTrackDataServiceImpl implements TrackDataServiceGrpc.TrackDataService {
    @Override
    public void getTracks(GetTracksRequest request, StreamObserver < GetTracksResponse > responseObserver) {
        GetTracksResponse reply = GetTracksResponse
            .newBuilder()
            .addTracks(TrackInfo.newBuilder()
                .setOwner("test")
                .setTrackName("test")
                .build())
            .build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }
}

Answer 1

Unit testing is very straight forward using the InProcess transport mentioned by Eric above. Here is an example a bit more explicit on code:

We test a service based on this protobuff definition:

syntax = "proto3";

option java_multiple_files = true;
option java_package = "servers.dummy";
option java_outer_classname = "DummyProto";
option objc_class_prefix = "DMYS";

package dummy;

import "general.proto";

// The dummy service definition.
service DummyService {
  // # Misc
  // Returns the server version
  rpc getVersion (Empty) returns (ServerVersion) {}
  // Returns the java version
  rpc getJava (Empty) returns (JavaVersion) {}
}


// Transmission data types

(The following file is included above:)

syntax = "proto3";

option java_multiple_files = true;
option java_package = "general";
option java_outer_classname = "General";
option objc_class_prefix = "G";

// Transmission data types

message Empty {} // Empty Request or Reply

message ServerVersion {
  string version = 1;
}

message JavaVersion {
  string version = 1;
}

The DummyService based on the generated Java from the Protoc compiler is the following:

package servers.dummy;

import java.util.logging.Logger;

import general.Empty;
import general.JavaVersion;
import general.ServerVersion;
import io.grpc.stub.StreamObserver;

public class DummyService extends DummyServiceGrpc.DummyServiceImplBase {
  private static final Logger logger = Logger.getLogger(DummyService.class.getName());

  @Override
  public void getVersion(Empty req, StreamObserver<ServerVersion> responseObserver) {
    logger.info("Server Version-Request received...");
    ServerVersion version = ServerVersion.newBuilder().setVersion("1.0.0").build();
    responseObserver.onNext(version);
    responseObserver.onCompleted();
  }

  @Override
  public void getJava(Empty req, StreamObserver<JavaVersion> responseObserver) {
    logger.info("Java Version Request received...");
    JavaVersion version = JavaVersion.newBuilder().setVersion(Runtime.class.getPackage().getImplementationVersion() + " (" + Runtime.class.getPackage().getImplementationVendor() + ")").build();
    responseObserver.onNext(version);
    responseObserver.onCompleted();
  }
}

Now we build an InProcessServer that runs our Dummy service (or any other service you want to test):

package servers;

import io.grpc.Server;
import io.grpc.inprocess.InProcessServerBuilder;

import java.io.IOException;
import java.util.logging.Logger;

import servers.util.PortServer;

/**
 * InProcessServer that manages startup/shutdown of a service within the same process as the client is running. Used for unit testing purposes.
 * @author be
 */
public class InProcessServer<T extends io.grpc.BindableService> {
  private static final Logger logger = Logger.getLogger(PortServer.class.getName());

  private Server server;
    
  private Class<T> clazz;
    
  public InProcessServer(Class<T> clazz){
    this.clazz = clazz;
  }

  public void start() throws IOException, InstantiationException, IllegalAccessException {
    server = InProcessServerBuilder
        .forName("test")
        .directExecutor()
        .addService(clazz.newInstance())
        .build()
        .start();
    logger.info("InProcessServer started.");
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        // Use stderr here since the logger may have been reset by its JVM shutdown hook.
        System.err.println("*** shutting down gRPC server since JVM is shutting down");
        InProcessServer.this.stop();
        System.err.println("*** server shut down");
      }
    });
  }

  void stop() {
    if (server != null) {
      server.shutdown();
    }
  }

  /**
   * Await termination on the main thread since the grpc library uses daemon threads.
   */
  public void blockUntilShutdown() throws InterruptedException {
    if (server != null) {
      server.awaitTermination();
    }
  }
}

We can now test the service using the following unit test:

package servers;

import static org.junit.Assert.*;
import general.ServerVersion;
import io.grpc.ManagedChannel;
import io.grpc.StatusRuntimeException;
import io.grpc.inprocess.InProcessChannelBuilder;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import servers.dummy.DummyService;
import servers.dummy.DummyServiceGrpc;
import servers.dummy.DummyServiceGrpc.DummyServiceBlockingStub;
import servers.dummy.DummyServiceGrpc.DummyServiceStub;

public class InProcessServerTest {
  private static final Logger logger = Logger.getLogger(InProcessServerTest.class.getName());

  private InProcessServer<DummyService> inprocessServer;
  private ManagedChannel channel;
  private DummyServiceBlockingStub blockingStub;
  private DummyServiceStub asyncStub;
    
  public InProcessServerTest() {
    super();
  }
    
  @Test
  public void testInProcessServer() throws InterruptedException{
    try {
      String version = getServerVersion();
      assertEquals("1.0.0", version);
    } finally {
      shutdown();
    }
  }

  /** Ask for the server version */
  public String getServerVersion() {
    logger.info("Will try to get server version...");
    ServerVersion response;
    try {
      response = blockingStub.getVersion(null);
    } catch (StatusRuntimeException e) {
      logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
      fail();
      return "";
    }
    return response.getVersion();
  }

  @Before
  public void beforeEachTest() throws InstantiationException, IllegalAccessException, IOException {
    inprocessServer = new InProcessServer<DummyService>(DummyService.class);
    inprocessServer.start();       
    channel = InProcessChannelBuilder
        .forName("test")
        .directExecutor()
        // Channels are secure by default (via SSL/TLS). For the example we disable TLS to avoid
        // needing certificates.
        .usePlaintext(true)
        .build();
    blockingStub = DummyServiceGrpc.newBlockingStub(channel);
    asyncStub = DummyServiceGrpc.newStub(channel);
  }

  @After
  public void afterEachTest(){
  channel.shutdownNow();
    inprocessServer.stop();
  }
    
  public void shutdown() throws InterruptedException {
    channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
  }
}

The test does only test one of two methods, as it is just for illustration purposes. The other method can be tested accordingly.

See the RouteGuideExample for more information on how to test both server and client: https://github.com/grpc/grpc-java/blob/master/examples/src/test/java/io/grpc/examples/routeguide/RouteGuideServerTest.java

Answer 2

First, refactor the code so it's easier to unit test:

public class RpcTrackDataServiceImpl implements TrackDataServiceGrpc.TrackDataService {
  @Override
  public void getTracks(GetTracksRequest request, StreamObserver<GetTracksResponse> responseObserver) {
    GetTracksResponse reply = getTracks(request);

    responseObserver.onNext(reply);
    responseObserver.onCompleted();
  }

  @VisibleForTesting
  GetTracksResponse getTracks(GetTracksRequest request) {
    return GetTracksResponse
        .newBuilder()
        .addTracks(TrackInfo.newBuilder()
            .setOwner("test")
            .setTrackName("test")
            .build())
        .build();
  }
}

Small tests can then be written for each (more easily if using Spring Boot):

public class UnitTest {
  private final ApplicationContextRunner applicationContextRunner = new ApplicationContextRunner();

  @Configuration
  public static class GetTracksConfiguration {
    @Bean
    public GetTracksService getTracksService() {
      return new GetTracksService();
    }
  }

  @Test
  public void replyShouldBeSent() {
    final GetTracksRequest request = GetTracksRequest.newBuilder().build();
    final StreamObserver<GetTracksResponse> response = mock(StreamObserver.class);

    applicationContextRunner
        .withUserConfiguration(RequestTracksConfiguration.class)
        .run(context -> {
          assertThat(context) 
              .hasSingleBean(RequestTracksService.class);
          context.getBean(RequestTracksService.class)
              .getTracks(request, response);

          verify(response, times(1)).onNext(any(GetTracksResponse.class));
          verify(response, times(1)).onCompleted();
          verify(response, never()).onError(any(Throwable.class));
        });
  }

  @Test
  public void shouldTestLogic {
    assertLogicInFactoredOutMethodIsCorrect();
  }

The larger test should then only test the startup and wiring:

@RunWith(SpringRunner.class)
@SpringBootTest(
    classes = {GetTracksService.class}
)
@EnableAutoConfiguration
public class SmokeTest {
  private GetTracksServiceGrpc.GetTracksServiceBlockingStub blockingStub;

  @Test
  public void springClientConnects() {
    final GetTracksRequest request = GetTracksRequest.newBuilder()
        .build();

    assertNotNull(blockingStub.getTracks(request));
  }
}

Note: The above code may not work OOTB since I've left out some annotations we use internally. The major point is there's no need to pay for the cost of bringing up a server for unit tests that are meant to test logic.

Answer 3

@RunWith(JUnit4.class)
public class HelloWorldServerTest {
  /**
   * This rule manages automatic graceful shutdown for the registered servers and channels at the
   * end of test.
   */
  @Rule
  public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();

  /**
   * To test the server, make calls with a real stub using the in-process channel, and verify
   * behaviors or state changes from the client side.
   */
  @Test
  public void greeterImpl_replyMessage() throws Exception {
    // Generate a unique in-process server name.
    String serverName = InProcessServerBuilder.generateName();

    // Create a server, add service, start, and register for automatic graceful shutdown.
    grpcCleanup.register(InProcessServerBuilder
        .forName(serverName).directExecutor().addService(new GreeterImpl()).build().start());

    GreeterGrpc.GreeterBlockingStub blockingStub = GreeterGrpc.newBlockingStub(
        // Create a client channel and register for automatic graceful shutdown.
        grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()));


    HelloReply reply =
        blockingStub.sayHello(HelloRequest.newBuilder().setName( "test name").build());

    assertEquals("Hello test name", reply.getMessage());
  }
}

https://github.com/grpc/grpc-java/blob/master/examples/src/test/java/io/grpc/examples/helloworld/HelloWorldServerTest.java

Answer 4

I will insert snippets from the official gRPC examples. I have successfully created tests based on these HelloWorld example.

The HelloWorldService:

/*
 * Copyright 2015, gRPC Authors All rights reserved.
 *
 * Licensed 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 io.grpc.examples.helloworld;

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;
import java.io.IOException;
import java.util.logging.Logger;

/**
 * Server that manages startup/shutdown of a {@code Greeter} server.
 */
public class HelloWorldServer {
  private static final Logger logger = Logger.getLogger(HelloWorldServer.class.getName());

  private Server server;

  private void start() throws IOException {
    /* The port on which the server should run */
    int port = 50051;
    server = ServerBuilder.forPort(port)
        .addService(new GreeterImpl())
        .build()
        .start();
    logger.info("Server started, listening on " + port);
    Runtime.getRuntime().addShutdownHook(new Thread() {
      @Override
      public void run() {
        // Use stderr here since the logger may have been reset by its JVM shutdown hook.
        System.err.println("*** shutting down gRPC server since JVM is shutting down");
        HelloWorldServer.this.stop();
        System.err.println("*** server shut down");
      }
    });
  }

  private void stop() {
    if (server != null) {
      server.shutdown();
    }
  }

  /**
   * Await termination on the main thread since the grpc library uses daemon threads.
   */
  private void blockUntilShutdown() throws InterruptedException {
    if (server != null) {
      server.awaitTermination();
    }
  }

  /**
   * Main launches the server from the command line.
   */
  public static void main(String[] args) throws IOException, InterruptedException {
    final HelloWorldServer server = new HelloWorldServer();
    server.start();
    server.blockUntilShutdown();
  }

  static class GreeterImpl extends GreeterGrpc.GreeterImplBase {

    @Override
    public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
      HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
      responseObserver.onNext(reply);
      responseObserver.onCompleted();
    }
  }
}

And the test:

/*
 * Copyright 2016, gRPC Authors All rights reserved.
 *
 * Licensed 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 io.grpc.examples.helloworld;

import static org.junit.Assert.assertEquals;

import io.grpc.examples.helloworld.HelloWorldServer.GreeterImpl;
import io.grpc.testing.GrpcServerRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/**
 * Unit tests for {@link HelloWorldServer}.
 * For demonstrating how to write gRPC unit test only.
 * Not intended to provide a high code coverage or to test every major usecase.
 *
 * <p>For more unit test examples see {@link io.grpc.examples.routeguide.RouteGuideClientTest} and
 * {@link io.grpc.examples.routeguide.RouteGuideServerTest}.
 */
@RunWith(JUnit4.class)
public class HelloWorldServerTest {
  /**
   * This creates and starts an in-process server, and creates a client with an in-process channel.
   * When the test is done, it also shuts down the in-process client and server.
   */
  @Rule
  public final GrpcServerRule grpcServerRule = new GrpcServerRule().directExecutor();

  /**
   * To test the server, make calls with a real stub using the in-process channel, and verify
   * behaviors or state changes from the client side.
   */
  @Test
  public void greeterImpl_replyMessage() throws Exception {
    // Add the service to the in-process server.
    grpcServerRule.getServiceRegistry().addService(new GreeterImpl());

    GreeterGrpc.GreeterBlockingStub blockingStub =
        GreeterGrpc.newBlockingStub(grpcServerRule.getChannel());
    String testName = "test name";

    HelloReply reply = blockingStub.sayHello(HelloRequest.newBuilder().setName(testName).build());

    assertEquals("Hello " + testName, reply.getMessage());
  }
}

You can fin other examples, if you clone the examples repository as they describe it here:

https://grpc.io/docs/tutorials/basic/java.html

I hope it will help you, too.

Br, Renato

Answer 5

I ended up with a solution to create a FakeStreamObserver that implements the StreamObserver interface.
The FakeStreamObserver is passed in to execute onNext, onCompleted etc.
I'm not sure if this is the best way or not.

Answer 6

I'd suggest using the InProcess transport. The InProcess transport is very lightweight but also is using much of the "real" code, so the behavior closely matches a real transport. If you also use directExecutor() for the Channel and Server then the test is essentially single-threaded and will be deterministic. (Although another thread would still be used for deadline handling.)

Although the question is for unit testing a service, InProcess is also great for unit testing a client.