unit-testing - 如何对grpc-java服务器实现功能进行单元测试？

Question

我有一个 GRPC-java 服务器代码的实现，但我没有找到对 StreamObserver 进行单元测试的示例代码。有谁知道对函数进行单元测试的正确方法？

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

使用上面 Eric 提到的 InProcess 传输，单元测试非常简单。这是一个更明确的代码示例:
我们根据这个 protobuff 定义测试一个服务:

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

(以下文件包含在上面:)

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;
}

基于从 Protoc 编译器生成的 Java 的 DummyService 如下:

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();
  }
}

现在我们构建一个 InProcessServer 来运行我们的 Dummy 服务(或您想要测试的任何其他服务):

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();
    }
  }
}

我们现在可以使用以下单元测试来测试服务:

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);
  }
}

该测试仅测试两种方法中的一种，因为它仅用于说明目的。另一种方法可以相应地进行测试。
有关如何同时测试服务器和客户端的更多信息，请参阅 RouteGuideExample:
https://github.com/grpc/grpc-java/blob/master/examples/src/test/java/io/grpc/examples/routeguide/RouteGuideServerTest.java