wait for call to close before asserting

marcoferrer · Jan 12, 2020 · 9bd2cde · 9bd2cde
1 parent 979174d
commit 9bd2cde
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 15 deletions.
diff --git a/...kotlin/com/github/marcoferrer/krotoplus/coroutines/client/ClientCallBidiStreamingTests.kt b/...kotlin/com/github/marcoferrer/krotoplus/coroutines/client/ClientCallBidiStreamingTests.kt
@@ -136,6 +136,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 +182,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 +226,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 +278,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" }
@@ -317,6 +325,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 +355,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 +393,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 +439,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" }

diff --git a/...test/kotlin/com/github/marcoferrer/krotoplus/coroutines/integration/BidiStreamingTests.kt b/...test/kotlin/com/github/marcoferrer/krotoplus/coroutines/integration/BidiStreamingTests.kt
@@ -112,7 +112,7 @@ class BidiStreamingTests : RpcCallTest<HelloRequest, HelloReply>(GreeterCoroutin
         })
         lateinit var reqChanSpy: SendChannel<HelloRequest>
         runTest(5000L) {
-            val stub = GreeterCoroutineGrpc.newStub(nonDirectGrpcServerRule.channel)
+            val stub = GreeterCoroutineGrpc.newStub(grpcServerRule.channel)
                 .withInterceptors(callState)
                 .withCoroutineContext()
 
@@ -133,9 +133,6 @@ class BidiStreamingTests : RpcCallTest<HelloRequest, HelloReply>(GreeterCoroutin
 
         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 +159,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()