Repository: cashapp/turbine (kotlin)

Issue:
`test` and `testIn` do not behave consistently 
I've noticed that if a `Flow` is canceled "externally" (e.g. a `StateFlow` that is canceled only when the `CoroutineScope` it was launched in is canceled) then `test` and `testIn` do not behave the same.

While `test` reports all unconsumed events and fails the test, `testIn` does not and passes the test.

```kt
private val flow = flow {
  emit(1)
  emit(2)
  awaitCancellation()
}

// Throws
//   app.cash.turbine.TurbineAssertionError: Unconsumed events found:
//    - Item(2)
@Test
fun test() = runTest {
  flow.test {
    assertEquals(1, awaitItem())
  }
}

// Passes
@Test
fun testIn() = runTest {
  turbineScope {
    val turbine = flow.testIn(backgroundScope)
    assertEquals(1, turbine.awaitItem())
  }
}
```

This is not necessarily caused by the `backgroundScope` as it can be replicated with a custom `CoroutineScope()` + `scope.cancel()` or even when canceling the `Job` the turbine was launched in.

I've figured out this behavior is caused by not handling `CancellationException` as a special case. From the docs of `CancellationException`: "_It indicates normal cancellation of a coroutine_".

Pull request description:
This potentially fixes https://github.com/cashapp/turbine/issues/247.

Patch:
diff --git a/src/commonMain/kotlin/app/cash/turbine/Turbine.kt b/src/commonMain/kotlin/app/cash/turbine/Turbine.kt
index a191be8e..138f450a 100644
--- a/src/commonMain/kotlin/app/cash/turbine/Turbine.kt
+++ b/src/commonMain/kotlin/app/cash/turbine/Turbine.kt
@@ -211,6 +211,7 @@ internal class ChannelTurbine<T>(
     var cause: Throwable? = null
     while (true) {
       val event = channel.takeEventUnsafe() ?: break
+      if (event is Event.Error && event.throwable is CancellationException) break
       if (!(ignoreTerminalEvents && event.isTerminal)) unconsumed += event
       if (event is Event.Error) {
         cause = event.throwable
diff --git a/src/commonMain/kotlin/app/cash/turbine/flow.kt b/src/commonMain/kotlin/app/cash/turbine/flow.kt
index 104fccda..1f948647 100644
--- a/src/commonMain/kotlin/app/cash/turbine/flow.kt
+++ b/src/commonMain/kotlin/app/cash/turbine/flow.kt
@@ -195,7 +195,8 @@ private fun <T> testInInternal(flow: Flow<T>, timeout: Duration?, scope: Corouti
     if (debug) println("Scope ending ${exception ?: ""}")
 
     // Only validate events were consumed if the scope is exiting normally.
-    if (exception == null) {
+    // CancellationException also indicates _normal_ cancellation of a coroutine.
+    if (exception == null || exception is CancellationException) {
       turbine.ensureAllEventsConsumed()
     }
   }


Test patch:
diff --git a/src/commonTest/kotlin/app/cash/turbine/FlowInScopeTest.kt b/src/commonTest/kotlin/app/cash/turbine/FlowInScopeTest.kt
index b9e65230..9c6b11e1 100644
--- a/src/commonTest/kotlin/app/cash/turbine/FlowInScopeTest.kt
+++ b/src/commonTest/kotlin/app/cash/turbine/FlowInScopeTest.kt
@@ -12,7 +12,9 @@ import kotlin.time.Duration.Companion.seconds
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletionHandlerException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Dispatchers.Default
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
@@ -96,6 +98,31 @@ class FlowInScopeTest {
     assertFalse(collecting)
   }
 
+  @Test fun unconsumedItemThrowsWhenCancelledExternally() = runTestTurbine {
+    // We have to use an exception handler rather than assertFailsWith because runTest also uses
+    // one which defers throwing until its block completes.
+    val exceptionHandler = RecordingExceptionHandler()
+    launch(start = CoroutineStart.UNDISPATCHED) {
+      withContext(exceptionHandler) {
+        flow {
+          emit("item!")
+          emitAll(neverFlow()) // Avoid emitting complete
+        }.testIn(this)
+      }
+    }.cancel()
+    val exception = exceptionHandler.exceptions.removeFirst()
+    assertTrue(exception is CompletionHandlerException)
+    val cause = exception.cause
+    assertTrue(cause is AssertionError)
+    assertEquals(
+      """
+      |Unconsumed events found:
+      | - Item(item!)
+      """.trimMargin(),
+      cause.message,
+    )
+  }
+
   @Test fun unconsumedItemThrows() = runTestTurbine {
     // We have to use an exception handler rather than assertFailsWith because runTest also uses
     // one which defers throwing until its block completes.