wait for call to close before asserting

Author: marcoferrer

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/client/ClientCallBidiStreamingTests.kt

@@ -20,7 +20,6 @@ package com.github.marcoferrer.krotoplus.coroutines.client
 import com.github.marcoferrer.krotoplus.coroutines.RpcCallTest
 import com.github.marcoferrer.krotoplus.coroutines.utils.assertFails
 import com.github.marcoferrer.krotoplus.coroutines.utils.assertFailsWithCancellation
-import com.github.marcoferrer.krotoplus.coroutines.utils.assertFailsWithStatus
 import com.github.marcoferrer.krotoplus.coroutines.utils.assertFailsWithStatus2
 import com.github.marcoferrer.krotoplus.coroutines.utils.matchStatus
 import com.github.marcoferrer.krotoplus.coroutines.utils.matchThrowable
@@ -136,6 +135,8 @@ class ClientCallBidiStreamingTests :
             responseChannel.consumeAsFlow().map { it.message }.toList()
         }
 
+        callState.blockUntilClosed()
+
         assertEquals(3,result.size)
         result.forEachIndexed { index, message ->
             assertEquals("Req:#$index/Resp:#$index",message)
@@ -180,6 +181,8 @@ class ClientCallBidiStreamingTests :
             }
         }
 
+        callState.blockUntilClosed()
+
         assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }
         assert(responseChannel.isClosedForReceive) { "Response channel should be closed for receive" }
     }
@@ -222,6 +225,8 @@ class ClientCallBidiStreamingTests :
             }
         }
 
+        callState.blockUntilCancellation()
+
         verify(exactly = 1) { rpcSpy.call.cancel(any(), any()) }
         assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }
         assert(responseChannel.isClosedForReceive) { "Response channel should be closed for receive" }
@@ -272,6 +277,8 @@ class ClientCallBidiStreamingTests :
             }
         }
 
+        callState.blockUntilCancellation()
+
         verify(exactly = 1) { rpcSpy.call.cancel(any(), any()) }
         assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }
         assert(responseChannel.isClosedForReceive) { "Response channel should be closed for receive" }
@@ -286,7 +293,7 @@ class ClientCallBidiStreamingTests :
         lateinit var requestChannel: SendChannel<HelloRequest>
         lateinit var responseChannel: ReceiveChannel<HelloReply>
         assertFailsWith(IllegalStateException::class, "cancel") {
-            runBlocking {
+            runBlocking(Dispatchers.Default) {
                 val callChannel = stub
                     .withCoroutineContext()
                     .clientCallBidiStreaming(methodDescriptor)
@@ -317,6 +324,8 @@ class ClientCallBidiStreamingTests :
             }
         }
 
+        callState.blockUntilCancellation()
+
         verify(exactly = 1) { rpcSpy.call.cancel(any(), any()) }
         assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }
         assert(responseChannel.isClosedForReceive) { "Response channel should be closed for receive" }
@@ -345,18 +354,15 @@ class ClientCallBidiStreamingTests :
                 requestChannel.close(expectedException)
             }
 
-            //TODO: Cleanup
-//            cause = Status.CANCELLED
-//                .withDescription("Cancelled by client with StreamObserver.onError()")
-//                .withCause(expectedException)
-//                .asRuntimeException()
             assertFailsWithStatus2(Status.CANCELLED, "CANCELLED: $expectedCancelMessage") {
                 responseChannel.consumeAsFlow()
                     .map { it.message }
                     .collect { result.add(it) }
             }
         }
 
+        callState.blockUntilCancellation()
+
         assert(result.isNotEmpty())
         result.forEachIndexed { index, message ->
             assertEquals("Req:#$index/Resp:#$index",message)
@@ -386,11 +392,6 @@ class ClientCallBidiStreamingTests :
             result.add(responseChannel.receive().message)
             requestChannel.close(expectedException)
 
-            //TODO clean up
-//            cause = Status.CANCELLED
-//                .withDescription("Cancelled by client with StreamObserver.onError()")
-//                .withCause(expectedException)
-//                .asRuntimeException()
             assertFailsWithStatus2(Status.CANCELLED, "CANCELLED: $expectedCancelMessage") {
                 responseChannel.consumeAsFlow()
                     .collect { result.add(it.message) }
@@ -437,6 +438,7 @@ class ClientCallBidiStreamingTests :
             responseChannel.cancel()
         }
 
+        callState.blockUntilCancellation()
 
         verify(exactly = 1) { rpcSpy.call.cancel(MESSAGE_CLIENT_CANCELLED_CALL,matchStatus(Status.CANCELLED)) }
         assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/integration/BidiStreamingTests.kt

@@ -111,8 +111,8 @@ class BidiStreamingTests : RpcCallTest<HelloRequest, HelloReply>(GreeterCoroutin
             }
         })
         lateinit var reqChanSpy: SendChannel<HelloRequest>
-        runTest(5000L) {
-            val stub = GreeterCoroutineGrpc.newStub(nonDirectGrpcServerRule.channel)
+        runTest(30_000) {
+            val stub = GreeterCoroutineGrpc.newStub(grpcServerRule.channel)
                 .withInterceptors(callState)
                 .withCoroutineContext()
 
@@ -131,11 +131,6 @@ class BidiStreamingTests : RpcCallTest<HelloRequest, HelloReply>(GreeterCoroutin
             reqJob.join()
         }
 
-        runBlocking { serverJob.join() }
-
-        // 1 - Server requests and receives
-        // 2 - Message is loaded into outbound buffer
-        // 3 - Suspending invocation awaiting next onReady
         coVerify(exactly = 3) { reqChanSpy.send(any()) }
         assert(reqChanSpy.isClosedForSend) { "Request channel should be closed after response channel is closed" }
     }
@@ -162,7 +157,7 @@ class BidiStreamingTests : RpcCallTest<HelloRequest, HelloReply>(GreeterCoroutin
 
         val numMessages = 500000
         val receivedCount = AtomicInteger()
-        runTest(timeout = 60_000) {
+        runTest(timeout = 60_000 * 2) {
             val req = HelloRequest.newBuilder()
                 .setName("test").build()
 

kroto-plus-coroutines/src/test/kotlin/com/github/marcoferrer/krotoplus/coroutines/integration/ClientStreamingBackPressureTests.kt

@@ -201,13 +201,15 @@ class ClientStreamingBackPressureTests :
         }
 
         callState {
-//            blockUntilCancellation()
             client.closed.assertBlocking { "Client must be closed" }
         }
 
         verify(exactly = 1) { rpcSpy.call.cancel(expectedCancelMessage, matchThrowable(expectedException)) }
         assert(requestChannel.isClosedForSend) { "Request channel should be closed for send" }
-        assertExEquals(expectedException, response.getCompletionExceptionOrNull()?.cause)
+        assertExEquals(
+            expectedException,
+            response.getCompletionExceptionOrNull()?.cause ?:response.getCompletionExceptionOrNull()
+        )
         assert(response.isCancelled) { "Response should not be cancelled" }
         assert(runBlocking {serverJob.await() }.isCancelled){ "Server job should be cancelled" }
     }