aboutsummaryrefslogtreecommitdiffhomepage
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/core/iomgr/ev_epoll_linux_test.c244
-rw-r--r--test/core/security/credentials_test.c49
-rw-r--r--test/core/security/oauth2_utils.c3
-rw-r--r--test/core/security/print_google_default_creds_token.c3
-rw-r--r--test/cpp/end2end/end2end_test.cc6
-rw-r--r--test/cpp/interop/client.cc102
-rw-r--r--test/cpp/interop/interop_client.cc357
-rw-r--r--test/cpp/interop/interop_client.h6
-rw-r--r--test/cpp/interop/interop_server.cc (renamed from test/cpp/interop/server_main.cc)111
-rw-r--r--test/cpp/interop/rnd.datbin524288 -> 0 bytes
-rw-r--r--test/cpp/interop/server_helper.cc4
-rw-r--r--test/cpp/interop/server_helper.h1
-rw-r--r--test/cpp/interop/stress_interop_client.cc14
-rw-r--r--test/cpp/interop/stress_interop_client.h36
14 files changed, 664 insertions, 272 deletions
diff --git a/test/core/iomgr/ev_epoll_linux_test.c b/test/core/iomgr/ev_epoll_linux_test.c
new file mode 100644
index 0000000000..2547dc9871
--- /dev/null
+++ b/test/core/iomgr/ev_epoll_linux_test.c
@@ -0,0 +1,244 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+#include <grpc/support/port_platform.h>
+
+/* This test only relevant on linux systems where epoll() is available */
+#ifdef GPR_LINUX_EPOLL
+#include "src/core/lib/iomgr/ev_epoll_linux.h"
+#include "src/core/lib/iomgr/ev_posix.h"
+
+#include <errno.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+
+#include "src/core/lib/iomgr/iomgr.h"
+#include "test/core/util/test_config.h"
+
+typedef struct test_pollset {
+ grpc_pollset *pollset;
+ gpr_mu *mu;
+} test_pollset;
+
+typedef struct test_fd {
+ int inner_fd;
+ grpc_fd *fd;
+} test_fd;
+
+/* num_fds should be an even number */
+static void test_fd_init(test_fd *tfds, int *fds, int num_fds) {
+ int i;
+ for (i = 0; i < num_fds; i++) {
+ tfds[i].inner_fd = fds[i];
+ tfds[i].fd = grpc_fd_create(fds[i], "test_fd");
+ }
+}
+
+static void test_fd_cleanup(grpc_exec_ctx *exec_ctx, test_fd *tfds,
+ int num_fds) {
+ int release_fd;
+ int i;
+
+ for (i = 0; i < num_fds; i++) {
+ grpc_fd_shutdown(exec_ctx, tfds[i].fd);
+ grpc_exec_ctx_flush(exec_ctx);
+
+ grpc_fd_orphan(exec_ctx, tfds[i].fd, NULL, &release_fd, "test_fd_cleanup");
+ grpc_exec_ctx_flush(exec_ctx);
+
+ GPR_ASSERT(release_fd == tfds[i].inner_fd);
+ close(tfds[i].inner_fd);
+ }
+}
+
+static void test_pollset_init(test_pollset *pollsets, int num_pollsets) {
+ int i;
+ for (i = 0; i < num_pollsets; i++) {
+ pollsets[i].pollset = gpr_malloc(grpc_pollset_size());
+ grpc_pollset_init(pollsets[i].pollset, &pollsets[i].mu);
+ }
+}
+
+static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p,
+ grpc_error *error) {
+ grpc_pollset_destroy(p);
+}
+
+static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx,
+ test_pollset *pollsets, int num_pollsets) {
+ grpc_closure destroyed;
+ int i;
+
+ for (i = 0; i < num_pollsets; i++) {
+ grpc_closure_init(&destroyed, destroy_pollset, pollsets[i].pollset);
+ grpc_pollset_shutdown(exec_ctx, pollsets[i].pollset, &destroyed);
+
+ grpc_exec_ctx_flush(exec_ctx);
+ gpr_free(pollsets[i].pollset);
+ }
+}
+
+#define NUM_FDS 8
+#define NUM_POLLSETS 4
+/*
+ * Cases to test:
+ * case 1) Polling islands of both fd and pollset are NULL
+ * case 2) Polling island of fd is NULL but that of pollset is not-NULL
+ * case 3) Polling island of fd is not-NULL but that of pollset is NULL
+ * case 4) Polling islands of both fd and pollset are not-NULL and:
+ * case 4.1) Polling islands of fd and pollset are equal
+ * case 4.2) Polling islands of fd and pollset are NOT-equal (This results
+ * in a merge)
+ * */
+static void test_add_fd_to_pollset() {
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ test_fd tfds[NUM_FDS];
+ int fds[NUM_FDS];
+ test_pollset pollsets[NUM_POLLSETS];
+ void *expected_pi = NULL;
+ int i;
+ int r;
+
+ /* Create some dummy file descriptors. Currently using pipe file descriptors
+ * for this test but we could use any other type of file descriptors. Also,
+ * since pipe() used in this test creates two fds in each call, NUM_FDS should
+ * be an even number */
+ for (i = 0; i < NUM_FDS; i = i + 2) {
+ r = pipe(fds + i);
+ if (r != 0) {
+ gpr_log(GPR_ERROR, "Error in creating pipe. %d (%s)", errno,
+ strerror(errno));
+ return;
+ }
+ }
+
+ test_fd_init(tfds, fds, NUM_FDS);
+ test_pollset_init(pollsets, NUM_POLLSETS);
+
+ /*Step 1.
+ * Create three polling islands (This will exercise test case 1 and 2) with
+ * the following configuration:
+ * polling island 0 = { fds:0,1,2, pollsets:0}
+ * polling island 1 = { fds:3,4, pollsets:1}
+ * polling island 2 = { fds:5,6,7 pollsets:2}
+ *
+ *Step 2.
+ * Add pollset 3 to polling island 0 (by adding fds 0 and 1 to pollset 3)
+ * (This will exercise test cases 3 and 4.1). The configuration becomes:
+ * polling island 0 = { fds:0,1,2, pollsets:0,3} <<< pollset 3 added here
+ * polling island 1 = { fds:3,4, pollsets:1}
+ * polling island 2 = { fds:5,6,7 pollsets:2}
+ *
+ *Step 3.
+ * Merge polling islands 0 and 1 by adding fd 0 to pollset 1 (This will
+ * exercise test case 4.2). The configuration becomes:
+ * polling island (merged) = {fds: 0,1,2,3,4, pollsets: 0,1,3}
+ * polling island 2 = {fds: 5,6,7 pollsets: 2}
+ *
+ *Step 4.
+ * Finally do one more merge by adding fd 3 to pollset 2.
+ * polling island (merged) = {fds: 0,1,2,3,4,5,6,7, pollsets: 0,1,2,3}
+ */
+
+ /* == Step 1 == */
+ for (i = 0; i <= 2; i++) {
+ grpc_pollset_add_fd(&exec_ctx, pollsets[0].pollset, tfds[i].fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+ }
+
+ for (i = 3; i <= 4; i++) {
+ grpc_pollset_add_fd(&exec_ctx, pollsets[1].pollset, tfds[i].fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+ }
+
+ for (i = 5; i <= 7; i++) {
+ grpc_pollset_add_fd(&exec_ctx, pollsets[2].pollset, tfds[i].fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+ }
+
+ /* == Step 2 == */
+ for (i = 0; i <= 1; i++) {
+ grpc_pollset_add_fd(&exec_ctx, pollsets[3].pollset, tfds[i].fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+ }
+
+ /* == Step 3 == */
+ grpc_pollset_add_fd(&exec_ctx, pollsets[1].pollset, tfds[0].fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+
+ /* == Step 4 == */
+ grpc_pollset_add_fd(&exec_ctx, pollsets[2].pollset, tfds[3].fd);
+ grpc_exec_ctx_flush(&exec_ctx);
+
+ /* All polling islands are merged at this point */
+
+ /* Compare Fd:0's polling island with that of all other Fds */
+ expected_pi = grpc_fd_get_polling_island(tfds[0].fd);
+ for (i = 1; i < NUM_FDS; i++) {
+ GPR_ASSERT(grpc_are_polling_islands_equal(
+ expected_pi, grpc_fd_get_polling_island(tfds[i].fd)));
+ }
+
+ /* Compare Fd:0's polling island with that of all other pollsets */
+ for (i = 0; i < NUM_POLLSETS; i++) {
+ GPR_ASSERT(grpc_are_polling_islands_equal(
+ expected_pi, grpc_pollset_get_polling_island(pollsets[i].pollset)));
+ }
+
+ test_fd_cleanup(&exec_ctx, tfds, NUM_FDS);
+ test_pollset_cleanup(&exec_ctx, pollsets, NUM_POLLSETS);
+ grpc_exec_ctx_finish(&exec_ctx);
+}
+
+int main(int argc, char **argv) {
+ const char *poll_strategy = NULL;
+ grpc_test_init(argc, argv);
+ grpc_iomgr_init();
+
+ poll_strategy = grpc_get_poll_strategy_name();
+ if (poll_strategy != NULL && strcmp(poll_strategy, "epoll") == 0) {
+ test_add_fd_to_pollset();
+ } else {
+ gpr_log(GPR_INFO,
+ "Skipping the test. The test is only relevant for 'epoll' "
+ "strategy. and the current strategy is: '%s'",
+ poll_strategy);
+ }
+ grpc_iomgr_shutdown();
+ return 0;
+}
+#else /* defined(GPR_LINUX_EPOLL) */
+int main(int argc, char **argv) { return 0; }
+#endif /* !defined(GPR_LINUX_EPOLL) */
diff --git a/test/core/security/credentials_test.c b/test/core/security/credentials_test.c
index e703dbdeb6..7043953154 100644
--- a/test/core/security/credentials_test.c
+++ b/test/core/security/credentials_test.c
@@ -348,13 +348,15 @@ static void check_metadata(expected_md *expected, grpc_credentials_md *md_elems,
static void check_google_iam_metadata(grpc_exec_ctx *exec_ctx, void *user_data,
grpc_credentials_md *md_elems,
size_t num_md,
- grpc_credentials_status status) {
+ grpc_credentials_status status,
+ const char *error_details) {
grpc_call_credentials *c = (grpc_call_credentials *)user_data;
expected_md emd[] = {{GRPC_IAM_AUTHORIZATION_TOKEN_METADATA_KEY,
test_google_iam_authorization_token},
{GRPC_IAM_AUTHORITY_SELECTOR_METADATA_KEY,
test_google_iam_authority_selector}};
GPR_ASSERT(status == GRPC_CREDENTIALS_OK);
+ GPR_ASSERT(error_details == NULL);
GPR_ASSERT(num_md == 2);
check_metadata(emd, md_elems, num_md);
grpc_call_credentials_unref(c);
@@ -372,14 +374,13 @@ static void test_google_iam_creds(void) {
grpc_exec_ctx_finish(&exec_ctx);
}
-static void check_access_token_metadata(grpc_exec_ctx *exec_ctx,
- void *user_data,
- grpc_credentials_md *md_elems,
- size_t num_md,
- grpc_credentials_status status) {
+static void check_access_token_metadata(
+ grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
grpc_call_credentials *c = (grpc_call_credentials *)user_data;
expected_md emd[] = {{GRPC_AUTHORIZATION_METADATA_KEY, "Bearer blah"}};
GPR_ASSERT(status == GRPC_CREDENTIALS_OK);
+ GPR_ASSERT(error_details == NULL);
GPR_ASSERT(num_md == 1);
check_metadata(emd, md_elems, num_md);
grpc_call_credentials_unref(c);
@@ -428,7 +429,7 @@ static void test_channel_oauth2_composite_creds(void) {
static void check_oauth2_google_iam_composite_metadata(
grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
- size_t num_md, grpc_credentials_status status) {
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
grpc_call_credentials *c = (grpc_call_credentials *)user_data;
expected_md emd[] = {
{GRPC_AUTHORIZATION_METADATA_KEY, test_oauth2_bearer_token},
@@ -437,6 +438,7 @@ static void check_oauth2_google_iam_composite_metadata(
{GRPC_IAM_AUTHORITY_SELECTOR_METADATA_KEY,
test_google_iam_authority_selector}};
GPR_ASSERT(status == GRPC_CREDENTIALS_OK);
+ GPR_ASSERT(error_details == NULL);
GPR_ASSERT(num_md == 3);
check_metadata(emd, md_elems, num_md);
grpc_call_credentials_unref(c);
@@ -521,8 +523,9 @@ static void test_channel_oauth2_google_iam_composite_creds(void) {
static void on_oauth2_creds_get_metadata_success(
grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
- size_t num_md, grpc_credentials_status status) {
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
GPR_ASSERT(status == GRPC_CREDENTIALS_OK);
+ GPR_ASSERT(error_details == NULL);
GPR_ASSERT(num_md == 1);
GPR_ASSERT(gpr_slice_str_cmp(md_elems[0].key, "authorization") == 0);
GPR_ASSERT(gpr_slice_str_cmp(md_elems[0].value,
@@ -534,7 +537,7 @@ static void on_oauth2_creds_get_metadata_success(
static void on_oauth2_creds_get_metadata_failure(
grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
- size_t num_md, grpc_credentials_status status) {
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
GPR_ASSERT(status == GRPC_CREDENTIALS_ERROR);
GPR_ASSERT(num_md == 0);
GPR_ASSERT(user_data != NULL);
@@ -769,14 +772,13 @@ static char *encode_and_sign_jwt_should_not_be_called(
return NULL;
}
-static void on_jwt_creds_get_metadata_success(grpc_exec_ctx *exec_ctx,
- void *user_data,
- grpc_credentials_md *md_elems,
- size_t num_md,
- grpc_credentials_status status) {
+static void on_jwt_creds_get_metadata_success(
+ grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
char *expected_md_value;
gpr_asprintf(&expected_md_value, "Bearer %s", test_signed_jwt);
GPR_ASSERT(status == GRPC_CREDENTIALS_OK);
+ GPR_ASSERT(error_details == NULL);
GPR_ASSERT(num_md == 1);
GPR_ASSERT(gpr_slice_str_cmp(md_elems[0].key, "authorization") == 0);
GPR_ASSERT(gpr_slice_str_cmp(md_elems[0].value, expected_md_value) == 0);
@@ -785,11 +787,9 @@ static void on_jwt_creds_get_metadata_success(grpc_exec_ctx *exec_ctx,
gpr_free(expected_md_value);
}
-static void on_jwt_creds_get_metadata_failure(grpc_exec_ctx *exec_ctx,
- void *user_data,
- grpc_credentials_md *md_elems,
- size_t num_md,
- grpc_credentials_status status) {
+static void on_jwt_creds_get_metadata_failure(
+ grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
GPR_ASSERT(status == GRPC_CREDENTIALS_ERROR);
GPR_ASSERT(num_md == 0);
GPR_ASSERT(user_data != NULL);
@@ -1033,6 +1033,8 @@ static void plugin_get_metadata_success(void *state,
cb(user_data, md, GPR_ARRAY_SIZE(md), GRPC_STATUS_OK, NULL);
}
+static const char *plugin_error_details = "Could not get metadata for plugin.";
+
static void plugin_get_metadata_failure(void *state,
grpc_auth_metadata_context context,
grpc_credentials_plugin_metadata_cb cb,
@@ -1043,13 +1045,12 @@ static void plugin_get_metadata_failure(void *state,
GPR_ASSERT(context.channel_auth_context == NULL);
GPR_ASSERT(context.reserved == NULL);
*s = PLUGIN_GET_METADATA_CALLED_STATE;
- cb(user_data, NULL, 0, GRPC_STATUS_UNAUTHENTICATED,
- "Could not get metadata for plugin.");
+ cb(user_data, NULL, 0, GRPC_STATUS_UNAUTHENTICATED, plugin_error_details);
}
static void on_plugin_metadata_received_success(
grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
- size_t num_md, grpc_credentials_status status) {
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
size_t i = 0;
GPR_ASSERT(user_data == NULL);
GPR_ASSERT(md_elems != NULL);
@@ -1062,11 +1063,13 @@ static void on_plugin_metadata_received_success(
static void on_plugin_metadata_received_failure(
grpc_exec_ctx *exec_ctx, void *user_data, grpc_credentials_md *md_elems,
- size_t num_md, grpc_credentials_status status) {
+ size_t num_md, grpc_credentials_status status, const char *error_details) {
GPR_ASSERT(user_data == NULL);
GPR_ASSERT(md_elems == NULL);
GPR_ASSERT(num_md == 0);
GPR_ASSERT(status == GRPC_CREDENTIALS_ERROR);
+ GPR_ASSERT(error_details != NULL);
+ GPR_ASSERT(strcmp(error_details, plugin_error_details) == 0);
}
static void plugin_destroy(void *state) {
diff --git a/test/core/security/oauth2_utils.c b/test/core/security/oauth2_utils.c
index a334edc32d..9b97c38fcb 100644
--- a/test/core/security/oauth2_utils.c
+++ b/test/core/security/oauth2_utils.c
@@ -53,7 +53,8 @@ typedef struct {
static void on_oauth2_response(grpc_exec_ctx *exec_ctx, void *user_data,
grpc_credentials_md *md_elems, size_t num_md,
- grpc_credentials_status status) {
+ grpc_credentials_status status,
+ const char *error_details) {
oauth2_request *request = user_data;
char *token = NULL;
gpr_slice token_slice;
diff --git a/test/core/security/print_google_default_creds_token.c b/test/core/security/print_google_default_creds_token.c
index 18fbc3c41c..a391c0876b 100644
--- a/test/core/security/print_google_default_creds_token.c
+++ b/test/core/security/print_google_default_creds_token.c
@@ -54,7 +54,8 @@ typedef struct {
static void on_metadata_response(grpc_exec_ctx *exec_ctx, void *user_data,
grpc_credentials_md *md_elems, size_t num_md,
- grpc_credentials_status status) {
+ grpc_credentials_status status,
+ const char *error_details) {
synchronizer *sync = user_data;
if (status == GRPC_CREDENTIALS_ERROR) {
fprintf(stderr, "Fetching token failed.\n");
diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 8de9d339f6..354a59cedd 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -75,6 +75,8 @@ bool CheckIsLocalhost(const grpc::string& addr) {
addr.substr(0, kIpv6.size()) == kIpv6;
}
+const char kTestCredsPluginErrorMsg[] = "Could not find plugin metadata.";
+
class TestMetadataCredentialsPlugin : public MetadataCredentialsPlugin {
public:
static const char kMetadataKey[];
@@ -99,7 +101,7 @@ class TestMetadataCredentialsPlugin : public MetadataCredentialsPlugin {
metadata->insert(std::make_pair(kMetadataKey, metadata_value_));
return Status::OK;
} else {
- return Status(StatusCode::NOT_FOUND, "Could not find plugin metadata.");
+ return Status(StatusCode::NOT_FOUND, kTestCredsPluginErrorMsg);
}
}
@@ -1331,6 +1333,7 @@ TEST_P(SecureEnd2endTest, NonBlockingAuthMetadataPluginFailure) {
Status s = stub_->Echo(&context, request, &response);
EXPECT_FALSE(s.ok());
EXPECT_EQ(s.error_code(), StatusCode::UNAUTHENTICATED);
+ EXPECT_EQ(s.error_message(), kTestCredsPluginErrorMsg);
}
TEST_P(SecureEnd2endTest, NonBlockingAuthMetadataPluginAndProcessorSuccess) {
@@ -1388,6 +1391,7 @@ TEST_P(SecureEnd2endTest, BlockingAuthMetadataPluginFailure) {
Status s = stub_->Echo(&context, request, &response);
EXPECT_FALSE(s.ok());
EXPECT_EQ(s.error_code(), StatusCode::UNAUTHENTICATED);
+ EXPECT_EQ(s.error_message(), kTestCredsPluginErrorMsg);
}
TEST_P(SecureEnd2endTest, ClientAuthContext) {
diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc
index addaf174f2..e8ae6ee572 100644
--- a/test/cpp/interop/client.cc
+++ b/test/cpp/interop/client.cc
@@ -40,7 +40,9 @@
#include <grpc++/client_context.h>
#include <grpc/grpc.h>
#include <grpc/support/log.h>
+#include <grpc/support/useful.h>
+#include "src/core/lib/support/string.h"
#include "test/cpp/interop/client_helper.h"
#include "test/cpp/interop/interop_client.h"
#include "test/cpp/util/test_config.h"
@@ -52,30 +54,31 @@ DEFINE_string(server_host, "127.0.0.1", "Server host to connect to");
DEFINE_string(server_host_override, "foo.test.google.fr",
"Override the server host which is sent in HTTP header");
DEFINE_string(test_case, "large_unary",
- "Configure different test cases. Valid options are: "
- "empty_unary : empty (zero bytes) request and response; "
- "large_unary : single request and (large) response; "
- "large_compressed_unary : single request and compressed (large) "
- "response; "
- "client_streaming : request streaming with single response; "
- "server_streaming : single request with response streaming; "
+ "Configure different test cases. Valid options are:\n\n"
+ "all : all test cases;\n"
+ "cancel_after_begin : cancel stream after starting it;\n"
+ "cancel_after_first_response: cancel on first response;\n"
+ "client_compressed_streaming : compressed request streaming with "
+ "client_compressed_unary : single compressed request;\n"
+ "client_streaming : request streaming with single response;\n"
+ "compute_engine_creds: large_unary with compute engine auth;\n"
+ "custom_metadata: server will echo custom metadata;\n"
+ "empty_stream : bi-di stream with no request/response;\n"
+ "empty_unary : empty (zero bytes) request and response;\n"
+ "half_duplex : half-duplex streaming;\n"
+ "jwt_token_creds: large_unary with JWT token auth;\n"
+ "large_unary : single request and (large) response;\n"
+ "oauth2_auth_token: raw oauth2 access token auth;\n"
+ "per_rpc_creds: raw oauth2 access token on a single rpc;\n"
+ "ping_pong : full-duplex streaming;\n"
+ "response streaming;\n"
"server_compressed_streaming : single request with compressed "
- "response streaming; "
- "slow_consumer : single request with response; "
- " streaming with slow client consumer; "
- "half_duplex : half-duplex streaming; "
- "ping_pong : full-duplex streaming; "
- "cancel_after_begin : cancel stream after starting it; "
- "cancel_after_first_response: cancel on first response; "
- "timeout_on_sleeping_server: deadline exceeds on stream; "
- "empty_stream : bi-di stream with no request/response; "
- "compute_engine_creds: large_unary with compute engine auth; "
- "jwt_token_creds: large_unary with JWT token auth; "
- "oauth2_auth_token: raw oauth2 access token auth; "
- "per_rpc_creds: raw oauth2 access token on a single rpc; "
- "status_code_and_message: verify status code & message; "
- "custom_metadata: server will echo custom metadata;"
- "all : all of above.");
+ "server_compressed_unary : single compressed response;\n"
+ "server_streaming : single request with response streaming;\n"
+ "slow_consumer : single request with response streaming with "
+ "slow client consumer;\n"
+ "status_code_and_message: verify status code & message;\n"
+ "timeout_on_sleeping_server: deadline exceeds on stream;\n");
DEFINE_string(default_service_account, "",
"Email of GCE default service account");
DEFINE_string(service_account_key_file, "",
@@ -104,14 +107,18 @@ int main(int argc, char** argv) {
client.DoEmpty();
} else if (FLAGS_test_case == "large_unary") {
client.DoLargeUnary();
- } else if (FLAGS_test_case == "large_compressed_unary") {
- client.DoLargeCompressedUnary();
+ } else if (FLAGS_test_case == "server_compressed_unary") {
+ client.DoServerCompressedUnary();
+ } else if (FLAGS_test_case == "client_compressed_unary") {
+ client.DoClientCompressedUnary();
} else if (FLAGS_test_case == "client_streaming") {
client.DoRequestStreaming();
} else if (FLAGS_test_case == "server_streaming") {
client.DoResponseStreaming();
} else if (FLAGS_test_case == "server_compressed_streaming") {
- client.DoResponseCompressedStreaming();
+ client.DoServerCompressedStreaming();
+ } else if (FLAGS_test_case == "client_compressed_streaming") {
+ client.DoClientCompressedStreaming();
} else if (FLAGS_test_case == "slow_consumer") {
client.DoResponseStreamingWithSlowConsumer();
} else if (FLAGS_test_case == "half_duplex") {
@@ -144,9 +151,12 @@ int main(int argc, char** argv) {
} else if (FLAGS_test_case == "all") {
client.DoEmpty();
client.DoLargeUnary();
+ client.DoClientCompressedUnary();
+ client.DoServerCompressedUnary();
client.DoRequestStreaming();
client.DoResponseStreaming();
- client.DoResponseCompressedStreaming();
+ client.DoClientCompressedStreaming();
+ client.DoServerCompressedStreaming();
client.DoHalfDuplex();
client.DoPingPong();
client.DoCancelAfterBegin();
@@ -165,15 +175,35 @@ int main(int argc, char** argv) {
}
// compute_engine_creds only runs in GCE.
} else {
- gpr_log(
- GPR_ERROR,
- "Unsupported test case %s. Valid options are all|empty_unary|"
- "large_unary|large_compressed_unary|client_streaming|server_streaming|"
- "server_compressed_streaming|half_duplex|ping_pong|cancel_after_begin|"
- "cancel_after_first_response|timeout_on_sleeping_server|empty_stream|"
- "compute_engine_creds|jwt_token_creds|oauth2_auth_token|per_rpc_creds|"
- "status_code_and_message|custom_metadata",
- FLAGS_test_case.c_str());
+ const char* testcases[] = {"all",
+ "cancel_after_begin",
+ "cancel_after_first_response",
+ "client_compressed_streaming",
+ "client_compressed_unary",
+ "client_streaming",
+ "compute_engine_creds",
+ "custom_metadata",
+ "empty_stream",
+ "empty_unary",
+ "half_duplex",
+ "jwt_token_creds",
+ "large_unary",
+ "oauth2_auth_token",
+ "oauth2_auth_token",
+ "per_rpc_creds",
+ "per_rpc_creds",
+ "ping_pong",
+ "server_compressed_streaming",
+ "server_compressed_unary",
+ "server_streaming",
+ "status_code_and_message",
+ "timeout_on_sleeping_server"};
+ char* joined_testcases =
+ gpr_strjoin_sep(testcases, GPR_ARRAY_SIZE(testcases), "\n", NULL);
+
+ gpr_log(GPR_ERROR, "Unsupported test case %s. Valid options are\n%s",
+ FLAGS_test_case.c_str(), joined_testcases);
+ gpr_free(joined_testcases);
ret = 1;
}
diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc
index 608f902d6f..89f841dbe9 100644
--- a/test/cpp/interop/interop_client.cc
+++ b/test/cpp/interop/interop_client.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -57,7 +57,7 @@ namespace testing {
namespace {
// The same value is defined by the Java client.
const std::vector<int> request_stream_sizes = {27182, 8, 1828, 45904};
-const std::vector<int> response_stream_sizes = {31415, 59, 2653, 58979};
+const std::vector<int> response_stream_sizes = {31415, 9, 2653, 58979};
const int kNumResponseMessages = 2000;
const int kResponseMessageSize = 1030;
const int kReceiveDelayMilliSeconds = 20;
@@ -67,28 +67,23 @@ const int kLargeResponseSize = 314159;
void NoopChecks(const InteropClientContextInspector& inspector,
const SimpleRequest* request, const SimpleResponse* response) {}
-void CompressionChecks(const InteropClientContextInspector& inspector,
- const SimpleRequest* request,
- const SimpleResponse* response) {
+void UnaryCompressionChecks(const InteropClientContextInspector& inspector,
+ const SimpleRequest* request,
+ const SimpleResponse* response) {
const grpc_compression_algorithm received_compression =
inspector.GetCallCompressionAlgorithm();
- if (request->request_compressed_response() &&
- received_compression == GRPC_COMPRESS_NONE) {
- if (request->request_compressed_response() &&
- received_compression == GRPC_COMPRESS_NONE) {
+ if (request->response_compressed().value()) {
+ if (received_compression == GRPC_COMPRESS_NONE) {
// Requested some compression, got NONE. This is an error.
gpr_log(GPR_ERROR,
"Failure: Requested compression but got uncompressed response "
"from server.");
abort();
}
- }
- if (!request->request_compressed_response()) {
- GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
- } else if (request->response_type() == PayloadType::COMPRESSABLE) {
- // requested compression and compressable response => results should always
- // be compressed.
GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
+ } else {
+ // Didn't request compression -> make sure the response is uncompressed
+ GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
}
}
} // namespace
@@ -190,11 +185,16 @@ bool InteropClient::PerformLargeUnary(SimpleRequest* request,
CheckerFn custom_checks_fn) {
ClientContext context;
InteropClientContextInspector inspector(context);
- // If the request doesn't already specify the response type, default to
- // COMPRESSABLE.
request->set_response_size(kLargeResponseSize);
grpc::string payload(kLargeRequestSize, '\0');
request->mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
+ if (request->has_expect_compressed()) {
+ if (request->expect_compressed().value()) {
+ context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
+ } else {
+ context.set_compression_algorithm(GRPC_COMPRESS_NONE);
+ }
+ }
Status s = serviceStub_.Get()->UnaryCall(&context, *request, response);
if (!AssertStatusOk(s)) {
@@ -204,27 +204,8 @@ bool InteropClient::PerformLargeUnary(SimpleRequest* request,
custom_checks_fn(inspector, request, response);
// Payload related checks.
- GPR_ASSERT(response->payload().type() == request->response_type());
- switch (response->payload().type()) {
- case PayloadType::COMPRESSABLE:
- GPR_ASSERT(response->payload().body() ==
- grpc::string(kLargeResponseSize, '\0'));
- break;
- case PayloadType::UNCOMPRESSABLE: {
- // We don't really check anything: We can't assert that the payload is
- // uncompressed because it's the server's prerogative to decide on that,
- // and different implementations decide differently (ie, Java always
- // compresses when requested to do so, whereas C core throws away the
- // compressed payload if the output is larger than the input).
- // In addition, we don't compare the actual random bytes received because
- // asserting that data is sent/received properly isn't the purpose of this
- // test. Moreover, different implementations are also free to use
- // different sets of random bytes.
- } break;
- default:
- GPR_ASSERT(false);
- }
-
+ GPR_ASSERT(response->payload().body() ==
+ grpc::string(kLargeResponseSize, '\0'));
return true;
}
@@ -237,7 +218,6 @@ bool InteropClient::DoComputeEngineCreds(
SimpleResponse response;
request.set_fill_username(true);
request.set_fill_oauth_scope(true);
- request.set_response_type(PayloadType::COMPRESSABLE);
if (!PerformLargeUnary(&request, &response)) {
return false;
@@ -311,7 +291,6 @@ bool InteropClient::DoJwtTokenCreds(const grpc::string& username) {
SimpleRequest request;
SimpleResponse response;
request.set_fill_username(true);
- request.set_response_type(PayloadType::COMPRESSABLE);
if (!PerformLargeUnary(&request, &response)) {
return false;
@@ -327,7 +306,6 @@ bool InteropClient::DoLargeUnary() {
gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
SimpleRequest request;
SimpleResponse response;
- request.set_response_type(PayloadType::COMPRESSABLE);
if (!PerformLargeUnary(&request, &response)) {
return false;
}
@@ -335,32 +313,73 @@ bool InteropClient::DoLargeUnary() {
return true;
}
-bool InteropClient::DoLargeCompressedUnary() {
- const bool request_compression[] = {false, true};
- const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE};
- for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
- for (size_t j = 0; j < GPR_ARRAY_SIZE(request_compression); j++) {
- char* log_suffix;
- gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
- request_compression[j] ? "true" : "false",
- PayloadType_Name(payload_types[i]).c_str());
-
- gpr_log(GPR_DEBUG, "Sending a large compressed unary rpc %s.",
- log_suffix);
- SimpleRequest request;
- SimpleResponse response;
- request.set_response_type(payload_types[i]);
- request.set_request_compressed_response(request_compression[j]);
-
- if (!PerformLargeUnary(&request, &response, CompressionChecks)) {
- gpr_log(GPR_ERROR, "Large compressed unary failed %s", log_suffix);
- gpr_free(log_suffix);
- return false;
- }
-
- gpr_log(GPR_DEBUG, "Large compressed unary done %s.", log_suffix);
+bool InteropClient::DoClientCompressedUnary() {
+ // Probing for compression-checks support.
+ ClientContext probe_context;
+ SimpleRequest probe_req;
+ SimpleResponse probe_res;
+
+ probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
+ probe_req.mutable_expect_compressed()->set_value(true); // lies!
+
+ probe_req.set_response_size(kLargeResponseSize);
+ probe_req.mutable_payload()->set_body(grpc::string(kLargeRequestSize, '\0'));
+
+ gpr_log(GPR_DEBUG, "Sending probe for compressed unary request.");
+ const Status s =
+ serviceStub_.Get()->UnaryCall(&probe_context, probe_req, &probe_res);
+ if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
+ // The server isn't able to evaluate incoming compression, making the rest
+ // of this test moot.
+ gpr_log(GPR_DEBUG, "Compressed unary request probe failed");
+ return false;
+ }
+ gpr_log(GPR_DEBUG, "Compressed unary request probe succeeded. Proceeding.");
+
+ const std::vector<bool> compressions = {true, false};
+ for (size_t i = 0; i < compressions.size(); i++) {
+ char* log_suffix;
+ gpr_asprintf(&log_suffix, "(compression=%s)",
+ compressions[i] ? "true" : "false");
+
+ gpr_log(GPR_DEBUG, "Sending compressed unary request %s.", log_suffix);
+ SimpleRequest request;
+ SimpleResponse response;
+ request.mutable_expect_compressed()->set_value(compressions[i]);
+ if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
+ gpr_log(GPR_ERROR, "Compressed unary request failed %s", log_suffix);
+ gpr_free(log_suffix);
+ return false;
+ }
+
+ gpr_log(GPR_DEBUG, "Compressed unary request failed %s", log_suffix);
+ gpr_free(log_suffix);
+ }
+
+ return true;
+}
+
+bool InteropClient::DoServerCompressedUnary() {
+ const std::vector<bool> compressions = {true, false};
+ for (size_t i = 0; i < compressions.size(); i++) {
+ char* log_suffix;
+ gpr_asprintf(&log_suffix, "(compression=%s)",
+ compressions[i] ? "true" : "false");
+
+ gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.",
+ log_suffix);
+ SimpleRequest request;
+ SimpleResponse response;
+ request.mutable_response_compressed()->set_value(compressions[i]);
+
+ if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
+ gpr_log(GPR_ERROR, "Request for compressed unary failed %s", log_suffix);
gpr_free(log_suffix);
+ return false;
}
+
+ gpr_log(GPR_DEBUG, "Request for compressed unary failed %s", log_suffix);
+ gpr_free(log_suffix);
}
return true;
@@ -387,7 +406,7 @@ bool InteropClient::DoRequestStreaming() {
serviceStub_.Get()->StreamingInputCall(&context, &response));
int aggregated_payload_size = 0;
- for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
+ for (size_t i = 0; i < request_stream_sizes.size(); ++i) {
Payload* payload = request.mutable_payload();
payload->set_body(grpc::string(request_stream_sizes[i], '\0'));
if (!stream->Write(request)) {
@@ -396,7 +415,7 @@ bool InteropClient::DoRequestStreaming() {
}
aggregated_payload_size += request_stream_sizes[i];
}
- stream->WritesDone();
+ GPR_ASSERT(stream->WritesDone());
Status s = stream->Finish();
if (!AssertStatusOk(s)) {
@@ -446,92 +465,129 @@ bool InteropClient::DoResponseStreaming() {
return true;
}
-bool InteropClient::DoResponseCompressedStreaming() {
- const bool request_compression[] = {false, true};
- const PayloadType payload_types[] = {COMPRESSABLE, UNCOMPRESSABLE};
- for (size_t i = 0; i < GPR_ARRAY_SIZE(payload_types); i++) {
- for (size_t j = 0; j < GPR_ARRAY_SIZE(request_compression); j++) {
- ClientContext context;
- InteropClientContextInspector inspector(context);
- StreamingOutputCallRequest request;
-
- char* log_suffix;
- gpr_asprintf(&log_suffix, "(compression=%s; payload=%s)",
- request_compression[j] ? "true" : "false",
- PayloadType_Name(payload_types[i]).c_str());
-
- gpr_log(GPR_DEBUG, "Receiving response streaming rpc %s.", log_suffix);
-
- request.set_response_type(payload_types[i]);
- request.set_request_compressed_response(request_compression[j]);
-
- for (size_t k = 0; k < response_stream_sizes.size(); ++k) {
- ResponseParameters* response_parameter =
- request.add_response_parameters();
- response_parameter->set_size(response_stream_sizes[k]);
- }
- StreamingOutputCallResponse response;
-
- std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
- serviceStub_.Get()->StreamingOutputCall(&context, request));
-
- size_t k = 0;
- while (stream->Read(&response)) {
- // Payload related checks.
- GPR_ASSERT(response.payload().type() == request.response_type());
- switch (response.payload().type()) {
- case PayloadType::COMPRESSABLE:
- GPR_ASSERT(response.payload().body() ==
- grpc::string(response_stream_sizes[k], '\0'));
- break;
- case PayloadType::UNCOMPRESSABLE:
- break;
- default:
- GPR_ASSERT(false);
- }
-
- // Compression related checks.
- if (request.request_compressed_response()) {
- GPR_ASSERT(inspector.GetCallCompressionAlgorithm() >
- GRPC_COMPRESS_NONE);
- if (request.response_type() == PayloadType::COMPRESSABLE) {
- // requested compression and compressable response => results should
- // always be compressed.
- GPR_ASSERT(inspector.GetMessageFlags() &
- GRPC_WRITE_INTERNAL_COMPRESS);
- }
- } else {
- // requested *no* compression.
- GPR_ASSERT(
- !(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
- }
-
- ++k;
- }
-
- gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix);
- gpr_free(log_suffix);
+bool InteropClient::DoClientCompressedStreaming() {
+ // Probing for compression-checks support.
+ ClientContext probe_context;
+ StreamingInputCallRequest probe_req;
+ StreamingInputCallResponse probe_res;
+
+ probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
+ probe_req.mutable_expect_compressed()->set_value(true); // lies!
+ probe_req.mutable_payload()->set_body(grpc::string(27182, '\0'));
+
+ gpr_log(GPR_DEBUG, "Sending probe for compressed streaming request.");
+
+ std::unique_ptr<ClientWriter<StreamingInputCallRequest>> probe_stream(
+ serviceStub_.Get()->StreamingInputCall(&probe_context, &probe_res));
+
+ if (!probe_stream->Write(probe_req)) {
+ gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
+ return TransientFailureOrAbort();
+ }
+ Status s = probe_stream->Finish();
+ if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
+ // The server isn't able to evaluate incoming compression, making the rest
+ // of this test moot.
+ gpr_log(GPR_DEBUG, "Compressed streaming request probe failed");
+ return false;
+ }
+ gpr_log(GPR_DEBUG,
+ "Compressed streaming request probe succeeded. Proceeding.");
+
+ ClientContext context;
+ StreamingInputCallRequest request;
+ StreamingInputCallResponse response;
+
+ context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
+ std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
+ serviceStub_.Get()->StreamingInputCall(&context, &response));
+
+ request.mutable_payload()->set_body(grpc::string(27182, '\0'));
+ request.mutable_expect_compressed()->set_value(true);
+ gpr_log(GPR_DEBUG, "Sending streaming request with compression enabled");
+ if (!stream->Write(request)) {
+ gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
+ return TransientFailureOrAbort();
+ }
+
+ WriteOptions wopts;
+ wopts.set_no_compression();
+ request.mutable_payload()->set_body(grpc::string(45904, '\0'));
+ request.mutable_expect_compressed()->set_value(false);
+ gpr_log(GPR_DEBUG, "Sending streaming request with compression disabled");
+ if (!stream->Write(request, wopts)) {
+ gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
+ return TransientFailureOrAbort();
+ }
+ GPR_ASSERT(stream->WritesDone());
+
+ s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
+
+ return true;
+}
- if (k < response_stream_sizes.size()) {
- // stream->Read() failed before reading all the expected messages. This
- // is most likely due to a connection failure.
- gpr_log(GPR_ERROR,
- "DoResponseCompressedStreaming(): Responses read (k=%" PRIuPTR
- ") is "
- "less than the expected messages (i.e "
- "response_stream_sizes.size() (%" PRIuPTR ")). (i=%" PRIuPTR
- ", j=%" PRIuPTR ")",
- k, response_stream_sizes.size(), i, j);
- return TransientFailureOrAbort();
- }
-
- Status s = stream->Finish();
- if (!AssertStatusOk(s)) {
- return false;
- }
+bool InteropClient::DoServerCompressedStreaming() {
+ const std::vector<bool> compressions = {true, false};
+ const std::vector<int> sizes = {31415, 92653};
+
+ ClientContext context;
+ InteropClientContextInspector inspector(context);
+ StreamingOutputCallRequest request;
+
+ GPR_ASSERT(compressions.size() == sizes.size());
+ for (size_t i = 0; i < sizes.size(); i++) {
+ char* log_suffix;
+ gpr_asprintf(&log_suffix, "(compression=%s; size=%d)",
+ compressions[i] ? "true" : "false", sizes[i]);
+
+ gpr_log(GPR_DEBUG, "Sending request streaming rpc %s.", log_suffix);
+ gpr_free(log_suffix);
+
+ ResponseParameters* const response_parameter =
+ request.add_response_parameters();
+ response_parameter->mutable_compressed()->set_value(compressions[i]);
+ response_parameter->set_size(sizes[i]);
+ }
+ std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
+ serviceStub_.Get()->StreamingOutputCall(&context, request));
+
+ size_t k = 0;
+ StreamingOutputCallResponse response;
+ while (stream->Read(&response)) {
+ // Payload size checks.
+ GPR_ASSERT(response.payload().body() ==
+ grpc::string(request.response_parameters(k).size(), '\0'));
+
+ // Compression checks.
+ GPR_ASSERT(request.response_parameters(k).has_compressed());
+ if (request.response_parameters(k).compressed().value()) {
+ GPR_ASSERT(inspector.GetCallCompressionAlgorithm() > GRPC_COMPRESS_NONE);
+ GPR_ASSERT(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS);
+ } else {
+ // requested *no* compression.
+ GPR_ASSERT(!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS));
}
+ ++k;
+ }
+
+ if (k < sizes.size()) {
+ // stream->Read() failed before reading all the expected messages. This
+ // is most likely due to a connection failure.
+ gpr_log(GPR_ERROR, "%s(): Responses read (k=%" PRIuPTR
+ ") is "
+ "less than the expected messages (i.e "
+ "response_stream_sizes.size() (%" PRIuPTR ")).",
+ __func__, k, response_stream_sizes.size());
+ return TransientFailureOrAbort();
}
+ Status s = stream->Finish();
+ if (!AssertStatusOk(s)) {
+ return false;
+ }
return true;
}
@@ -632,7 +688,6 @@ bool InteropClient::DoPingPong() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
Payload* payload = request.mutable_payload();
StreamingOutputCallResponse response;
@@ -699,7 +754,6 @@ bool InteropClient::DoCancelAfterFirstResponse() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
response_parameter->set_size(31415);
request.mutable_payload()->set_body(grpc::string(27182, '\0'));
@@ -839,7 +893,6 @@ bool InteropClient::DoCustomMetadata() {
stream(serviceStub_.Get()->FullDuplexCall(&context));
StreamingOutputCallRequest request;
- request.set_response_type(PayloadType::COMPRESSABLE);
ResponseParameters* response_parameter = request.add_response_parameters();
response_parameter->set_size(kLargeResponseSize);
grpc::string payload(kLargeRequestSize, '\0');
diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h
index c8d6810c12..eb886fcb7e 100644
--- a/test/cpp/interop/interop_client.h
+++ b/test/cpp/interop/interop_client.h
@@ -64,12 +64,14 @@ class InteropClient {
bool DoEmpty();
bool DoLargeUnary();
- bool DoLargeCompressedUnary();
+ bool DoServerCompressedUnary();
+ bool DoClientCompressedUnary();
bool DoPingPong();
bool DoHalfDuplex();
bool DoRequestStreaming();
bool DoResponseStreaming();
- bool DoResponseCompressedStreaming();
+ bool DoServerCompressedStreaming();
+ bool DoClientCompressedStreaming();
bool DoResponseStreamingWithSlowConsumer();
bool DoCancelAfterBegin();
bool DoCancelAfterFirstResponse();
diff --git a/test/cpp/interop/server_main.cc b/test/cpp/interop/interop_server.cc
index 062857f3d6..ebef0002a3 100644
--- a/test/cpp/interop/server_main.cc
+++ b/test/cpp/interop/interop_server.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -48,6 +48,7 @@
#include <grpc/support/log.h>
#include <grpc/support/useful.h>
+#include "src/core/lib/transport/byte_stream.h"
#include "src/proto/grpc/testing/empty.grpc.pb.h"
#include "src/proto/grpc/testing/messages.grpc.pb.h"
#include "src/proto/grpc/testing/test.grpc.pb.h"
@@ -64,10 +65,10 @@ using grpc::ServerCredentials;
using grpc::ServerReader;
using grpc::ServerReaderWriter;
using grpc::ServerWriter;
+using grpc::WriteOptions;
using grpc::SslServerCredentialsOptions;
using grpc::testing::InteropServerContextInspector;
using grpc::testing::Payload;
-using grpc::testing::PayloadType;
using grpc::testing::SimpleRequest;
using grpc::testing::SimpleResponse;
using grpc::testing::StreamingInputCallRequest;
@@ -78,7 +79,6 @@ using grpc::testing::TestService;
using grpc::Status;
static bool got_sigint = false;
-static const char* kRandomFile = "test/cpp/interop/rnd.dat";
const char kEchoInitialMetadataKey[] = "x-grpc-test-echo-initial";
const char kEchoTrailingBinMetadataKey[] = "x-grpc-test-echo-trailing-bin";
@@ -110,34 +110,41 @@ void MaybeEchoMetadata(ServerContext* context) {
}
}
-bool SetPayload(PayloadType response_type, int size, Payload* payload) {
- payload->set_type(response_type);
- switch (response_type) {
- case PayloadType::COMPRESSABLE: {
- std::unique_ptr<char[]> body(new char[size]());
- payload->set_body(body.get(), size);
- } break;
- case PayloadType::UNCOMPRESSABLE: {
- std::unique_ptr<char[]> body(new char[size]());
- std::ifstream rnd_file(kRandomFile);
- GPR_ASSERT(rnd_file.good());
- rnd_file.read(body.get(), size);
- GPR_ASSERT(!rnd_file.eof()); // Requested more rnd bytes than available
- payload->set_body(body.get(), size);
- } break;
- default:
- GPR_ASSERT(false);
- }
+bool SetPayload(int size, Payload* payload) {
+ std::unique_ptr<char[]> body(new char[size]());
+ payload->set_body(body.get(), size);
return true;
}
-template <typename RequestType>
-void SetResponseCompression(ServerContext* context,
- const RequestType& request) {
- if (request.request_compressed_response()) {
- // Any level would do, let's go for HIGH because we are overachievers.
- context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
+bool CheckExpectedCompression(const ServerContext& context,
+ const bool compression_expected) {
+ const InteropServerContextInspector inspector(context);
+ const grpc_compression_algorithm received_compression =
+ inspector.GetCallCompressionAlgorithm();
+
+ if (compression_expected) {
+ if (received_compression == GRPC_COMPRESS_NONE) {
+ // Expected some compression, got NONE. This is an error.
+ gpr_log(GPR_ERROR,
+ "Expected compression but got uncompressed request from client.");
+ return false;
+ }
+ if (!(inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS)) {
+ gpr_log(GPR_ERROR,
+ "Failure: Requested compression in a compressable request, but "
+ "compression bit in message flags not set.");
+ return false;
+ }
+ } else {
+ // Didn't expect compression -> make sure the request is uncompressed
+ if (inspector.GetMessageFlags() & GRPC_WRITE_INTERNAL_COMPRESS) {
+ gpr_log(GPR_ERROR,
+ "Failure: Didn't requested compression, but compression bit in "
+ "message flags set.");
+ return false;
+ }
}
+ return true;
}
class TestServiceImpl : public TestService::Service {
@@ -151,11 +158,26 @@ class TestServiceImpl : public TestService::Service {
Status UnaryCall(ServerContext* context, const SimpleRequest* request,
SimpleResponse* response) {
MaybeEchoMetadata(context);
- SetResponseCompression(context, *request);
+ if (request->has_response_compressed()) {
+ const bool compression_requested = request->response_compressed().value();
+ gpr_log(GPR_DEBUG, "Request for compression (%s) present for %s",
+ compression_requested ? "enabled" : "disabled", __func__);
+ if (compression_requested) {
+ // Any level would do, let's go for HIGH because we are overachievers.
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
+ } else {
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_NONE);
+ }
+ }
+ if (!CheckExpectedCompression(*context,
+ request->expect_compressed().value())) {
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Compressed request expectation not met.");
+ }
if (request->response_size() > 0) {
- if (!SetPayload(request->response_type(), request->response_size(),
- response->mutable_payload())) {
- return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+ if (!SetPayload(request->response_size(), response->mutable_payload())) {
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Error creating payload.");
}
}
@@ -171,15 +193,26 @@ class TestServiceImpl : public TestService::Service {
Status StreamingOutputCall(
ServerContext* context, const StreamingOutputCallRequest* request,
ServerWriter<StreamingOutputCallResponse>* writer) {
- SetResponseCompression(context, *request);
StreamingOutputCallResponse response;
bool write_success = true;
for (int i = 0; write_success && i < request->response_parameters_size();
i++) {
- if (!SetPayload(request->response_type(),
- request->response_parameters(i).size(),
+ if (!SetPayload(request->response_parameters(i).size(),
response.mutable_payload())) {
- return Status(grpc::StatusCode::INTERNAL, "Error creating payload.");
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Error creating payload.");
+ }
+ WriteOptions wopts;
+ if (request->response_parameters(i).has_compressed()) {
+ // Compress by default. Disabled on a per-message basis.
+ context->set_compression_level(GRPC_COMPRESS_LEVEL_HIGH);
+ const bool compression_requested =
+ request->response_parameters(i).compressed().value();
+ gpr_log(GPR_DEBUG, "Request for compression (%s) present for %s",
+ compression_requested ? "enabled" : "disabled", __func__);
+ if (!compression_requested) {
+ wopts.set_no_compression();
+ } // else, compression is already enabled via the context.
}
int time_us;
if ((time_us = request->response_parameters(i).interval_us()) > 0) {
@@ -189,7 +222,7 @@ class TestServiceImpl : public TestService::Service {
gpr_time_from_micros(time_us, GPR_TIMESPAN));
gpr_sleep_until(sleep_time);
}
- write_success = writer->Write(response);
+ write_success = writer->Write(response, wopts);
}
if (write_success) {
return Status::OK;
@@ -204,6 +237,11 @@ class TestServiceImpl : public TestService::Service {
StreamingInputCallRequest request;
int aggregated_payload_size = 0;
while (reader->Read(&request)) {
+ if (!CheckExpectedCompression(*context,
+ request.expect_compressed().value())) {
+ return Status(grpc::StatusCode::INVALID_ARGUMENT,
+ "Compressed request expectation not met.");
+ }
if (request.has_payload()) {
aggregated_payload_size += request.payload().body().size();
}
@@ -221,7 +259,6 @@ class TestServiceImpl : public TestService::Service {
StreamingOutputCallResponse response;
bool write_success = true;
while (write_success && stream->Read(&request)) {
- SetResponseCompression(context, request);
if (request.response_parameters_size() != 0) {
response.mutable_payload()->set_type(request.payload().type());
response.mutable_payload()->set_body(
diff --git a/test/cpp/interop/rnd.dat b/test/cpp/interop/rnd.dat
deleted file mode 100644
index 8c7f38f9e0..0000000000
--- a/test/cpp/interop/rnd.dat
+++ /dev/null
Binary files differ
diff --git a/test/cpp/interop/server_helper.cc b/test/cpp/interop/server_helper.cc
index c6d891ad71..8b0b511bcb 100644
--- a/test/cpp/interop/server_helper.cc
+++ b/test/cpp/interop/server_helper.cc
@@ -72,6 +72,10 @@ uint32_t InteropServerContextInspector::GetEncodingsAcceptedByClient() const {
return grpc_call_test_only_get_encodings_accepted_by_peer(context_.call_);
}
+uint32_t InteropServerContextInspector::GetMessageFlags() const {
+ return grpc_call_test_only_get_message_flags(context_.call_);
+}
+
std::shared_ptr<const AuthContext>
InteropServerContextInspector::GetAuthContext() const {
return context_.auth_context();
diff --git a/test/cpp/interop/server_helper.h b/test/cpp/interop/server_helper.h
index 12865e4032..a1da14a4c8 100644
--- a/test/cpp/interop/server_helper.h
+++ b/test/cpp/interop/server_helper.h
@@ -54,6 +54,7 @@ class InteropServerContextInspector {
bool IsCancelled() const;
grpc_compression_algorithm GetCallCompressionAlgorithm() const;
uint32_t GetEncodingsAcceptedByClient() const;
+ uint32_t GetMessageFlags() const;
private:
const ::grpc::ServerContext& context_;
diff --git a/test/cpp/interop/stress_interop_client.cc b/test/cpp/interop/stress_interop_client.cc
index aa95682e74..1d5fc80cf2 100644
--- a/test/cpp/interop/stress_interop_client.cc
+++ b/test/cpp/interop/stress_interop_client.cc
@@ -138,8 +138,12 @@ bool StressTestInteropClient::RunTest(TestCaseType test_case) {
is_success = interop_client_->DoLargeUnary();
break;
}
- case LARGE_COMPRESSED_UNARY: {
- is_success = interop_client_->DoLargeCompressedUnary();
+ case CLIENT_COMPRESSED_UNARY: {
+ is_success = interop_client_->DoClientCompressedUnary();
+ break;
+ }
+ case CLIENT_COMPRESSED_STREAMING: {
+ is_success = interop_client_->DoClientCompressedStreaming();
break;
}
case CLIENT_STREAMING: {
@@ -150,8 +154,12 @@ bool StressTestInteropClient::RunTest(TestCaseType test_case) {
is_success = interop_client_->DoResponseStreaming();
break;
}
+ case SERVER_COMPRESSED_UNARY: {
+ is_success = interop_client_->DoServerCompressedUnary();
+ break;
+ }
case SERVER_COMPRESSED_STREAMING: {
- is_success = interop_client_->DoResponseCompressedStreaming();
+ is_success = interop_client_->DoServerCompressedStreaming();
break;
}
case SLOW_CONSUMER: {
diff --git a/test/cpp/interop/stress_interop_client.h b/test/cpp/interop/stress_interop_client.h
index aa93b58b4a..cf6a713473 100644
--- a/test/cpp/interop/stress_interop_client.h
+++ b/test/cpp/interop/stress_interop_client.h
@@ -51,29 +51,33 @@ using std::vector;
enum TestCaseType {
UNKNOWN_TEST = -1,
- EMPTY_UNARY = 0,
- LARGE_UNARY = 1,
- LARGE_COMPRESSED_UNARY = 2,
- CLIENT_STREAMING = 3,
- SERVER_STREAMING = 4,
- SERVER_COMPRESSED_STREAMING = 5,
- SLOW_CONSUMER = 6,
- HALF_DUPLEX = 7,
- PING_PONG = 8,
- CANCEL_AFTER_BEGIN = 9,
- CANCEL_AFTER_FIRST_RESPONSE = 10,
- TIMEOUT_ON_SLEEPING_SERVER = 11,
- EMPTY_STREAM = 12,
- STATUS_CODE_AND_MESSAGE = 13,
- CUSTOM_METADATA = 14
+ EMPTY_UNARY,
+ LARGE_UNARY,
+ CLIENT_COMPRESSED_UNARY,
+ CLIENT_COMPRESSED_STREAMING,
+ CLIENT_STREAMING,
+ SERVER_STREAMING,
+ SERVER_COMPRESSED_UNARY,
+ SERVER_COMPRESSED_STREAMING,
+ SLOW_CONSUMER,
+ HALF_DUPLEX,
+ PING_PONG,
+ CANCEL_AFTER_BEGIN,
+ CANCEL_AFTER_FIRST_RESPONSE,
+ TIMEOUT_ON_SLEEPING_SERVER,
+ EMPTY_STREAM,
+ STATUS_CODE_AND_MESSAGE,
+ CUSTOM_METADATA
};
const vector<pair<TestCaseType, grpc::string>> kTestCaseList = {
{EMPTY_UNARY, "empty_unary"},
{LARGE_UNARY, "large_unary"},
- {LARGE_COMPRESSED_UNARY, "large_compressed_unary"},
+ {CLIENT_COMPRESSED_UNARY, "client_compressed_unary"},
+ {CLIENT_COMPRESSED_STREAMING, "client_compressed_streaming"},
{CLIENT_STREAMING, "client_streaming"},
{SERVER_STREAMING, "server_streaming"},
+ {SERVER_COMPRESSED_UNARY, "server_compressed_unary"},
{SERVER_COMPRESSED_STREAMING, "server_compressed_streaming"},
{SLOW_CONSUMER, "slow_consumer"},
{HALF_DUPLEX, "half_duplex"},
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-24 23:24:49 +0000