如何对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

我建议使用InProcess传输。 InProcess传输非常轻量级，但也使用了很多“真实”代码，因此其行为与真实传输密切相关。如果你还为Channel和Server使用directExecutor()，那么测试本质上是单线程的，并且是确定性的。 （虽然另一个线程仍将用于最后期限处理）。

尽管问题是单元测试服务，但InProcess对单元测试客户端也很有用。

Answer 2

我最终创建了一个解决方案来创建一个实现StreamObserver接口的FakeStreamObserver。
FakeStreamObserver被传入以执行onNext，onCompleted等。
我不确定这是否是最好的方法。

Answer 3

使用上面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; 
} 

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

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（或任何你想要的其他服务进行测试）：

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(); 
     assertTrue(version == "1.0.0"); 
    } 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

Answer 4

我会插入来自官方GRPC例子片断。我已经成功创建了基于这些HelloWorld示例的测试。

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

而且测试：

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

您可以鳍其他例子，如果你克隆实例存储库，因为他们在这里形容它：

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

我希望它也能帮助你。

BR， 雷纳托