Add client streaming, server streaming and bidi streaming tests

4 files changed, 267 insertions, 134 deletions

diff --git a/test/cpp/end2end/BUILD b/test/cpp/end2end/BUILD
index d246f41ff8..235249e8bf 100644
--- a/test/cpp/end2end/BUILD
+++ b/test/cpp/end2end/BUILD
@@ -35,6 +35,19 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "interceptors_util",
+    testonly = True,
+    hdrs = ["interceptors_util.h"],
+    external_deps = [
+        "gtest",
+    ],
+    deps = [
+        "//src/proto/grpc/testing:echo_proto",
+        "//test/cpp/util:test_util",
+    ],
+)
+
 grpc_cc_test(
     name = "async_end2end_test",
     srcs = ["async_end2end_test.cc"],
@@ -124,6 +137,7 @@ grpc_cc_test(
         "gtest",
     ],
     deps = [
+        ":interceptors_util",
         ":test_service_impl",
         "//:gpr",
         "//:grpc",
@@ -495,6 +509,7 @@ grpc_cc_test(
         "gtest",
     ],
     deps = [
+        ":interceptors_util",
         ":test_service_impl",
         "//:gpr",
         "//:grpc",
diff --git a/test/cpp/end2end/client_interceptors_end2end_test.cc b/test/cpp/end2end/client_interceptors_end2end_test.cc
index 3716582c58..ff012f6f48 100644
--- a/test/cpp/end2end/client_interceptors_end2end_test.cc
+++ b/test/cpp/end2end/client_interceptors_end2end_test.cc
@@ -32,6 +32,7 @@
 #include "src/proto/grpc/testing/echo.grpc.pb.h"
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
+#include "test/cpp/end2end/interceptors_util.h"
 #include "test/cpp/end2end/test_service_impl.h"
 #include "test/cpp/util/byte_buffer_proto_helper.h"
 #include "test/cpp/util/string_ref_helper.h"
@@ -42,28 +43,6 @@ namespace grpc {
 namespace testing {
 namespace {
 
-class EchoTestServiceStreamingImpl : public EchoTestService::Service {
- public:
-  ~EchoTestServiceStreamingImpl() override {}
-
-  Status BidiStream(
-      ServerContext* context,
-      grpc::ServerReaderWriter<EchoResponse, EchoRequest>* stream) override {
-    EchoRequest req;
-    EchoResponse resp;
-    auto client_metadata = context->client_metadata();
-    for (const auto& pair : client_metadata) {
-      context->AddTrailingMetadata(ToString(pair.first), ToString(pair.second));
-    }
-
-    while (stream->Read(&req)) {
-      resp.set_message(req.message());
-      stream->Write(resp, grpc::WriteOptions());
-    }
-    return Status::OK;
-  }
-};
-
 class ClientInterceptorsStreamingEnd2endTest : public ::testing::Test {
  protected:
   ClientInterceptorsStreamingEnd2endTest() {
@@ -157,7 +136,6 @@ class HijackingInterceptor : public experimental::Interceptor {
   }
 
   virtual void Intercept(experimental::InterceptorBatchMethods* methods) {
-    // gpr_log(GPR_ERROR, "ran this");
     bool hijack = false;
     if (methods->QueryInterceptionHookPoint(
             experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) {
@@ -261,7 +239,6 @@ class HijackingInterceptorMakesAnotherCall : public experimental::Interceptor {
   }
 
   virtual void Intercept(experimental::InterceptorBatchMethods* methods) {
-    // gpr_log(GPR_ERROR, "ran this");
     if (methods->QueryInterceptionHookPoint(
             experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) {
       auto* map = methods->GetSendInitialMetadata();
@@ -329,6 +306,7 @@ class HijackingInterceptorMakesAnotherCall : public experimental::Interceptor {
     }
     if (methods->QueryInterceptionHookPoint(
             experimental::InterceptionHookPoints::PRE_RECV_INITIAL_METADATA)) {
+      gpr_log(GPR_ERROR, "hijacked");
       auto* map = methods->GetRecvInitialMetadata();
       // Got nothing better to do here at the moment
       EXPECT_EQ(map->size(), static_cast<unsigned>(0));
@@ -345,7 +323,7 @@ class HijackingInterceptorMakesAnotherCall : public experimental::Interceptor {
       auto* map = methods->GetRecvTrailingMetadata();
       // insert the metadata that we want
       EXPECT_EQ(map->size(), static_cast<unsigned>(0));
-      *map = ctx_.GetServerTrailingMetadata();
+      map->insert(std::make_pair("testkey", "testvalue"));
       auto* status = methods->GetRecvStatus();
       *status = Status(StatusCode::OK, "");
     }
@@ -376,7 +354,6 @@ class LoggingInterceptor : public experimental::Interceptor {
   LoggingInterceptor(experimental::ClientRpcInfo* info) { info_ = info; }
 
   virtual void Intercept(experimental::InterceptorBatchMethods* methods) {
-    // gpr_log(GPR_ERROR, "ran this");
     if (methods->QueryInterceptionHookPoint(
             experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) {
       auto* map = methods->GetSendInitialMetadata();
@@ -440,63 +417,6 @@ class LoggingInterceptorFactory
   }
 };
 
-void MakeCall(const std::shared_ptr<Channel>& channel) {
-  auto stub = grpc::testing::EchoTestService::NewStub(channel);
-  ClientContext ctx;
-  EchoRequest req;
-  req.mutable_param()->set_echo_metadata(true);
-  ctx.AddMetadata("testkey", "testvalue");
-  req.set_message("Hello");
-  EchoResponse resp;
-  Status s = stub->Echo(&ctx, req, &resp);
-  EXPECT_EQ(s.ok(), true);
-  EXPECT_EQ(resp.message(), "Hello");
-}
-
-void MakeCallbackCall(const std::shared_ptr<Channel>& channel) {
-  auto stub = grpc::testing::EchoTestService::NewStub(channel);
-  ClientContext ctx;
-  EchoRequest req;
-  std::mutex mu;
-  std::condition_variable cv;
-  bool done = false;
-  req.mutable_param()->set_echo_metadata(true);
-  ctx.AddMetadata("testkey", "testvalue");
-  req.set_message("Hello");
-  EchoResponse resp;
-  stub->experimental_async()->Echo(&ctx, &req, &resp,
-                                   [&resp, &mu, &done, &cv](Status s) {
-                                     // gpr_log(GPR_ERROR, "got the callback");
-                                     EXPECT_EQ(s.ok(), true);
-                                     EXPECT_EQ(resp.message(), "Hello");
-                                     std::lock_guard<std::mutex> l(mu);
-                                     done = true;
-                                     cv.notify_one();
-                                   });
-  std::unique_lock<std::mutex> l(mu);
-  while (!done) {
-    cv.wait(l);
-  }
-}
-
-void MakeStreamingCall(const std::shared_ptr<Channel>& channel) {
-  auto stub = grpc::testing::EchoTestService::NewStub(channel);
-  ClientContext ctx;
-  EchoRequest req;
-  EchoResponse resp;
-  ctx.AddMetadata("testkey", "testvalue");
-  auto stream = stub->BidiStream(&ctx);
-  for (auto i = 0; i < 10; i++) {
-    req.set_message("Hello" + std::to_string(i));
-    stream->Write(req);
-    stream->Read(&resp);
-    EXPECT_EQ(req.message(), resp.message());
-  }
-  ASSERT_TRUE(stream->WritesDone());
-  Status s = stream->Finish();
-  EXPECT_EQ(s.ok(), true);
-}
-
 TEST_F(ClientInterceptorsEnd2endTest, ClientInterceptorLoggingTest) {
   ChannelArguments args;
   DummyInterceptor::Reset();
@@ -582,9 +502,6 @@ TEST_F(ClientInterceptorsEnd2endTest,
     creators->push_back(std::unique_ptr<DummyInterceptorFactory>(
         new DummyInterceptorFactory()));
   }
-  // auto channel = experimental::CreateCustomChannelWithInterceptors(
-  //    server_address_, InsecureChannelCredentials(), args,
-  //    std::move(creators));
   auto channel = server_->experimental().InProcessChannelWithInterceptors(
       args, std::move(creators));
 
@@ -616,7 +533,49 @@ TEST_F(ClientInterceptorsEnd2endTest,
   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 }
 
-TEST_F(ClientInterceptorsStreamingEnd2endTest, ClientInterceptorLoggingTest) {
+TEST_F(ClientInterceptorsStreamingEnd2endTest, ClientStreamingTest) {
+  ChannelArguments args;
+  DummyInterceptor::Reset();
+  auto creators = std::unique_ptr<std::vector<
+      std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>(
+      new std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>());
+  creators->push_back(std::unique_ptr<LoggingInterceptorFactory>(
+      new LoggingInterceptorFactory()));
+  // Add 20 dummy interceptors
+  for (auto i = 0; i < 20; i++) {
+    creators->push_back(std::unique_ptr<DummyInterceptorFactory>(
+        new DummyInterceptorFactory()));
+  }
+  auto channel = experimental::CreateCustomChannelWithInterceptors(
+      server_address_, InsecureChannelCredentials(), args, std::move(creators));
+  MakeClientStreamingCall(channel);
+  // Make sure all 20 dummy interceptors were run
+  EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
+}
+
+TEST_F(ClientInterceptorsStreamingEnd2endTest, ServerStreamingTest) {
+  ChannelArguments args;
+  DummyInterceptor::Reset();
+  auto creators = std::unique_ptr<std::vector<
+      std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>(
+      new std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>());
+  creators->push_back(std::unique_ptr<LoggingInterceptorFactory>(
+      new LoggingInterceptorFactory()));
+  // Add 20 dummy interceptors
+  for (auto i = 0; i < 20; i++) {
+    creators->push_back(std::unique_ptr<DummyInterceptorFactory>(
+        new DummyInterceptorFactory()));
+  }
+  auto channel = experimental::CreateCustomChannelWithInterceptors(
+      server_address_, InsecureChannelCredentials(), args, std::move(creators));
+  MakeServerStreamingCall(channel);
+  // Make sure all 20 dummy interceptors were run
+  EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
+}
+
+TEST_F(ClientInterceptorsStreamingEnd2endTest, BidiStreamingTest) {
   ChannelArguments args;
   DummyInterceptor::Reset();
   auto creators = std::unique_ptr<std::vector<
@@ -632,7 +591,7 @@ TEST_F(ClientInterceptorsStreamingEnd2endTest, ClientInterceptorLoggingTest) {
   }
   auto channel = experimental::CreateCustomChannelWithInterceptors(
       server_address_, InsecureChannelCredentials(), args, std::move(creators));
-  MakeStreamingCall(channel);
+  MakeBidiStreamingCall(channel);
   // Make sure all 20 dummy interceptors were run
   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 }
diff --git a/test/cpp/end2end/interceptors_util.h b/test/cpp/end2end/interceptors_util.h
new file mode 100644
index 0000000000..bc6211517d
--- /dev/null
+++ b/test/cpp/end2end/interceptors_util.h
@@ -0,0 +1,178 @@
+/*
+ *
+ * Copyright 2018 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.
+ *
+ */
+
+#include "src/proto/grpc/testing/echo.grpc.pb.h"
+#include "test/cpp/util/string_ref_helper.h"
+
+#include <gtest/gtest.h>
+
+namespace grpc {
+namespace testing {
+class EchoTestServiceStreamingImpl : public EchoTestService::Service {
+ public:
+  ~EchoTestServiceStreamingImpl() override {}
+
+  Status BidiStream(
+      ServerContext* context,
+      grpc::ServerReaderWriter<EchoResponse, EchoRequest>* stream) override {
+    EchoRequest req;
+    EchoResponse resp;
+    auto client_metadata = context->client_metadata();
+    for (const auto& pair : client_metadata) {
+      context->AddTrailingMetadata(ToString(pair.first), ToString(pair.second));
+    }
+
+    while (stream->Read(&req)) {
+      resp.set_message(req.message());
+      EXPECT_TRUE(stream->Write(resp, grpc::WriteOptions()));
+    }
+    return Status::OK;
+  }
+
+  Status RequestStream(ServerContext* context,
+                       ServerReader<EchoRequest>* reader,
+                       EchoResponse* resp) override {
+    auto client_metadata = context->client_metadata();
+    for (const auto& pair : client_metadata) {
+      context->AddTrailingMetadata(ToString(pair.first), ToString(pair.second));
+    }
+
+    EchoRequest req;
+    string response_str = "";
+    while (reader->Read(&req)) {
+      response_str += req.message();
+    }
+    resp->set_message(response_str);
+    return Status::OK;
+  }
+
+  Status ResponseStream(ServerContext* context, const EchoRequest* req,
+                        ServerWriter<EchoResponse>* writer) override {
+    auto client_metadata = context->client_metadata();
+    for (const auto& pair : client_metadata) {
+      context->AddTrailingMetadata(ToString(pair.first), ToString(pair.second));
+    }
+
+    EchoResponse resp;
+    resp.set_message(req->message());
+    for (int i = 0; i < 10; i++) {
+      EXPECT_TRUE(writer->Write(resp));
+    }
+    return Status::OK;
+  }
+};
+
+void MakeCall(const std::shared_ptr<Channel>& channel) {
+  auto stub = grpc::testing::EchoTestService::NewStub(channel);
+  ClientContext ctx;
+  EchoRequest req;
+  req.mutable_param()->set_echo_metadata(true);
+  ctx.AddMetadata("testkey", "testvalue");
+  req.set_message("Hello");
+  EchoResponse resp;
+  Status s = stub->Echo(&ctx, req, &resp);
+  EXPECT_EQ(s.ok(), true);
+  EXPECT_EQ(resp.message(), "Hello");
+}
+
+void MakeClientStreamingCall(const std::shared_ptr<Channel>& channel) {
+  auto stub = grpc::testing::EchoTestService::NewStub(channel);
+  ClientContext ctx;
+  EchoRequest req;
+  req.mutable_param()->set_echo_metadata(true);
+  ctx.AddMetadata("testkey", "testvalue");
+  req.set_message("Hello");
+  EchoResponse resp;
+  string expected_resp = "";
+  auto writer = stub->RequestStream(&ctx, &resp);
+  for (int i = 0; i < 10; i++) {
+    writer->Write(req);
+    expected_resp += "Hello";
+  }
+  writer->WritesDone();
+  Status s = writer->Finish();
+  EXPECT_EQ(s.ok(), true);
+  EXPECT_EQ(resp.message(), expected_resp);
+}
+
+void MakeServerStreamingCall(const std::shared_ptr<Channel>& channel) {
+  auto stub = grpc::testing::EchoTestService::NewStub(channel);
+  ClientContext ctx;
+  EchoRequest req;
+  req.mutable_param()->set_echo_metadata(true);
+  ctx.AddMetadata("testkey", "testvalue");
+  req.set_message("Hello");
+  EchoResponse resp;
+  string expected_resp = "";
+  auto reader = stub->ResponseStream(&ctx, req);
+  int count = 0;
+  while (reader->Read(&resp)) {
+    EXPECT_EQ(resp.message(), "Hello");
+    count++;
+  }
+  ASSERT_EQ(count, 10);
+  Status s = reader->Finish();
+  EXPECT_EQ(s.ok(), true);
+}
+
+void MakeBidiStreamingCall(const std::shared_ptr<Channel>& channel) {
+  auto stub = grpc::testing::EchoTestService::NewStub(channel);
+  ClientContext ctx;
+  EchoRequest req;
+  EchoResponse resp;
+  ctx.AddMetadata("testkey", "testvalue");
+  auto stream = stub->BidiStream(&ctx);
+  for (auto i = 0; i < 10; i++) {
+    req.set_message("Hello" + std::to_string(i));
+    stream->Write(req);
+    stream->Read(&resp);
+    EXPECT_EQ(req.message(), resp.message());
+  }
+  ASSERT_TRUE(stream->WritesDone());
+  Status s = stream->Finish();
+  EXPECT_EQ(s.ok(), true);
+}
+
+void MakeCallbackCall(const std::shared_ptr<Channel>& channel) {
+  auto stub = grpc::testing::EchoTestService::NewStub(channel);
+  ClientContext ctx;
+  EchoRequest req;
+  std::mutex mu;
+  std::condition_variable cv;
+  bool done = false;
+  req.mutable_param()->set_echo_metadata(true);
+  ctx.AddMetadata("testkey", "testvalue");
+  req.set_message("Hello");
+  EchoResponse resp;
+  stub->experimental_async()->Echo(&ctx, &req, &resp,
+                                   [&resp, &mu, &done, &cv](Status s) {
+                                     // gpr_log(GPR_ERROR, "got the callback");
+                                     EXPECT_EQ(s.ok(), true);
+                                     EXPECT_EQ(resp.message(), "Hello");
+                                     std::lock_guard<std::mutex> l(mu);
+                                     done = true;
+                                     cv.notify_one();
+                                   });
+  std::unique_lock<std::mutex> l(mu);
+  while (!done) {
+    cv.wait(l);
+  }
+}
+
+}  // namespace testing
+}  // namespace grpc
\ No newline at end of file
diff --git a/test/cpp/end2end/server_interceptors_end2end_test.cc b/test/cpp/end2end/server_interceptors_end2end_test.cc
index f59fc63f06..57b85a479e 100644
--- a/test/cpp/end2end/server_interceptors_end2end_test.cc
+++ b/test/cpp/end2end/server_interceptors_end2end_test.cc
@@ -32,9 +32,9 @@
 #include "src/proto/grpc/testing/echo.grpc.pb.h"
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
+#include "test/cpp/end2end/interceptors_util.h"
 #include "test/cpp/end2end/test_service_impl.h"
 #include "test/cpp/util/byte_buffer_proto_helper.h"
-#include "test/cpp/util/string_ref_helper.h"
 
 #include <gtest/gtest.h>
 
@@ -42,28 +42,6 @@ namespace grpc {
 namespace testing {
 namespace {
 
-class EchoTestServiceStreamingImpl : public EchoTestService::Service {
- public:
-  ~EchoTestServiceStreamingImpl() override {}
-
-  Status BidiStream(
-      ServerContext* context,
-      grpc::ServerReaderWriter<EchoResponse, EchoRequest>* stream) override {
-    EchoRequest req;
-    EchoResponse resp;
-    auto client_metadata = context->client_metadata();
-    for (const auto& pair : client_metadata) {
-      context->AddTrailingMetadata(ToString(pair.first), ToString(pair.second));
-    }
-
-    while (stream->Read(&req)) {
-      resp.set_message(req.message());
-      stream->Write(resp, grpc::WriteOptions());
-    }
-    return Status::OK;
-  }
-};
-
 /* This interceptor does nothing. Just keeps a global count on the number of
  * times it was invoked. */
 class DummyInterceptor : public experimental::Interceptor {
@@ -181,20 +159,7 @@ class LoggingInterceptorFactory
   }
 };
 
-void MakeCall(const std::shared_ptr<Channel>& channel) {
-  auto stub = grpc::testing::EchoTestService::NewStub(channel);
-  ClientContext ctx;
-  EchoRequest req;
-  req.mutable_param()->set_echo_metadata(true);
-  ctx.AddMetadata("testkey", "testvalue");
-  req.set_message("Hello");
-  EchoResponse resp;
-  Status s = stub->Echo(&ctx, req, &resp);
-  EXPECT_EQ(s.ok(), true);
-  EXPECT_EQ(resp.message(), "Hello");
-}
-
-/*void MakeStreamingCall(const std::shared_ptr<Channel>& channel) {
+void MakeBidiStreamingCall(const std::shared_ptr<Channel>& channel) {
   auto stub = grpc::testing::EchoTestService::NewStub(channel);
   ClientContext ctx;
   EchoRequest req;
@@ -210,7 +175,7 @@ void MakeCall(const std::shared_ptr<Channel>& channel) {
   ASSERT_TRUE(stream->WritesDone());
   Status s = stream->Finish();
   EXPECT_EQ(s.ok(), true);
-}*/
+}
 
 class ServerInterceptorsEnd2endSyncUnaryTest : public ::testing::Test {
  protected:
@@ -240,7 +205,7 @@ class ServerInterceptorsEnd2endSyncUnaryTest : public ::testing::Test {
   std::unique_ptr<Server> server_;
 };
 
-TEST_F(ServerInterceptorsEnd2endSyncUnaryTest, ServerInterceptorTest) {
+TEST_F(ServerInterceptorsEnd2endSyncUnaryTest, UnaryTest) {
   ChannelArguments args;
   DummyInterceptor::Reset();
   auto channel = CreateChannel(server_address_, InsecureChannelCredentials());
@@ -249,10 +214,9 @@ TEST_F(ServerInterceptorsEnd2endSyncUnaryTest, ServerInterceptorTest) {
   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 }
 
-class ServerInterceptorsEnd2endSyncClientStreamingTest
-    : public ::testing::Test {
+class ServerInterceptorsEnd2endSyncStreamingTest : public ::testing::Test {
  protected:
-  ServerInterceptorsEnd2endSyncClientStreamingTest() {
+  ServerInterceptorsEnd2endSyncStreamingTest() {
     int port = grpc_pick_unused_port_or_die();
 
     ServerBuilder builder;
@@ -274,16 +238,33 @@ class ServerInterceptorsEnd2endSyncClientStreamingTest
     server_ = builder.BuildAndStart();
   }
   std::string server_address_;
-  TestServiceImpl service_;
+  EchoTestServiceStreamingImpl service_;
   std::unique_ptr<Server> server_;
 };
 
-TEST_F(ServerInterceptorsEnd2endSyncClientStreamingTest,
-       ServerInterceptorTest) {
+TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, ClientStreamingTest) {
   ChannelArguments args;
   DummyInterceptor::Reset();
   auto channel = CreateChannel(server_address_, InsecureChannelCredentials());
-  MakeCall(channel);
+  MakeClientStreamingCall(channel);
+  // Make sure all 20 dummy interceptors were run
+  EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
+}
+
+TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, ServerStreamingTest) {
+  ChannelArguments args;
+  DummyInterceptor::Reset();
+  auto channel = CreateChannel(server_address_, InsecureChannelCredentials());
+  MakeServerStreamingCall(channel);
+  // Make sure all 20 dummy interceptors were run
+  EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
+}
+
+TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, BidiStreamingTest) {
+  ChannelArguments args;
+  DummyInterceptor::Reset();
+  auto channel = CreateChannel(server_address_, InsecureChannelCredentials());
+  MakeBidiStreamingCall(channel);
   // Make sure all 20 dummy interceptors were run
   EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20);
 }