Completed interop spec, as well as code for c++

7 files changed, 167 insertions, 67 deletions

diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc
index 7727824979..c7d081100e 100644
--- a/test/cpp/interop/client.cc
+++ b/test/cpp/interop/client.cc
@@ -40,7 +40,9 @@
 #include <grpc++/client_context.h>
 #include <grpc/grpc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/useful.h>
 
+#include "src/core/lib/support/string.h"
 #include "test/cpp/interop/client_helper.h"
 #include "test/cpp/interop/interop_client.h"
 #include "test/cpp/util/test_config.h"
@@ -55,10 +57,14 @@ DEFINE_string(test_case, "large_unary",
               "Configure different test cases. Valid options are: "
               "empty_unary : empty (zero bytes) request and response; "
               "large_unary : single request and (large) response; "
-              "large_compressed_unary : single request and compressed (large) "
-              "response; "
+
+              "client_compressed_unary : single compressed request; "
+              "server_compressed_unary : single compressed response; "
+
               "client_streaming : request streaming with single response; "
               "server_streaming : single request with response streaming; "
+              "client_compressed_streaming : compressed request streaming with "
+              "single response; "
               "server_compressed_streaming : single request with compressed "
               "response streaming; "
               "slow_consumer : single request with response; "
@@ -104,14 +110,18 @@ int main(int argc, char** argv) {
     client.DoEmpty();
   } else if (FLAGS_test_case == "large_unary") {
     client.DoLargeUnary();
-  } else if (FLAGS_test_case == "large_compressed_unary") {
-    client.DoLargeCompressedUnary();
+  } else if (FLAGS_test_case == "server_compressed_unary") {
+    client.DoServerCompressedUnary();
+  } else if (FLAGS_test_case == "client_compressed_unary") {
+    client.DoClientCompressedUnary();
   } else if (FLAGS_test_case == "client_streaming") {
     client.DoRequestStreaming();
   } else if (FLAGS_test_case == "server_streaming") {
     client.DoResponseStreaming();
   } else if (FLAGS_test_case == "server_compressed_streaming") {
-    client.DoResponseCompressedStreaming();
+    client.DoServerCompressedStreaming();
+  } else if (FLAGS_test_case == "client_compressed_streaming") {
+    client.DoClientCompressedStreaming();
   } else if (FLAGS_test_case == "slow_consumer") {
     client.DoResponseStreamingWithSlowConsumer();
   } else if (FLAGS_test_case == "half_duplex") {
@@ -144,9 +154,12 @@ int main(int argc, char** argv) {
   } else if (FLAGS_test_case == "all") {
     client.DoEmpty();
     client.DoLargeUnary();
+    client.DoClientCompressedUnary();
+    client.DoServerCompressedUnary();
     client.DoRequestStreaming();
     client.DoResponseStreaming();
-    client.DoResponseCompressedStreaming();
+    client.DoClientCompressedStreaming();
+    client.DoServerCompressedStreaming();
     client.DoHalfDuplex();
     client.DoPingPong();
     client.DoCancelAfterBegin();
@@ -165,14 +178,36 @@ int main(int argc, char** argv) {
     }
     // compute_engine_creds only runs in GCE.
   } else {
-    gpr_log(
-        GPR_ERROR,
-        "Unsupported test case %s. Valid options are all|empty_unary|"
-        "large_unary|large_compressed_unary|client_streaming|server_streaming|"
-        "server_compressed_streaming|half_duplex|ping_pong|cancel_after_begin|"
-        "cancel_after_first_response|timeout_on_sleeping_server|empty_stream|"
-        "compute_engine_creds|jwt_token_creds|oauth2_auth_token|per_rpc_creds",
-        "status_code_and_message|custom_metadata", FLAGS_test_case.c_str());
+    const char* testcases[] =
+    { "all",
+      "cancel_after_begin",
+      "cancel_after_first_response",
+      "client_compressed_streaming",
+      "client_compressed_unary",
+      "client_streaming",
+      "compute_engine_creds",
+      "custom_metadata",
+      "empty_stream",
+      "empty_unary",
+      "half_duplex",
+      "jwt_token_creds",
+      "large_unary",
+      "oauth2_auth_token",
+      "oauth2_auth_token",
+      "per_rpc_creds",
+      "per_rpc_creds",
+      "ping_pong",
+      "server_compressed_streaming",
+      "server_compressed_unary",
+      "server_streaming",
+      "status_code_and_message",
+      "timeout_on_sleeping_server"};
+    char* joined_testcases =
+        gpr_strjoin_sep(testcases, GPR_ARRAY_SIZE(testcases), "\n", NULL);
+
+    gpr_log(GPR_ERROR, "Unsupported test case %s. Valid options are\n%s",
+            FLAGS_test_case.c_str(), joined_testcases);
+    gpr_free(joined_testcases);
     ret = 1;
   }
 
diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc
index 7705bb1592..e5d3751402 100644
--- a/test/cpp/interop/interop_client.cc
+++ b/test/cpp/interop/interop_client.cc
@@ -73,23 +73,22 @@ void CompressionChecks(const InteropClientContextInspector& inspector,
                        const SimpleResponse* response) {
   const grpc_compression_algorithm received_compression =
       inspector.GetCallCompressionAlgorithm();
-  if (request->request_compressed_response() &&
-      received_compression == GRPC_COMPRESS_NONE) {
-    if (request->request_compressed_response() &&
-        received_compression == GRPC_COMPRESS_NONE) {
+  if (request->request_compressed_response()) {
+    if (received_compression == GRPC_COMPRESS_NONE) {
       // Requested some compression, got NONE. This is an error.
       gpr_log(GPR_ERROR,
               "Failure: Requested compression but got uncompressed response "
               "from server.");
       abort();
     }
-  }
-  if (!request->request_compressed_response()) {
+    if (request->response_type() == PayloadType::COMPRESSABLE) {
+      // requested compression and compressable response => results should
+      // always be compressed.
+      GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
+    }
+  } else {
+    // Didn't request compression -> make sure the response is uncompressed
     GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
-  } else if (request->response_type() == PayloadType::COMPRESSABLE) {
-    // requested compression and compressable response => results should always
-    // be compressed.
-    GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
   }
 }
 }  // namespace
@@ -211,17 +210,6 @@ bool InteropClient::PerformLargeUnary(SimpleRequest* request,
       GPR_ASSERT(response->payload().body() ==
                  grpc::string(kLargeResponseSize, '\0'));
       break;
-    case PayloadType::UNCOMPRESSABLE: {
-      // We don't really check anything: We can't assert that the payload is
-      // uncompressed because it's the server's prerogative to decide on that,
-      // and different implementations decide differently (ie, Java always
-      // compresses when requested to do so, whereas C core throws away the
-      // compressed payload if the output is larger than the input).
-      // In addition, we don't compare the actual random bytes received because
-      // asserting that data is sent/received properly isn't the purpose of this
-      // test. Moreover, different implementations are also free to use
-      // different sets of random bytes.
-    } break;
     default:
       GPR_ASSERT(false);
   }
@@ -336,9 +324,39 @@ bool InteropClient::DoLargeUnary() {
   return true;
 }
 
-bool InteropClient::DoLargeCompressedUnary() {
+bool InteropClient::DoClientCompressedUnary() {
+  const bool expect_compression[] = {false, true};
+  const PayloadType payload_types[] = {COMPRESSABLE};
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
+    for (size_t j = 0; j < GPR_ARRAY_SIZE(expect_compression); j++) {
+      char* log_suffix;
+      gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
+                   expect_compression[j] ? "true" : "false",
+                   PayloadType_Name(payload_types[i]).c_str());
+
+      gpr_log(GPR_DEBUG, "Sending compressed unary request %s.", log_suffix);
+      SimpleRequest request;
+      SimpleResponse response;
+      request.set_response_type(payload_types[i]);
+      request.set_expect_compressed_request(expect_compression[j]);
+
+      if (!PerformLargeUnary(&request, &response, CompressionChecks)) {
+        gpr_log(GPR_ERROR, "Compressed unary request failed %s", log_suffix);
+        gpr_free(log_suffix);
+        return false;
+      }
+
+      gpr_log(GPR_DEBUG, "Compressed unary request failed %s", log_suffix);
+      gpr_free(log_suffix);
+    }
+  }
+
+  return true;
+}
+
+bool InteropClient::DoServerCompressedUnary() {
   const bool request_compression[] = {false, true};
-  const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE};
+  const PayloadType payload_types[] = {COMPRESSABLE};
   for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
     for (size_t j = 0; j < GPR_ARRAY_SIZE(request_compression); j++) {
       char* log_suffix;
@@ -346,7 +364,7 @@ bool InteropClient::DoLargeCompressedUnary() {
                    request_compression[j] ? "true" : "false",
                    PayloadType_Name(payload_types[i]).c_str());
 
-      gpr_log(GPR_DEBUG, "Sending a large compressed unary rpc %s.",
+      gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.",
               log_suffix);
       SimpleRequest request;
       SimpleResponse response;
@@ -354,12 +372,13 @@ bool InteropClient::DoLargeCompressedUnary() {
       request.set_request_compressed_response(request_compression[j]);
 
       if (!PerformLargeUnary(&request, &response, CompressionChecks)) {
-        gpr_log(GPR_ERROR, "Large compressed unary failed %s", log_suffix);
+        gpr_log(GPR_ERROR, "Request for compressed unary failed %s",
+                log_suffix);
         gpr_free(log_suffix);
         return false;
       }
 
-      gpr_log(GPR_DEBUG, "Large compressed unary done %s.", log_suffix);
+      gpr_log(GPR_DEBUG, "Request for compressed unary failed %s", log_suffix);
       gpr_free(log_suffix);
     }
   }
@@ -447,9 +466,16 @@ bool InteropClient::DoResponseStreaming() {
   return true;
 }
 
-bool InteropClient::DoResponseCompressedStreaming() {
+bool InteropClient::DoClientCompressedStreaming() {
+  // XXX
+  return false;
+}
+
+bool InteropClient::DoServerCompressedStreaming() {
   const bool request_compression[] = {false, true};
-  const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE};
+  const PayloadType payload_types[] = {COMPRESSABLE};
+  const std::vector<int> response_stream_sizes = {31415, 58979};
+
   for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
     for (size_t j = 0; j < GPR_ARRAY_SIZE(request_compression); j++) {
       ClientContext context;
@@ -466,11 +492,10 @@ bool InteropClient::DoResponseCompressedStreaming() {
       request.set_response_type(payload_types[i]);
       request.set_request_compressed_response(request_compression[j]);
 
-      for (size_t k = 0; k < response_stream_sizes.size() / 2; ++k) {
+      for (size_t k = 0; k < response_stream_sizes.size(); ++k) {
         ResponseParameters* response_parameter =
             request.add_response_parameters();
-        response_parameter->set_size(response_stream_sizes[k] +
-                                     response_stream_sizes[k + 1]);
+        response_parameter->set_size(response_stream_sizes[k]);
       }
       StreamingOutputCallResponse response;
 
@@ -484,11 +509,7 @@ bool InteropClient::DoResponseCompressedStreaming() {
         switch (response.payload().type()) {
           case PayloadType::COMPRESSABLE:
             GPR_ASSERT(response.payload().body() ==
-                       grpc::string(response_stream_sizes[k] +
-                                        response_stream_sizes[k + 1],
-                                    '\0'));
-            break;
-          case PayloadType::UNCOMPRESSABLE:
+                       grpc::string(response_stream_sizes[k], '\0'));
             break;
           default:
             GPR_ASSERT(false);
@@ -516,14 +537,14 @@ bool InteropClient::DoResponseCompressedStreaming() {
       gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix);
       gpr_free(log_suffix);
 
-      if (k < response_stream_sizes.size() / 2) {
+      if (k < response_stream_sizes.size()) {
         // stream->Read() failed before reading all the expected messages. This
         // is most likely due to a connection failure.
         gpr_log(GPR_ERROR,
-                "DoResponseCompressedStreaming(): Responses read (k=%d) is "
+                "DoServerCompressedStreaming(): Responses read (k=%d) is "
                 "less than the expected messages (i.e "
                 "response_stream_sizes.size()/2 (%d)). (i=%d, j=%d)",
-                k, response_stream_sizes.size() / 2, i, j);
+                k, response_stream_sizes.size(), i, j);
         return TransientFailureOrAbort();
       }
 
diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h
index ae75762bb8..ea44986fbc 100644
--- a/test/cpp/interop/interop_client.h
+++ b/test/cpp/interop/interop_client.h
@@ -64,12 +64,14 @@ class InteropClient {
 
   bool DoEmpty();
   bool DoLargeUnary();
-  bool DoLargeCompressedUnary();
+  bool DoServerCompressedUnary();
+  bool DoClientCompressedUnary();
   bool DoPingPong();
   bool DoHalfDuplex();
   bool DoRequestStreaming();
   bool DoResponseStreaming();
-  bool DoResponseCompressedStreaming();
+  bool DoServerCompressedStreaming();
+  bool DoClientCompressedStreaming();
   bool DoResponseStreamingWithSlowConsumer();
   bool DoCancelAfterBegin();
   bool DoCancelAfterFirstResponse();
diff --git a/test/cpp/interop/server_main.cc b/test/cpp/interop/interop_server.cc
index bbedda14d2..b328f478fa 100644
--- a/test/cpp/interop/server_main.cc
+++ b/test/cpp/interop/interop_server.cc
@@ -48,6 +48,7 @@
 #include <grpc/support/log.h>
 #include <grpc/support/useful.h>
 
+#include "src/core/lib/transport/byte_stream.h"
 #include "src/proto/grpc/testing/empty.grpc.pb.h"
 #include "src/proto/grpc/testing/messages.grpc.pb.h"
 #include "src/proto/grpc/testing/test.grpc.pb.h"
@@ -78,7 +79,6 @@ using grpc::testing::TestService;
 using grpc::Status;
 
 static bool got_sigint = false;
-static const char* kRandomFile = "test/cpp/interop/rnd.dat";
 
 const char kEchoInitialMetadataKey[] = "x-grpc-test-echo-initial";
 const char kEchoTrailingBinMetadataKey[] = "x-grpc-test-echo-trailing-bin";
@@ -117,16 +117,8 @@ bool SetPayload(PayloadType response_type, int size, Payload* payload) {
       std::unique_ptr<char[]> body(new char[size]());
       payload->set_body(body.get(), size);
     } break;
-    case PayloadType::UNCOMPRESSABLE: {
-      std::unique_ptr<char[]> body(new char[size]());
-      std::ifstream rnd_file(kRandomFile);
-      GPR_ASSERT(rnd_file.good());
-      rnd_file.read(body.get(), size);
-      GPR_ASSERT(!rnd_file.eof());  // Requested more rnd bytes than available
-      payload->set_body(body.get(), size);
-    } break;
     default:
-      GPR_ASSERT(false);
+      return false;
   }
   return true;
 }
@@ -140,6 +132,41 @@ void SetResponseCompression(ServerContext* context,
   }
 }
 
+template <typename RequestType>
+bool CheckExpectedCompression(const ServerContext& context,
+                              const RequestType& request) {
+  const InteropServerContextInspector inspector(context);
+  const grpc_compression_algorithm received_compression =
+      inspector.GetCallCompressionAlgorithm();
+
+  if (request.expect_compressed_request()) {
+    if (received_compression == GRPC_COMPRESS_NONE) {
+      // Expected some compression, got NONE. This is an error.
+      gpr_log(GPR_ERROR,
+              "Failure: Expected compression but got uncompressed request "
+              "from client.");
+      return false;
+    }
+    if (request.payload_type() == PayloadType::COMPRESSABLE) {
+      if (!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS)) {
+        gpr_log(GPR_ERROR,
+                "Failure: Requested compression in a compressable request, but "
+                "compression bit in message flags not set.");
+        return false;
+      }
+    }
+  } else {
+    // Didn't expect compression -> make sure the request is uncompressed
+    if (inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS) {
+      gpr_log(GPR_ERROR,
+              "Failure: Didn't requested compression, but compression bit in "
+              "message flags set.");
+      return false;
+    }
+  }
+  return true;
+}
+
 class TestServiceImpl : public TestService::Service {
  public:
   Status EmptyCall(ServerContext* context, const grpc::testing::Empty* request,
@@ -152,10 +179,15 @@ class TestServiceImpl : public TestService::Service {
                    SimpleResponse* response) {
     MaybeEchoMetadata(context);
     SetResponseCompression(context, *request);
+    if (!CheckExpectedCompression(*context, *request)) {
+      return Status(grpc::StatusCode::INVALID_ARGUMENT,
+                    "Compressed request expectation not met.");
+    }
     if (request->response_size() > 0) {
       if (!SetPayload(request->response_type(), request->response_size(),
                       response->mutable_payload())) {
-        return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+        return Status(grpc::StatusCode::INVALID_ARGUMENT,
+                      "Error creating payload.");
       }
     }
 
@@ -179,7 +211,8 @@ class TestServiceImpl : public TestService::Service {
       if (!SetPayload(request->response_type(),
                       request->response_parameters(i).size(),
                       response.mutable_payload())) {
-        return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+        return Status(grpc::StatusCode::INVALID_ARGUMENT,
+                      "Error creating payload.");
       }
       write_success = writer->Write(response);
     }
@@ -196,6 +229,10 @@ class TestServiceImpl : public TestService::Service {
     StreamingInputCallRequest request;
     int aggregated_payload_size = 0;
     while (reader->Read(&request)) {
+      if (!CheckExpectedCompression(*context, request)) {
+        return Status(grpc::StatusCode::INVALID_ARGUMENT,
+                      "Compressed request expectation not met.");
+      }
       if (request.has_payload()) {
         aggregated_payload_size += request.payload().body().size();
       }
diff --git a/test/cpp/interop/rnd.dat b/test/cpp/interop/rnd.dat
deleted file mode 100644
index 8c7f38f9e0..0000000000
--- a/test/cpp/interop/rnd.dat
+++ /dev/null
diff --git a/test/cpp/interop/server_helper.cc b/test/cpp/interop/server_helper.cc
index c6d891ad71..8b0b511bcb 100644
--- a/test/cpp/interop/server_helper.cc
+++ b/test/cpp/interop/server_helper.cc
@@ -72,6 +72,10 @@ uint32_t InteropServerContextInspector::GetEncodingsAcceptedByClient() const {
   return grpc_call_test_only_get_encodings_accepted_by_peer(context_.call_);
 }
 
+uint32_t InteropServerContextInspector::GetMessageFlags() const {
+  return grpc_call_test_only_get_message_flags(context_.call_);
+}
+
 std::shared_ptr<const AuthContext>
 InteropServerContextInspector::GetAuthContext() const {
   return context_.auth_context();
diff --git a/test/cpp/interop/server_helper.h b/test/cpp/interop/server_helper.h
index 12865e4032..a1da14a4c8 100644
--- a/test/cpp/interop/server_helper.h
+++ b/test/cpp/interop/server_helper.h
@@ -54,6 +54,7 @@ class InteropServerContextInspector {
   bool IsCancelled() const;
   grpc_compression_algorithm GetCallCompressionAlgorithm() const;
   uint32_t GetEncodingsAcceptedByClient() const;
+  uint32_t GetMessageFlags() const;
 
  private:
   const ::grpc::ServerContext& context_;