diff options
Diffstat (limited to 'test/cpp')
68 files changed, 3721 insertions, 262 deletions
diff --git a/test/cpp/client/client_channel_stress_test.cc b/test/cpp/client/client_channel_stress_test.cc index 976eeb6aea..bf321d8a89 100644 --- a/test/cpp/client/client_channel_stress_test.cc +++ b/test/cpp/client/client_channel_stress_test.cc @@ -321,7 +321,7 @@ class ClientChannelStressTest { int main(int argc, char** argv) { grpc_init(); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); grpc::testing::ClientChannelStressTest test; test.Run(); grpc_shutdown(); diff --git a/test/cpp/codegen/compiler_test_golden b/test/cpp/codegen/compiler_test_golden index 93e1e68654..1871e1375e 100644 --- a/test/cpp/codegen/compiler_test_golden +++ b/test/cpp/codegen/compiler_test_golden @@ -30,9 +30,11 @@ #include <grpcpp/impl/codegen/async_generic_service.h> #include <grpcpp/impl/codegen/async_stream.h> #include <grpcpp/impl/codegen/async_unary_call.h> +#include <grpcpp/impl/codegen/client_callback.h> #include <grpcpp/impl/codegen/method_handler_impl.h> #include <grpcpp/impl/codegen/proto_utils.h> #include <grpcpp/impl/codegen/rpc_method.h> +#include <grpcpp/impl/codegen/server_callback.h> #include <grpcpp/impl/codegen/service_type.h> #include <grpcpp/impl/codegen/status.h> #include <grpcpp/impl/codegen/stub_options.h> @@ -116,10 +118,13 @@ class ServiceA final { // // Method A2 leading comment 1 // Method A2 leading comment 2 + virtual void MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor< ::grpc::testing::Request>* reactor) = 0; // MethodA2 trailing comment 1 // Method A3 leading comment 1 + virtual void MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor< ::grpc::testing::Response>* reactor) = 0; // Method A3 trailing comment 1 // Method A4 leading comment 1 + virtual void MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor< ::grpc::testing::Request,::grpc::testing::Response>* reactor) = 0; // Method A4 trailing comment 1 }; virtual class experimental_async_interface* experimental_async() { return nullptr; } @@ -177,6 +182,9 @@ class ServiceA final { public StubInterface::experimental_async_interface { public: void MethodA1(::grpc::ClientContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, std::function<void(::grpc::Status)>) override; + void MethodA2(::grpc::ClientContext* context, ::grpc::testing::Response* response, ::grpc::experimental::ClientWriteReactor< ::grpc::testing::Request>* reactor) override; + void MethodA3(::grpc::ClientContext* context, ::grpc::testing::Request* request, ::grpc::experimental::ClientReadReactor< ::grpc::testing::Response>* reactor) override; + void MethodA4(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor< ::grpc::testing::Request,::grpc::testing::Response>* reactor) override; private: friend class Stub; explicit experimental_async(Stub* stub): stub_(stub) { } @@ -308,6 +316,98 @@ class ServiceA final { }; typedef WithAsyncMethod_MethodA1<WithAsyncMethod_MethodA2<WithAsyncMethod_MethodA3<WithAsyncMethod_MethodA4<Service > > > > AsyncService; template <class BaseClass> + class ExperimentalWithCallbackMethod_MethodA1 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithCallbackMethod_MethodA1() { + ::grpc::Service::experimental().MarkMethodCallback(0, + new ::grpc::internal::CallbackUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this](::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + return this->MethodA1(context, request, response, controller); + })); + } + ~ExperimentalWithCallbackMethod_MethodA1() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual void MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + }; + template <class BaseClass> + class ExperimentalWithCallbackMethod_MethodA2 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithCallbackMethod_MethodA2() { + ::grpc::Service::experimental().MarkMethodCallback(1, + new ::grpc::internal::CallbackClientStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this] { return this->MethodA2(); })); + } + ~ExperimentalWithCallbackMethod_MethodA2() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual ::grpc::experimental::ServerReadReactor< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA2() { + return new ::grpc::internal::UnimplementedReadReactor< + ::grpc::testing::Request, ::grpc::testing::Response>;} + }; + template <class BaseClass> + class ExperimentalWithCallbackMethod_MethodA3 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithCallbackMethod_MethodA3() { + ::grpc::Service::experimental().MarkMethodCallback(2, + new ::grpc::internal::CallbackServerStreamingHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this] { return this->MethodA3(); })); + } + ~ExperimentalWithCallbackMethod_MethodA3() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual ::grpc::experimental::ServerWriteReactor< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA3() { + return new ::grpc::internal::UnimplementedWriteReactor< + ::grpc::testing::Request, ::grpc::testing::Response>;} + }; + template <class BaseClass> + class ExperimentalWithCallbackMethod_MethodA4 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithCallbackMethod_MethodA4() { + ::grpc::Service::experimental().MarkMethodCallback(3, + new ::grpc::internal::CallbackBidiHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this] { return this->MethodA4(); })); + } + ~ExperimentalWithCallbackMethod_MethodA4() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual ::grpc::experimental::ServerBidiReactor< ::grpc::testing::Request, ::grpc::testing::Response>* MethodA4() { + return new ::grpc::internal::UnimplementedBidiReactor< + ::grpc::testing::Request, ::grpc::testing::Response>;} + }; + typedef ExperimentalWithCallbackMethod_MethodA1<ExperimentalWithCallbackMethod_MethodA2<ExperimentalWithCallbackMethod_MethodA3<ExperimentalWithCallbackMethod_MethodA4<Service > > > > ExperimentalCallbackService; + template <class BaseClass> class WithGenericMethod_MethodA1 : public BaseClass { private: void BaseClassMustBeDerivedFromService(const Service *service) {} @@ -456,6 +556,97 @@ class ServiceA final { } }; template <class BaseClass> + class ExperimentalWithRawCallbackMethod_MethodA1 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithRawCallbackMethod_MethodA1() { + ::grpc::Service::experimental().MarkMethodRawCallback(0, + new ::grpc::internal::CallbackUnaryHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this](::grpc::ServerContext* context, + const ::grpc::ByteBuffer* request, + ::grpc::ByteBuffer* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + this->MethodA1(context, request, response, controller); + })); + } + ~ExperimentalWithRawCallbackMethod_MethodA1() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual void MethodA1(::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, ::grpc::ByteBuffer* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + }; + template <class BaseClass> + class ExperimentalWithRawCallbackMethod_MethodA2 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithRawCallbackMethod_MethodA2() { + ::grpc::Service::experimental().MarkMethodRawCallback(1, + new ::grpc::internal::CallbackClientStreamingHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this] { return this->MethodA2(); })); + } + ~ExperimentalWithRawCallbackMethod_MethodA2() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA2(::grpc::ServerContext* context, ::grpc::ServerReader< ::grpc::testing::Request>* reader, ::grpc::testing::Response* response) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual ::grpc::experimental::ServerReadReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* MethodA2() { + return new ::grpc::internal::UnimplementedReadReactor< + ::grpc::ByteBuffer, ::grpc::ByteBuffer>;} + }; + template <class BaseClass> + class ExperimentalWithRawCallbackMethod_MethodA3 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithRawCallbackMethod_MethodA3() { + ::grpc::Service::experimental().MarkMethodRawCallback(2, + new ::grpc::internal::CallbackServerStreamingHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this] { return this->MethodA3(); })); + } + ~ExperimentalWithRawCallbackMethod_MethodA3() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA3(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::ServerWriter< ::grpc::testing::Response>* writer) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual ::grpc::experimental::ServerWriteReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* MethodA3() { + return new ::grpc::internal::UnimplementedWriteReactor< + ::grpc::ByteBuffer, ::grpc::ByteBuffer>;} + }; + template <class BaseClass> + class ExperimentalWithRawCallbackMethod_MethodA4 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithRawCallbackMethod_MethodA4() { + ::grpc::Service::experimental().MarkMethodRawCallback(3, + new ::grpc::internal::CallbackBidiHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this] { return this->MethodA4(); })); + } + ~ExperimentalWithRawCallbackMethod_MethodA4() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodA4(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::grpc::testing::Response, ::grpc::testing::Request>* stream) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual ::grpc::experimental::ServerBidiReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* MethodA4() { + return new ::grpc::internal::UnimplementedBidiReactor< + ::grpc::ByteBuffer, ::grpc::ByteBuffer>;} + }; + template <class BaseClass> class WithStreamedUnaryMethod_MethodA1 : public BaseClass { private: void BaseClassMustBeDerivedFromService(const Service *service) {} @@ -591,6 +782,32 @@ class ServiceB final { }; typedef WithAsyncMethod_MethodB1<Service > AsyncService; template <class BaseClass> + class ExperimentalWithCallbackMethod_MethodB1 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithCallbackMethod_MethodB1() { + ::grpc::Service::experimental().MarkMethodCallback(0, + new ::grpc::internal::CallbackUnaryHandler< ::grpc::testing::Request, ::grpc::testing::Response>( + [this](::grpc::ServerContext* context, + const ::grpc::testing::Request* request, + ::grpc::testing::Response* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + return this->MethodB1(context, request, response, controller); + })); + } + ~ExperimentalWithCallbackMethod_MethodB1() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual void MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + }; + typedef ExperimentalWithCallbackMethod_MethodB1<Service > ExperimentalCallbackService; + template <class BaseClass> class WithGenericMethod_MethodB1 : public BaseClass { private: void BaseClassMustBeDerivedFromService(const Service *service) {} @@ -628,6 +845,31 @@ class ServiceB final { } }; template <class BaseClass> + class ExperimentalWithRawCallbackMethod_MethodB1 : public BaseClass { + private: + void BaseClassMustBeDerivedFromService(const Service *service) {} + public: + ExperimentalWithRawCallbackMethod_MethodB1() { + ::grpc::Service::experimental().MarkMethodRawCallback(0, + new ::grpc::internal::CallbackUnaryHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>( + [this](::grpc::ServerContext* context, + const ::grpc::ByteBuffer* request, + ::grpc::ByteBuffer* response, + ::grpc::experimental::ServerCallbackRpcController* controller) { + this->MethodB1(context, request, response, controller); + })); + } + ~ExperimentalWithRawCallbackMethod_MethodB1() override { + BaseClassMustBeDerivedFromService(this); + } + // disable synchronous version of this method + ::grpc::Status MethodB1(::grpc::ServerContext* context, const ::grpc::testing::Request* request, ::grpc::testing::Response* response) override { + abort(); + return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, ""); + } + virtual void MethodB1(::grpc::ServerContext* context, const ::grpc::ByteBuffer* request, ::grpc::ByteBuffer* response, ::grpc::experimental::ServerCallbackRpcController* controller) { controller->Finish(::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "")); } + }; + template <class BaseClass> class WithStreamedUnaryMethod_MethodB1 : public BaseClass { private: void BaseClassMustBeDerivedFromService(const Service *service) {} diff --git a/test/cpp/common/alarm_test.cc b/test/cpp/common/alarm_test.cc index e909d03658..802cdc209a 100644 --- a/test/cpp/common/alarm_test.cc +++ b/test/cpp/common/alarm_test.cc @@ -313,7 +313,7 @@ TEST(AlarmTest, UnsetDestruction) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/BUILD b/test/cpp/end2end/BUILD index 0415efc1ef..eb600ffb17 100644 --- a/test/cpp/end2end/BUILD +++ b/test/cpp/end2end/BUILD @@ -35,6 +35,32 @@ grpc_cc_library( ], ) +grpc_cc_library( + name = "test_health_check_service_impl", + testonly = True, + srcs = ["test_health_check_service_impl.cc"], + hdrs = ["test_health_check_service_impl.h"], + deps = [ + "//:grpc", + "//:grpc++", + "//src/proto/grpc/health/v1:health_proto", + ], +) + +grpc_cc_library( + name = "interceptors_util", + testonly = True, + srcs = ["interceptors_util.cc"], + 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"], @@ -110,6 +136,27 @@ grpc_cc_test( "//:grpc", "//:grpc++", "//src/proto/grpc/testing:echo_messages_proto", + "//src/proto/grpc/testing:simple_messages_proto", + "//src/proto/grpc/testing:echo_proto", + "//test/core/util:gpr_test_util", + "//test/core/util:grpc_test_util", + "//test/cpp/util:test_util", + ], +) + +grpc_cc_test( + name = "client_interceptors_end2end_test", + srcs = ["client_interceptors_end2end_test.cc"], + external_deps = [ + "gtest", + ], + deps = [ + ":interceptors_util", + ":test_service_impl", + "//:gpr", + "//:grpc", + "//:grpc++", + "//src/proto/grpc/testing:echo_messages_proto", "//src/proto/grpc/testing:echo_proto", "//test/core/util:gpr_test_util", "//test/core/util:grpc_test_util", @@ -125,6 +172,7 @@ grpc_cc_library( "gtest", ], deps = [ + ":interceptors_util", ":test_service_impl", "//:gpr", "//:grpc", @@ -247,6 +295,7 @@ grpc_cc_test( "gtest", ], deps = [ + ":test_health_check_service_impl", ":test_service_impl", "//:gpr", "//:grpc", @@ -470,6 +519,26 @@ grpc_cc_binary( ) grpc_cc_test( + name = "server_interceptors_end2end_test", + srcs = ["server_interceptors_end2end_test.cc"], + external_deps = [ + "gtest", + ], + deps = [ + ":interceptors_util", + ":test_service_impl", + "//:gpr", + "//:grpc", + "//:grpc++", + "//src/proto/grpc/testing:echo_messages_proto", + "//src/proto/grpc/testing:echo_proto", + "//test/core/util:gpr_test_util", + "//test/core/util:grpc_test_util", + "//test/cpp/util:test_util", + ], +) + +grpc_cc_test( name = "server_load_reporting_end2end_test", srcs = ["server_load_reporting_end2end_test.cc"], external_deps = [ diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 6ecb957801..e09f54dcc3 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -1884,7 +1884,7 @@ int main(int argc, char** argv) { // Change the backup poll interval from 5s to 100ms to speed up the // ReconnectChannel test gpr_setenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS", "100"); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); int ret = RUN_ALL_TESTS(); return ret; diff --git a/test/cpp/end2end/channelz_service_test.cc b/test/cpp/end2end/channelz_service_test.cc index 7a7a716e79..29b59e4e5e 100644 --- a/test/cpp/end2end/channelz_service_test.cc +++ b/test/cpp/end2end/channelz_service_test.cc @@ -41,6 +41,8 @@ using grpc::channelz::v1::GetChannelRequest; using grpc::channelz::v1::GetChannelResponse; +using grpc::channelz::v1::GetServerRequest; +using grpc::channelz::v1::GetServerResponse; using grpc::channelz::v1::GetServerSocketsRequest; using grpc::channelz::v1::GetServerSocketsResponse; using grpc::channelz::v1::GetServersRequest; @@ -460,6 +462,29 @@ TEST_F(ChannelzServerTest, BasicServerTest) { EXPECT_EQ(response.server_size(), 1); } +TEST_F(ChannelzServerTest, BasicGetServerTest) { + ResetStubs(); + ConfigureProxy(1); + GetServersRequest get_servers_request; + GetServersResponse get_servers_response; + get_servers_request.set_start_server_id(0); + ClientContext get_servers_context; + Status s = channelz_stub_->GetServers( + &get_servers_context, get_servers_request, &get_servers_response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); + EXPECT_EQ(get_servers_response.server_size(), 1); + GetServerRequest get_server_request; + GetServerResponse get_server_response; + get_server_request.set_server_id( + get_servers_response.server(0).ref().server_id()); + ClientContext get_server_context; + s = channelz_stub_->GetServer(&get_server_context, get_server_request, + &get_server_response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); + EXPECT_EQ(get_servers_response.server(0).ref().server_id(), + get_server_response.server().ref().server_id()); +} + TEST_F(ChannelzServerTest, ServerCallTest) { ResetStubs(); ConfigureProxy(1); @@ -626,11 +651,33 @@ TEST_F(ChannelzServerTest, GetServerSocketsTest) { EXPECT_EQ(get_server_sockets_response.socket_ref_size(), 1); } +TEST_F(ChannelzServerTest, GetServerListenSocketsTest) { + ResetStubs(); + ConfigureProxy(1); + GetServersRequest get_server_request; + GetServersResponse get_server_response; + get_server_request.set_start_server_id(0); + ClientContext get_server_context; + Status s = channelz_stub_->GetServers(&get_server_context, get_server_request, + &get_server_response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); + EXPECT_EQ(get_server_response.server_size(), 1); + EXPECT_EQ(get_server_response.server(0).listen_socket_size(), 1); + GetSocketRequest get_socket_request; + GetSocketResponse get_socket_response; + get_socket_request.set_socket_id( + get_server_response.server(0).listen_socket(0).socket_id()); + ClientContext get_socket_context; + s = channelz_stub_->GetSocket(&get_socket_context, get_socket_request, + &get_socket_response); + EXPECT_TRUE(s.ok()) << "s.error_message() = " << s.error_message(); +} + } // namespace testing } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/client_callback_end2end_test.cc b/test/cpp/end2end/client_callback_end2end_test.cc index 62a85641c7..a999321992 100644 --- a/test/cpp/end2end/client_callback_end2end_test.cc +++ b/test/cpp/end2end/client_callback_end2end_test.cc @@ -34,6 +34,7 @@ #include "test/core/util/test_config.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> @@ -41,13 +42,38 @@ namespace grpc { namespace testing { namespace { -class ClientCallbackEnd2endTest : public ::testing::Test { +class TestScenario { + public: + TestScenario(bool serve_callback) : callback_server(serve_callback) {} + void Log() const; + bool callback_server; +}; + +static std::ostream& operator<<(std::ostream& out, + const TestScenario& scenario) { + return out << "TestScenario{callback_server=" + << (scenario.callback_server ? "true" : "false") << "}"; +} + +void TestScenario::Log() const { + std::ostringstream out; + out << *this; + gpr_log(GPR_DEBUG, "%s", out.str().c_str()); +} + +class ClientCallbackEnd2endTest + : public ::testing::TestWithParam<TestScenario> { protected: - ClientCallbackEnd2endTest() {} + ClientCallbackEnd2endTest() { GetParam().Log(); } void SetUp() override { ServerBuilder builder; - builder.RegisterService(&service_); + + if (!GetParam().callback_server) { + builder.RegisterService(&service_); + } else { + builder.RegisterService(&callback_service_); + } server_ = builder.BuildAndStart(); is_server_started_ = true; @@ -75,11 +101,13 @@ class ClientCallbackEnd2endTest : public ::testing::Test { test_string += "Hello world. "; request.set_message(test_string); - + grpc::string val; if (with_binary_metadata) { + request.mutable_param()->set_echo_metadata(true); char bytes[8] = {'\0', '\1', '\2', '\3', '\4', '\5', '\6', static_cast<char>(i)}; - cli_ctx.AddMetadata("custom-bin", grpc::string(bytes, 8)); + val = grpc::string(bytes, 8); + cli_ctx.AddMetadata("custom-bin", val); } cli_ctx.set_compression_algorithm(GRPC_COMPRESS_GZIP); @@ -89,10 +117,18 @@ class ClientCallbackEnd2endTest : public ::testing::Test { bool done = false; stub_->experimental_async()->Echo( &cli_ctx, &request, &response, - [&request, &response, &done, &mu, &cv](Status s) { + [&cli_ctx, &request, &response, &done, &mu, &cv, val, + with_binary_metadata](Status s) { GPR_ASSERT(s.ok()); EXPECT_EQ(request.message(), response.message()); + if (with_binary_metadata) { + EXPECT_EQ( + 1u, cli_ctx.GetServerTrailingMetadata().count("custom-bin")); + EXPECT_EQ(val, ToString(cli_ctx.GetServerTrailingMetadata() + .find("custom-bin") + ->second)); + } std::lock_guard<std::mutex> l(mu); done = true; cv.notify_one(); @@ -146,42 +182,145 @@ class ClientCallbackEnd2endTest : public ::testing::Test { } } + void SendGenericEchoAsBidi(int num_rpcs, int reuses) { + const grpc::string kMethodName("/grpc.testing.EchoTestService/Echo"); + grpc::string test_string(""); + for (int i = 0; i < num_rpcs; i++) { + test_string += "Hello world. "; + class Client : public grpc::experimental::ClientBidiReactor<ByteBuffer, + ByteBuffer> { + public: + Client(ClientCallbackEnd2endTest* test, const grpc::string& method_name, + const grpc::string& test_str, int reuses) + : reuses_remaining_(reuses) { + activate_ = [this, test, method_name, test_str] { + if (reuses_remaining_ > 0) { + cli_ctx_.reset(new ClientContext); + reuses_remaining_--; + test->generic_stub_->experimental().PrepareBidiStreamingCall( + cli_ctx_.get(), method_name, this); + request_.set_message(test_str); + send_buf_ = SerializeToByteBuffer(&request_); + StartWrite(send_buf_.get()); + StartRead(&recv_buf_); + StartCall(); + } else { + std::unique_lock<std::mutex> l(mu_); + done_ = true; + cv_.notify_one(); + } + }; + activate_(); + } + void OnWriteDone(bool ok) override { StartWritesDone(); } + void OnReadDone(bool ok) override { + EchoResponse response; + EXPECT_TRUE(ParseFromByteBuffer(&recv_buf_, &response)); + EXPECT_EQ(request_.message(), response.message()); + }; + void OnDone(const Status& s) override { + EXPECT_TRUE(s.ok()); + activate_(); + } + void Await() { + std::unique_lock<std::mutex> l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + EchoRequest request_; + std::unique_ptr<ByteBuffer> send_buf_; + ByteBuffer recv_buf_; + std::unique_ptr<ClientContext> cli_ctx_; + int reuses_remaining_; + std::function<void()> activate_; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } rpc{this, kMethodName, test_string, reuses}; + + rpc.Await(); + } + } bool is_server_started_; std::shared_ptr<Channel> channel_; std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_; std::unique_ptr<grpc::GenericStub> generic_stub_; TestServiceImpl service_; + CallbackTestServiceImpl callback_service_; std::unique_ptr<Server> server_; }; -TEST_F(ClientCallbackEnd2endTest, SimpleRpc) { +TEST_P(ClientCallbackEnd2endTest, SimpleRpc) { ResetStub(); SendRpcs(1, false); } -TEST_F(ClientCallbackEnd2endTest, SequentialRpcs) { +TEST_P(ClientCallbackEnd2endTest, SequentialRpcs) { ResetStub(); SendRpcs(10, false); } -TEST_F(ClientCallbackEnd2endTest, SequentialRpcsWithVariedBinaryMetadataValue) { +TEST_P(ClientCallbackEnd2endTest, SendClientInitialMetadata) { + ResetStub(); + SimpleRequest request; + SimpleResponse response; + ClientContext cli_ctx; + + cli_ctx.AddMetadata(kCheckClientInitialMetadataKey, + kCheckClientInitialMetadataVal); + + std::mutex mu; + std::condition_variable cv; + bool done = false; + stub_->experimental_async()->CheckClientInitialMetadata( + &cli_ctx, &request, &response, [&done, &mu, &cv](Status s) { + GPR_ASSERT(s.ok()); + + std::lock_guard<std::mutex> l(mu); + done = true; + cv.notify_one(); + }); + std::unique_lock<std::mutex> l(mu); + while (!done) { + cv.wait(l); + } +} + +TEST_P(ClientCallbackEnd2endTest, SimpleRpcWithBinaryMetadata) { + ResetStub(); + SendRpcs(1, true); +} + +TEST_P(ClientCallbackEnd2endTest, SequentialRpcsWithVariedBinaryMetadataValue) { ResetStub(); SendRpcs(10, true); } -TEST_F(ClientCallbackEnd2endTest, SequentialGenericRpcs) { +TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcs) { ResetStub(); SendRpcsGeneric(10, false); } +TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcsAsBidi) { + ResetStub(); + SendGenericEchoAsBidi(10, 1); +} + +TEST_P(ClientCallbackEnd2endTest, SequentialGenericRpcsAsBidiWithReactorReuse) { + ResetStub(); + SendGenericEchoAsBidi(10, 10); +} + #if GRPC_ALLOW_EXCEPTIONS -TEST_F(ClientCallbackEnd2endTest, ExceptingRpc) { +TEST_P(ClientCallbackEnd2endTest, ExceptingRpc) { ResetStub(); SendRpcsGeneric(10, true); } #endif -TEST_F(ClientCallbackEnd2endTest, MultipleRpcsWithVariedBinaryMetadataValue) { +TEST_P(ClientCallbackEnd2endTest, MultipleRpcsWithVariedBinaryMetadataValue) { ResetStub(); std::vector<std::thread> threads; threads.reserve(10); @@ -193,7 +332,7 @@ TEST_F(ClientCallbackEnd2endTest, MultipleRpcsWithVariedBinaryMetadataValue) { } } -TEST_F(ClientCallbackEnd2endTest, MultipleRpcs) { +TEST_P(ClientCallbackEnd2endTest, MultipleRpcs) { ResetStub(); std::vector<std::thread> threads; threads.reserve(10); @@ -205,7 +344,7 @@ TEST_F(ClientCallbackEnd2endTest, MultipleRpcs) { } } -TEST_F(ClientCallbackEnd2endTest, CancelRpcBeforeStart) { +TEST_P(ClientCallbackEnd2endTest, CancelRpcBeforeStart) { ResetStub(); EchoRequest request; EchoResponse response; @@ -230,12 +369,167 @@ TEST_F(ClientCallbackEnd2endTest, CancelRpcBeforeStart) { } } +TEST_P(ClientCallbackEnd2endTest, RequestStream) { + ResetStub(); + class Client : public grpc::experimental::ClientWriteReactor<EchoRequest> { + public: + explicit Client(grpc::testing::EchoTestService::Stub* stub) { + context_.set_initial_metadata_corked(true); + stub->experimental_async()->RequestStream(&context_, &response_, this); + StartCall(); + request_.set_message("Hello server."); + StartWrite(&request_); + } + void OnWriteDone(bool ok) override { + writes_left_--; + if (writes_left_ > 1) { + StartWrite(&request_); + } else if (writes_left_ == 1) { + StartWriteLast(&request_, WriteOptions()); + } + } + void OnDone(const Status& s) override { + EXPECT_TRUE(s.ok()); + EXPECT_EQ(response_.message(), "Hello server.Hello server.Hello server."); + std::unique_lock<std::mutex> l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock<std::mutex> l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + private: + EchoRequest request_; + EchoResponse response_; + ClientContext context_; + int writes_left_{3}; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } test{stub_.get()}; + + test.Await(); +} + +TEST_P(ClientCallbackEnd2endTest, ResponseStream) { + ResetStub(); + class Client : public grpc::experimental::ClientReadReactor<EchoResponse> { + public: + explicit Client(grpc::testing::EchoTestService::Stub* stub) { + request_.set_message("Hello client "); + stub->experimental_async()->ResponseStream(&context_, &request_, this); + StartCall(); + StartRead(&response_); + } + void OnReadDone(bool ok) override { + if (!ok) { + EXPECT_EQ(reads_complete_, kServerDefaultResponseStreamsToSend); + } else { + EXPECT_LE(reads_complete_, kServerDefaultResponseStreamsToSend); + EXPECT_EQ(response_.message(), + request_.message() + grpc::to_string(reads_complete_)); + reads_complete_++; + StartRead(&response_); + } + } + void OnDone(const Status& s) override { + EXPECT_TRUE(s.ok()); + std::unique_lock<std::mutex> l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock<std::mutex> l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + private: + EchoRequest request_; + EchoResponse response_; + ClientContext context_; + int reads_complete_{0}; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } test{stub_.get()}; + + test.Await(); +} + +TEST_P(ClientCallbackEnd2endTest, BidiStream) { + ResetStub(); + class Client : public grpc::experimental::ClientBidiReactor<EchoRequest, + EchoResponse> { + public: + explicit Client(grpc::testing::EchoTestService::Stub* stub) { + request_.set_message("Hello fren "); + stub->experimental_async()->BidiStream(&context_, this); + StartCall(); + StartRead(&response_); + StartWrite(&request_); + } + void OnReadDone(bool ok) override { + if (!ok) { + EXPECT_EQ(reads_complete_, kServerDefaultResponseStreamsToSend); + } else { + EXPECT_LE(reads_complete_, kServerDefaultResponseStreamsToSend); + EXPECT_EQ(response_.message(), request_.message()); + reads_complete_++; + StartRead(&response_); + } + } + void OnWriteDone(bool ok) override { + EXPECT_TRUE(ok); + if (++writes_complete_ == kServerDefaultResponseStreamsToSend) { + StartWritesDone(); + } else { + StartWrite(&request_); + } + } + void OnDone(const Status& s) override { + EXPECT_TRUE(s.ok()); + std::unique_lock<std::mutex> l(mu_); + done_ = true; + cv_.notify_one(); + } + void Await() { + std::unique_lock<std::mutex> l(mu_); + while (!done_) { + cv_.wait(l); + } + } + + private: + EchoRequest request_; + EchoResponse response_; + ClientContext context_; + int reads_complete_{0}; + int writes_complete_{0}; + std::mutex mu_; + std::condition_variable cv_; + bool done_ = false; + } test{stub_.get()}; + + test.Await(); +} + +TestScenario scenarios[] = {TestScenario{false}, TestScenario{true}}; + +INSTANTIATE_TEST_CASE_P(ClientCallbackEnd2endTest, ClientCallbackEnd2endTest, + ::testing::ValuesIn(scenarios)); + } // namespace } // namespace testing } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/client_crash_test.cc b/test/cpp/end2end/client_crash_test.cc index 2a06f44c60..992f3c488f 100644 --- a/test/cpp/end2end/client_crash_test.cc +++ b/test/cpp/end2end/client_crash_test.cc @@ -135,7 +135,7 @@ int main(int argc, char** argv) { g_root = "."; } - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); // Order seems to matter on these tests: run three times to eliminate that for (int i = 0; i < 3; i++) { diff --git a/test/cpp/end2end/client_interceptors_end2end_test.cc b/test/cpp/end2end/client_interceptors_end2end_test.cc new file mode 100644 index 0000000000..968b5d49d1 --- /dev/null +++ b/test/cpp/end2end/client_interceptors_end2end_test.cc @@ -0,0 +1,651 @@ +/* + * + * 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 <memory> +#include <vector> + +#include <grpcpp/channel.h> +#include <grpcpp/client_context.h> +#include <grpcpp/create_channel.h> +#include <grpcpp/generic/generic_stub.h> +#include <grpcpp/impl/codegen/client_interceptor.h> +#include <grpcpp/impl/codegen/proto_utils.h> +#include <grpcpp/server.h> +#include <grpcpp/server_builder.h> +#include <grpcpp/server_context.h> + +#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> + +namespace grpc { +namespace testing { +namespace { + +/* Hijacks Echo RPC and fills in the expected values */ +class HijackingInterceptor : public experimental::Interceptor { + public: + HijackingInterceptor(experimental::ClientRpcInfo* info) { + info_ = info; + // Make sure it is the right method + EXPECT_EQ(strcmp("/grpc.testing.EchoTestService/Echo", info->method()), 0); + EXPECT_EQ(info->type(), experimental::ClientRpcInfo::Type::UNARY); + } + + virtual void Intercept(experimental::InterceptorBatchMethods* methods) { + bool hijack = false; + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) { + auto* map = methods->GetSendInitialMetadata(); + // Check that we can see the test metadata + ASSERT_EQ(map->size(), static_cast<unsigned>(1)); + auto iterator = map->begin(); + EXPECT_EQ("testkey", iterator->first); + EXPECT_EQ("testvalue", iterator->second); + hijack = true; + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) { + EchoRequest req; + auto* buffer = methods->GetSendMessage(); + auto copied_buffer = *buffer; + EXPECT_TRUE( + SerializationTraits<EchoRequest>::Deserialize(&copied_buffer, &req) + .ok()); + EXPECT_EQ(req.message(), "Hello"); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_CLOSE)) { + // Got nothing to do here for now + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA)) { + auto* map = methods->GetRecvInitialMetadata(); + // Got nothing better to do here for now + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_MESSAGE)) { + EchoResponse* resp = + static_cast<EchoResponse*>(methods->GetRecvMessage()); + // Check that we got the hijacked message, and re-insert the expected + // message + EXPECT_EQ(resp->message(), "Hello1"); + resp->set_message("Hello"); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_STATUS)) { + auto* map = methods->GetRecvTrailingMetadata(); + bool found = false; + // Check that we received the metadata as an echo + for (const auto& pair : *map) { + found = pair.first.starts_with("testkey") && + pair.second.starts_with("testvalue"); + if (found) break; + } + EXPECT_EQ(found, true); + auto* status = methods->GetRecvStatus(); + EXPECT_EQ(status->ok(), true); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_INITIAL_METADATA)) { + auto* map = methods->GetRecvInitialMetadata(); + // Got nothing better to do here at the moment + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_MESSAGE)) { + // Insert a different message than expected + EchoResponse* resp = + static_cast<EchoResponse*>(methods->GetRecvMessage()); + resp->set_message("Hello1"); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_STATUS)) { + auto* map = methods->GetRecvTrailingMetadata(); + // insert the metadata that we want + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + map->insert(std::make_pair("testkey", "testvalue")); + auto* status = methods->GetRecvStatus(); + *status = Status(StatusCode::OK, ""); + } + if (hijack) { + methods->Hijack(); + } else { + methods->Proceed(); + } + } + + private: + experimental::ClientRpcInfo* info_; +}; + +class HijackingInterceptorFactory + : public experimental::ClientInterceptorFactoryInterface { + public: + virtual experimental::Interceptor* CreateClientInterceptor( + experimental::ClientRpcInfo* info) override { + return new HijackingInterceptor(info); + } +}; + +class HijackingInterceptorMakesAnotherCall : public experimental::Interceptor { + public: + HijackingInterceptorMakesAnotherCall(experimental::ClientRpcInfo* info) { + info_ = info; + // Make sure it is the right method + EXPECT_EQ(strcmp("/grpc.testing.EchoTestService/Echo", info->method()), 0); + } + + virtual void Intercept(experimental::InterceptorBatchMethods* methods) { + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) { + auto* map = methods->GetSendInitialMetadata(); + // Check that we can see the test metadata + ASSERT_EQ(map->size(), static_cast<unsigned>(1)); + auto iterator = map->begin(); + EXPECT_EQ("testkey", iterator->first); + EXPECT_EQ("testvalue", iterator->second); + // Make a copy of the map + metadata_map_ = *map; + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) { + EchoRequest req; + auto* buffer = methods->GetSendMessage(); + auto copied_buffer = *buffer; + EXPECT_TRUE( + SerializationTraits<EchoRequest>::Deserialize(&copied_buffer, &req) + .ok()); + EXPECT_EQ(req.message(), "Hello"); + req_ = req; + stub_ = grpc::testing::EchoTestService::NewStub( + methods->GetInterceptedChannel()); + ctx_.AddMetadata(metadata_map_.begin()->first, + metadata_map_.begin()->second); + stub_->experimental_async()->Echo(&ctx_, &req_, &resp_, + [this, methods](Status s) { + EXPECT_EQ(s.ok(), true); + EXPECT_EQ(resp_.message(), "Hello"); + methods->Hijack(); + }); + // There isn't going to be any other interesting operation in this batch, + // so it is fine to return + return; + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_CLOSE)) { + // Got nothing to do here for now + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA)) { + auto* map = methods->GetRecvInitialMetadata(); + // Got nothing better to do here for now + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_MESSAGE)) { + EchoResponse* resp = + static_cast<EchoResponse*>(methods->GetRecvMessage()); + // Check that we got the hijacked message, and re-insert the expected + // message + EXPECT_EQ(resp->message(), "Hello"); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_STATUS)) { + auto* map = methods->GetRecvTrailingMetadata(); + bool found = false; + // Check that we received the metadata as an echo + for (const auto& pair : *map) { + found = pair.first.starts_with("testkey") && + pair.second.starts_with("testvalue"); + if (found) break; + } + EXPECT_EQ(found, true); + auto* status = methods->GetRecvStatus(); + EXPECT_EQ(status->ok(), true); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_INITIAL_METADATA)) { + auto* map = methods->GetRecvInitialMetadata(); + // Got nothing better to do here at the moment + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_MESSAGE)) { + // Insert a different message than expected + EchoResponse* resp = + static_cast<EchoResponse*>(methods->GetRecvMessage()); + resp->set_message(resp_.message()); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_RECV_STATUS)) { + auto* map = methods->GetRecvTrailingMetadata(); + // insert the metadata that we want + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + map->insert(std::make_pair("testkey", "testvalue")); + auto* status = methods->GetRecvStatus(); + *status = Status(StatusCode::OK, ""); + } + + methods->Proceed(); + } + + private: + experimental::ClientRpcInfo* info_; + std::multimap<grpc::string, grpc::string> metadata_map_; + ClientContext ctx_; + EchoRequest req_; + EchoResponse resp_; + std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_; +}; + +class HijackingInterceptorMakesAnotherCallFactory + : public experimental::ClientInterceptorFactoryInterface { + public: + virtual experimental::Interceptor* CreateClientInterceptor( + experimental::ClientRpcInfo* info) override { + return new HijackingInterceptorMakesAnotherCall(info); + } +}; + +class LoggingInterceptor : public experimental::Interceptor { + public: + LoggingInterceptor(experimental::ClientRpcInfo* info) { info_ = info; } + + virtual void Intercept(experimental::InterceptorBatchMethods* methods) { + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) { + auto* map = methods->GetSendInitialMetadata(); + // Check that we can see the test metadata + ASSERT_EQ(map->size(), static_cast<unsigned>(1)); + auto iterator = map->begin(); + EXPECT_EQ("testkey", iterator->first); + EXPECT_EQ("testvalue", iterator->second); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) { + EchoRequest req; + auto* buffer = methods->GetSendMessage(); + auto copied_buffer = *buffer; + EXPECT_TRUE( + SerializationTraits<EchoRequest>::Deserialize(&copied_buffer, &req) + .ok()); + EXPECT_TRUE(req.message().find("Hello") == 0); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_CLOSE)) { + // Got nothing to do here for now + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA)) { + auto* map = methods->GetRecvInitialMetadata(); + // Got nothing better to do here for now + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_MESSAGE)) { + EchoResponse* resp = + static_cast<EchoResponse*>(methods->GetRecvMessage()); + EXPECT_TRUE(resp->message().find("Hello") == 0); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_STATUS)) { + auto* map = methods->GetRecvTrailingMetadata(); + bool found = false; + // Check that we received the metadata as an echo + for (const auto& pair : *map) { + found = pair.first.starts_with("testkey") && + pair.second.starts_with("testvalue"); + if (found) break; + } + EXPECT_EQ(found, true); + auto* status = methods->GetRecvStatus(); + EXPECT_EQ(status->ok(), true); + } + methods->Proceed(); + } + + private: + experimental::ClientRpcInfo* info_; +}; + +class LoggingInterceptorFactory + : public experimental::ClientInterceptorFactoryInterface { + public: + virtual experimental::Interceptor* CreateClientInterceptor( + experimental::ClientRpcInfo* info) override { + return new LoggingInterceptor(info); + } +}; + +class ClientInterceptorsEnd2endTest : public ::testing::Test { + protected: + ClientInterceptorsEnd2endTest() { + int port = grpc_pick_unused_port_or_die(); + + ServerBuilder builder; + server_address_ = "localhost:" + std::to_string(port); + builder.AddListeningPort(server_address_, InsecureServerCredentials()); + builder.RegisterService(&service_); + server_ = builder.BuildAndStart(); + } + + ~ClientInterceptorsEnd2endTest() { server_->Shutdown(); } + + std::string server_address_; + TestServiceImpl service_; + std::unique_ptr<Server> server_; +}; + +TEST_F(ClientInterceptorsEnd2endTest, ClientInterceptorLoggingTest) { + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + 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)); + MakeCall(channel); + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); +} + +TEST_F(ClientInterceptorsEnd2endTest, ClientInterceptorHijackingTest) { + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + // Add 20 dummy interceptors before hijacking interceptor + creators.reserve(20); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + creators.push_back(std::unique_ptr<HijackingInterceptorFactory>( + new HijackingInterceptorFactory())); + // Add 20 dummy interceptors after hijacking interceptor + 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)); + + MakeCall(channel); + // Make sure only 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); +} + +TEST_F(ClientInterceptorsEnd2endTest, ClientInterceptorLogThenHijackTest) { + ChannelArguments args; + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + creators.push_back(std::unique_ptr<LoggingInterceptorFactory>( + new LoggingInterceptorFactory())); + creators.push_back(std::unique_ptr<HijackingInterceptorFactory>( + new HijackingInterceptorFactory())); + auto channel = experimental::CreateCustomChannelWithInterceptors( + server_address_, InsecureChannelCredentials(), args, std::move(creators)); + + MakeCall(channel); +} + +TEST_F(ClientInterceptorsEnd2endTest, + ClientInterceptorHijackingMakesAnotherCallTest) { + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + // Add 5 dummy interceptors before hijacking interceptor + creators.reserve(5); + for (auto i = 0; i < 5; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + creators.push_back( + std::unique_ptr<experimental::ClientInterceptorFactoryInterface>( + new HijackingInterceptorMakesAnotherCallFactory())); + // Add 7 dummy interceptors after hijacking interceptor + for (auto i = 0; i < 7; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + auto channel = server_->experimental().InProcessChannelWithInterceptors( + args, std::move(creators)); + + MakeCall(channel); + // Make sure all interceptors were run once, since the hijacking interceptor + // makes an RPC on the intercepted channel + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 12); +} + +TEST_F(ClientInterceptorsEnd2endTest, + ClientInterceptorLoggingTestWithCallback) { + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + 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 = server_->experimental().InProcessChannelWithInterceptors( + args, std::move(creators)); + MakeCallbackCall(channel); + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); +} + +class ClientInterceptorsStreamingEnd2endTest : public ::testing::Test { + protected: + ClientInterceptorsStreamingEnd2endTest() { + int port = grpc_pick_unused_port_or_die(); + + ServerBuilder builder; + server_address_ = "localhost:" + std::to_string(port); + builder.AddListeningPort(server_address_, InsecureServerCredentials()); + builder.RegisterService(&service_); + server_ = builder.BuildAndStart(); + } + + ~ClientInterceptorsStreamingEnd2endTest() { server_->Shutdown(); } + + std::string server_address_; + EchoTestServiceStreamingImpl service_; + std::unique_ptr<Server> server_; +}; + +TEST_F(ClientInterceptorsStreamingEnd2endTest, ClientStreamingTest) { + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + 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(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + 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(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + 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)); + MakeBidiStreamingCall(channel); + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); +} + +class ClientGlobalInterceptorEnd2endTest : public ::testing::Test { + protected: + ClientGlobalInterceptorEnd2endTest() { + int port = grpc_pick_unused_port_or_die(); + + ServerBuilder builder; + server_address_ = "localhost:" + std::to_string(port); + builder.AddListeningPort(server_address_, InsecureServerCredentials()); + builder.RegisterService(&service_); + server_ = builder.BuildAndStart(); + } + + ~ClientGlobalInterceptorEnd2endTest() { server_->Shutdown(); } + + std::string server_address_; + TestServiceImpl service_; + std::unique_ptr<Server> server_; +}; + +TEST_F(ClientGlobalInterceptorEnd2endTest, DummyGlobalInterceptor) { + // We should ideally be registering a global interceptor only once per + // process, but for the purposes of testing, it should be fine to modify the + // registered global interceptor when there are no ongoing gRPC operations + DummyInterceptorFactory global_factory; + experimental::RegisterGlobalClientInterceptorFactory(&global_factory); + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + // Add 20 dummy interceptors + creators.reserve(20); + 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)); + MakeCall(channel); + // Make sure all 20 dummy interceptors were run with the global interceptor + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 21); + // Reset the global interceptor. This is again 'safe' because there are no + // other ongoing gRPC operations + experimental::RegisterGlobalClientInterceptorFactory(nullptr); +} + +TEST_F(ClientGlobalInterceptorEnd2endTest, LoggingGlobalInterceptor) { + // We should ideally be registering a global interceptor only once per + // process, but for the purposes of testing, it should be fine to modify the + // registered global interceptor when there are no ongoing gRPC operations + LoggingInterceptorFactory global_factory; + experimental::RegisterGlobalClientInterceptorFactory(&global_factory); + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + // Add 20 dummy interceptors + creators.reserve(20); + 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)); + MakeCall(channel); + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + // Reset the global interceptor. This is again 'safe' because there are no + // other ongoing gRPC operations + experimental::RegisterGlobalClientInterceptorFactory(nullptr); +} + +TEST_F(ClientGlobalInterceptorEnd2endTest, HijackingGlobalInterceptor) { + // We should ideally be registering a global interceptor only once per + // process, but for the purposes of testing, it should be fine to modify the + // registered global interceptor when there are no ongoing gRPC operations + HijackingInterceptorFactory global_factory; + experimental::RegisterGlobalClientInterceptorFactory(&global_factory); + ChannelArguments args; + DummyInterceptor::Reset(); + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + // Add 20 dummy interceptors + creators.reserve(20); + 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)); + MakeCall(channel); + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + // Reset the global interceptor. This is again 'safe' because there are no + // other ongoing gRPC operations + experimental::RegisterGlobalClientInterceptorFactory(nullptr); +} + +} // namespace +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc::testing::TestEnvironment env(argc, argv); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/cpp/end2end/client_lb_end2end_test.cc b/test/cpp/end2end/client_lb_end2end_test.cc index 77c25c1416..1dfa91a76c 100644 --- a/test/cpp/end2end/client_lb_end2end_test.cc +++ b/test/cpp/end2end/client_lb_end2end_test.cc @@ -42,6 +42,9 @@ #include "src/core/lib/gprpp/debug_location.h" #include "src/core/lib/gprpp/ref_counted_ptr.h" #include "src/core/lib/iomgr/tcp_client.h" +#include "src/core/lib/security/credentials/fake/fake_credentials.h" +#include "src/cpp/client/secure_credentials.h" +#include "src/cpp/server/secure_server_credentials.h" #include "src/proto/grpc/testing/echo.grpc.pb.h" #include "test/core/util/port.h" @@ -113,7 +116,10 @@ class MyTestServiceImpl : public TestServiceImpl { class ClientLbEnd2endTest : public ::testing::Test { protected: ClientLbEnd2endTest() - : server_host_("localhost"), kRequestMessage_("Live long and prosper.") { + : server_host_("localhost"), + kRequestMessage_("Live long and prosper."), + creds_(new SecureChannelCredentials( + grpc_fake_transport_security_credentials_create())) { // Make the backup poller poll very frequently in order to pick up // updates from all the subchannels's FDs. gpr_setenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS", "1"); @@ -146,7 +152,7 @@ class ClientLbEnd2endTest : public ::testing::Test { void StartServers(size_t num_servers, std::vector<int> ports = std::vector<int>()) { - CreateServers(num_servers, ports); + CreateServers(num_servers, std::move(ports)); for (size_t i = 0; i < num_servers; ++i) { StartServer(i); } @@ -188,6 +194,11 @@ class ClientLbEnd2endTest : public ::testing::Test { grpc_channel_args_destroy(fake_results); } + void SetFailureOnReresolution() { + grpc_core::ExecCtx exec_ctx; + response_generator_->SetFailureOnReresolution(); + } + std::vector<int> GetServersPorts() { std::vector<int> ports; for (const auto& server : servers_) ports.push_back(server->port_); @@ -207,19 +218,20 @@ class ClientLbEnd2endTest : public ::testing::Test { } // else, default to pick first args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR, response_generator_.get()); - return CreateCustomChannel("fake:///", InsecureChannelCredentials(), args); + return CreateCustomChannel("fake:///", creds_, args); } bool SendRpc( const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub, EchoResponse* response = nullptr, int timeout_ms = 1000, - Status* result = nullptr) { + Status* result = nullptr, bool wait_for_ready = false) { const bool local_response = (response == nullptr); if (local_response) response = new EchoResponse; EchoRequest request; request.set_message(kRequestMessage_); ClientContext context; context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms)); + if (wait_for_ready) context.set_wait_for_ready(true); Status status = stub->Echo(&context, request, response); if (result != nullptr) *result = status; if (local_response) delete response; @@ -228,10 +240,11 @@ class ClientLbEnd2endTest : public ::testing::Test { void CheckRpcSendOk( const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub, - const grpc_core::DebugLocation& location) { + const grpc_core::DebugLocation& location, bool wait_for_ready = false) { EchoResponse response; Status status; - const bool success = SendRpc(stub, &response, 2000, &status); + const bool success = + SendRpc(stub, &response, 2000, &status, wait_for_ready); ASSERT_TRUE(success) << "From " << location.file() << ":" << location.line() << "\n" << "Error: " << status.error_message() << " " @@ -253,6 +266,7 @@ class ClientLbEnd2endTest : public ::testing::Test { MyTestServiceImpl service_; std::unique_ptr<std::thread> thread_; bool server_ready_ = false; + bool started_ = false; explicit ServerData(int port = 0) { port_ = port > 0 ? port : grpc_pick_unused_port_or_die(); @@ -260,6 +274,7 @@ class ClientLbEnd2endTest : public ::testing::Test { void Start(const grpc::string& server_host) { gpr_log(GPR_INFO, "starting server on port %d", port_); + started_ = true; std::mutex mu; std::unique_lock<std::mutex> lock(mu); std::condition_variable cond; @@ -275,8 +290,9 @@ class ClientLbEnd2endTest : public ::testing::Test { std::ostringstream server_address; server_address << server_host << ":" << port_; ServerBuilder builder; - builder.AddListeningPort(server_address.str(), - InsecureServerCredentials()); + std::shared_ptr<ServerCredentials> creds(new SecureServerCredentials( + grpc_fake_transport_security_server_credentials_create())); + builder.AddListeningPort(server_address.str(), std::move(creds)); builder.RegisterService(&service_); server_ = builder.BuildAndStart(); std::lock_guard<std::mutex> lock(*mu); @@ -284,9 +300,11 @@ class ClientLbEnd2endTest : public ::testing::Test { cond->notify_one(); } - void Shutdown(bool join = true) { + void Shutdown() { + if (!started_) return; server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0)); - if (join) thread_->join(); + thread_->join(); + started_ = false; } void SetServingStatus(const grpc::string& service, bool serving) { @@ -306,7 +324,7 @@ class ClientLbEnd2endTest : public ::testing::Test { if (ignore_failure) { SendRpc(stub); } else { - CheckRpcSendOk(stub, location); + CheckRpcSendOk(stub, location, true); } } while (servers_[server_idx]->service_.request_count() == 0); ResetCounters(); @@ -365,6 +383,7 @@ class ClientLbEnd2endTest : public ::testing::Test { grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator> response_generator_; const grpc::string kRequestMessage_; + std::shared_ptr<ChannelCredentials> creds_; }; TEST_F(ClientLbEnd2endTest, PickFirst) { @@ -409,6 +428,30 @@ TEST_F(ClientLbEnd2endTest, PickFirstProcessPending) { CheckRpcSendOk(second_stub, DEBUG_LOCATION); } +TEST_F(ClientLbEnd2endTest, PickFirstSelectsReadyAtStartup) { + ChannelArguments args; + constexpr int kInitialBackOffMs = 5000; + args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs); + // Create 2 servers, but start only the second one. + std::vector<int> ports = {grpc_pick_unused_port_or_die(), + grpc_pick_unused_port_or_die()}; + CreateServers(2, ports); + StartServer(1); + auto channel1 = BuildChannel("pick_first", args); + auto stub1 = BuildStub(channel1); + SetNextResolution(ports); + // Wait for second server to be ready. + WaitForServer(stub1, 1, DEBUG_LOCATION); + // Create a second channel with the same addresses. Its PF instance + // should immediately pick the second subchannel, since it's already + // in READY state. + auto channel2 = BuildChannel("pick_first", args); + SetNextResolution(ports); + // Check that the channel reports READY without waiting for the + // initial backoff. + EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1 /* timeout_seconds */)); +} + TEST_F(ClientLbEnd2endTest, PickFirstBackOffInitialReconnect) { ChannelArguments args; constexpr int kInitialBackOffMs = 100; @@ -494,6 +537,51 @@ TEST_F(ClientLbEnd2endTest, PickFirstResetConnectionBackoff) { EXPECT_LT(waited_ms, kInitialBackOffMs); } +TEST_F(ClientLbEnd2endTest, + PickFirstResetConnectionBackoffNextAttemptStartsImmediately) { + ChannelArguments args; + constexpr int kInitialBackOffMs = 1000; + args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs); + const std::vector<int> ports = {grpc_pick_unused_port_or_die()}; + auto channel = BuildChannel("pick_first", args); + auto stub = BuildStub(channel); + SetNextResolution(ports); + // Wait for connect, which should fail ~immediately, because the server + // is not up. + gpr_log(GPR_INFO, "=== INITIAL CONNECTION ATTEMPT"); + EXPECT_FALSE( + channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10))); + // Reset connection backoff. + // Note that the time at which the third attempt will be started is + // actually computed at this point, so we record the start time here. + gpr_log(GPR_INFO, "=== RESETTING BACKOFF"); + const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC); + experimental::ChannelResetConnectionBackoff(channel.get()); + // Trigger a second connection attempt. This should also fail + // ~immediately, but the retry should be scheduled for + // kInitialBackOffMs instead of applying the multiplier. + gpr_log(GPR_INFO, "=== POLLING FOR SECOND CONNECTION ATTEMPT"); + EXPECT_FALSE( + channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10))); + // Bring up a server on the chosen port. + gpr_log(GPR_INFO, "=== STARTING BACKEND"); + StartServers(1, ports); + // Wait for connect. Should happen within kInitialBackOffMs. + // Give an extra 100ms to account for the time spent in the second and + // third connection attempts themselves (since what we really want to + // measure is the time between the two). As long as this is less than + // the 1.6x increase we would see if the backoff state was not reset + // properly, the test is still proving that the backoff was reset. + constexpr int kWaitMs = kInitialBackOffMs + 100; + gpr_log(GPR_INFO, "=== POLLING FOR THIRD CONNECTION ATTEMPT"); + EXPECT_TRUE(channel->WaitForConnected( + grpc_timeout_milliseconds_to_deadline(kWaitMs))); + const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC); + const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0)); + gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms); + EXPECT_LT(waited_ms, kWaitMs); +} + TEST_F(ClientLbEnd2endTest, PickFirstUpdates) { // Start servers and send one RPC per server. const int kNumServers = 3; @@ -518,7 +606,7 @@ TEST_F(ClientLbEnd2endTest, PickFirstUpdates) { do { channel_state = channel->GetState(true /* try to connect */); } while (channel_state == GRPC_CHANNEL_READY); - GPR_ASSERT(channel_state != GRPC_CHANNEL_READY); + ASSERT_NE(channel_state, GRPC_CHANNEL_READY); servers_[0]->service_.ResetCounters(); // Next update introduces servers_[1], making the channel recover. @@ -720,6 +808,23 @@ TEST_F(ClientLbEnd2endTest, PickFirstCheckStateBeforeStartWatch) { EXPECT_EQ("pick_first", channel_2->GetLoadBalancingPolicyName()); } +TEST_F(ClientLbEnd2endTest, PickFirstIdleOnDisconnect) { + // Start server, send RPC, and make sure channel is READY. + const int kNumServers = 1; + StartServers(kNumServers); + auto channel = BuildChannel(""); // pick_first is the default. + auto stub = BuildStub(channel); + SetNextResolution(GetServersPorts()); + CheckRpcSendOk(stub, DEBUG_LOCATION); + EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY); + // Stop server. Channel should go into state IDLE. + SetFailureOnReresolution(); + servers_[0]->Shutdown(); + EXPECT_TRUE(WaitForChannelNotReady(channel.get())); + EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE); + servers_.clear(); +} + TEST_F(ClientLbEnd2endTest, RoundRobin) { // Start servers and send one RPC per server. const int kNumServers = 3; @@ -835,7 +940,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinUpdates) { do { channel_state = channel->GetState(true /* try to connect */); } while (channel_state == GRPC_CHANNEL_READY); - GPR_ASSERT(channel_state != GRPC_CHANNEL_READY); + ASSERT_NE(channel_state, GRPC_CHANNEL_READY); servers_[0]->service_.ResetCounters(); // Next update introduces servers_[1], making the channel recover. @@ -844,7 +949,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinUpdates) { SetNextResolution(ports); WaitForServer(stub, 1, DEBUG_LOCATION); channel_state = channel->GetState(false /* try to connect */); - GPR_ASSERT(channel_state == GRPC_CHANNEL_READY); + ASSERT_EQ(channel_state, GRPC_CHANNEL_READY); // Check LB policy name for the channel. EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName()); @@ -869,7 +974,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinUpdateInError) { servers_[0]->service_.ResetCounters(); // Shutdown one of the servers to be sent in the update. - servers_[1]->Shutdown(false); + servers_[1]->Shutdown(); ports.emplace_back(servers_[1]->port_); ports.emplace_back(servers_[2]->port_); SetNextResolution(ports); @@ -928,7 +1033,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinReresolve) { // Kill all servers gpr_log(GPR_INFO, "****** ABOUT TO KILL SERVERS *******"); for (size_t i = 0; i < servers_.size(); ++i) { - servers_[i]->Shutdown(true); + servers_[i]->Shutdown(); } gpr_log(GPR_INFO, "****** SERVERS KILLED *******"); gpr_log(GPR_INFO, "****** SENDING DOOMED REQUESTS *******"); @@ -954,7 +1059,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinReresolve) { if (SendRpc(stub)) break; now = gpr_now(GPR_CLOCK_MONOTONIC); } - GPR_ASSERT(gpr_time_cmp(deadline, now) > 0); + ASSERT_GT(gpr_time_cmp(deadline, now), 0); } TEST_F(ClientLbEnd2endTest, RoundRobinSingleReconnect) { @@ -976,7 +1081,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinSingleReconnect) { } const auto pre_death = servers_[0]->service_.request_count(); // Kill the first server. - servers_[0]->Shutdown(true); + servers_[0]->Shutdown(); // Client request still succeed. May need retrying if RR had returned a pick // before noticing the change in the server's connectivity. while (!SendRpc(stub)) { @@ -1123,7 +1228,7 @@ TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingInhibitPerChannel) { int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); const auto result = RUN_ALL_TESTS(); return result; } diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index fc07681535..05cd4330c6 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -40,6 +40,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/string_ref_helper.h" #include "test/cpp/util/test_credentials_provider.h" @@ -179,7 +180,7 @@ class Proxy : public ::grpc::testing::EchoTestService::Service { } private: - std::unique_ptr< ::grpc::testing::EchoTestService::Stub> stub_; + std::unique_ptr<::grpc::testing::EchoTestService::Stub> stub_; }; class TestServiceImplDupPkg @@ -194,19 +195,29 @@ class TestServiceImplDupPkg class TestScenario { public: - TestScenario(bool proxy, bool inproc_stub, const grpc::string& creds_type) - : use_proxy(proxy), inproc(inproc_stub), credentials_type(creds_type) {} + TestScenario(bool interceptors, bool proxy, bool inproc_stub, + const grpc::string& creds_type, bool use_callback_server) + : use_interceptors(interceptors), + use_proxy(proxy), + inproc(inproc_stub), + credentials_type(creds_type), + callback_server(use_callback_server) {} void Log() const; + bool use_interceptors; bool use_proxy; bool inproc; const grpc::string credentials_type; + bool callback_server; }; static std::ostream& operator<<(std::ostream& out, const TestScenario& scenario) { - return out << "TestScenario{use_proxy=" - << (scenario.use_proxy ? "true" : "false") + return out << "TestScenario{use_interceptors=" + << (scenario.use_interceptors ? "true" : "false") + << ", use_proxy=" << (scenario.use_proxy ? "true" : "false") << ", inproc=" << (scenario.inproc ? "true" : "false") + << ", server_type=" + << (scenario.callback_server ? "callback" : "sync") << ", credentials='" << scenario.credentials_type << "'}"; } @@ -260,8 +271,24 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> { if (GetParam().credentials_type != kInsecureCredentialsType) { server_creds->SetAuthMetadataProcessor(processor); } + if (GetParam().use_interceptors) { + std::vector< + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + // Add 20 dummy server interceptors + creators.reserve(20); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + } builder.AddListeningPort(server_address_.str(), server_creds); - builder.RegisterService(&service_); + if (!GetParam().callback_server) { + builder.RegisterService(&service_); + } else { + builder.RegisterService(&callback_service_); + } builder.RegisterService("foo.test.youtube.com", &special_service_); builder.RegisterService(&dup_pkg_service_); @@ -292,10 +319,21 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> { args.SetString(GRPC_ARG_SECONDARY_USER_AGENT_STRING, "end2end_test"); if (!GetParam().inproc) { - channel_ = - CreateCustomChannel(server_address_.str(), channel_creds, args); + if (!GetParam().use_interceptors) { + channel_ = + CreateCustomChannel(server_address_.str(), channel_creds, args); + } else { + channel_ = CreateCustomChannelWithInterceptors( + server_address_.str(), channel_creds, args, + CreateDummyClientInterceptors()); + } } else { - channel_ = server_->InProcessChannel(args); + if (!GetParam().use_interceptors) { + channel_ = server_->InProcessChannel(args); + } else { + channel_ = server_->experimental().InProcessChannelWithInterceptors( + args, CreateDummyClientInterceptors()); + } } } @@ -320,6 +358,7 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> { } stub_ = grpc::testing::EchoTestService::NewStub(channel_); + DummyInterceptor::Reset(); } bool is_server_started_; @@ -331,6 +370,7 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> { std::ostringstream server_address_; const int kMaxMessageSize_; TestServiceImpl service_; + CallbackTestServiceImpl callback_service_; TestServiceImpl special_service_; TestServiceImplDupPkg dup_pkg_service_; grpc::string user_agent_prefix_; @@ -376,6 +416,7 @@ class End2endServerTryCancelTest : public End2endTest { // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL. void TestRequestStreamServerCancel( ServerTryCancelRequestPhase server_try_cancel, int num_msgs_to_send) { + RestartServer(std::shared_ptr<AuthMetadataProcessor>()); ResetStub(); EchoRequest request; EchoResponse response; @@ -432,6 +473,10 @@ class End2endServerTryCancelTest : public End2endTest { EXPECT_FALSE(s.ok()); EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); + // Make sure that the server interceptors were notified + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } // Helper for testing server-streaming RPCs which are cancelled on the server. @@ -449,6 +494,7 @@ class End2endServerTryCancelTest : public End2endTest { // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL. void TestResponseStreamServerCancel( ServerTryCancelRequestPhase server_try_cancel) { + RestartServer(std::shared_ptr<AuthMetadataProcessor>()); ResetStub(); EchoRequest request; EchoResponse response; @@ -508,7 +554,10 @@ class End2endServerTryCancelTest : public End2endTest { } EXPECT_FALSE(s.ok()); - EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); + // Make sure that the server interceptors were notified + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } // Helper for testing bidirectional-streaming RPCs which are cancelled on the @@ -526,6 +575,7 @@ class End2endServerTryCancelTest : public End2endTest { // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL. void TestBidiStreamServerCancel(ServerTryCancelRequestPhase server_try_cancel, int num_messages) { + RestartServer(std::shared_ptr<AuthMetadataProcessor>()); ResetStub(); EchoRequest request; EchoResponse response; @@ -592,6 +642,10 @@ class End2endServerTryCancelTest : public End2endTest { EXPECT_FALSE(s.ok()); EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); + // Make sure that the server interceptors were notified + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } }; @@ -971,7 +1025,8 @@ TEST_P(End2endTest, DiffPackageServices) { EXPECT_TRUE(s.ok()); } -void CancelRpc(ClientContext* context, int delay_us, TestServiceImpl* service) { +template <class ServiceType> +void CancelRpc(ClientContext* context, int delay_us, ServiceType* service) { gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_micros(delay_us, GPR_TIMESPAN))); while (!service->signal_client()) { @@ -989,6 +1044,9 @@ TEST_P(End2endTest, CancelRpcBeforeStart) { Status s = stub_->Echo(&context, request, &response); EXPECT_EQ("", response.message()); EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } // Client cancels request stream after sending two messages @@ -1009,6 +1067,9 @@ TEST_P(End2endTest, ClientCancelsRequestStream) { EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); EXPECT_EQ(response.message(), ""); + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } // Client cancels server stream after sending some messages @@ -1041,6 +1102,9 @@ TEST_P(End2endTest, ClientCancelsResponseStream) { // The final status could be either of CANCELLED or OK depending on // who won the race. EXPECT_GE(grpc::StatusCode::CANCELLED, s.error_code()); + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } // Client cancels bidi stream after sending some messages @@ -1074,6 +1138,9 @@ TEST_P(End2endTest, ClientCancelsBidi) { Status s = stream->Finish(); EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code()); + if (GetParam().use_interceptors) { + EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel()); + } } TEST_P(End2endTest, RpcMaxMessageSize) { @@ -1389,7 +1456,24 @@ TEST_P(ProxyEnd2endTest, ClientCancelsRpc) { request.mutable_param()->set_client_cancel_after_us(kCancelDelayUs); ClientContext context; - std::thread cancel_thread(CancelRpc, &context, kCancelDelayUs, &service_); + std::thread cancel_thread; + if (!GetParam().callback_server) { + cancel_thread = std::thread( + [&context, this](int delay) { CancelRpc(&context, delay, &service_); }, + kCancelDelayUs); + // Note: the unusual pattern above (and below) is caused by a conflict + // between two sets of compiler expectations. clang allows const to be + // captured without mention, so there is no need to capture kCancelDelayUs + // (and indeed clang-tidy complains if you do so). OTOH, a Windows compiler + // in our tests requires an explicit capture even for const. We square this + // circle by passing the const value in as an argument to the lambda. + } else { + cancel_thread = std::thread( + [&context, this](int delay) { + CancelRpc(&context, delay, &callback_service_); + }, + kCancelDelayUs); + } Status s = stub_->Echo(&context, request, &response); cancel_thread.join(); EXPECT_EQ(StatusCode::CANCELLED, s.error_code()); @@ -1781,10 +1865,12 @@ TEST_P(ResourceQuotaEnd2endTest, SimpleRequest) { EXPECT_TRUE(s.ok()); } +// TODO(vjpai): refactor arguments into a struct if it makes sense std::vector<TestScenario> CreateTestScenarios(bool use_proxy, bool test_insecure, bool test_secure, - bool test_inproc) { + bool test_inproc, + bool test_callback_server) { std::vector<TestScenario> scenarios; std::vector<grpc::string> credentials_types; if (test_secure) { @@ -1800,38 +1886,48 @@ std::vector<TestScenario> CreateTestScenarios(bool use_proxy, if (test_insecure && insec_ok()) { credentials_types.push_back(kInsecureCredentialsType); } + + // For now test callback server only with inproc GPR_ASSERT(!credentials_types.empty()); for (const auto& cred : credentials_types) { - scenarios.emplace_back(false, false, cred); + scenarios.emplace_back(false, false, false, cred, false); + scenarios.emplace_back(true, false, false, cred, false); if (use_proxy) { - scenarios.emplace_back(true, false, cred); + scenarios.emplace_back(false, true, false, cred, false); + scenarios.emplace_back(true, true, false, cred, false); } } if (test_inproc && insec_ok()) { - scenarios.emplace_back(false, true, kInsecureCredentialsType); + scenarios.emplace_back(false, false, true, kInsecureCredentialsType, false); + scenarios.emplace_back(true, false, true, kInsecureCredentialsType, false); + if (test_callback_server) { + scenarios.emplace_back(false, false, true, kInsecureCredentialsType, + true); + scenarios.emplace_back(true, false, true, kInsecureCredentialsType, true); + } } return scenarios; } -INSTANTIATE_TEST_CASE_P(End2end, End2endTest, - ::testing::ValuesIn(CreateTestScenarios(false, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + End2end, End2endTest, + ::testing::ValuesIn(CreateTestScenarios(false, true, true, true, true))); -INSTANTIATE_TEST_CASE_P(End2endServerTryCancel, End2endServerTryCancelTest, - ::testing::ValuesIn(CreateTestScenarios(false, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + End2endServerTryCancel, End2endServerTryCancelTest, + ::testing::ValuesIn(CreateTestScenarios(false, true, true, true, true))); -INSTANTIATE_TEST_CASE_P(ProxyEnd2end, ProxyEnd2endTest, - ::testing::ValuesIn(CreateTestScenarios(true, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + ProxyEnd2end, ProxyEnd2endTest, + ::testing::ValuesIn(CreateTestScenarios(true, true, true, true, false))); -INSTANTIATE_TEST_CASE_P(SecureEnd2end, SecureEnd2endTest, - ::testing::ValuesIn(CreateTestScenarios(false, false, - true, false))); +INSTANTIATE_TEST_CASE_P( + SecureEnd2end, SecureEnd2endTest, + ::testing::ValuesIn(CreateTestScenarios(false, false, true, false, true))); -INSTANTIATE_TEST_CASE_P(ResourceQuotaEnd2end, ResourceQuotaEnd2endTest, - ::testing::ValuesIn(CreateTestScenarios(false, true, - true, true))); +INSTANTIATE_TEST_CASE_P( + ResourceQuotaEnd2end, ResourceQuotaEnd2endTest, + ::testing::ValuesIn(CreateTestScenarios(false, true, true, true, false))); } // namespace } // namespace testing @@ -1839,7 +1935,7 @@ INSTANTIATE_TEST_CASE_P(ResourceQuotaEnd2end, ResourceQuotaEnd2endTest, int main(int argc, char** argv) { gpr_setenv("GRPC_CLIENT_CHANNEL_BACKUP_POLL_INTERVAL_MS", "200"); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/exception_test.cc b/test/cpp/end2end/exception_test.cc index 5343997663..0d2c00263b 100644 --- a/test/cpp/end2end/exception_test.cc +++ b/test/cpp/end2end/exception_test.cc @@ -117,7 +117,7 @@ TEST_F(ExceptionTest, RequestStream) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/filter_end2end_test.cc b/test/cpp/end2end/filter_end2end_test.cc index 88f8f380c3..ad67402e3d 100644 --- a/test/cpp/end2end/filter_end2end_test.cc +++ b/test/cpp/end2end/filter_end2end_test.cc @@ -331,7 +331,7 @@ void RegisterFilter() { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); grpc::testing::RegisterFilter(); return RUN_ALL_TESTS(); diff --git a/test/cpp/end2end/generic_end2end_test.cc b/test/cpp/end2end/generic_end2end_test.cc index 88a1227ca2..015862bfe8 100644 --- a/test/cpp/end2end/generic_end2end_test.cc +++ b/test/cpp/end2end/generic_end2end_test.cc @@ -333,7 +333,7 @@ TEST_F(GenericEnd2endTest, Deadline) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/grpclb_end2end_test.cc b/test/cpp/end2end/grpclb_end2end_test.cc index b69b861fcf..9c4cd05061 100644 --- a/test/cpp/end2end/grpclb_end2end_test.cc +++ b/test/cpp/end2end/grpclb_end2end_test.cc @@ -539,22 +539,25 @@ class GrpclbEnd2endTest : public ::testing::Test { balancers_.at(i)->add_response(response, delay_ms); } - Status SendRpc(EchoResponse* response = nullptr, int timeout_ms = 1000) { + Status SendRpc(EchoResponse* response = nullptr, int timeout_ms = 1000, + bool wait_for_ready = false) { const bool local_response = (response == nullptr); if (local_response) response = new EchoResponse; EchoRequest request; request.set_message(kRequestMessage_); ClientContext context; context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms)); + if (wait_for_ready) context.set_wait_for_ready(true); Status status = stub_->Echo(&context, request, response); if (local_response) delete response; return status; } - void CheckRpcSendOk(const size_t times = 1, const int timeout_ms = 1000) { + void CheckRpcSendOk(const size_t times = 1, const int timeout_ms = 1000, + bool wait_for_ready = false) { for (size_t i = 0; i < times; ++i) { EchoResponse response; - const Status status = SendRpc(&response, timeout_ms); + const Status status = SendRpc(&response, timeout_ms, wait_for_ready); EXPECT_TRUE(status.ok()) << "code=" << status.error_code() << " message=" << status.error_message(); EXPECT_EQ(response.message(), kRequestMessage_); @@ -715,10 +718,9 @@ TEST_F(SingleBalancerTest, InitiallyEmptyServerlist) { ScheduleResponseForBalancer( 0, BalancerServiceImpl::BuildResponseForBackends(GetBackendPorts(), {}), kServerlistDelayMs); - const auto t0 = system_clock::now(); // Client will block: LB will initially send empty serverlist. - CheckRpcSendOk(1, kCallDeadlineMs); + CheckRpcSendOk(1, kCallDeadlineMs, true /* wait_for_ready */); const auto ellapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>( system_clock::now() - t0); @@ -1366,7 +1368,7 @@ TEST_F(SingleBalancerTest, DropAllFirst) { {}, {{"rate_limiting", num_of_drop_by_rate_limiting_addresses}, {"load_balancing", num_of_drop_by_load_balancing_addresses}}), 0); - const Status status = SendRpc(); + const Status status = SendRpc(nullptr, 1000, true); EXPECT_FALSE(status.ok()); EXPECT_EQ(status.error_message(), "Call dropped by load balancing policy"); } @@ -1391,7 +1393,7 @@ TEST_F(SingleBalancerTest, DropAll) { // fail. Status status; do { - status = SendRpc(); + status = SendRpc(nullptr, 1000, true); } while (status.ok()); EXPECT_FALSE(status.ok()); EXPECT_EQ(status.error_message(), "Call dropped by load balancing policy"); @@ -1516,7 +1518,7 @@ TEST_F(SingleBalancerWithClientLoadReportingTest, Drop) { int main(int argc, char** argv) { grpc_init(); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); const auto result = RUN_ALL_TESTS(); grpc_shutdown(); diff --git a/test/cpp/end2end/health_service_end2end_test.cc b/test/cpp/end2end/health_service_end2end_test.cc index fca65dfc13..b96ff53a3e 100644 --- a/test/cpp/end2end/health_service_end2end_test.cc +++ b/test/cpp/end2end/health_service_end2end_test.cc @@ -37,6 +37,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/test_health_check_service_impl.h" #include "test/cpp/end2end/test_service_impl.h" #include <gtest/gtest.h> @@ -49,62 +50,6 @@ namespace grpc { namespace testing { namespace { -// A sample sync implementation of the health checking service. This does the -// same thing as the default one. -class HealthCheckServiceImpl : public ::grpc::health::v1::Health::Service { - public: - Status Check(ServerContext* context, const HealthCheckRequest* request, - HealthCheckResponse* response) override { - std::lock_guard<std::mutex> lock(mu_); - auto iter = status_map_.find(request->service()); - if (iter == status_map_.end()) { - return Status(StatusCode::NOT_FOUND, ""); - } - response->set_status(iter->second); - return Status::OK; - } - - Status Watch(ServerContext* context, const HealthCheckRequest* request, - ::grpc::ServerWriter<HealthCheckResponse>* writer) override { - auto last_state = HealthCheckResponse::UNKNOWN; - while (!context->IsCancelled()) { - { - std::lock_guard<std::mutex> lock(mu_); - HealthCheckResponse response; - auto iter = status_map_.find(request->service()); - if (iter == status_map_.end()) { - response.set_status(response.SERVICE_UNKNOWN); - } else { - response.set_status(iter->second); - } - if (response.status() != last_state) { - writer->Write(response, ::grpc::WriteOptions()); - } - } - gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), - gpr_time_from_millis(1000, GPR_TIMESPAN))); - } - return Status::OK; - } - - void SetStatus(const grpc::string& service_name, - HealthCheckResponse::ServingStatus status) { - std::lock_guard<std::mutex> lock(mu_); - status_map_[service_name] = status; - } - - void SetAll(HealthCheckResponse::ServingStatus status) { - std::lock_guard<std::mutex> lock(mu_); - for (auto iter = status_map_.begin(); iter != status_map_.end(); ++iter) { - iter->second = status; - } - } - - private: - std::mutex mu_; - std::map<const grpc::string, HealthCheckResponse::ServingStatus> status_map_; -}; - // A custom implementation of the health checking service interface. This is // used to test that it prevents the server from creating a default service and // also serves as an example of how to override the default service. @@ -125,6 +70,8 @@ class CustomHealthCheckService : public HealthCheckServiceInterface { : HealthCheckResponse::NOT_SERVING); } + void Shutdown() override { impl_->Shutdown(); } + private: HealthCheckServiceImpl* impl_; // not owned }; @@ -260,6 +207,74 @@ class HealthServiceEnd2endTest : public ::testing::Test { context.TryCancel(); } + // Verify that after HealthCheckServiceInterface::Shutdown is called + // 1. unary client will see NOT_SERVING. + // 2. unary client still sees NOT_SERVING after a SetServing(true) is called. + // 3. streaming (Watch) client will see an update. + // 4. setting a new service to serving after shutdown will add the service + // name but return NOT_SERVING to client. + // This has to be called last. + void VerifyHealthCheckServiceShutdown() { + HealthCheckServiceInterface* service = server_->GetHealthCheckService(); + EXPECT_TRUE(service != nullptr); + const grpc::string kHealthyService("healthy_service"); + const grpc::string kUnhealthyService("unhealthy_service"); + const grpc::string kNotRegisteredService("not_registered"); + const grpc::string kNewService("add_after_shutdown"); + service->SetServingStatus(kHealthyService, true); + service->SetServingStatus(kUnhealthyService, false); + + ResetStubs(); + + // Start Watch for service. + ClientContext context; + HealthCheckRequest request; + request.set_service(kHealthyService); + std::unique_ptr<::grpc::ClientReaderInterface<HealthCheckResponse>> reader = + hc_stub_->Watch(&context, request); + + HealthCheckResponse response; + EXPECT_TRUE(reader->Read(&response)); + EXPECT_EQ(response.SERVING, response.status()); + + SendHealthCheckRpc("", Status::OK, HealthCheckResponse::SERVING); + SendHealthCheckRpc(kHealthyService, Status::OK, + HealthCheckResponse::SERVING); + SendHealthCheckRpc(kUnhealthyService, Status::OK, + HealthCheckResponse::NOT_SERVING); + SendHealthCheckRpc(kNotRegisteredService, + Status(StatusCode::NOT_FOUND, "")); + SendHealthCheckRpc(kNewService, Status(StatusCode::NOT_FOUND, "")); + + // Shutdown health check service. + service->Shutdown(); + + // Watch client gets another update. + EXPECT_TRUE(reader->Read(&response)); + EXPECT_EQ(response.NOT_SERVING, response.status()); + // Finish Watch call. + context.TryCancel(); + + SendHealthCheckRpc("", Status::OK, HealthCheckResponse::NOT_SERVING); + SendHealthCheckRpc(kHealthyService, Status::OK, + HealthCheckResponse::NOT_SERVING); + SendHealthCheckRpc(kUnhealthyService, Status::OK, + HealthCheckResponse::NOT_SERVING); + SendHealthCheckRpc(kNotRegisteredService, + Status(StatusCode::NOT_FOUND, "")); + + // Setting status after Shutdown has no effect. + service->SetServingStatus(kHealthyService, true); + SendHealthCheckRpc(kHealthyService, Status::OK, + HealthCheckResponse::NOT_SERVING); + + // Adding serving status for a new service after shutdown will return + // NOT_SERVING. + service->SetServingStatus(kNewService, true); + SendHealthCheckRpc(kNewService, Status::OK, + HealthCheckResponse::NOT_SERVING); + } + TestServiceImpl echo_test_service_; HealthCheckServiceImpl health_check_service_impl_; std::unique_ptr<Health::Stub> hc_stub_; @@ -295,6 +310,13 @@ TEST_F(HealthServiceEnd2endTest, DefaultHealthService) { Status(StatusCode::INVALID_ARGUMENT, "")); } +TEST_F(HealthServiceEnd2endTest, DefaultHealthServiceShutdown) { + EnableDefaultHealthCheckService(true); + EXPECT_TRUE(DefaultHealthCheckServiceEnabled()); + SetUpServer(true, false, false, nullptr); + VerifyHealthCheckServiceShutdown(); +} + // Provide an empty service to disable the default service. TEST_F(HealthServiceEnd2endTest, ExplicitlyDisableViaOverride) { EnableDefaultHealthCheckService(true); @@ -326,12 +348,27 @@ TEST_F(HealthServiceEnd2endTest, ExplicitlyOverride) { VerifyHealthCheckServiceStreaming(); } +TEST_F(HealthServiceEnd2endTest, ExplicitlyHealthServiceShutdown) { + EnableDefaultHealthCheckService(true); + EXPECT_TRUE(DefaultHealthCheckServiceEnabled()); + std::unique_ptr<HealthCheckServiceInterface> override_service( + new CustomHealthCheckService(&health_check_service_impl_)); + HealthCheckServiceInterface* underlying_service = override_service.get(); + SetUpServer(false, false, true, std::move(override_service)); + HealthCheckServiceInterface* service = server_->GetHealthCheckService(); + EXPECT_TRUE(service == underlying_service); + + ResetStubs(); + + VerifyHealthCheckServiceShutdown(); +} + } // namespace } // namespace testing } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/hybrid_end2end_test.cc b/test/cpp/end2end/hybrid_end2end_test.cc index 339eadde92..18bb1ff4b9 100644 --- a/test/cpp/end2end/hybrid_end2end_test.cc +++ b/test/cpp/end2end/hybrid_end2end_test.cc @@ -894,7 +894,7 @@ TEST_F(HybridEnd2endTest, GenericMethodWithoutGenericService) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/interceptors_util.cc b/test/cpp/end2end/interceptors_util.cc new file mode 100644 index 0000000000..e0ad7d1526 --- /dev/null +++ b/test/cpp/end2end/interceptors_util.cc @@ -0,0 +1,149 @@ +/* + * + * 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 "test/cpp/end2end/interceptors_util.h" + +namespace grpc { +namespace testing { + +std::atomic<int> DummyInterceptor::num_times_run_; +std::atomic<int> DummyInterceptor::num_times_run_reverse_; +std::atomic<int> DummyInterceptor::num_times_cancel_; + +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); + } +} + +bool CheckMetadata(const std::multimap<grpc::string_ref, grpc::string_ref>& map, + const string& key, const string& value) { + for (const auto& pair : map) { + if (pair.first.starts_with(key) && pair.second.starts_with(value)) { + return true; + } + } + return false; +} + +std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> +CreateDummyClientInterceptors() { + std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> + creators; + // Add 20 dummy interceptors before hijacking interceptor + creators.reserve(20); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + return creators; +} + +} // namespace testing +} // namespace grpc diff --git a/test/cpp/end2end/interceptors_util.h b/test/cpp/end2end/interceptors_util.h new file mode 100644 index 0000000000..d886e32494 --- /dev/null +++ b/test/cpp/end2end/interceptors_util.h @@ -0,0 +1,278 @@ +/* + * + * 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 <condition_variable> + +#include <grpcpp/channel.h> + +#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 { +/* This interceptor does nothing. Just keeps a global count on the number of + * times it was invoked. */ +class DummyInterceptor : public experimental::Interceptor { + public: + DummyInterceptor() {} + + virtual void Intercept(experimental::InterceptorBatchMethods* methods) { + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) { + num_times_run_++; + } else if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints:: + POST_RECV_INITIAL_METADATA)) { + num_times_run_reverse_++; + } else if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_CANCEL)) { + num_times_cancel_++; + } + methods->Proceed(); + } + + static void Reset() { + num_times_run_.store(0); + num_times_run_reverse_.store(0); + num_times_cancel_.store(0); + } + + static int GetNumTimesRun() { + EXPECT_EQ(num_times_run_.load(), num_times_run_reverse_.load()); + return num_times_run_.load(); + } + + static int GetNumTimesCancel() { return num_times_cancel_.load(); } + + private: + static std::atomic<int> num_times_run_; + static std::atomic<int> num_times_run_reverse_; + static std::atomic<int> num_times_cancel_; +}; + +class DummyInterceptorFactory + : public experimental::ClientInterceptorFactoryInterface, + public experimental::ServerInterceptorFactoryInterface { + public: + virtual experimental::Interceptor* CreateClientInterceptor( + experimental::ClientRpcInfo* info) override { + return new DummyInterceptor(); + } + + virtual experimental::Interceptor* CreateServerInterceptor( + experimental::ServerRpcInfo* info) override { + return new DummyInterceptor(); + } +}; + +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); + +void MakeClientStreamingCall(const std::shared_ptr<Channel>& channel); + +void MakeServerStreamingCall(const std::shared_ptr<Channel>& channel); + +void MakeBidiStreamingCall(const std::shared_ptr<Channel>& channel); + +void MakeCallbackCall(const std::shared_ptr<Channel>& channel); + +bool CheckMetadata(const std::multimap<grpc::string_ref, grpc::string_ref>& map, + const string& key, const string& value); + +std::vector<std::unique_ptr<experimental::ClientInterceptorFactoryInterface>> +CreateDummyClientInterceptors(); + +inline void* tag(int i) { return (void*)static_cast<intptr_t>(i); } +inline int detag(void* p) { + return static_cast<int>(reinterpret_cast<intptr_t>(p)); +} + +class Verifier { + public: + Verifier() : lambda_run_(false) {} + // Expect sets the expected ok value for a specific tag + Verifier& Expect(int i, bool expect_ok) { + return ExpectUnless(i, expect_ok, false); + } + // ExpectUnless sets the expected ok value for a specific tag + // unless the tag was already marked seen (as a result of ExpectMaybe) + Verifier& ExpectUnless(int i, bool expect_ok, bool seen) { + if (!seen) { + expectations_[tag(i)] = expect_ok; + } + return *this; + } + // ExpectMaybe sets the expected ok value for a specific tag, but does not + // require it to appear + // If it does, sets *seen to true + Verifier& ExpectMaybe(int i, bool expect_ok, bool* seen) { + if (!*seen) { + maybe_expectations_[tag(i)] = MaybeExpect{expect_ok, seen}; + } + return *this; + } + + // Next waits for 1 async tag to complete, checks its + // expectations, and returns the tag + int Next(CompletionQueue* cq, bool ignore_ok) { + bool ok; + void* got_tag; + EXPECT_TRUE(cq->Next(&got_tag, &ok)); + GotTag(got_tag, ok, ignore_ok); + return detag(got_tag); + } + + template <typename T> + CompletionQueue::NextStatus DoOnceThenAsyncNext( + CompletionQueue* cq, void** got_tag, bool* ok, T deadline, + std::function<void(void)> lambda) { + if (lambda_run_) { + return cq->AsyncNext(got_tag, ok, deadline); + } else { + lambda_run_ = true; + return cq->DoThenAsyncNext(lambda, got_tag, ok, deadline); + } + } + + // Verify keeps calling Next until all currently set + // expected tags are complete + void Verify(CompletionQueue* cq) { Verify(cq, false); } + + // This version of Verify allows optionally ignoring the + // outcome of the expectation + void Verify(CompletionQueue* cq, bool ignore_ok) { + GPR_ASSERT(!expectations_.empty() || !maybe_expectations_.empty()); + while (!expectations_.empty()) { + Next(cq, ignore_ok); + } + } + + // This version of Verify stops after a certain deadline, and uses the + // DoThenAsyncNext API + // to call the lambda + void Verify(CompletionQueue* cq, + std::chrono::system_clock::time_point deadline, + const std::function<void(void)>& lambda) { + if (expectations_.empty()) { + bool ok; + void* got_tag; + EXPECT_EQ(DoOnceThenAsyncNext(cq, &got_tag, &ok, deadline, lambda), + CompletionQueue::TIMEOUT); + } else { + while (!expectations_.empty()) { + bool ok; + void* got_tag; + EXPECT_EQ(DoOnceThenAsyncNext(cq, &got_tag, &ok, deadline, lambda), + CompletionQueue::GOT_EVENT); + GotTag(got_tag, ok, false); + } + } + } + + private: + void GotTag(void* got_tag, bool ok, bool ignore_ok) { + auto it = expectations_.find(got_tag); + if (it != expectations_.end()) { + if (!ignore_ok) { + EXPECT_EQ(it->second, ok); + } + expectations_.erase(it); + } else { + auto it2 = maybe_expectations_.find(got_tag); + if (it2 != maybe_expectations_.end()) { + if (it2->second.seen != nullptr) { + EXPECT_FALSE(*it2->second.seen); + *it2->second.seen = true; + } + if (!ignore_ok) { + EXPECT_EQ(it2->second.ok, ok); + } + } else { + gpr_log(GPR_ERROR, "Unexpected tag: %p", got_tag); + abort(); + } + } + } + + struct MaybeExpect { + bool ok; + bool* seen; + }; + + std::map<void*, bool> expectations_; + std::map<void*, MaybeExpect> maybe_expectations_; + bool lambda_run_; +}; + +} // namespace testing +} // namespace grpc diff --git a/test/cpp/end2end/mock_test.cc b/test/cpp/end2end/mock_test.cc index ba3122c895..917ca28020 100644 --- a/test/cpp/end2end/mock_test.cc +++ b/test/cpp/end2end/mock_test.cc @@ -345,7 +345,7 @@ TEST_F(MockTest, BidiStream) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/nonblocking_test.cc b/test/cpp/end2end/nonblocking_test.cc index d8337baca2..36dea1fcb3 100644 --- a/test/cpp/end2end/nonblocking_test.cc +++ b/test/cpp/end2end/nonblocking_test.cc @@ -187,7 +187,7 @@ int main(int argc, char** argv) { grpc_poll_function = maybe_assert_non_blocking_poll; #endif // GRPC_POSIX_SOCKET - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); int ret = RUN_ALL_TESTS(); return ret; diff --git a/test/cpp/end2end/proto_server_reflection_test.cc b/test/cpp/end2end/proto_server_reflection_test.cc index 21a275ef62..ff097aa9a7 100644 --- a/test/cpp/end2end/proto_server_reflection_test.cc +++ b/test/cpp/end2end/proto_server_reflection_test.cc @@ -144,7 +144,7 @@ TEST_F(ProtoServerReflectionTest, CheckResponseWithLocalDescriptorPool) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/raw_end2end_test.cc b/test/cpp/end2end/raw_end2end_test.cc index a413905ef7..c8556bae95 100644 --- a/test/cpp/end2end/raw_end2end_test.cc +++ b/test/cpp/end2end/raw_end2end_test.cc @@ -363,7 +363,7 @@ TEST_F(RawEnd2EndTest, CompileTest) { int main(int argc, char** argv) { // Change the backup poll interval from 5s to 100ms to speed up the // ReconnectChannel test - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); int ret = RUN_ALL_TESTS(); return ret; diff --git a/test/cpp/end2end/server_builder_plugin_test.cc b/test/cpp/end2end/server_builder_plugin_test.cc index d54523fcbb..d744a93912 100644 --- a/test/cpp/end2end/server_builder_plugin_test.cc +++ b/test/cpp/end2end/server_builder_plugin_test.cc @@ -264,7 +264,7 @@ INSTANTIATE_TEST_CASE_P(ServerBuilderPluginTest, ServerBuilderPluginTest, } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/server_crash_test.cc b/test/cpp/end2end/server_crash_test.cc index 93257b2705..353ebf713a 100644 --- a/test/cpp/end2end/server_crash_test.cc +++ b/test/cpp/end2end/server_crash_test.cc @@ -153,7 +153,7 @@ int main(int argc, char** argv) { g_root = "."; } - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/server_early_return_test.cc b/test/cpp/end2end/server_early_return_test.cc index 8948e5b854..c47e25052e 100644 --- a/test/cpp/end2end/server_early_return_test.cc +++ b/test/cpp/end2end/server_early_return_test.cc @@ -226,7 +226,7 @@ TEST_F(ServerEarlyReturnTest, RequestStreamEarlyCancel) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/server_interceptors_end2end_test.cc b/test/cpp/end2end/server_interceptors_end2end_test.cc new file mode 100644 index 0000000000..9460a7d6c6 --- /dev/null +++ b/test/cpp/end2end/server_interceptors_end2end_test.cc @@ -0,0 +1,610 @@ +/* + * + * 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 <memory> +#include <vector> + +#include <grpcpp/channel.h> +#include <grpcpp/client_context.h> +#include <grpcpp/create_channel.h> +#include <grpcpp/generic/generic_stub.h> +#include <grpcpp/impl/codegen/proto_utils.h> +#include <grpcpp/impl/codegen/server_interceptor.h> +#include <grpcpp/server.h> +#include <grpcpp/server_builder.h> +#include <grpcpp/server_context.h> + +#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 <gtest/gtest.h> + +namespace grpc { +namespace testing { +namespace { + +class LoggingInterceptor : public experimental::Interceptor { + public: + LoggingInterceptor(experimental::ServerRpcInfo* info) { + info_ = info; + + // Check the method name and compare to the type + const char* method = info->method(); + experimental::ServerRpcInfo::Type type = info->type(); + + // Check that we use one of our standard methods with expected type. + // Also allow the health checking service. + // We accept BIDI_STREAMING for Echo in case it's an AsyncGenericService + // being tested (the GenericRpc test). + // The empty method is for the Unimplemented requests that arise + // when draining the CQ. + EXPECT_TRUE( + strstr(method, "/grpc.health") == method || + (strcmp(method, "/grpc.testing.EchoTestService/Echo") == 0 && + (type == experimental::ServerRpcInfo::Type::UNARY || + type == experimental::ServerRpcInfo::Type::BIDI_STREAMING)) || + (strcmp(method, "/grpc.testing.EchoTestService/RequestStream") == 0 && + type == experimental::ServerRpcInfo::Type::CLIENT_STREAMING) || + (strcmp(method, "/grpc.testing.EchoTestService/ResponseStream") == 0 && + type == experimental::ServerRpcInfo::Type::SERVER_STREAMING) || + (strcmp(method, "/grpc.testing.EchoTestService/BidiStream") == 0 && + type == experimental::ServerRpcInfo::Type::BIDI_STREAMING) || + strcmp(method, "/grpc.testing.EchoTestService/Unimplemented") == 0 || + (strcmp(method, "") == 0 && + type == experimental::ServerRpcInfo::Type::BIDI_STREAMING)); + } + + virtual void Intercept(experimental::InterceptorBatchMethods* methods) { + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_INITIAL_METADATA)) { + auto* map = methods->GetSendInitialMetadata(); + // Got nothing better to do here for now + EXPECT_EQ(map->size(), static_cast<unsigned>(0)); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_MESSAGE)) { + EchoRequest req; + auto* buffer = methods->GetSendMessage(); + auto copied_buffer = *buffer; + EXPECT_TRUE( + SerializationTraits<EchoRequest>::Deserialize(&copied_buffer, &req) + .ok()); + EXPECT_TRUE(req.message().find("Hello") == 0); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::PRE_SEND_STATUS)) { + auto* map = methods->GetSendTrailingMetadata(); + bool found = false; + // Check that we received the metadata as an echo + for (const auto& pair : *map) { + found = pair.first.find("testkey") == 0 && + pair.second.find("testvalue") == 0; + if (found) break; + } + EXPECT_EQ(found, true); + auto status = methods->GetSendStatus(); + EXPECT_EQ(status.ok(), true); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_INITIAL_METADATA)) { + auto* map = methods->GetRecvInitialMetadata(); + bool found = false; + // Check that we received the metadata as an echo + for (const auto& pair : *map) { + found = pair.first.find("testkey") == 0 && + pair.second.find("testvalue") == 0; + if (found) break; + } + EXPECT_EQ(found, true); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_MESSAGE)) { + EchoResponse* resp = + static_cast<EchoResponse*>(methods->GetRecvMessage()); + EXPECT_TRUE(resp->message().find("Hello") == 0); + } + if (methods->QueryInterceptionHookPoint( + experimental::InterceptionHookPoints::POST_RECV_CLOSE)) { + // Got nothing interesting to do here + } + methods->Proceed(); + } + + private: + experimental::ServerRpcInfo* info_; +}; + +class LoggingInterceptorFactory + : public experimental::ServerInterceptorFactoryInterface { + public: + virtual experimental::Interceptor* CreateServerInterceptor( + experimental::ServerRpcInfo* info) override { + return new LoggingInterceptor(info); + } +}; + +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); +} + +class ServerInterceptorsEnd2endSyncUnaryTest : public ::testing::Test { + protected: + ServerInterceptorsEnd2endSyncUnaryTest() { + int port = grpc_pick_unused_port_or_die(); + + ServerBuilder builder; + server_address_ = "localhost:" + std::to_string(port); + builder.AddListeningPort(server_address_, InsecureServerCredentials()); + builder.RegisterService(&service_); + + std::vector< + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.push_back( + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>( + new LoggingInterceptorFactory())); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + server_ = builder.BuildAndStart(); + } + std::string server_address_; + TestServiceImpl service_; + std::unique_ptr<Server> server_; +}; + +TEST_F(ServerInterceptorsEnd2endSyncUnaryTest, UnaryTest) { + ChannelArguments args; + DummyInterceptor::Reset(); + auto channel = CreateChannel(server_address_, InsecureChannelCredentials()); + MakeCall(channel); + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); +} + +class ServerInterceptorsEnd2endSyncStreamingTest : public ::testing::Test { + protected: + ServerInterceptorsEnd2endSyncStreamingTest() { + int port = grpc_pick_unused_port_or_die(); + + ServerBuilder builder; + server_address_ = "localhost:" + std::to_string(port); + builder.AddListeningPort(server_address_, InsecureServerCredentials()); + builder.RegisterService(&service_); + + std::vector< + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.push_back( + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>( + new LoggingInterceptorFactory())); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + server_ = builder.BuildAndStart(); + } + std::string server_address_; + EchoTestServiceStreamingImpl service_; + std::unique_ptr<Server> server_; +}; + +TEST_F(ServerInterceptorsEnd2endSyncStreamingTest, ClientStreamingTest) { + ChannelArguments args; + DummyInterceptor::Reset(); + auto channel = CreateChannel(server_address_, InsecureChannelCredentials()); + 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); +} + +class ServerInterceptorsAsyncEnd2endTest : public ::testing::Test {}; + +TEST_F(ServerInterceptorsAsyncEnd2endTest, UnaryTest) { + DummyInterceptor::Reset(); + int port = grpc_pick_unused_port_or_die(); + string server_address = "localhost:" + std::to_string(port); + ServerBuilder builder; + EchoTestService::AsyncService service; + builder.AddListeningPort(server_address, InsecureServerCredentials()); + builder.RegisterService(&service); + std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.push_back( + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>( + new LoggingInterceptorFactory())); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + auto cq = builder.AddCompletionQueue(); + auto server = builder.BuildAndStart(); + + ChannelArguments args; + auto channel = CreateChannel(server_address, InsecureChannelCredentials()); + auto stub = grpc::testing::EchoTestService::NewStub(channel); + + EchoRequest send_request; + EchoRequest recv_request; + EchoResponse send_response; + EchoResponse recv_response; + Status recv_status; + + ClientContext cli_ctx; + ServerContext srv_ctx; + grpc::ServerAsyncResponseWriter<EchoResponse> response_writer(&srv_ctx); + + send_request.set_message("Hello"); + cli_ctx.AddMetadata("testkey", "testvalue"); + std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader( + stub->AsyncEcho(&cli_ctx, send_request, cq.get())); + + service.RequestEcho(&srv_ctx, &recv_request, &response_writer, cq.get(), + cq.get(), tag(2)); + + response_reader->Finish(&recv_response, &recv_status, tag(4)); + + Verifier().Expect(2, true).Verify(cq.get()); + EXPECT_EQ(send_request.message(), recv_request.message()); + + EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue")); + srv_ctx.AddTrailingMetadata("testkey", "testvalue"); + + send_response.set_message(recv_request.message()); + response_writer.Finish(send_response, Status::OK, tag(3)); + Verifier().Expect(3, true).Expect(4, true).Verify(cq.get()); + + EXPECT_EQ(send_response.message(), recv_response.message()); + EXPECT_TRUE(recv_status.ok()); + EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey", + "testvalue")); + + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + + server->Shutdown(); + cq->Shutdown(); + void* ignored_tag; + bool ignored_ok; + while (cq->Next(&ignored_tag, &ignored_ok)) + ; + grpc_recycle_unused_port(port); +} + +TEST_F(ServerInterceptorsAsyncEnd2endTest, BidiStreamingTest) { + DummyInterceptor::Reset(); + int port = grpc_pick_unused_port_or_die(); + string server_address = "localhost:" + std::to_string(port); + ServerBuilder builder; + EchoTestService::AsyncService service; + builder.AddListeningPort(server_address, InsecureServerCredentials()); + builder.RegisterService(&service); + std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.push_back( + std::unique_ptr<experimental::ServerInterceptorFactoryInterface>( + new LoggingInterceptorFactory())); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + auto cq = builder.AddCompletionQueue(); + auto server = builder.BuildAndStart(); + + ChannelArguments args; + auto channel = CreateChannel(server_address, InsecureChannelCredentials()); + auto stub = grpc::testing::EchoTestService::NewStub(channel); + + EchoRequest send_request; + EchoRequest recv_request; + EchoResponse send_response; + EchoResponse recv_response; + Status recv_status; + + ClientContext cli_ctx; + ServerContext srv_ctx; + grpc::ServerAsyncReaderWriter<EchoResponse, EchoRequest> srv_stream(&srv_ctx); + + send_request.set_message("Hello"); + cli_ctx.AddMetadata("testkey", "testvalue"); + std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse>> + cli_stream(stub->AsyncBidiStream(&cli_ctx, cq.get(), tag(1))); + + service.RequestBidiStream(&srv_ctx, &srv_stream, cq.get(), cq.get(), tag(2)); + + Verifier().Expect(1, true).Expect(2, true).Verify(cq.get()); + + EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue")); + srv_ctx.AddTrailingMetadata("testkey", "testvalue"); + + cli_stream->Write(send_request, tag(3)); + srv_stream.Read(&recv_request, tag(4)); + Verifier().Expect(3, true).Expect(4, true).Verify(cq.get()); + EXPECT_EQ(send_request.message(), recv_request.message()); + + send_response.set_message(recv_request.message()); + srv_stream.Write(send_response, tag(5)); + cli_stream->Read(&recv_response, tag(6)); + Verifier().Expect(5, true).Expect(6, true).Verify(cq.get()); + EXPECT_EQ(send_response.message(), recv_response.message()); + + cli_stream->WritesDone(tag(7)); + srv_stream.Read(&recv_request, tag(8)); + Verifier().Expect(7, true).Expect(8, false).Verify(cq.get()); + + srv_stream.Finish(Status::OK, tag(9)); + cli_stream->Finish(&recv_status, tag(10)); + Verifier().Expect(9, true).Expect(10, true).Verify(cq.get()); + + EXPECT_TRUE(recv_status.ok()); + EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey", + "testvalue")); + + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + + server->Shutdown(); + cq->Shutdown(); + void* ignored_tag; + bool ignored_ok; + while (cq->Next(&ignored_tag, &ignored_ok)) + ; + grpc_recycle_unused_port(port); +} + +TEST_F(ServerInterceptorsAsyncEnd2endTest, GenericRPCTest) { + DummyInterceptor::Reset(); + int port = grpc_pick_unused_port_or_die(); + string server_address = "localhost:" + std::to_string(port); + ServerBuilder builder; + AsyncGenericService service; + builder.AddListeningPort(server_address, InsecureServerCredentials()); + builder.RegisterAsyncGenericService(&service); + std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.reserve(20); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + auto cq = builder.AddCompletionQueue(); + auto server = builder.BuildAndStart(); + + ChannelArguments args; + auto channel = CreateChannel(server_address, InsecureChannelCredentials()); + GenericStub generic_stub(channel); + + const grpc::string kMethodName("/grpc.cpp.test.util.EchoTestService/Echo"); + EchoRequest send_request; + EchoRequest recv_request; + EchoResponse send_response; + EchoResponse recv_response; + Status recv_status; + + ClientContext cli_ctx; + GenericServerContext srv_ctx; + GenericServerAsyncReaderWriter stream(&srv_ctx); + + // The string needs to be long enough to test heap-based slice. + send_request.set_message("Hello"); + cli_ctx.AddMetadata("testkey", "testvalue"); + + std::unique_ptr<GenericClientAsyncReaderWriter> call = + generic_stub.PrepareCall(&cli_ctx, kMethodName, cq.get()); + call->StartCall(tag(1)); + Verifier().Expect(1, true).Verify(cq.get()); + std::unique_ptr<ByteBuffer> send_buffer = + SerializeToByteBuffer(&send_request); + call->Write(*send_buffer, tag(2)); + // Send ByteBuffer can be destroyed after calling Write. + send_buffer.reset(); + Verifier().Expect(2, true).Verify(cq.get()); + call->WritesDone(tag(3)); + Verifier().Expect(3, true).Verify(cq.get()); + + service.RequestCall(&srv_ctx, &stream, cq.get(), cq.get(), tag(4)); + + Verifier().Expect(4, true).Verify(cq.get()); + EXPECT_EQ(kMethodName, srv_ctx.method()); + EXPECT_TRUE(CheckMetadata(srv_ctx.client_metadata(), "testkey", "testvalue")); + srv_ctx.AddTrailingMetadata("testkey", "testvalue"); + + ByteBuffer recv_buffer; + stream.Read(&recv_buffer, tag(5)); + Verifier().Expect(5, true).Verify(cq.get()); + EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_request)); + EXPECT_EQ(send_request.message(), recv_request.message()); + + send_response.set_message(recv_request.message()); + send_buffer = SerializeToByteBuffer(&send_response); + stream.Write(*send_buffer, tag(6)); + send_buffer.reset(); + Verifier().Expect(6, true).Verify(cq.get()); + + stream.Finish(Status::OK, tag(7)); + Verifier().Expect(7, true).Verify(cq.get()); + + recv_buffer.Clear(); + call->Read(&recv_buffer, tag(8)); + Verifier().Expect(8, true).Verify(cq.get()); + EXPECT_TRUE(ParseFromByteBuffer(&recv_buffer, &recv_response)); + + call->Finish(&recv_status, tag(9)); + Verifier().Expect(9, true).Verify(cq.get()); + + EXPECT_EQ(send_response.message(), recv_response.message()); + EXPECT_TRUE(recv_status.ok()); + EXPECT_TRUE(CheckMetadata(cli_ctx.GetServerTrailingMetadata(), "testkey", + "testvalue")); + + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + + server->Shutdown(); + cq->Shutdown(); + void* ignored_tag; + bool ignored_ok; + while (cq->Next(&ignored_tag, &ignored_ok)) + ; + grpc_recycle_unused_port(port); +} + +TEST_F(ServerInterceptorsAsyncEnd2endTest, UnimplementedRpcTest) { + DummyInterceptor::Reset(); + int port = grpc_pick_unused_port_or_die(); + string server_address = "localhost:" + std::to_string(port); + ServerBuilder builder; + builder.AddListeningPort(server_address, InsecureServerCredentials()); + std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.reserve(20); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + auto cq = builder.AddCompletionQueue(); + auto server = builder.BuildAndStart(); + + ChannelArguments args; + std::shared_ptr<Channel> channel = + CreateChannel(server_address, InsecureChannelCredentials()); + std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub; + stub = grpc::testing::UnimplementedEchoService::NewStub(channel); + EchoRequest send_request; + EchoResponse recv_response; + Status recv_status; + + ClientContext cli_ctx; + send_request.set_message("Hello"); + std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader( + stub->AsyncUnimplemented(&cli_ctx, send_request, cq.get())); + + response_reader->Finish(&recv_response, &recv_status, tag(4)); + Verifier().Expect(4, true).Verify(cq.get()); + + EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code()); + EXPECT_EQ("", recv_status.error_message()); + + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + + server->Shutdown(); + cq->Shutdown(); + void* ignored_tag; + bool ignored_ok; + while (cq->Next(&ignored_tag, &ignored_ok)) + ; + grpc_recycle_unused_port(port); +} + +class ServerInterceptorsSyncUnimplementedEnd2endTest : public ::testing::Test { +}; + +TEST_F(ServerInterceptorsSyncUnimplementedEnd2endTest, UnimplementedRpcTest) { + DummyInterceptor::Reset(); + int port = grpc_pick_unused_port_or_die(); + string server_address = "localhost:" + std::to_string(port); + ServerBuilder builder; + TestServiceImpl service; + builder.RegisterService(&service); + builder.AddListeningPort(server_address, InsecureServerCredentials()); + std::vector<std::unique_ptr<experimental::ServerInterceptorFactoryInterface>> + creators; + creators.reserve(20); + for (auto i = 0; i < 20; i++) { + creators.push_back(std::unique_ptr<DummyInterceptorFactory>( + new DummyInterceptorFactory())); + } + builder.experimental().SetInterceptorCreators(std::move(creators)); + auto server = builder.BuildAndStart(); + + ChannelArguments args; + std::shared_ptr<Channel> channel = + CreateChannel(server_address, InsecureChannelCredentials()); + std::unique_ptr<grpc::testing::UnimplementedEchoService::Stub> stub; + stub = grpc::testing::UnimplementedEchoService::NewStub(channel); + EchoRequest send_request; + EchoResponse recv_response; + + ClientContext cli_ctx; + send_request.set_message("Hello"); + Status recv_status = + stub->Unimplemented(&cli_ctx, send_request, &recv_response); + + EXPECT_EQ(StatusCode::UNIMPLEMENTED, recv_status.error_code()); + EXPECT_EQ("", recv_status.error_message()); + + // Make sure all 20 dummy interceptors were run + EXPECT_EQ(DummyInterceptor::GetNumTimesRun(), 20); + + server->Shutdown(); + grpc_recycle_unused_port(port); +} + +} // namespace +} // namespace testing +} // namespace grpc + +int main(int argc, char** argv) { + grpc::testing::TestEnvironment env(argc, argv); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} diff --git a/test/cpp/end2end/shutdown_test.cc b/test/cpp/end2end/shutdown_test.cc index a53de691bc..da42178d67 100644 --- a/test/cpp/end2end/shutdown_test.cc +++ b/test/cpp/end2end/shutdown_test.cc @@ -164,7 +164,7 @@ TEST_P(ShutdownTest, ShutdownTest) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/streaming_throughput_test.cc b/test/cpp/end2end/streaming_throughput_test.cc index 898d1ec118..440656588b 100644 --- a/test/cpp/end2end/streaming_throughput_test.cc +++ b/test/cpp/end2end/streaming_throughput_test.cc @@ -187,7 +187,7 @@ TEST_F(End2endTest, StreamingThroughput) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/end2end/test_health_check_service_impl.cc b/test/cpp/end2end/test_health_check_service_impl.cc new file mode 100644 index 0000000000..0801e30199 --- /dev/null +++ b/test/cpp/end2end/test_health_check_service_impl.cc @@ -0,0 +1,97 @@ +/* + * + * 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 "test/cpp/end2end/test_health_check_service_impl.h" + +#include <grpc/grpc.h> + +using grpc::health::v1::HealthCheckRequest; +using grpc::health::v1::HealthCheckResponse; + +namespace grpc { +namespace testing { + +Status HealthCheckServiceImpl::Check(ServerContext* context, + const HealthCheckRequest* request, + HealthCheckResponse* response) { + std::lock_guard<std::mutex> lock(mu_); + auto iter = status_map_.find(request->service()); + if (iter == status_map_.end()) { + return Status(StatusCode::NOT_FOUND, ""); + } + response->set_status(iter->second); + return Status::OK; +} + +Status HealthCheckServiceImpl::Watch( + ServerContext* context, const HealthCheckRequest* request, + ::grpc::ServerWriter<HealthCheckResponse>* writer) { + auto last_state = HealthCheckResponse::UNKNOWN; + while (!context->IsCancelled()) { + { + std::lock_guard<std::mutex> lock(mu_); + HealthCheckResponse response; + auto iter = status_map_.find(request->service()); + if (iter == status_map_.end()) { + response.set_status(response.SERVICE_UNKNOWN); + } else { + response.set_status(iter->second); + } + if (response.status() != last_state) { + writer->Write(response, ::grpc::WriteOptions()); + } + } + gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), + gpr_time_from_millis(1000, GPR_TIMESPAN))); + } + return Status::OK; +} + +void HealthCheckServiceImpl::SetStatus( + const grpc::string& service_name, + HealthCheckResponse::ServingStatus status) { + std::lock_guard<std::mutex> lock(mu_); + if (shutdown_) { + status = HealthCheckResponse::NOT_SERVING; + } + status_map_[service_name] = status; +} + +void HealthCheckServiceImpl::SetAll(HealthCheckResponse::ServingStatus status) { + std::lock_guard<std::mutex> lock(mu_); + if (shutdown_) { + return; + } + for (auto iter = status_map_.begin(); iter != status_map_.end(); ++iter) { + iter->second = status; + } +} + +void HealthCheckServiceImpl::Shutdown() { + std::lock_guard<std::mutex> lock(mu_); + if (shutdown_) { + return; + } + shutdown_ = true; + for (auto iter = status_map_.begin(); iter != status_map_.end(); ++iter) { + iter->second = HealthCheckResponse::NOT_SERVING; + } +} + +} // namespace testing +} // namespace grpc diff --git a/test/cpp/end2end/test_health_check_service_impl.h b/test/cpp/end2end/test_health_check_service_impl.h new file mode 100644 index 0000000000..5d36ce5320 --- /dev/null +++ b/test/cpp/end2end/test_health_check_service_impl.h @@ -0,0 +1,58 @@ +/* + * + * 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. + * + */ +#ifndef GRPC_TEST_CPP_END2END_TEST_HEALTH_CHECK_SERVICE_IMPL_H +#define GRPC_TEST_CPP_END2END_TEST_HEALTH_CHECK_SERVICE_IMPL_H + +#include <map> +#include <mutex> + +#include <grpcpp/server_context.h> +#include <grpcpp/support/status.h> + +#include "src/proto/grpc/health/v1/health.grpc.pb.h" + +namespace grpc { +namespace testing { + +// A sample sync implementation of the health checking service. This does the +// same thing as the default one. +class HealthCheckServiceImpl : public health::v1::Health::Service { + public: + Status Check(ServerContext* context, + const health::v1::HealthCheckRequest* request, + health::v1::HealthCheckResponse* response) override; + Status Watch(ServerContext* context, + const health::v1::HealthCheckRequest* request, + ServerWriter<health::v1::HealthCheckResponse>* writer) override; + void SetStatus(const grpc::string& service_name, + health::v1::HealthCheckResponse::ServingStatus status); + void SetAll(health::v1::HealthCheckResponse::ServingStatus status); + + void Shutdown(); + + private: + std::mutex mu_; + bool shutdown_ = false; + std::map<const grpc::string, health::v1::HealthCheckResponse::ServingStatus> + status_map_; +}; + +} // namespace testing +} // namespace grpc + +#endif // GRPC_TEST_CPP_END2END_TEST_HEALTH_CHECK_SERVICE_IMPL_H diff --git a/test/cpp/end2end/test_service_impl.cc b/test/cpp/end2end/test_service_impl.cc index 3c3a5d9cd4..6729ad14f4 100644 --- a/test/cpp/end2end/test_service_impl.cc +++ b/test/cpp/end2end/test_service_impl.cc @@ -69,6 +69,62 @@ void CheckServerAuthContext( EXPECT_EQ(expected_client_identity, identity[0]); } } + +// Returns the number of pairs in metadata that exactly match the given +// key-value pair. Returns -1 if the pair wasn't found. +int MetadataMatchCount( + const std::multimap<grpc::string_ref, grpc::string_ref>& metadata, + const grpc::string& key, const grpc::string& value) { + int count = 0; + for (std::multimap<grpc::string_ref, grpc::string_ref>::const_iterator iter = + metadata.begin(); + iter != metadata.end(); ++iter) { + if (ToString(iter->first) == key && ToString(iter->second) == value) { + count++; + } + } + return count; +} +} // namespace + +namespace { +int GetIntValueFromMetadataHelper( + const char* key, + const std::multimap<grpc::string_ref, grpc::string_ref>& metadata, + int default_value) { + if (metadata.find(key) != metadata.end()) { + std::istringstream iss(ToString(metadata.find(key)->second)); + iss >> default_value; + gpr_log(GPR_INFO, "%s : %d", key, default_value); + } + + return default_value; +} + +int GetIntValueFromMetadata( + const char* key, + const std::multimap<grpc::string_ref, grpc::string_ref>& metadata, + int default_value) { + return GetIntValueFromMetadataHelper(key, metadata, default_value); +} + +void ServerTryCancel(ServerContext* context) { + EXPECT_FALSE(context->IsCancelled()); + context->TryCancel(); + gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); + // Now wait until it's really canceled + while (!context->IsCancelled()) { + gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_micros(1000, GPR_TIMESPAN))); + } +} + +void ServerTryCancelNonblocking(ServerContext* context) { + EXPECT_FALSE(context->IsCancelled()); + context->TryCancel(); + gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); +} + } // namespace Status TestServiceImpl::Echo(ServerContext* context, const EchoRequest* request, @@ -165,6 +221,166 @@ Status TestServiceImpl::Echo(ServerContext* context, const EchoRequest* request, return Status::OK; } +Status TestServiceImpl::CheckClientInitialMetadata(ServerContext* context, + const SimpleRequest* request, + SimpleResponse* response) { + EXPECT_EQ(MetadataMatchCount(context->client_metadata(), + kCheckClientInitialMetadataKey, + kCheckClientInitialMetadataVal), + 1); + EXPECT_EQ(1u, + context->client_metadata().count(kCheckClientInitialMetadataKey)); + return Status::OK; +} + +void CallbackTestServiceImpl::Echo( + ServerContext* context, const EchoRequest* request, EchoResponse* response, + experimental::ServerCallbackRpcController* controller) { + // A bit of sleep to make sure that short deadline tests fail + if (request->has_param() && request->param().server_sleep_us() > 0) { + // Set an alarm for that much time + alarm_.experimental().Set( + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), + gpr_time_from_micros(request->param().server_sleep_us(), + GPR_TIMESPAN)), + [this, context, request, response, controller](bool) { + EchoNonDelayed(context, request, response, controller); + }); + } else { + EchoNonDelayed(context, request, response, controller); + } +} + +void CallbackTestServiceImpl::CheckClientInitialMetadata( + ServerContext* context, const SimpleRequest* request, + SimpleResponse* response, + experimental::ServerCallbackRpcController* controller) { + EXPECT_EQ(MetadataMatchCount(context->client_metadata(), + kCheckClientInitialMetadataKey, + kCheckClientInitialMetadataVal), + 1); + EXPECT_EQ(1u, + context->client_metadata().count(kCheckClientInitialMetadataKey)); + controller->Finish(Status::OK); +} + +void CallbackTestServiceImpl::EchoNonDelayed( + ServerContext* context, const EchoRequest* request, EchoResponse* response, + experimental::ServerCallbackRpcController* controller) { + if (request->has_param() && request->param().server_die()) { + gpr_log(GPR_ERROR, "The request should not reach application handler."); + GPR_ASSERT(0); + } + if (request->has_param() && request->param().has_expected_error()) { + const auto& error = request->param().expected_error(); + controller->Finish(Status(static_cast<StatusCode>(error.code()), + error.error_message(), + error.binary_error_details())); + return; + } + int server_try_cancel = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + if (server_try_cancel > DO_NOT_CANCEL) { + // Since this is a unary RPC, by the time this server handler is called, + // the 'request' message is already read from the client. So the scenarios + // in server_try_cancel don't make much sense. Just cancel the RPC as long + // as server_try_cancel is not DO_NOT_CANCEL + EXPECT_FALSE(context->IsCancelled()); + context->TryCancel(); + gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); + // Now wait until it's really canceled + + std::function<void(bool)> recurrence = [this, context, controller, + &recurrence](bool) { + if (!context->IsCancelled()) { + alarm_.experimental().Set( + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_micros(1000, GPR_TIMESPAN)), + recurrence); + } else { + controller->Finish(Status::CANCELLED); + } + }; + recurrence(true); + return; + } + + gpr_log(GPR_DEBUG, "Request message was %s", request->message().c_str()); + response->set_message(request->message()); + MaybeEchoDeadline(context, request, response); + if (host_) { + response->mutable_param()->set_host(*host_); + } + if (request->has_param() && request->param().client_cancel_after_us()) { + { + std::unique_lock<std::mutex> lock(mu_); + signal_client_ = true; + } + std::function<void(bool)> recurrence = [this, context, request, controller, + &recurrence](bool) { + if (!context->IsCancelled()) { + alarm_.experimental().Set( + gpr_time_add( + gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_micros(request->param().client_cancel_after_us(), + GPR_TIMESPAN)), + recurrence); + } else { + controller->Finish(Status::CANCELLED); + } + }; + recurrence(true); + return; + } else if (request->has_param() && + request->param().server_cancel_after_us()) { + alarm_.experimental().Set( + gpr_time_add( + gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_micros(request->param().server_cancel_after_us(), + GPR_TIMESPAN)), + [controller](bool) { controller->Finish(Status::CANCELLED); }); + return; + } else if (!request->has_param() || + !request->param().skip_cancelled_check()) { + EXPECT_FALSE(context->IsCancelled()); + } + + if (request->has_param() && request->param().echo_metadata()) { + const std::multimap<grpc::string_ref, grpc::string_ref>& client_metadata = + context->client_metadata(); + for (std::multimap<grpc::string_ref, grpc::string_ref>::const_iterator + iter = client_metadata.begin(); + iter != client_metadata.end(); ++iter) { + context->AddTrailingMetadata(ToString(iter->first), + ToString(iter->second)); + } + // Terminate rpc with error and debug info in trailer. + if (request->param().debug_info().stack_entries_size() || + !request->param().debug_info().detail().empty()) { + grpc::string serialized_debug_info = + request->param().debug_info().SerializeAsString(); + context->AddTrailingMetadata(kDebugInfoTrailerKey, serialized_debug_info); + controller->Finish(Status::CANCELLED); + return; + } + } + if (request->has_param() && + (request->param().expected_client_identity().length() > 0 || + request->param().check_auth_context())) { + CheckServerAuthContext(context, + request->param().expected_transport_security_type(), + request->param().expected_client_identity()); + } + if (request->has_param() && request->param().response_message_length() > 0) { + response->set_message( + grpc::string(request->param().response_message_length(), '\0')); + } + if (request->has_param() && request->param().echo_peer()) { + response->mutable_param()->set_peer(context->peer()); + } + controller->Finish(Status::OK); +} + // Unimplemented is left unimplemented to test the returned error. Status TestServiceImpl::RequestStream(ServerContext* context, @@ -192,7 +408,7 @@ Status TestServiceImpl::RequestStream(ServerContext* context, std::thread* server_try_cancel_thd = nullptr; if (server_try_cancel == CANCEL_DURING_PROCESSING) { server_try_cancel_thd = - new std::thread(&TestServiceImpl::ServerTryCancel, this, context); + new std::thread([context] { ServerTryCancel(context); }); } int num_msgs_read = 0; @@ -247,7 +463,7 @@ Status TestServiceImpl::ResponseStream(ServerContext* context, std::thread* server_try_cancel_thd = nullptr; if (server_try_cancel == CANCEL_DURING_PROCESSING) { server_try_cancel_thd = - new std::thread(&TestServiceImpl::ServerTryCancel, this, context); + new std::thread([context] { ServerTryCancel(context); }); } for (int i = 0; i < server_responses_to_send; i++) { @@ -298,7 +514,7 @@ Status TestServiceImpl::BidiStream( std::thread* server_try_cancel_thd = nullptr; if (server_try_cancel == CANCEL_DURING_PROCESSING) { server_try_cancel_thd = - new std::thread(&TestServiceImpl::ServerTryCancel, this, context); + new std::thread([context] { ServerTryCancel(context); }); } // kServerFinishAfterNReads suggests after how many reads, the server should @@ -332,28 +548,244 @@ Status TestServiceImpl::BidiStream( return Status::OK; } -int TestServiceImpl::GetIntValueFromMetadata( - const char* key, - const std::multimap<grpc::string_ref, grpc::string_ref>& metadata, - int default_value) { - if (metadata.find(key) != metadata.end()) { - std::istringstream iss(ToString(metadata.find(key)->second)); - iss >> default_value; - gpr_log(GPR_INFO, "%s : %d", key, default_value); - } +experimental::ServerReadReactor<EchoRequest, EchoResponse>* +CallbackTestServiceImpl::RequestStream() { + class Reactor : public ::grpc::experimental::ServerReadReactor<EchoRequest, + EchoResponse> { + public: + Reactor() {} + void OnStarted(ServerContext* context, EchoResponse* response) override { + ctx_ = context; + response_ = response; + // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by + // the server by calling ServerContext::TryCancel() depending on the + // value: + // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server + // reads any message from the client CANCEL_DURING_PROCESSING: The RPC + // is cancelled while the server is reading messages from the client + // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads + // all the messages from the client + server_try_cancel_ = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + + response_->set_message(""); + + if (server_try_cancel_ == CANCEL_BEFORE_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + ctx_->TryCancel(); + // Don't wait for it here + } + + StartRead(&request_); + } + void OnDone() override { delete this; } + void OnCancel() override { FinishOnce(Status::CANCELLED); } + void OnReadDone(bool ok) override { + if (ok) { + response_->mutable_message()->append(request_.message()); + num_msgs_read_++; + StartRead(&request_); + } else { + gpr_log(GPR_INFO, "Read: %d messages", num_msgs_read_); + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + // Let OnCancel recover this + return; + } + if (server_try_cancel_ == CANCEL_AFTER_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + FinishOnce(Status::OK); + } + } - return default_value; + private: + void FinishOnce(const Status& s) { + std::lock_guard<std::mutex> l(finish_mu_); + if (!finished_) { + Finish(s); + finished_ = true; + } + } + + ServerContext* ctx_; + EchoResponse* response_; + EchoRequest request_; + int num_msgs_read_{0}; + int server_try_cancel_; + std::mutex finish_mu_; + bool finished_{false}; + }; + + return new Reactor; } -void TestServiceImpl::ServerTryCancel(ServerContext* context) { - EXPECT_FALSE(context->IsCancelled()); - context->TryCancel(); - gpr_log(GPR_INFO, "Server called TryCancel() to cancel the request"); - // Now wait until it's really canceled - while (!context->IsCancelled()) { - gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), - gpr_time_from_micros(1000, GPR_TIMESPAN))); - } +// Return 'kNumResponseStreamMsgs' messages. +// TODO(yangg) make it generic by adding a parameter into EchoRequest +experimental::ServerWriteReactor<EchoRequest, EchoResponse>* +CallbackTestServiceImpl::ResponseStream() { + class Reactor + : public ::grpc::experimental::ServerWriteReactor<EchoRequest, + EchoResponse> { + public: + Reactor() {} + void OnStarted(ServerContext* context, + const EchoRequest* request) override { + ctx_ = context; + request_ = request; + // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by + // the server by calling ServerContext::TryCancel() depending on the + // value: + // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server + // reads any message from the client CANCEL_DURING_PROCESSING: The RPC + // is cancelled while the server is reading messages from the client + // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads + // all the messages from the client + server_try_cancel_ = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + server_coalescing_api_ = GetIntValueFromMetadata( + kServerUseCoalescingApi, context->client_metadata(), 0); + server_responses_to_send_ = GetIntValueFromMetadata( + kServerResponseStreamsToSend, context->client_metadata(), + kServerDefaultResponseStreamsToSend); + if (server_try_cancel_ == CANCEL_BEFORE_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + ctx_->TryCancel(); + } + if (num_msgs_sent_ < server_responses_to_send_) { + NextWrite(); + } + } + void OnDone() override { delete this; } + void OnCancel() override { FinishOnce(Status::CANCELLED); } + void OnWriteDone(bool ok) override { + if (num_msgs_sent_ < server_responses_to_send_) { + NextWrite(); + } else if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + // Let OnCancel recover this + } else if (server_try_cancel_ == CANCEL_AFTER_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + } else { + FinishOnce(Status::OK); + } + } + + private: + void FinishOnce(const Status& s) { + std::lock_guard<std::mutex> l(finish_mu_); + if (!finished_) { + Finish(s); + finished_ = true; + } + } + + void NextWrite() { + response_.set_message(request_->message() + + grpc::to_string(num_msgs_sent_)); + if (num_msgs_sent_ == server_responses_to_send_ - 1 && + server_coalescing_api_ != 0) { + num_msgs_sent_++; + StartWriteLast(&response_, WriteOptions()); + } else { + num_msgs_sent_++; + StartWrite(&response_); + } + } + ServerContext* ctx_; + const EchoRequest* request_; + EchoResponse response_; + int num_msgs_sent_{0}; + int server_try_cancel_; + int server_coalescing_api_; + int server_responses_to_send_; + std::mutex finish_mu_; + bool finished_{false}; + }; + return new Reactor; +} + +experimental::ServerBidiReactor<EchoRequest, EchoResponse>* +CallbackTestServiceImpl::BidiStream() { + class Reactor : public ::grpc::experimental::ServerBidiReactor<EchoRequest, + EchoResponse> { + public: + Reactor() {} + void OnStarted(ServerContext* context) override { + ctx_ = context; + // If 'server_try_cancel' is set in the metadata, the RPC is cancelled by + // the server by calling ServerContext::TryCancel() depending on the + // value: + // CANCEL_BEFORE_PROCESSING: The RPC is cancelled before the server + // reads any message from the client CANCEL_DURING_PROCESSING: The RPC + // is cancelled while the server is reading messages from the client + // CANCEL_AFTER_PROCESSING: The RPC is cancelled after the server reads + // all the messages from the client + server_try_cancel_ = GetIntValueFromMetadata( + kServerTryCancelRequest, context->client_metadata(), DO_NOT_CANCEL); + server_write_last_ = GetIntValueFromMetadata( + kServerFinishAfterNReads, context->client_metadata(), 0); + if (server_try_cancel_ == CANCEL_BEFORE_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + return; + } + + if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + ctx_->TryCancel(); + } + + StartRead(&request_); + } + void OnDone() override { delete this; } + void OnCancel() override { FinishOnce(Status::CANCELLED); } + void OnReadDone(bool ok) override { + if (ok) { + num_msgs_read_++; + gpr_log(GPR_INFO, "recv msg %s", request_.message().c_str()); + response_.set_message(request_.message()); + if (num_msgs_read_ == server_write_last_) { + StartWriteLast(&response_, WriteOptions()); + } else { + StartWrite(&response_); + } + } else if (server_try_cancel_ == CANCEL_DURING_PROCESSING) { + // Let OnCancel handle this + } else if (server_try_cancel_ == CANCEL_AFTER_PROCESSING) { + ServerTryCancelNonblocking(ctx_); + } else { + FinishOnce(Status::OK); + } + } + void OnWriteDone(bool ok) override { StartRead(&request_); } + + private: + void FinishOnce(const Status& s) { + std::lock_guard<std::mutex> l(finish_mu_); + if (!finished_) { + Finish(s); + finished_ = true; + } + } + + ServerContext* ctx_; + EchoRequest request_; + EchoResponse response_; + int num_msgs_read_{0}; + int server_try_cancel_; + int server_write_last_; + std::mutex finish_mu_; + bool finished_{false}; + }; + + return new Reactor; } } // namespace testing diff --git a/test/cpp/end2end/test_service_impl.h b/test/cpp/end2end/test_service_impl.h index 052543a03e..e36423d44e 100644 --- a/test/cpp/end2end/test_service_impl.h +++ b/test/cpp/end2end/test_service_impl.h @@ -22,6 +22,7 @@ #include <mutex> #include <grpc/grpc.h> +#include <grpcpp/alarm.h> #include <grpcpp/server_context.h> #include "src/proto/grpc/testing/echo.grpc.pb.h" @@ -35,6 +36,8 @@ const char* const kServerTryCancelRequest = "server_try_cancel"; const char* const kDebugInfoTrailerKey = "debug-info-bin"; const char* const kServerFinishAfterNReads = "server_finish_after_n_reads"; const char* const kServerUseCoalescingApi = "server_use_coalescing_api"; +const char* const kCheckClientInitialMetadataKey = "custom_client_metadata"; +const char* const kCheckClientInitialMetadataVal = "Value for client metadata"; typedef enum { DO_NOT_CANCEL = 0, @@ -52,6 +55,10 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service { Status Echo(ServerContext* context, const EchoRequest* request, EchoResponse* response) override; + Status CheckClientInitialMetadata(ServerContext* context, + const SimpleRequest* request, + SimpleResponse* response) override; + // Unimplemented is left unimplemented to test the returned error. Status RequestStream(ServerContext* context, @@ -71,14 +78,48 @@ class TestServiceImpl : public ::grpc::testing::EchoTestService::Service { } private: - int GetIntValueFromMetadata( - const char* key, - const std::multimap<grpc::string_ref, grpc::string_ref>& metadata, - int default_value); + bool signal_client_; + std::mutex mu_; + std::unique_ptr<grpc::string> host_; +}; + +class CallbackTestServiceImpl + : public ::grpc::testing::EchoTestService::ExperimentalCallbackService { + public: + CallbackTestServiceImpl() : signal_client_(false), host_() {} + explicit CallbackTestServiceImpl(const grpc::string& host) + : signal_client_(false), host_(new grpc::string(host)) {} + + void Echo(ServerContext* context, const EchoRequest* request, + EchoResponse* response, + experimental::ServerCallbackRpcController* controller) override; + + void CheckClientInitialMetadata( + ServerContext* context, const SimpleRequest* request, + SimpleResponse* response, + experimental::ServerCallbackRpcController* controller) override; - void ServerTryCancel(ServerContext* context); + experimental::ServerReadReactor<EchoRequest, EchoResponse>* RequestStream() + override; + + experimental::ServerWriteReactor<EchoRequest, EchoResponse>* ResponseStream() + override; + + experimental::ServerBidiReactor<EchoRequest, EchoResponse>* BidiStream() + override; + + // Unimplemented is left unimplemented to test the returned error. + bool signal_client() { + std::unique_lock<std::mutex> lock(mu_); + return signal_client_; + } private: + void EchoNonDelayed(ServerContext* context, const EchoRequest* request, + EchoResponse* response, + experimental::ServerCallbackRpcController* controller); + + Alarm alarm_; bool signal_client_; std::mutex mu_; std::unique_ptr<grpc::string> host_; diff --git a/test/cpp/end2end/thread_stress_test.cc b/test/cpp/end2end/thread_stress_test.cc index 1a5ed28a2c..e30ce0dbcb 100644 --- a/test/cpp/end2end/thread_stress_test.cc +++ b/test/cpp/end2end/thread_stress_test.cc @@ -429,7 +429,7 @@ TYPED_TEST(AsyncClientEnd2endTest, ThreadStress) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/ext/filters/census/stats_plugin_end2end_test.cc b/test/cpp/ext/filters/census/stats_plugin_end2end_test.cc index 664504a090..7339402830 100644 --- a/test/cpp/ext/filters/census/stats_plugin_end2end_test.cc +++ b/test/cpp/ext/filters/census/stats_plugin_end2end_test.cc @@ -370,7 +370,7 @@ TEST_F(StatsPluginEnd2EndTest, RequestReceivedMessagesPerRpc) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/interop/client_helper.h b/test/cpp/interop/client_helper.h index eada2f671f..7dee85cc98 100644 --- a/test/cpp/interop/client_helper.h +++ b/test/cpp/interop/client_helper.h @@ -19,10 +19,12 @@ #ifndef GRPC_TEST_CPP_INTEROP_CLIENT_HELPER_H #define GRPC_TEST_CPP_INTEROP_CLIENT_HELPER_H +#include <functional> #include <memory> #include <unordered_map> #include <grpcpp/channel.h> +#include <grpcpp/client_context.h> #include "src/core/lib/surface/call_test_only.h" diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc index a99cf8122f..4ff153f980 100644 --- a/test/cpp/interop/interop_client.cc +++ b/test/cpp/interop/interop_client.cc @@ -76,7 +76,7 @@ void UnaryCompressionChecks(const InteropClientContextInspector& inspector, InteropClient::ServiceStub::ServiceStub( ChannelCreationFunc channel_creation_func, bool new_stub_every_call) - : channel_creation_func_(channel_creation_func), + : channel_creation_func_(std::move(channel_creation_func)), channel_(channel_creation_func_()), new_stub_every_call_(new_stub_every_call) { // If new_stub_every_call is false, then this is our chance to initialize @@ -112,7 +112,7 @@ void InteropClient::ServiceStub::ResetChannel() { InteropClient::InteropClient(ChannelCreationFunc channel_creation_func, bool new_stub_every_test_case, bool do_not_abort_on_transient_failures) - : serviceStub_(channel_creation_func, new_stub_every_test_case), + : serviceStub_(std::move(channel_creation_func), new_stub_every_test_case), do_not_abort_on_transient_failures_(do_not_abort_on_transient_failures) {} bool InteropClient::AssertStatusOk(const Status& s, diff --git a/test/cpp/interop/stress_interop_client.cc b/test/cpp/interop/stress_interop_client.cc index 7dc1956f78..5bbe913365 100644 --- a/test/cpp/interop/stress_interop_client.cc +++ b/test/cpp/interop/stress_interop_client.cc @@ -73,7 +73,7 @@ StressTestInteropClient::StressTestInteropClient( long sleep_duration_ms, bool do_not_abort_on_transient_failures) : test_id_(test_id), server_address_(server_address), - channel_creation_func_(channel_creation_func), + channel_creation_func_(std::move(channel_creation_func)), interop_client_(new InteropClient(channel_creation_func_, false, do_not_abort_on_transient_failures)), test_selector_(test_selector), diff --git a/test/cpp/microbenchmarks/bm_call_create.cc b/test/cpp/microbenchmarks/bm_call_create.cc index 1f7831096c..8d12606434 100644 --- a/test/cpp/microbenchmarks/bm_call_create.cc +++ b/test/cpp/microbenchmarks/bm_call_create.cc @@ -135,7 +135,8 @@ static void BM_LameChannelCallCreateCpp(benchmark::State& state) { "", grpc_lame_client_channel_create("localhost:1234", GRPC_STATUS_UNAUTHENTICATED, "blah"), - nullptr)); + std::vector<std::unique_ptr< + grpc::experimental::ClientInterceptorFactoryInterface>>())); grpc::CompletionQueue cq; grpc::testing::EchoRequest send_request; grpc::testing::EchoResponse recv_response; @@ -468,7 +469,7 @@ class NoOp { class SendEmptyMetadata { public: - SendEmptyMetadata() { + SendEmptyMetadata() : op_payload_(nullptr) { memset(&op_, 0, sizeof(op_)); op_.on_complete = GRPC_CLOSURE_INIT(&closure_, DoNothing, nullptr, grpc_schedule_on_exec_ctx); diff --git a/test/cpp/microbenchmarks/bm_chttp2_hpack.cc b/test/cpp/microbenchmarks/bm_chttp2_hpack.cc index ba4b57ad22..85d233dd26 100644 --- a/test/cpp/microbenchmarks/bm_chttp2_hpack.cc +++ b/test/cpp/microbenchmarks/bm_chttp2_hpack.cc @@ -778,8 +778,7 @@ static void free_timeout(void* p) { gpr_free(p); } // Benchmark the current on_initial_header implementation static void OnInitialHeader(void* user_data, grpc_mdelem md) { // Setup for benchmark. This will bloat the absolute values of this benchmark - grpc_chttp2_incoming_metadata_buffer buffer; - grpc_chttp2_incoming_metadata_buffer_init(&buffer, (gpr_arena*)user_data); + grpc_chttp2_incoming_metadata_buffer buffer((gpr_arena*)user_data); bool seen_error = false; // Below here is the code we actually care about benchmarking @@ -822,7 +821,6 @@ static void OnInitialHeader(void* user_data, grpc_mdelem md) { GPR_ASSERT(0); } } - grpc_chttp2_incoming_metadata_buffer_destroy(&buffer); } // Benchmark timeout handling diff --git a/test/cpp/microbenchmarks/bm_chttp2_transport.cc b/test/cpp/microbenchmarks/bm_chttp2_transport.cc index 189923a841..650152ecc0 100644 --- a/test/cpp/microbenchmarks/bm_chttp2_transport.cc +++ b/test/cpp/microbenchmarks/bm_chttp2_transport.cc @@ -54,7 +54,8 @@ class DummyEndpoint : public grpc_endpoint { destroy, get_resource_user, get_peer, - get_fd}; + get_fd, + can_track_err}; grpc_endpoint::vtable = &my_vtable; ru_ = grpc_resource_user_create(Library::get().rq(), "dummy_endpoint"); } @@ -125,6 +126,7 @@ class DummyEndpoint : public grpc_endpoint { } static char* get_peer(grpc_endpoint* ep) { return gpr_strdup("test"); } static int get_fd(grpc_endpoint* ep) { return 0; } + static bool can_track_err(grpc_endpoint* ep) { return false; } }; class Fixture { @@ -262,7 +264,7 @@ static void BM_StreamCreateDestroy(benchmark::State& state) { Fixture f(grpc::ChannelArguments(), true); Stream s(&f); grpc_transport_stream_op_batch op; - grpc_transport_stream_op_batch_payload op_payload; + grpc_transport_stream_op_batch_payload op_payload(nullptr); memset(&op, 0, sizeof(op)); op.cancel_stream = true; op.payload = &op_payload; @@ -308,8 +310,7 @@ static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State& state) { Fixture f(grpc::ChannelArguments(), true); Stream s(&f); grpc_transport_stream_op_batch op; - grpc_transport_stream_op_batch_payload op_payload; - memset(&op_payload, 0, sizeof(op_payload)); + grpc_transport_stream_op_batch_payload op_payload(nullptr); std::unique_ptr<Closure> start; std::unique_ptr<Closure> done; @@ -360,8 +361,7 @@ static void BM_TransportEmptyOp(benchmark::State& state) { Stream s(&f); s.Init(state); grpc_transport_stream_op_batch op; - grpc_transport_stream_op_batch_payload op_payload; - memset(&op_payload, 0, sizeof(op_payload)); + grpc_transport_stream_op_batch_payload op_payload(nullptr); auto reset_op = [&]() { memset(&op, 0, sizeof(op)); op.payload = &op_payload; @@ -393,8 +393,7 @@ static void BM_TransportStreamSend(benchmark::State& state) { auto s = std::unique_ptr<Stream>(new Stream(&f)); s->Init(state); grpc_transport_stream_op_batch op; - grpc_transport_stream_op_batch_payload op_payload; - memset(&op_payload, 0, sizeof(op_payload)); + grpc_transport_stream_op_batch_payload op_payload(nullptr); auto reset_op = [&]() { memset(&op, 0, sizeof(op)); op.payload = &op_payload; @@ -526,8 +525,7 @@ static void BM_TransportStreamRecv(benchmark::State& state) { Fixture f(grpc::ChannelArguments(), true); Stream s(&f); s.Init(state); - grpc_transport_stream_op_batch_payload op_payload; - memset(&op_payload, 0, sizeof(op_payload)); + grpc_transport_stream_op_batch_payload op_payload(nullptr); grpc_transport_stream_op_batch op; grpc_core::OrphanablePtr<grpc_core::ByteStream> recv_stream; grpc_slice incoming_data = CreateIncomingDataSlice(state.range(0), 16384); diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc index 85767c8758..dca97c85b1 100644 --- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc +++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc @@ -94,6 +94,7 @@ static const grpc_event_engine_vtable* init_engine_vtable(bool) { g_vtable.pollset_destroy = pollset_destroy; g_vtable.pollset_work = pollset_work; g_vtable.pollset_kick = pollset_kick; + g_vtable.shutdown_background_closure = [] {}; g_vtable.shutdown_engine = [] {}; return &g_vtable; diff --git a/test/cpp/microbenchmarks/fullstack_fixtures.h b/test/cpp/microbenchmarks/fullstack_fixtures.h index e57eb6ddd1..6bbf553bbd 100644 --- a/test/cpp/microbenchmarks/fullstack_fixtures.h +++ b/test/cpp/microbenchmarks/fullstack_fixtures.h @@ -200,7 +200,7 @@ class EndpointPairFixture : public BaseFixture { } grpc_server_setup_transport(server_->c_server(), server_transport_, - nullptr, server_args, 0); + nullptr, server_args, nullptr); grpc_chttp2_transport_start_reading(server_transport_, nullptr, nullptr); } @@ -218,7 +218,10 @@ class EndpointPairFixture : public BaseFixture { "target", &c_args, GRPC_CLIENT_DIRECT_CHANNEL, client_transport_); grpc_chttp2_transport_start_reading(client_transport_, nullptr, nullptr); - channel_ = CreateChannelInternal("", channel, nullptr); + channel_ = CreateChannelInternal( + "", channel, + std::vector<std::unique_ptr< + experimental::ClientInterceptorFactoryInterface>>()); } } diff --git a/test/cpp/naming/address_sorting_test.cc b/test/cpp/naming/address_sorting_test.cc index fc6721d0ba..3eb0e7d725 100644 --- a/test/cpp/naming/address_sorting_test.cc +++ b/test/cpp/naming/address_sorting_test.cc @@ -838,7 +838,7 @@ int main(int argc, char** argv) { gpr_log(GPR_INFO, "GRPC_DNS_RESOLVER != ares: %s.", resolver); } gpr_free(resolver); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); auto result = RUN_ALL_TESTS(); // Test sequential and nested inits and shutdowns. diff --git a/test/cpp/naming/cancel_ares_query_test.cc b/test/cpp/naming/cancel_ares_query_test.cc index dec7c171dc..3e789f0b14 100644 --- a/test/cpp/naming/cancel_ares_query_test.cc +++ b/test/cpp/naming/cancel_ares_query_test.cc @@ -260,8 +260,15 @@ TEST(CancelDuringAresQuery, TestFdsAreDeletedFromPollsetSet) { grpc_pollset_set_destroy(fake_other_pollset_set); } -TEST(CancelDuringAresQuery, - TestHitDeadlineAndDestroyChannelDuringAresResolutionIsGraceful) { +// Settings for TestCancelDuringActiveQuery test +typedef enum { + NONE, + SHORT, + ZERO, +} cancellation_test_query_timeout_setting; + +void TestCancelDuringActiveQuery( + cancellation_test_query_timeout_setting query_timeout_setting) { // Start up fake non responsive DNS server int fake_dns_port = grpc_pick_unused_port_or_die(); FakeNonResponsiveDNSServer fake_dns_server(fake_dns_port); @@ -271,9 +278,33 @@ TEST(CancelDuringAresQuery, &client_target, "dns://[::1]:%d/dont-care-since-wont-be-resolved.test.com:1234", fake_dns_port)); + gpr_log(GPR_DEBUG, "TestCancelActiveDNSQuery. query timeout setting: %d", + query_timeout_setting); + grpc_channel_args* client_args = nullptr; + grpc_status_code expected_status_code = GRPC_STATUS_OK; + if (query_timeout_setting == NONE) { + expected_status_code = GRPC_STATUS_DEADLINE_EXCEEDED; + client_args = nullptr; + } else if (query_timeout_setting == SHORT) { + expected_status_code = GRPC_STATUS_UNAVAILABLE; + grpc_arg arg; + arg.type = GRPC_ARG_INTEGER; + arg.key = const_cast<char*>(GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS); + arg.value.integer = + 1; // Set this shorter than the call deadline so that it goes off. + client_args = grpc_channel_args_copy_and_add(nullptr, &arg, 1); + } else if (query_timeout_setting == ZERO) { + expected_status_code = GRPC_STATUS_DEADLINE_EXCEEDED; + grpc_arg arg; + arg.type = GRPC_ARG_INTEGER; + arg.key = const_cast<char*>(GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS); + arg.value.integer = 0; // Set this to zero to disable query timeouts. + client_args = grpc_channel_args_copy_and_add(nullptr, &arg, 1); + } else { + abort(); + } grpc_channel* client = - grpc_insecure_channel_create(client_target, - /* client_args */ nullptr, nullptr); + grpc_insecure_channel_create(client_target, client_args, nullptr); gpr_free(client_target); grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr); cq_verifier* cqv = cq_verifier_create(cq); @@ -325,8 +356,9 @@ TEST(CancelDuringAresQuery, EXPECT_EQ(GRPC_CALL_OK, error); CQ_EXPECT_COMPLETION(cqv, Tag(1), 1); cq_verify(cqv); - EXPECT_EQ(status, GRPC_STATUS_DEADLINE_EXCEEDED); + EXPECT_EQ(status, expected_status_code); // Teardown + grpc_channel_args_destroy(client_args); grpc_slice_unref(details); gpr_free((void*)error_string); grpc_metadata_array_destroy(&initial_metadata_recv); @@ -338,10 +370,27 @@ TEST(CancelDuringAresQuery, EndTest(client, cq); } +TEST(CancelDuringAresQuery, + TestHitDeadlineAndDestroyChannelDuringAresResolutionIsGraceful) { + TestCancelDuringActiveQuery(NONE /* don't set query timeouts */); +} + +TEST( + CancelDuringAresQuery, + TestHitDeadlineAndDestroyChannelDuringAresResolutionWithQueryTimeoutIsGraceful) { + TestCancelDuringActiveQuery(SHORT /* set short query timeout */); +} + +TEST( + CancelDuringAresQuery, + TestHitDeadlineAndDestroyChannelDuringAresResolutionWithZeroQueryTimeoutIsGraceful) { + TestCancelDuringActiveQuery(ZERO /* disable query timeouts */); +} + } // namespace int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); gpr_setenv("GRPC_DNS_RESOLVER", "ares"); // Sanity check the time that it takes to run the test diff --git a/test/cpp/naming/resolver_component_test.cc b/test/cpp/naming/resolver_component_test.cc index 3dc6e7178c..fe6fcb8d9c 100644 --- a/test/cpp/naming/resolver_component_test.cc +++ b/test/cpp/naming/resolver_component_test.cc @@ -477,7 +477,7 @@ TEST(ResolverComponentTest, TestResolvesRelevantRecordsWithConcurrentFdStress) { int main(int argc, char** argv) { grpc_init(); - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); ParseCommandLineFlags(&argc, &argv, true); if (FLAGS_target_name == "") { diff --git a/test/cpp/performance/writes_per_rpc_test.cc b/test/cpp/performance/writes_per_rpc_test.cc index 32eab1fc44..7b22f23cf0 100644 --- a/test/cpp/performance/writes_per_rpc_test.cc +++ b/test/cpp/performance/writes_per_rpc_test.cc @@ -100,7 +100,7 @@ class EndpointPairFixture { } grpc_server_setup_transport(server_->c_server(), transport, nullptr, - server_args, 0); + server_args, nullptr); grpc_chttp2_transport_start_reading(transport, nullptr, nullptr); } @@ -118,7 +118,10 @@ class EndpointPairFixture { "target", &c_args, GRPC_CLIENT_DIRECT_CHANNEL, transport); grpc_chttp2_transport_start_reading(transport, nullptr, nullptr); - channel_ = CreateChannelInternal("", channel, nullptr); + channel_ = CreateChannelInternal( + "", channel, + std::vector<std::unique_ptr< + experimental::ClientInterceptorFactoryInterface>>()); } } @@ -252,7 +255,7 @@ TEST(WritesPerRpcTest, UnaryPingPong) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/qps/BUILD b/test/cpp/qps/BUILD index 2ef7441371..626ac5f3f2 100644 --- a/test/cpp/qps/BUILD +++ b/test/cpp/qps/BUILD @@ -36,6 +36,7 @@ grpc_cc_library( "qps_server_builder.cc", "qps_worker.cc", "server_async.cc", + "server_callback.cc", "server_sync.cc", ], hdrs = [ @@ -169,7 +170,7 @@ grpc_cc_test( grpc_cc_test( name = "qps_openloop_test", srcs = ["qps_openloop_test.cc"], - data = ["//third_party/toolchains:RBE_USE_MACHINE_TYPE_LARGE"], + exec_compatible_with = ["//third_party/toolchains/machine_size:large"], deps = [ ":benchmark_config", ":driver_impl", diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index 4ed34e0405..668d941916 100644 --- a/test/cpp/qps/client.h +++ b/test/cpp/qps/client.h @@ -94,9 +94,11 @@ class ClientRequestCreator<ByteBuffer> { public: ClientRequestCreator(ByteBuffer* req, const PayloadConfig& payload_config) { if (payload_config.has_bytebuf_params()) { - std::unique_ptr<char[]> buf( - new char[payload_config.bytebuf_params().req_size()]); - Slice slice(buf.get(), payload_config.bytebuf_params().req_size()); + size_t req_sz = + static_cast<size_t>(payload_config.bytebuf_params().req_size()); + std::unique_ptr<char[]> buf(new char[req_sz]); + memset(buf.get(), 0, req_sz); + Slice slice(buf.get(), req_sz); *req = ByteBuffer(&slice, 1); } else { GPR_ASSERT(false); // not appropriate for this specialization diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc index 11cfb4aa05..181e11f12b 100644 --- a/test/cpp/qps/driver.cc +++ b/test/cpp/qps/driver.cc @@ -95,6 +95,17 @@ static deque<string> get_workers(const string& env_name) { return out; } +std::string GetCredType( + const std::string& worker_addr, + const std::map<std::string, std::string>& per_worker_credential_types, + const std::string& credential_type) { + auto it = per_worker_credential_types.find(worker_addr); + if (it != per_worker_credential_types.end()) { + return it->second; + } + return credential_type; +} + // helpers for postprocess_scenario_result static double WallTime(const ClientStats& s) { return s.time_elapsed(); } static double SystemTime(const ClientStats& s) { return s.time_system(); } @@ -198,8 +209,9 @@ std::unique_ptr<ScenarioResult> RunScenario( const ServerConfig& initial_server_config, size_t num_servers, int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count, const grpc::string& qps_server_target_override, - const grpc::string& credential_type, bool run_inproc, - int32_t median_latency_collection_interval_millis) { + const grpc::string& credential_type, + const std::map<std::string, std::string>& per_worker_credential_types, + bool run_inproc, int32_t median_latency_collection_interval_millis) { if (run_inproc) { g_inproc_servers = new std::vector<grpc::testing::Server*>; } @@ -278,7 +290,9 @@ std::unique_ptr<ScenarioResult> RunScenario( if (!run_inproc) { servers[i].stub = WorkerService::NewStub(CreateChannel( workers[i], GetCredentialsProvider()->GetChannelCredentials( - credential_type, &channel_args))); + GetCredType(workers[i], per_worker_credential_types, + credential_type), + &channel_args))); } else { servers[i].stub = WorkerService::NewStub( local_workers[i]->InProcessChannel(channel_args)); @@ -335,9 +349,11 @@ std::unique_ptr<ScenarioResult> RunScenario( gpr_log(GPR_INFO, "Starting client on %s (worker #%" PRIuPTR ")", worker.c_str(), i + num_servers); if (!run_inproc) { - clients[i].stub = WorkerService::NewStub( - CreateChannel(worker, GetCredentialsProvider()->GetChannelCredentials( - credential_type, &channel_args))); + clients[i].stub = WorkerService::NewStub(CreateChannel( + worker, + GetCredentialsProvider()->GetChannelCredentials( + GetCredType(worker, per_worker_credential_types, credential_type), + &channel_args))); } else { clients[i].stub = WorkerService::NewStub( local_workers[i + num_servers]->InProcessChannel(channel_args)); @@ -529,7 +545,9 @@ std::unique_ptr<ScenarioResult> RunScenario( return result; } -bool RunQuit(const grpc::string& credential_type) { +bool RunQuit( + const grpc::string& credential_type, + const std::map<std::string, std::string>& per_worker_credential_types) { // Get client, server lists bool result = true; auto workers = get_workers("QPS_WORKERS"); @@ -541,7 +559,9 @@ bool RunQuit(const grpc::string& credential_type) { for (size_t i = 0; i < workers.size(); i++) { auto stub = WorkerService::NewStub(CreateChannel( workers[i], GetCredentialsProvider()->GetChannelCredentials( - credential_type, &channel_args))); + GetCredType(workers[i], per_worker_credential_types, + credential_type), + &channel_args))); Void dummy; grpc::ClientContext ctx; ctx.set_wait_for_ready(true); diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h index cda89f7ddf..568871d2da 100644 --- a/test/cpp/qps/driver.h +++ b/test/cpp/qps/driver.h @@ -32,10 +32,13 @@ std::unique_ptr<ScenarioResult> RunScenario( const grpc::testing::ServerConfig& server_config, size_t num_servers, int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count, const grpc::string& qps_server_target_override, - const grpc::string& credential_type, bool run_inproc, - int32_t median_latency_collection_interval_millis); + const grpc::string& credential_type, + const std::map<std::string, std::string>& per_worker_credential_types, + bool run_inproc, int32_t median_latency_collection_interval_millis); -bool RunQuit(const grpc::string& credential_type); +bool RunQuit( + const grpc::string& credential_type, + const std::map<std::string, std::string>& per_worker_credential_types); } // namespace testing } // namespace grpc diff --git a/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc b/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc index 56d1730252..6257e42ebf 100644 --- a/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc +++ b/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc @@ -48,7 +48,7 @@ static void RunSynchronousUnaryPingPong() { const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2, "", - kInsecureCredentialsType, true, 0); + kInsecureCredentialsType, {}, true, 0); GetReporter()->ReportQPS(*result); GetReporter()->ReportLatency(*result); diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc index eaa0dd992c..1d67e79b86 100644 --- a/test/cpp/qps/qps_json_driver.cc +++ b/test/cpp/qps/qps_json_driver.cc @@ -65,6 +65,16 @@ DEFINE_string(json_file_out, "", "File to write the JSON output to."); DEFINE_string(credential_type, grpc::testing::kInsecureCredentialsType, "Credential type for communication with workers"); +DEFINE_string( + per_worker_credential_types, "", + "A map of QPS worker addresses to credential types. When creating a " + "channel to a QPS worker's driver port, the qps_json_driver first checks " + "if the 'name:port' string is in the map, and it uses the corresponding " + "credential type if so. If the QPS worker's 'name:port' string is not " + "in the map, then the driver -> worker channel will be created with " + "the credentials specified in --credential_type. The value of this flag " + "is a semicolon-separated list of map entries, where each map entry is " + "a comma-separated pair."); DEFINE_bool(run_inproc, false, "Perform an in-process transport test"); DEFINE_int32( median_latency_collection_interval_millis, 0, @@ -75,16 +85,53 @@ DEFINE_int32( namespace grpc { namespace testing { -static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario, - bool* success) { +static std::map<std::string, std::string> +ConstructPerWorkerCredentialTypesMap() { + // Parse a list of the form: "addr1,cred_type1;addr2,cred_type2;..." into + // a map. + std::string remaining = FLAGS_per_worker_credential_types; + std::map<std::string, std::string> out; + while (!remaining.empty()) { + size_t next_semicolon = remaining.find(';'); + std::string next_entry = remaining.substr(0, next_semicolon); + if (next_semicolon == std::string::npos) { + remaining = ""; + } else { + remaining = remaining.substr(next_semicolon + 1, std::string::npos); + } + size_t comma = next_entry.find(','); + if (comma == std::string::npos) { + gpr_log(GPR_ERROR, + "Expectd --per_worker_credential_types to be a list " + "of the form: 'addr1,cred_type1;addr2,cred_type2;...' " + "into."); + abort(); + } + std::string addr = next_entry.substr(0, comma); + std::string cred_type = next_entry.substr(comma + 1, std::string::npos); + if (out.find(addr) != out.end()) { + gpr_log(GPR_ERROR, + "Found duplicate addr in per_worker_credential_types."); + abort(); + } + out[addr] = cred_type; + } + return out; +} + +static std::unique_ptr<ScenarioResult> RunAndReport( + const Scenario& scenario, + const std::map<std::string, std::string>& per_worker_credential_types, + bool* success) { std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n"; - auto result = RunScenario( - scenario.client_config(), scenario.num_clients(), - scenario.server_config(), scenario.num_servers(), - scenario.warmup_seconds(), scenario.benchmark_seconds(), - !FLAGS_run_inproc ? scenario.spawn_local_worker_count() : -2, - FLAGS_qps_server_target_override, FLAGS_credential_type, FLAGS_run_inproc, - FLAGS_median_latency_collection_interval_millis); + auto result = + RunScenario(scenario.client_config(), scenario.num_clients(), + scenario.server_config(), scenario.num_servers(), + scenario.warmup_seconds(), scenario.benchmark_seconds(), + !FLAGS_run_inproc ? scenario.spawn_local_worker_count() : -2, + FLAGS_qps_server_target_override, FLAGS_credential_type, + per_worker_credential_types, FLAGS_run_inproc, + FLAGS_median_latency_collection_interval_millis); // Amend the result with scenario config. Eventually we should adjust // RunScenario contract so we don't need to touch the result here. @@ -115,21 +162,26 @@ static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario, return result; } -static double GetCpuLoad(Scenario* scenario, double offered_load, - bool* success) { +static double GetCpuLoad( + Scenario* scenario, double offered_load, + const std::map<std::string, std::string>& per_worker_credential_types, + bool* success) { scenario->mutable_client_config() ->mutable_load_params() ->mutable_poisson() ->set_offered_load(offered_load); - auto result = RunAndReport(*scenario, success); + auto result = RunAndReport(*scenario, per_worker_credential_types, success); return result->summary().server_cpu_usage(); } -static double BinarySearch(Scenario* scenario, double targeted_cpu_load, - double low, double high, bool* success) { +static double BinarySearch( + Scenario* scenario, double targeted_cpu_load, double low, double high, + const std::map<std::string, std::string>& per_worker_credential_types, + bool* success) { while (low <= high * (1 - FLAGS_error_tolerance)) { double mid = low + (high - low) / 2; - double current_cpu_load = GetCpuLoad(scenario, mid, success); + double current_cpu_load = + GetCpuLoad(scenario, mid, per_worker_credential_types, success); gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f", mid); if (!*success) { gpr_log(GPR_ERROR, "Client/Server Failure"); @@ -145,12 +197,14 @@ static double BinarySearch(Scenario* scenario, double targeted_cpu_load, return low; } -static double SearchOfferedLoad(double initial_offered_load, - double targeted_cpu_load, Scenario* scenario, - bool* success) { +static double SearchOfferedLoad( + double initial_offered_load, double targeted_cpu_load, Scenario* scenario, + const std::map<std::string, std::string>& per_worker_credential_types, + bool* success) { std::cerr << "RUNNING SCENARIO: " << scenario->name() << "\n"; double current_offered_load = initial_offered_load; - double current_cpu_load = GetCpuLoad(scenario, current_offered_load, success); + double current_cpu_load = GetCpuLoad(scenario, current_offered_load, + per_worker_credential_types, success); if (current_cpu_load > targeted_cpu_load) { gpr_log(GPR_ERROR, "Initial offered load too high"); return -1; @@ -158,14 +212,15 @@ static double SearchOfferedLoad(double initial_offered_load, while (*success && (current_cpu_load < targeted_cpu_load)) { current_offered_load *= 2; - current_cpu_load = GetCpuLoad(scenario, current_offered_load, success); + current_cpu_load = GetCpuLoad(scenario, current_offered_load, + per_worker_credential_types, success); gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f", current_offered_load); } double targeted_offered_load = BinarySearch(scenario, targeted_cpu_load, current_offered_load / 2, - current_offered_load, success); + current_offered_load, per_worker_credential_types, success); return targeted_offered_load; } @@ -183,6 +238,7 @@ static bool QpsDriver() { abort(); } + auto per_worker_credential_types = ConstructPerWorkerCredentialTypesMap(); if (scfile) { // Read the json data from disk FILE* json_file = fopen(FLAGS_scenarios_file.c_str(), "r"); @@ -198,7 +254,7 @@ static bool QpsDriver() { } else if (scjson) { json = FLAGS_scenarios_json.c_str(); } else if (FLAGS_quit) { - return RunQuit(FLAGS_credential_type); + return RunQuit(FLAGS_credential_type, per_worker_credential_types); } // Parse into an array of scenarios @@ -212,15 +268,16 @@ static bool QpsDriver() { for (int i = 0; i < scenarios.scenarios_size(); i++) { if (FLAGS_search_param == "") { const Scenario& scenario = scenarios.scenarios(i); - RunAndReport(scenario, &success); + RunAndReport(scenario, per_worker_credential_types, &success); } else { if (FLAGS_search_param == "offered_load") { Scenario* scenario = scenarios.mutable_scenarios(i); - double targeted_offered_load = - SearchOfferedLoad(FLAGS_initial_search_value, - FLAGS_targeted_cpu_load, scenario, &success); + double targeted_offered_load = SearchOfferedLoad( + FLAGS_initial_search_value, FLAGS_targeted_cpu_load, scenario, + per_worker_credential_types, &success); gpr_log(GPR_INFO, "targeted_offered_load %f", targeted_offered_load); - GetCpuLoad(scenario, targeted_offered_load, &success); + GetCpuLoad(scenario, targeted_offered_load, per_worker_credential_types, + &success); } else { gpr_log(GPR_ERROR, "Unimplemented search param"); } diff --git a/test/cpp/qps/qps_openloop_test.cc b/test/cpp/qps/qps_openloop_test.cc index 6044f4265a..68062e66f2 100644 --- a/test/cpp/qps/qps_openloop_test.cc +++ b/test/cpp/qps/qps_openloop_test.cc @@ -52,7 +52,7 @@ static void RunQPS() { const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2, "", - kInsecureCredentialsType, false, 0); + kInsecureCredentialsType, {}, false, 0); GetReporter()->ReportQPSPerCore(*result); GetReporter()->ReportLatency(*result); diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc index d97d95d8f3..23fe72316a 100644 --- a/test/cpp/qps/qps_worker.cc +++ b/test/cpp/qps/qps_worker.cc @@ -79,6 +79,8 @@ static std::unique_ptr<Server> CreateServer(const ServerConfig& config) { return CreateAsyncServer(config); case ServerType::ASYNC_GENERIC_SERVER: return CreateAsyncGenericServer(config); + case ServerType::CALLBACK_SERVER: + return CreateCallbackServer(config); default: abort(); } diff --git a/test/cpp/qps/secure_sync_unary_ping_pong_test.cc b/test/cpp/qps/secure_sync_unary_ping_pong_test.cc index a559c82cc8..422bd617eb 100644 --- a/test/cpp/qps/secure_sync_unary_ping_pong_test.cc +++ b/test/cpp/qps/secure_sync_unary_ping_pong_test.cc @@ -55,7 +55,7 @@ static void RunSynchronousUnaryPingPong() { const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2, "", - kInsecureCredentialsType, false, 0); + kInsecureCredentialsType, {}, false, 0); GetReporter()->ReportQPS(*result); GetReporter()->ReportLatency(*result); diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h index 6b1ef93617..89b0e3af4b 100644 --- a/test/cpp/qps/server.h +++ b/test/cpp/qps/server.h @@ -152,6 +152,7 @@ class Server { std::unique_ptr<Server> CreateSynchronousServer(const ServerConfig& config); std::unique_ptr<Server> CreateAsyncServer(const ServerConfig& config); std::unique_ptr<Server> CreateAsyncGenericServer(const ServerConfig& config); +std::unique_ptr<Server> CreateCallbackServer(const ServerConfig& config); } // namespace testing } // namespace grpc diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc index 5cd975cf74..a5f8347c26 100644 --- a/test/cpp/qps/server_async.cc +++ b/test/cpp/qps/server_async.cc @@ -562,6 +562,7 @@ static Status ProcessGenericRPC(const PayloadConfig& payload_config, request->Clear(); int resp_size = payload_config.bytebuf_params().resp_size(); std::unique_ptr<char[]> buf(new char[resp_size]); + memset(buf.get(), 0, static_cast<size_t>(resp_size)); Slice slice(buf.get(), resp_size); *response = ByteBuffer(&slice, 1); return Status::OK; diff --git a/test/cpp/qps/server_callback.cc b/test/cpp/qps/server_callback.cc new file mode 100644 index 0000000000..8bedd44739 --- /dev/null +++ b/test/cpp/qps/server_callback.cc @@ -0,0 +1,96 @@ +/* + * + * 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. + * + */ +#include <grpc/grpc.h> +#include <grpc/support/alloc.h> +#include <grpcpp/security/server_credentials.h> +#include <grpcpp/server.h> +#include <grpcpp/server_context.h> + +#include "src/core/lib/gpr/host_port.h" +#include "src/proto/grpc/testing/benchmark_service.grpc.pb.h" +#include "test/cpp/qps/qps_server_builder.h" +#include "test/cpp/qps/server.h" +#include "test/cpp/qps/usage_timer.h" + +namespace grpc { +namespace testing { + +class BenchmarkCallbackServiceImpl final + : public BenchmarkService::ExperimentalCallbackService { + public: + void UnaryCall( + ServerContext* context, const SimpleRequest* request, + SimpleResponse* response, + experimental::ServerCallbackRpcController* controller) override { + auto s = SetResponse(request, response); + controller->Finish(s); + } + + private: + static Status SetResponse(const SimpleRequest* request, + SimpleResponse* response) { + if (request->response_size() > 0) { + if (!Server::SetPayload(request->response_type(), + request->response_size(), + response->mutable_payload())) { + return Status(grpc::StatusCode::INTERNAL, "Error creating payload."); + } + } + return Status::OK; + } +}; + +class CallbackServer final : public grpc::testing::Server { + public: + explicit CallbackServer(const ServerConfig& config) : Server(config) { + std::unique_ptr<ServerBuilder> builder = CreateQpsServerBuilder(); + + auto port_num = port(); + // Negative port number means inproc server, so no listen port needed + if (port_num >= 0) { + char* server_address = nullptr; + gpr_join_host_port(&server_address, "::", port_num); + builder->AddListeningPort(server_address, + Server::CreateServerCredentials(config)); + gpr_free(server_address); + } + + ApplyConfigToBuilder(config, builder.get()); + + builder->RegisterService(&service_); + + impl_ = builder->BuildAndStart(); + } + + std::shared_ptr<Channel> InProcessChannel( + const ChannelArguments& args) override { + return impl_->InProcessChannel(args); + } + + private: + BenchmarkCallbackServiceImpl service_; + std::unique_ptr<grpc::Server> impl_; +}; + +std::unique_ptr<grpc::testing::Server> CreateCallbackServer( + const ServerConfig& config) { + return std::unique_ptr<Server>(new CallbackServer(config)); +} + +} // namespace testing +} // namespace grpc diff --git a/test/cpp/server/load_reporter/get_cpu_stats_test.cc b/test/cpp/server/load_reporter/get_cpu_stats_test.cc index 5b1d5fa3a4..98c15b2c69 100644 --- a/test/cpp/server/load_reporter/get_cpu_stats_test.cc +++ b/test/cpp/server/load_reporter/get_cpu_stats_test.cc @@ -55,7 +55,7 @@ TEST(GetCpuStatsTest, Ascending) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/server/load_reporter/load_data_store_test.cc b/test/cpp/server/load_reporter/load_data_store_test.cc index c92c407e4f..a3a67a2851 100644 --- a/test/cpp/server/load_reporter/load_data_store_test.cc +++ b/test/cpp/server/load_reporter/load_data_store_test.cc @@ -475,7 +475,7 @@ TEST_F(PerBalancerStoreTest, DataAggregation) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/server/load_reporter/load_reporter_test.cc b/test/cpp/server/load_reporter/load_reporter_test.cc index 0d56cdf431..9d2ebfb0b4 100644 --- a/test/cpp/server/load_reporter/load_reporter_test.cc +++ b/test/cpp/server/load_reporter/load_reporter_test.cc @@ -501,7 +501,7 @@ TEST_F(LoadReportTest, BasicReport) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/util/BUILD b/test/cpp/util/BUILD index c8d4333ef0..61e65029ff 100644 --- a/test/cpp/util/BUILD +++ b/test/cpp/util/BUILD @@ -181,16 +181,19 @@ grpc_cc_test( data = [ "//src/proto/grpc/testing:echo.proto", "//src/proto/grpc/testing:echo_messages.proto", + "//src/proto/grpc/testing:simple_messages.proto", ], external_deps = [ "gtest", ], + tags = ["nomsan"], # death tests seem to be incompatible with msan deps = [ ":grpc_cli_libs", ":test_util", "//:grpc++_reflection", "//src/proto/grpc/testing:echo_messages_proto", "//src/proto/grpc/testing:echo_proto", + "//src/proto/grpc/testing:simple_messages_proto", "//test/core/end2end:ssl_test_data", "//test/core/util:grpc_test_util", ], diff --git a/test/cpp/util/byte_buffer_proto_helper.h b/test/cpp/util/byte_buffer_proto_helper.h index eb923eccb5..3d01fb2468 100644 --- a/test/cpp/util/byte_buffer_proto_helper.h +++ b/test/cpp/util/byte_buffer_proto_helper.h @@ -27,12 +27,13 @@ namespace grpc { namespace testing { -bool ParseFromByteBuffer(ByteBuffer* buffer, grpc::protobuf::Message* message); +bool ParseFromByteBuffer(ByteBuffer* buffer, + ::grpc::protobuf::Message* message); std::unique_ptr<ByteBuffer> SerializeToByteBuffer( - grpc::protobuf::Message* message); + ::grpc::protobuf::Message* message); -bool SerializeToByteBufferInPlace(grpc::protobuf::Message* message, +bool SerializeToByteBufferInPlace(::grpc::protobuf::Message* message, ByteBuffer* buffer); } // namespace testing diff --git a/test/cpp/util/channel_trace_proto_helper.cc b/test/cpp/util/channel_trace_proto_helper.cc index 42a436d49b..ff9d887385 100644 --- a/test/cpp/util/channel_trace_proto_helper.cc +++ b/test/cpp/util/channel_trace_proto_helper.cc @@ -82,6 +82,11 @@ void ValidateGetChannelResponseProtoJsonTranslation(char* json_c_str) { json_c_str); } +void ValidateGetServerResponseProtoJsonTranslation(char* json_c_str) { + VaidateProtoJsonTranslation<grpc::channelz::v1::GetServerResponse>( + json_c_str); +} + void ValidateSubchannelProtoJsonTranslation(char* json_c_str) { VaidateProtoJsonTranslation<grpc::channelz::v1::Subchannel>(json_c_str); } diff --git a/test/cpp/util/channel_trace_proto_helper.h b/test/cpp/util/channel_trace_proto_helper.h index 67c363e89b..4f74e02f10 100644 --- a/test/cpp/util/channel_trace_proto_helper.h +++ b/test/cpp/util/channel_trace_proto_helper.h @@ -26,6 +26,7 @@ void ValidateChannelTraceProtoJsonTranslation(char* json_c_str); void ValidateChannelProtoJsonTranslation(char* json_c_str); void ValidateGetTopChannelsResponseProtoJsonTranslation(char* json_c_str); void ValidateGetChannelResponseProtoJsonTranslation(char* json_c_str); +void ValidateGetServerResponseProtoJsonTranslation(char* json_c_str); void ValidateSubchannelProtoJsonTranslation(char* json_c_str); void ValidateServerProtoJsonTranslation(char* json_c_str); void ValidateGetServersResponseProtoJsonTranslation(char* json_c_str); diff --git a/test/cpp/util/cli_call_test.cc b/test/cpp/util/cli_call_test.cc index 516f3fa53d..1d641535e2 100644 --- a/test/cpp/util/cli_call_test.cc +++ b/test/cpp/util/cli_call_test.cc @@ -122,7 +122,7 @@ TEST_F(CliCallTest, SimpleRpc) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); } diff --git a/test/cpp/util/grpc_tool_test.cc b/test/cpp/util/grpc_tool_test.cc index be9a624a2c..b96b00f2db 100644 --- a/test/cpp/util/grpc_tool_test.cc +++ b/test/cpp/util/grpc_tool_test.cc @@ -48,6 +48,7 @@ using grpc::testing::EchoResponse; #define ECHO_TEST_SERVICE_SUMMARY \ "Echo\n" \ + "CheckClientInitialMetadata\n" \ "RequestStream\n" \ "ResponseStream\n" \ "BidiStream\n" \ @@ -59,6 +60,8 @@ using grpc::testing::EchoResponse; "service EchoTestService {\n" \ " rpc Echo(grpc.testing.EchoRequest) returns (grpc.testing.EchoResponse) " \ "{}\n" \ + " rpc CheckClientInitialMetadata(grpc.testing.SimpleRequest) returns " \ + "(grpc.testing.SimpleResponse) {}\n" \ " rpc RequestStream(stream grpc.testing.EchoRequest) returns " \ "(grpc.testing.EchoResponse) {}\n" \ " rpc ResponseStream(grpc.testing.EchoRequest) returns (stream " \ @@ -1173,7 +1176,7 @@ TEST_F(GrpcToolTest, ListCommand_OverrideSslHostName) { } // namespace grpc int main(int argc, char** argv) { - grpc_test_init(argc, argv); + grpc::testing::TestEnvironment env(argc, argv); ::testing::InitGoogleTest(&argc, argv); ::testing::FLAGS_gtest_death_test_style = "threadsafe"; return RUN_ALL_TESTS(); |