Add first pass at interop tests

Cargo.toml

@@ -4,5 +4,6 @@ members = [
     "tonic-macros",
     "tonic-build",
     "tonic-examples",
+    "tonic-interop",
     "tower-h2"
 ]

tonic-interop/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "tonic-interop"
+version = "0.1.0"
+authors = ["Lucio Franco <[email protected]>"]
+edition = "2018"
+
+[[bin]]
+name = "client"
+path = "src/bin/client.rs"
+
+[[bin]]
+name = "server"
+path = "src/bin/server.rs"
+
+[dependencies]
+tokio = "=0.2.0-alpha.1"
+tonic = { path = "../tonic" }
+prost = "0.5"
+prost-derive = "0.5"
+bytes = "0.4"
+tower-h2 = { path = "../tower-h2" }
+http = "0.1"
+
+console = "0.7"
+clap = "2.0"
+
+[build-dependencies]
+tonic-build = { path = "../tonic-build" }

tonic-interop/build.rs

@@ -0,0 +1,11 @@
+fn main() {
+    let files = &["proto/grpc/testing/test.proto"];
+    let dirs = &["proto/grpc/testing"];
+
+    tonic_build::compile_protos(files, dirs).unwrap();
+
+    // prevent needing to rebuild if files (or deps) haven't changed
+    for file in files {
+        println!("cargo:rerun-if-changed={}", file);
+    }
+}

tonic-interop/proto/grpc/testing/empty.proto

@@ -0,0 +1,27 @@
+// Copyright 2015 gRPC authors.
+//
+// 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.
+
+syntax = "proto3";
+
+package grpc.testing;
+
+// An empty message that you can re-use to avoid defining duplicated empty
+// messages in your project. A typical example is to use it as argument or the
+// return value of a service API. For instance:
+//
+//   service Foo {
+//     rpc Bar (grpc.testing.Empty) returns (grpc.testing.Empty) { };
+//   };
+//
+message Empty {}

tonic-interop/proto/grpc/testing/messages.proto

@@ -0,0 +1,168 @@
+// Copyright 2015-2016 gRPC authors.
+//
+// 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.
+
+// Message definitions to be used by integration test service definitions.
+
+syntax = "proto3";
+
+package grpc.testing;
+
+// TODO(dgq): Go back to using well-known types once
+// https://github.com/grpc/grpc/issues/6980 has been fixed.
+// import "google/protobuf/wrappers.proto";
+message BoolValue {
+  // The bool value.
+  bool value = 1;
+}
+
+// DEPRECATED, don't use. To be removed shortly.
+// The type of payload that should be returned.
+enum PayloadType {
+  // Compressable text format.
+  COMPRESSABLE = 0;
+}
+
+// A block of data, to simply increase gRPC message size.
+message Payload {
+  // DEPRECATED, don't use. To be removed shortly.
+  // The type of data in body.
+  PayloadType type = 1;
+  // Primary contents of payload.
+  bytes body = 2;
+}
+
+// A protobuf representation for grpc status. This is used by test
+// clients to specify a status that the server should attempt to return.
+message EchoStatus {
+  int32 code = 1;
+  string message = 2;
+}
+
+// Unary request.
+message SimpleRequest {
+  // DEPRECATED, don't use. To be removed shortly.
+  // Desired payload type in the response from the server.
+  // If response_type is RANDOM, server randomly chooses one from other formats.
+  PayloadType response_type = 1;
+
+  // Desired payload size in the response from the server.
+  int32 response_size = 2;
+
+  // Optional input payload sent along with the request.
+  Payload payload = 3;
+
+  // Whether SimpleResponse should include username.
+  bool fill_username = 4;
+
+  // Whether SimpleResponse should include OAuth scope.
+  bool fill_oauth_scope = 5;
+
+  // Whether to request the server to compress the response. This field is
+  // "nullable" in order to interoperate seamlessly with clients not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the response's compression status.
+  BoolValue response_compressed = 6;
+
+  // Whether server should return a given status
+  EchoStatus response_status = 7;
+
+  // Whether the server should expect this request to be compressed.
+  BoolValue expect_compressed = 8;
+}
+
+// Unary response, as configured by the request.
+message SimpleResponse {
+  // Payload to increase message size.
+  Payload payload = 1;
+  // The user the request came from, for verifying authentication was
+  // successful when the client expected it.
+  string username = 2;
+  // OAuth scope.
+  string oauth_scope = 3;
+}
+
+// Client-streaming request.
+message StreamingInputCallRequest {
+  // Optional input payload sent along with the request.
+  Payload payload = 1;
+
+  // Whether the server should expect this request to be compressed. This field
+  // is "nullable" in order to interoperate seamlessly with servers not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the request's compression status.
+  BoolValue expect_compressed = 2;
+
+  // Not expecting any payload from the response.
+}
+
+// Client-streaming response.
+message StreamingInputCallResponse {
+  // Aggregated size of payloads received from the client.
+  int32 aggregated_payload_size = 1;
+}
+
+// Configuration for a particular response.
+message ResponseParameters {
+  // Desired payload sizes in responses from the server.
+  int32 size = 1;
+
+  // Desired interval between consecutive responses in the response stream in
+  // microseconds.
+  int32 interval_us = 2;
+
+  // Whether to request the server to compress the response. This field is
+  // "nullable" in order to interoperate seamlessly with clients not able to
+  // implement the full compression tests by introspecting the call to verify
+  // the response's compression status.
+  BoolValue compressed = 3;
+}
+
+// Server-streaming request.
+message StreamingOutputCallRequest {
+  // DEPRECATED, don't use. To be removed shortly.
+  // Desired payload type in the response from the server.
+  // If response_type is RANDOM, the payload from each response in the stream
+  // might be of different types. This is to simulate a mixed type of payload
+  // stream.
+  PayloadType response_type = 1;
+
+  // Configuration for each expected response message.
+  repeated ResponseParameters response_parameters = 2;
+
+  // Optional input payload sent along with the request.
+  Payload payload = 3;
+
+  // Whether server should return a given status
+  EchoStatus response_status = 7;
+}
+
+// Server-streaming response, as configured by the request and parameters.
+message StreamingOutputCallResponse {
+  // Payload to increase response size.
+  Payload payload = 1;
+}
+
+// For reconnect interop test only.
+// Client tells server what reconnection parameters it used.
+message ReconnectParams {
+  int32 max_reconnect_backoff_ms = 1;
+}
+
+// For reconnect interop test only.
+// Server tells client whether its reconnects are following the spec and the
+// reconnect backoffs it saw.
+message ReconnectInfo {
+  bool passed = 1;
+  repeated int32 backoff_ms = 2;
+}

tonic-interop/proto/grpc/testing/test.proto

@@ -0,0 +1,78 @@
+// Copyright 2015-2016 gRPC authors.
+//
+// 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.
+
+// An integration test service that covers all the method signature permutations
+// of unary/streaming requests/responses.
+
+syntax = "proto3";
+
+import "empty.proto";
+import "messages.proto";
+
+package grpc.testing;
+
+// A simple service to test the various types of RPCs and experiment with
+// performance with various types of payload.
+service TestService {
+  // One empty request followed by one empty response.
+  rpc EmptyCall(grpc.testing.Empty) returns (grpc.testing.Empty);
+
+  // One request followed by one response.
+  rpc UnaryCall(SimpleRequest) returns (SimpleResponse);
+
+  // One request followed by one response. Response has cache control
+  // headers set such that a caching HTTP proxy (such as GFE) can
+  // satisfy subsequent requests.
+  rpc CacheableUnaryCall(SimpleRequest) returns (SimpleResponse);
+
+  // One request followed by a sequence of responses (streamed download).
+  // The server returns the payload with client desired type and sizes.
+  rpc StreamingOutputCall(StreamingOutputCallRequest)
+      returns (stream StreamingOutputCallResponse);
+
+  // A sequence of requests followed by one response (streamed upload).
+  // The server returns the aggregated size of client payload as the result.
+  rpc StreamingInputCall(stream StreamingInputCallRequest)
+      returns (StreamingInputCallResponse);
+
+  // A sequence of requests with each request served by the server immediately.
+  // As one request could lead to multiple responses, this interface
+  // demonstrates the idea of full duplexing.
+  rpc FullDuplexCall(stream StreamingOutputCallRequest)
+      returns (stream StreamingOutputCallResponse);
+
+  // A sequence of requests followed by a sequence of responses.
+  // The server buffers all the client requests and then serves them in order. A
+  // stream of responses are returned to the client when the server starts with
+  // first request.
+  rpc HalfDuplexCall(stream StreamingOutputCallRequest)
+      returns (stream StreamingOutputCallResponse);
+
+  // The test server will not implement this method. It will be used
+  // to test the behavior when clients call unimplemented methods.
+  rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty);
+}
+
+// A simple service NOT implemented at servers so clients can test for
+// that case.
+service UnimplementedService {
+  // A call that no server should implement
+  rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty);
+}
+
+// A service used to control reconnect server.
+service ReconnectService {
+  rpc Start(grpc.testing.ReconnectParams) returns (grpc.testing.Empty);
+  rpc Stop(grpc.testing.Empty) returns (grpc.testing.ReconnectInfo);
+}

tonic-interop/src/bin/client.rs

@@ -0,0 +1,79 @@
+#![feature(async_await)]
+
+use clap::{arg_enum, App, Arg, values_t};
+use tonic_interop::client;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let matches = App::new("My Super Program")
+        .version("1.0")
+        .about("Does awesome things")
+        .arg(
+            Arg::with_name("test_case")
+                .long("test_case")
+                .value_name("TESTCASE")
+                .help(
+                    "The name of the test case to execute. For example,
+                \"empty_unary\".",
+                )
+                .possible_values(&Testcase::variants())
+                .default_value("large_unary")
+                .takes_value(true)
+                .min_values(1)
+                .use_delimiter(true),
+        )
+        .get_matches();
+
+    let test_cases = values_t!(matches, "test_case", Testcase).unwrap_or_else(|e| e.exit());
+
+    let addr = "127.0.0.1:10000".parse()?;
+
+    let mut client = client::create(addr).await?;
+
+    for test_case in test_cases {
+        println!("{:?}:", test_case);
+        let mut test_results = Vec::new();
+
+        match test_case {
+            Testcase::empty_unary => client::unary_call(&mut client, &mut test_results).await,
+            _ => unimplemented!(),
+        }
+
+        for result in test_results {
+            println!("  {}", result);
+        }
+    }
+
+    Ok(())
+}
+
+arg_enum! {
+    #[derive(Debug, Copy, Clone)]
+    #[allow(non_camel_case_types)]
+    enum Testcase {
+        empty_unary,
+        cacheable_unary,
+        large_unary,
+        client_compressed_unary,
+        server_compressed_unary,
+        client_streaming,
+        client_compressed_streaming,
+        server_streaming,
+        server_compressed_streaming,
+        ping_pong,
+        empty_stream,
+        compute_engine_creds,
+        jwt_token_creds,
+        oauth2_auth_token,
+        per_rpc_creds,
+        custom_metadata,
+        status_code_and_message,
+        special_status_message,
+        unimplemented_method,
+        unimplemented_service,
+        cancel_after_begin,
+        cancel_after_first_response,
+        timeout_on_sleeping_server,
+        concurrent_large_unary
+    }
+}

tonic-interop/src/bin/server.rs

@@ -0,0 +1,49 @@
+#![feature(async_await)]
+
+use std::time::Duration;
+use tokio::{net::TcpListener, timer::Delay};
+use tonic::{Request, Response, Status};
+use tower_h2::Server;
+
+pub mod pb {
+    #![allow(dead_code)]
+    #![allow(unused_imports)]
+    include!(concat!(env!("OUT_DIR"), "/grpc.testing.rs"));
+    tonic::client!(service = "grpc.testing.TestService", proto = "self");
+}
+
+use pb::*;
+
+#[derive(Default, Clone)]
+pub struct TestService {
+    data: String,
+}
+
+#[tonic::server(service = "grpc.testing.TestService", proto = "pb")]
+impl TestService {
+    pub async fn empty_call(&self, request: Request<Empty>) -> Result<Response<Empty>, Status> {
+        println!("REQUEST={:?}", request);
+        Ok(Response::new(Empty {}))
+    }
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let addr = "[::1]:10000".parse().unwrap();
+    let mut bind = TcpListener::bind(&addr)?;
+
+    let greeter = TestService::default();
+    let mut server = Server::new(TestServiceServer::new(greeter), Default::default());
+
+    while let Ok((sock, _addr)) = bind.accept().await {
+        if let Err(e) = sock.set_nodelay(true) {
+            return Err(e.into());
+        }
+
+        if let Err(e) = server.serve(sock).await {
+            println!("H2 ERROR: {}", e);
+        }
+    }
+
+    Ok(())
+}

tonic-interop/src/client.rs

@@ -0,0 +1,39 @@
+use crate::{pb::*, test_assert, TestAssertion};
+use std::net::SocketAddr;
+use tokio::net::TcpStream;
+use tonic::Request;
+use tower_h2::{add_origin::AddOrigin, Connection};
+
+pub type Client = TestServiceClient<AddOrigin<Connection<tonic::BoxBody>>>;
+
+tonic::client!(service = "grpc.testing.TestService", proto = "crate::pb");
+
+pub async fn create(addr: SocketAddr) -> Result<Client, Box<dyn std::error::Error>> {
+    let io = TcpStream::connect(&addr).await?;
+
+    let origin = http::Uri::from_shared(format!("http://{}", addr).into()).unwrap();
+
+    let svc = Connection::handshake(io).await?;
+    let svc = AddOrigin::new(svc, origin);
+
+    Ok(TestServiceClient::new(svc))
+}
+
+pub async fn unary_call(client: &mut Client, assertions: &mut Vec<TestAssertion>) {
+    let result = client.empty_call(Request::new(Empty {})).await;
+
+    assertions.push(test_assert!(
+        "call must be successful",
+        result.is_ok(),
+        format!("result={:?}", result)
+    ));
+
+    if let Ok(response) = result {
+        let body = response.into_inner();
+        assertions.push(test_assert!(
+            "body must not be null",
+            body == Empty {},
+            format!("body={:?}", body)
+        ));
+    }
+}

tonic-interop/src/lib.rs

@@ -0,0 +1,90 @@
+#![feature(async_await)]
+
+pub mod client;
+
+pub mod pb {
+    #![allow(dead_code)]
+    #![allow(unused_imports)]
+    include!(concat!(env!("OUT_DIR"), "/grpc.testing.rs"));
+}
+
+use std::fmt;
+
+#[derive(Debug)]
+pub enum TestAssertion {
+    Passed {
+        description: &'static str,
+    },
+    Failed {
+        description: &'static str,
+        expression: &'static str,
+        why: Option<String>,
+    },
+}
+
+impl fmt::Display for TestAssertion {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        use console::{style, Emoji};
+        match *self {
+            TestAssertion::Passed { ref description } => write!(
+                f,
+                "{check} {desc}",
+                check = style(Emoji("✔", "+")).green(),
+                desc = style(description).green(),
+            ),
+            TestAssertion::Failed {
+                ref description,
+                ref expression,
+                why: Some(ref why),
+            } => write!(
+                f,
+                "{check} {desc}\n  in `{exp}`: {why}",
+                check = style(Emoji("✖", "x")).red(),
+                desc = style(description).red(),
+                exp = style(expression).red(),
+                why = style(why).red(),
+            ),
+            TestAssertion::Failed {
+                ref description,
+                ref expression,
+                why: None,
+            } => write!(
+                f,
+                "{check} {desc}\n  in `{exp}`",
+                check = style(Emoji("✖", "x")).red(),
+                desc = style(description).red(),
+                exp = style(expression).red(),
+            ),
+        }
+    }
+}
+
+#[macro_export]
+macro_rules! test_assert {
+    ($description:expr, $assertion:expr) => {
+        if $assertion {
+            crate::TestAssertion::Passed {
+                description: $description,
+            }
+        } else {
+            TestAssertion::Failed {
+                description: $description,
+                expression: stringify!($assertion),
+                why: None,
+            }
+        }
+    };
+    ($description:expr, $assertion:expr, $why:expr) => {
+        if $assertion {
+            crate::TestAssertion::Passed {
+                description: $description,
+            }
+        } else {
+            crate::TestAssertion::Failed {
+                description: $description,
+                expression: stringify!($assertion),
+                why: Some($why),
+            }
+        }
+    };
+}

tonic-macros/src/lib.rs

@@ -20,6 +20,8 @@ pub fn client(attr: TokenStream) -> TokenStream {
     let methods = client::generate(service, proto_path);
 
     let output = quote! {
+        use tonic::_codegen::*;
+
         pub struct #service_ident <T> {
             inner: tonic::client::Grpc<T>,
         }

tonic/src/client/grpc.rs

@@ -1,6 +1,6 @@
 use crate::{
     body::{Body, BoxBody},
-    codec::{decode_response, decode_empty, encode, Codec, Streaming},
+    codec::{decode_empty, decode_response, encode, Codec, Streaming},
     Code, GrpcService, Request, Response, Status,
 };
 use futures_core::Stream;

tonic/src/codec/prost.rs

@@ -27,7 +27,7 @@ where
     type Encoder = ProstEncoder<T>;
     type Decoder = ProstDecoder<U>;
 
-    const CONTENT_TYPE: &'static str = "application/groc+proto";
+    const CONTENT_TYPE: &'static str = "application/grpc+proto";
 
     fn encoder(&mut self) -> Self::Encoder {
         ProstEncoder(PhantomData)

tower-h2/src/flush.rs

@@ -17,6 +17,7 @@ where
     state: FlushState,
 }
 
+#[derive(Debug)]
 enum FlushState {
     Data,
     Trailers,