Merge pull request #4383 from jtattermusch/run_more_interop

run_interop_tests.py support for new interop tests

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);
         }
     }

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}}
+            };
+        }
     }
 }

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));
+        }
     }
 }

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);
+            }
+        }
     }
 }

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: