diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
index 0ade701a5387841c6da9fe05bbcc5781e910e5d3..de66759b94bdee24efa26d6209b663f401a5953c 100644
--- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs
@@ -284,7 +284,7 @@ namespace Grpc.Core.Internal
var finishedTask = asyncCall.ServerSideCallAsync();
var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall);
- await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."), Metadata.Empty).ConfigureAwait(false);
+ await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, ""), Metadata.Empty).ConfigureAwait(false);
await finishedTask.ConfigureAwait(false);
}
}
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
index 5eec11abf7f71b6a948f5dc29916bf06716c3f96..b0e33e49f77eefabf6a80ada6e0b22e2ab6f2fae 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs
@@ -34,6 +34,7 @@
using System;
using System.Collections.Generic;
using System.IO;
+using System.Linq;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
@@ -130,8 +131,7 @@ namespace Grpc.IntegrationTesting
};
}
var channel = new Channel(options.ServerHost, options.ServerPort, credentials, channelOptions);
- TestService.TestServiceClient client = new TestService.TestServiceClient(channel);
- await RunTestCaseAsync(client, options);
+ await RunTestCaseAsync(channel, options);
await channel.ShutdownAsync();
}
@@ -159,8 +159,9 @@ namespace Grpc.IntegrationTesting
return credentials;
}
- private async Task RunTestCaseAsync(TestService.TestServiceClient client, ClientOptions options)
+ private async Task RunTestCaseAsync(Channel channel, ClientOptions options)
{
+ var client = new TestService.TestServiceClient(channel);
switch (options.TestCase)
{
case "empty_unary":
@@ -202,8 +203,14 @@ namespace Grpc.IntegrationTesting
case "timeout_on_sleeping_server":
await RunTimeoutOnSleepingServerAsync(client);
break;
- case "benchmark_empty_unary":
- RunBenchmarkEmptyUnary(client);
+ case "custom_metadata":
+ await RunCustomMetadataAsync(client);
+ break;
+ case "status_code_and_message":
+ await RunStatusCodeAndMessageAsync(client);
+ break;
+ case "unimplemented_method":
+ RunUnimplementedMethod(new UnimplementedService.UnimplementedServiceClient(channel));
break;
default:
throw new ArgumentException("Unknown test case " + options.TestCase);
@@ -227,7 +234,6 @@ namespace Grpc.IntegrationTesting
ResponseSize = 314159,
Payload = CreateZerosPayload(271828)
};
-
var response = client.UnaryCall(request);
Assert.AreEqual(PayloadType.COMPRESSABLE, response.Payload.Type);
@@ -493,11 +499,95 @@ namespace Grpc.IntegrationTesting
Console.WriteLine("Passed!");
}
- // This is not an official interop test, but it's useful.
- public static void RunBenchmarkEmptyUnary(TestService.ITestServiceClient client)
+ public static async Task RunCustomMetadataAsync(TestService.ITestServiceClient client)
{
- BenchmarkUtil.RunBenchmark(10000, 10000,
- () => { client.EmptyCall(new Empty()); });
+ Console.WriteLine("running custom_metadata");
+ {
+ // step 1: test unary call
+ var request = new SimpleRequest
+ {
+ ResponseType = PayloadType.COMPRESSABLE,
+ ResponseSize = 314159,
+ Payload = CreateZerosPayload(271828)
+ };
+
+ var call = client.UnaryCallAsync(request, headers: CreateTestMetadata());
+ await call.ResponseAsync;
+
+ var responseHeaders = await call.ResponseHeadersAsync;
+ var responseTrailers = call.GetTrailers();
+
+ Assert.AreEqual("test_initial_metadata_value", responseHeaders.First((entry) => entry.Key == "x-grpc-test-echo-initial").Value);
+ CollectionAssert.AreEqual(new byte[] { 0xab, 0xab, 0xab }, responseTrailers.First((entry) => entry.Key == "x-grpc-test-echo-trailing-bin").ValueBytes);
+ }
+
+ {
+ // step 2: test full duplex call
+ var request = new StreamingOutputCallRequest
+ {
+ ResponseType = PayloadType.COMPRESSABLE,
+ ResponseParameters = { new ResponseParameters { Size = 31415 } },
+ Payload = CreateZerosPayload(27182)
+ };
+
+ var call = client.FullDuplexCall(headers: CreateTestMetadata());
+ var responseHeaders = await call.ResponseHeadersAsync;
+
+ await call.RequestStream.WriteAsync(request);
+ await call.RequestStream.CompleteAsync();
+ await call.ResponseStream.ToListAsync();
+
+ var responseTrailers = call.GetTrailers();
+
+ Assert.AreEqual("test_initial_metadata_value", responseHeaders.First((entry) => entry.Key == "x-grpc-test-echo-initial").Value);
+ CollectionAssert.AreEqual(new byte[] { 0xab, 0xab, 0xab }, responseTrailers.First((entry) => entry.Key == "x-grpc-test-echo-trailing-bin").ValueBytes);
+ }
+
+ Console.WriteLine("Passed!");
+ }
+
+ public static async Task RunStatusCodeAndMessageAsync(TestService.ITestServiceClient client)
+ {
+ Console.WriteLine("running status_code_and_message");
+ var echoStatus = new EchoStatus
+ {
+ Code = 2,
+ Message = "test status message"
+ };
+
+ {
+ // step 1: test unary call
+ var request = new SimpleRequest { ResponseStatus = echoStatus };
+
+ var e = Assert.Throws<RpcException>(() => client.UnaryCall(request));
+ Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode);
+ Assert.AreEqual(echoStatus.Message, e.Status.Detail);
+ }
+
+ {
+ // step 2: test full duplex call
+ var request = new StreamingOutputCallRequest { ResponseStatus = echoStatus };
+
+ var call = client.FullDuplexCall();
+ await call.RequestStream.WriteAsync(request);
+ await call.RequestStream.CompleteAsync();
+
+ var e = Assert.Throws<RpcException>(async () => await call.ResponseStream.ToListAsync());
+ Assert.AreEqual(StatusCode.Unknown, e.Status.StatusCode);
+ Assert.AreEqual(echoStatus.Message, e.Status.Detail);
+ }
+
+ Console.WriteLine("Passed!");
+ }
+
+ public static void RunUnimplementedMethod(UnimplementedService.IUnimplementedServiceClient client)
+ {
+ Console.WriteLine("running unimplemented_method");
+ var e = Assert.Throws<RpcException>(() => client.UnimplementedCall(new Empty()));
+
+ Assert.AreEqual(StatusCode.Unimplemented, e.Status.StatusCode);
+ Assert.AreEqual("", e.Status.Detail);
+ Console.WriteLine("Passed!");
}
private static Payload CreateZerosPayload(int size)
@@ -516,5 +606,14 @@ namespace Grpc.IntegrationTesting
Assert.IsTrue(email.Length > 0); // spec requires nonempty client email.
return email;
}
+
+ private static Metadata CreateTestMetadata()
+ {
+ return new Metadata
+ {
+ {"x-grpc-test-echo-initial", "test_initial_metadata_value"},
+ {"x-grpc-test-echo-trailing-bin", new byte[] {0xab, 0xab, 0xab}}
+ };
+ }
}
}
diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
index 837ae74c453db8bfa4c1ad921180c019a5872b3d..5facb87971ec04964bfbe7088b7f4c0d34d87cd7 100644
--- a/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
+++ b/src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs
@@ -128,9 +128,27 @@ namespace Grpc.IntegrationTesting
}
[Test]
- public async Task TimeoutOnSleepingServerAsync()
+ public async Task TimeoutOnSleepingServer()
{
await InteropClient.RunTimeoutOnSleepingServerAsync(client);
}
+
+ [Test]
+ public async Task CustomMetadata()
+ {
+ await InteropClient.RunCustomMetadataAsync(client);
+ }
+
+ [Test]
+ public async Task StatusCodeAndMessage()
+ {
+ await InteropClient.RunStatusCodeAndMessageAsync(client);
+ }
+
+ [Test]
+ public void UnimplementedMethod()
+ {
+ InteropClient.RunUnimplementedMethod(UnimplementedService.NewClient(channel));
+ }
}
}
diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
index c5bfcf08c0c0397f5adc0d442db579a44181499c..5a1b4cf319b696a57a0ec41badcdf6b2f598f280 100644
--- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
+++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs
@@ -33,6 +33,7 @@
using System;
using System.Collections.Generic;
+using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Google.Protobuf;
@@ -51,14 +52,20 @@ namespace Grpc.Testing
return Task.FromResult(new Empty());
}
- public Task<SimpleResponse> UnaryCall(SimpleRequest request, ServerCallContext context)
+ public async Task<SimpleResponse> UnaryCall(SimpleRequest request, ServerCallContext context)
{
+ await EnsureEchoMetadataAsync(context);
+ EnsureEchoStatus(request.ResponseStatus, context);
+
var response = new SimpleResponse { Payload = CreateZerosPayload(request.ResponseSize) };
- return Task.FromResult(response);
+ return response;
}
public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
+ await EnsureEchoMetadataAsync(context);
+ EnsureEchoStatus(request.ResponseStatus, context);
+
foreach (var responseParam in request.ResponseParameters)
{
var response = new StreamingOutputCallResponse { Payload = CreateZerosPayload(responseParam.Size) };
@@ -68,6 +75,8 @@ namespace Grpc.Testing
public async Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream, ServerCallContext context)
{
+ await EnsureEchoMetadataAsync(context);
+
int sum = 0;
await requestStream.ForEachAsync(async request =>
{
@@ -78,8 +87,11 @@ namespace Grpc.Testing
public async Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context)
{
+ await EnsureEchoMetadataAsync(context);
+
await requestStream.ForEachAsync(async request =>
{
+ EnsureEchoStatus(request.ResponseStatus, context);
foreach (var responseParam in request.ResponseParameters)
{
var response = new StreamingOutputCallResponse { Payload = CreateZerosPayload(responseParam.Size) };
@@ -97,5 +109,28 @@ namespace Grpc.Testing
{
return new Payload { Body = ByteString.CopyFrom(new byte[size]) };
}
+
+ private static async Task EnsureEchoMetadataAsync(ServerCallContext context)
+ {
+ var echoInitialList = context.RequestHeaders.Where((entry) => entry.Key == "x-grpc-test-echo-initial").ToList();
+ if (echoInitialList.Any()) {
+ var entry = echoInitialList.Single();
+ await context.WriteResponseHeadersAsync(new Metadata { entry });
+ }
+
+ var echoTrailingList = context.RequestHeaders.Where((entry) => entry.Key == "x-grpc-test-echo-trailing-bin").ToList();
+ if (echoTrailingList.Any()) {
+ context.ResponseTrailers.Add(echoTrailingList.Single());
+ }
+ }
+
+ private static void EnsureEchoStatus(EchoStatus responseStatus, ServerCallContext context)
+ {
+ if (responseStatus != null)
+ {
+ var statusCode = (StatusCode)responseStatus.Code;
+ context.Status = new Status(statusCode, responseStatus.Message);
+ }
+ }
}
}
diff --git a/tools/run_tests/run_interop_tests.py b/tools/run_tests/run_interop_tests.py
index 37b631bd0da96dc49f20632b056b14c9abfb6b35..763ff5615c2dfa1dfe2dda93e922d60991737e84 100755
--- a/tools/run_tests/run_interop_tests.py
+++ b/tools/run_tests/run_interop_tests.py
@@ -52,6 +52,11 @@ _DEFAULT_SERVER_PORT=8080
# supported by C core SslCredentials instead.
_SSL_CERT_ENV = { 'SSL_CERT_FILE':'/usr/local/share/grpc/roots.pem' }
+_SKIP_COMPRESSION = ['large_compressed_unary',
+ 'server_compressed_streaming']
+
+_SKIP_ADVANCED = ['custom_metadata', 'status_code_and_message',
+ 'unimplemented_method']
class CXXLanguage:
@@ -73,7 +78,10 @@ class CXXLanguage:
return {}
def unimplemented_test_cases(self):
- return []
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
def __str__(self):
return 'c++'
@@ -99,7 +107,11 @@ class CSharpLanguage:
return {}
def unimplemented_test_cases(self):
- return []
+ # TODO: status_code_and_message doesn't work against node_server
+ return _SKIP_COMPRESSION + ['status_code_and_message']
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_COMPRESSION
def __str__(self):
return 'csharp'
@@ -125,7 +137,10 @@ class JavaLanguage:
return {}
def unimplemented_test_cases(self):
- return []
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
def __str__(self):
return 'java'
@@ -152,7 +167,10 @@ class GoLanguage:
return {}
def unimplemented_test_cases(self):
- return []
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
def __str__(self):
return 'go'
@@ -180,6 +198,9 @@ class Http2Client:
def unimplemented_test_cases(self):
return _TEST_CASES
+ def unimplemented_test_cases_server(self):
+ return []
+
def __str__(self):
return 'http2'
@@ -203,7 +224,10 @@ class NodeLanguage:
return {}
def unimplemented_test_cases(self):
- return []
+ return _SKIP_COMPRESSION
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_COMPRESSION
def __str__(self):
return 'node'
@@ -225,6 +249,9 @@ class PHPLanguage:
return {}
def unimplemented_test_cases(self):
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
+
+ def unimplemented_test_cases_server(self):
return []
def __str__(self):
@@ -251,7 +278,10 @@ class RubyLanguage:
return {}
def unimplemented_test_cases(self):
- return []
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
def __str__(self):
return 'ruby'
@@ -289,7 +319,11 @@ class PythonLanguage:
return {'LD_LIBRARY_PATH': '{}/libs/opt'.format(DOCKER_WORKDIR_ROOT)}
def unimplemented_test_cases(self):
- return ['jwt_token_creds', 'per_rpc_creds']
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION + ['jwt_token_creds',
+ 'per_rpc_creds']
+
+ def unimplemented_test_cases_server(self):
+ return _SKIP_ADVANCED + _SKIP_COMPRESSION
def __str__(self):
return 'python'
@@ -312,7 +346,9 @@ _SERVERS = ['c++', 'node', 'csharp', 'java', 'go', 'ruby', 'python']
_TEST_CASES = ['large_unary', 'empty_unary', 'ping_pong',
'empty_stream', 'client_streaming', 'server_streaming',
'cancel_after_begin', 'cancel_after_first_response',
- 'timeout_on_sleeping_server']
+ 'timeout_on_sleeping_server', 'custom_metadata',
+ 'status_code_and_message', 'unimplemented_method',
+ 'large_compressed_unary', 'server_compressed_streaming']
_AUTH_TEST_CASES = ['compute_engine_creds', 'jwt_token_creds',
'oauth2_auth_token', 'per_rpc_creds']
@@ -635,9 +671,10 @@ try:
for language in languages:
for test_case in _TEST_CASES:
if not test_case in language.unimplemented_test_cases():
- test_job = cloud_to_prod_jobspec(language, test_case,
- docker_image=docker_images.get(str(language)))
- jobs.append(test_job)
+ if not test_case in _SKIP_ADVANCED + _SKIP_COMPRESSION:
+ test_job = cloud_to_prod_jobspec(language, test_case,
+ docker_image=docker_images.get(str(language)))
+ jobs.append(test_job)
# TODO(carl-mastrangelo): Currently prod TLS terminators aren't spec compliant. Reenable
# this once a better solution is in place.
@@ -664,16 +701,21 @@ try:
for server_name, server_address in server_addresses.iteritems():
(server_host, server_port) = server_address
+ server_language = _LANGUAGES.get(server_name, None)
+ skip_server = [] # test cases unimplemented by server
+ if server_language:
+ skip_server = server_language.unimplemented_test_cases_server()
for language in languages:
for test_case in _TEST_CASES:
if not test_case in language.unimplemented_test_cases():
- test_job = cloud_to_cloud_jobspec(language,
- test_case,
- server_name,
- server_host,
- server_port,
- docker_image=docker_images.get(str(language)))
- jobs.append(test_job)
+ if not test_case in skip_server:
+ test_job = cloud_to_cloud_jobspec(language,
+ test_case,
+ server_name,
+ server_host,
+ server_port,
+ docker_image=docker_images.get(str(language)))
+ jobs.append(test_job)
if args.http2_interop:
for test_case in _HTTP2_TEST_CASES: