Throw unconsumed events if the scope is cancelled

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

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

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.