make some interop tests run as async methods

pull/2664/head
Jan Tattermusch 9 years ago
parent 9b048e529e
commit b98e1dd7c4
  1. 254
      src/csharp/Grpc.IntegrationTesting/InteropClient.cs
  2. 24
      src/csharp/Grpc.IntegrationTesting/InteropClientServerTest.cs

@ -130,12 +130,12 @@ namespace Grpc.IntegrationTesting
client.HeaderInterceptor = OAuth2InterceptorFactory.Create(credential); client.HeaderInterceptor = OAuth2InterceptorFactory.Create(credential);
} }
RunTestCase(options.testCase, client); RunTestCaseAsync(options.testCase, client).Wait();
} }
GrpcEnvironment.Shutdown(); GrpcEnvironment.Shutdown();
} }
private void RunTestCase(string testCase, TestService.TestServiceClient client) private async Task RunTestCaseAsync(string testCase, TestService.TestServiceClient client)
{ {
switch (testCase) switch (testCase)
{ {
@ -146,16 +146,16 @@ namespace Grpc.IntegrationTesting
RunLargeUnary(client); RunLargeUnary(client);
break; break;
case "client_streaming": case "client_streaming":
RunClientStreaming(client); await RunClientStreamingAsync(client);
break; break;
case "server_streaming": case "server_streaming":
RunServerStreaming(client); await RunServerStreamingAsync(client);
break; break;
case "ping_pong": case "ping_pong":
RunPingPong(client); await RunPingPongAsync(client);
break; break;
case "empty_stream": case "empty_stream":
RunEmptyStream(client); await RunEmptyStreamAsync(client);
break; break;
case "service_account_creds": case "service_account_creds":
RunServiceAccountCreds(client); RunServiceAccountCreds(client);
@ -170,10 +170,10 @@ namespace Grpc.IntegrationTesting
RunPerRpcCreds(client); RunPerRpcCreds(client);
break; break;
case "cancel_after_begin": case "cancel_after_begin":
RunCancelAfterBegin(client); await RunCancelAfterBeginAsync(client);
break; break;
case "cancel_after_first_response": case "cancel_after_first_response":
RunCancelAfterFirstResponse(client); await RunCancelAfterFirstResponseAsync(client);
break; break;
case "benchmark_empty_unary": case "benchmark_empty_unary":
RunBenchmarkEmptyUnary(client); RunBenchmarkEmptyUnary(client);
@ -207,118 +207,106 @@ namespace Grpc.IntegrationTesting
Console.WriteLine("Passed!"); Console.WriteLine("Passed!");
} }
public static void RunClientStreaming(TestService.ITestServiceClient client) public static async Task RunClientStreamingAsync(TestService.ITestServiceClient client)
{ {
Task.Run(async () => Console.WriteLine("running client_streaming");
{
Console.WriteLine("running client_streaming");
var bodySizes = new List<int> { 27182, 8, 1828, 45904 }.ConvertAll((size) => StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build()); var bodySizes = new List<int> { 27182, 8, 1828, 45904 }.ConvertAll((size) => StreamingInputCallRequest.CreateBuilder().SetPayload(CreateZerosPayload(size)).Build());
using (var call = client.StreamingInputCall()) using (var call = client.StreamingInputCall())
{ {
await call.RequestStream.WriteAll(bodySizes); await call.RequestStream.WriteAll(bodySizes);
var response = await call.ResponseAsync; var response = await call.ResponseAsync;
Assert.AreEqual(74922, response.AggregatedPayloadSize); Assert.AreEqual(74922, response.AggregatedPayloadSize);
} }
Console.WriteLine("Passed!"); Console.WriteLine("Passed!");
}).Wait();
} }
public static void RunServerStreaming(TestService.ITestServiceClient client) public static async Task RunServerStreamingAsync(TestService.ITestServiceClient client)
{ {
Task.Run(async () => Console.WriteLine("running server_streaming");
{
Console.WriteLine("running server_streaming");
var bodySizes = new List<int> { 31415, 9, 2653, 58979 }; var bodySizes = new List<int> { 31415, 9, 2653, 58979 };
var request = StreamingOutputCallRequest.CreateBuilder() var request = StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE) .SetResponseType(PayloadType.COMPRESSABLE)
.AddRangeResponseParameters(bodySizes.ConvertAll( .AddRangeResponseParameters(bodySizes.ConvertAll(
(size) => ResponseParameters.CreateBuilder().SetSize(size).Build())) (size) => ResponseParameters.CreateBuilder().SetSize(size).Build()))
.Build(); .Build();
using (var call = client.StreamingOutputCall(request)) using (var call = client.StreamingOutputCall(request))
{
var responseList = await call.ResponseStream.ToList();
foreach (var res in responseList)
{ {
var responseList = await call.ResponseStream.ToList(); Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type);
foreach (var res in responseList)
{
Assert.AreEqual(PayloadType.COMPRESSABLE, res.Payload.Type);
}
CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length));
} }
Console.WriteLine("Passed!"); CollectionAssert.AreEqual(bodySizes, responseList.ConvertAll((item) => item.Payload.Body.Length));
}).Wait(); }
Console.WriteLine("Passed!");
} }
public static void RunPingPong(TestService.ITestServiceClient client) public static async Task RunPingPongAsync(TestService.ITestServiceClient client)
{ {
Task.Run(async () => Console.WriteLine("running ping_pong");
{
Console.WriteLine("running ping_pong");
using (var call = client.FullDuplexCall()) using (var call = client.FullDuplexCall())
{ {
await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder() await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE) .SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415)) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
.SetPayload(CreateZerosPayload(27182)).Build()); .SetPayload(CreateZerosPayload(27182)).Build());
Assert.IsTrue(await call.ResponseStream.MoveNext()); Assert.IsTrue(await call.ResponseStream.MoveNext());
Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type); Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length); Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length);
await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder() await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE) .SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(9)) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(9))
.SetPayload(CreateZerosPayload(8)).Build()); .SetPayload(CreateZerosPayload(8)).Build());
Assert.IsTrue(await call.ResponseStream.MoveNext()); Assert.IsTrue(await call.ResponseStream.MoveNext());
Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type); Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
Assert.AreEqual(9, call.ResponseStream.Current.Payload.Body.Length); Assert.AreEqual(9, call.ResponseStream.Current.Payload.Body.Length);
await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder() await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE) .SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653)) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(2653))
.SetPayload(CreateZerosPayload(1828)).Build()); .SetPayload(CreateZerosPayload(1828)).Build());
Assert.IsTrue(await call.ResponseStream.MoveNext()); Assert.IsTrue(await call.ResponseStream.MoveNext());
Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type); Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
Assert.AreEqual(2653, call.ResponseStream.Current.Payload.Body.Length); Assert.AreEqual(2653, call.ResponseStream.Current.Payload.Body.Length);
await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder() await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE) .SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(58979)) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(58979))
.SetPayload(CreateZerosPayload(45904)).Build()); .SetPayload(CreateZerosPayload(45904)).Build());
Assert.IsTrue(await call.ResponseStream.MoveNext()); Assert.IsTrue(await call.ResponseStream.MoveNext());
Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type); Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
Assert.AreEqual(58979, call.ResponseStream.Current.Payload.Body.Length); Assert.AreEqual(58979, call.ResponseStream.Current.Payload.Body.Length);
await call.RequestStream.CompleteAsync(); await call.RequestStream.CompleteAsync();
Assert.IsFalse(await call.ResponseStream.MoveNext()); Assert.IsFalse(await call.ResponseStream.MoveNext());
} }
Console.WriteLine("Passed!"); Console.WriteLine("Passed!");
}).Wait();
} }
public static void RunEmptyStream(TestService.ITestServiceClient client) public static async Task RunEmptyStreamAsync(TestService.ITestServiceClient client)
{ {
Task.Run(async () => Console.WriteLine("running empty_stream");
using (var call = client.FullDuplexCall())
{ {
Console.WriteLine("running empty_stream"); await call.RequestStream.CompleteAsync();
using (var call = client.FullDuplexCall())
{
await call.RequestStream.CompleteAsync();
var responseList = await call.ResponseStream.ToList(); var responseList = await call.ResponseStream.ToList();
Assert.AreEqual(0, responseList.Count); Assert.AreEqual(0, responseList.Count);
} }
Console.WriteLine("Passed!"); Console.WriteLine("Passed!");
}).Wait();
} }
public static void RunServiceAccountCreds(TestService.ITestServiceClient client) public static void RunServiceAccountCreds(TestService.ITestServiceClient client)
@ -406,65 +394,59 @@ namespace Grpc.IntegrationTesting
Console.WriteLine("Passed!"); Console.WriteLine("Passed!");
} }
public static void RunCancelAfterBegin(TestService.ITestServiceClient client) public static async Task RunCancelAfterBeginAsync(TestService.ITestServiceClient client)
{ {
Task.Run(async () => Console.WriteLine("running cancel_after_begin");
var cts = new CancellationTokenSource();
using (var call = client.StreamingInputCall(cancellationToken: cts.Token))
{ {
Console.WriteLine("running cancel_after_begin"); // TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it.
await Task.Delay(1000);
cts.Cancel();
var cts = new CancellationTokenSource(); try
using (var call = client.StreamingInputCall(cancellationToken: cts.Token))
{ {
// TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it. var response = await call.ResponseAsync;
await Task.Delay(1000); Assert.Fail();
cts.Cancel(); }
catch (RpcException e)
try {
{ Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
var response = await call.ResponseAsync;
Assert.Fail();
}
catch (RpcException e)
{
Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
}
} }
Console.WriteLine("Passed!"); }
}).Wait(); Console.WriteLine("Passed!");
} }
public static void RunCancelAfterFirstResponse(TestService.ITestServiceClient client) public static async Task RunCancelAfterFirstResponseAsync(TestService.ITestServiceClient client)
{ {
Task.Run(async () => Console.WriteLine("running cancel_after_first_response");
{
Console.WriteLine("running cancel_after_first_response");
var cts = new CancellationTokenSource(); var cts = new CancellationTokenSource();
using (var call = client.FullDuplexCall(cancellationToken: cts.Token)) using (var call = client.FullDuplexCall(cancellationToken: cts.Token))
{ {
await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder() await call.RequestStream.WriteAsync(StreamingOutputCallRequest.CreateBuilder()
.SetResponseType(PayloadType.COMPRESSABLE) .SetResponseType(PayloadType.COMPRESSABLE)
.AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415)) .AddResponseParameters(ResponseParameters.CreateBuilder().SetSize(31415))
.SetPayload(CreateZerosPayload(27182)).Build()); .SetPayload(CreateZerosPayload(27182)).Build());
Assert.IsTrue(await call.ResponseStream.MoveNext()); Assert.IsTrue(await call.ResponseStream.MoveNext());
Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type); Assert.AreEqual(PayloadType.COMPRESSABLE, call.ResponseStream.Current.Payload.Type);
Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length); Assert.AreEqual(31415, call.ResponseStream.Current.Payload.Body.Length);
cts.Cancel(); cts.Cancel();
try try
{ {
await call.ResponseStream.MoveNext(); await call.ResponseStream.MoveNext();
Assert.Fail(); Assert.Fail();
} }
catch (RpcException e) catch (RpcException e)
{ {
Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode); Assert.AreEqual(StatusCode.Cancelled, e.Status.StatusCode);
}
} }
Console.WriteLine("Passed!"); }
}).Wait(); Console.WriteLine("Passed!");
} }
// This is not an official interop test, but it's useful. // This is not an official interop test, but it's useful.

@ -89,39 +89,39 @@ namespace Grpc.IntegrationTesting
} }
[Test] [Test]
public void ClientStreaming() public async Task ClientStreaming()
{ {
InteropClient.RunClientStreaming(client); await InteropClient.RunClientStreamingAsync(client);
} }
[Test] [Test]
public void ServerStreaming() public async Task ServerStreaming()
{ {
InteropClient.RunServerStreaming(client); await InteropClient.RunServerStreamingAsync(client);
} }
[Test] [Test]
public void PingPong() public async Task PingPong()
{ {
InteropClient.RunPingPong(client); await InteropClient.RunPingPongAsync(client);
} }
[Test] [Test]
public void EmptyStream() public async Task EmptyStream()
{ {
InteropClient.RunEmptyStream(client); await InteropClient.RunEmptyStreamAsync(client);
} }
[Test] [Test]
public void CancelAfterBegin() public async Task CancelAfterBegin()
{ {
InteropClient.RunCancelAfterBegin(client); await InteropClient.RunCancelAfterBeginAsync(client);
} }
[Test] [Test]
public void CancelAfterFirstResponse() public async Task CancelAfterFirstResponse()
{ {
InteropClient.RunCancelAfterFirstResponse(client); await InteropClient.RunCancelAfterFirstResponseAsync(client);
} }
} }
} }

Loading…
Cancel
Save