Use test-specific unconfined when test scheduler is in use

This retains the test scheduler virtual time.

Unfortunately this requires taking a dependency on the test library and also relying on unstable API.

Author: JakeWharton

build.gradle

@@ -42,12 +42,12 @@ kotlin {
     commonMain {
       dependencies {
         api "org.jetbrains.kotlinx:kotlinx-coroutines-core:${versions.coroutines}"
+        implementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:${versions.coroutines}"
       }
     }
     commonTest {
       dependencies {
         implementation 'org.jetbrains.kotlin:kotlin-test'
-        implementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:${versions.coroutines}"
       }
     }
   }

src/commonMain/kotlin/app/cash/turbine/flow.kt

@@ -21,12 +21,15 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
 import kotlinx.coroutines.Dispatchers.Unconfined
+import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestCoroutineScheduler
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
 
 /**
  * Terminal flow operator that collects events from given flow and allows the [validate] lambda to
@@ -108,10 +111,16 @@ public fun <T> Flow<T>.testIn(scope: CoroutineScope): ReceiveTurbine<T> {
   return turbine
 }
 
+@OptIn(ExperimentalCoroutinesApi::class) // New kotlinx.coroutines test APIs are not stable 😬
 private fun <T> Flow<T>.collectTurbineIn(scope: CoroutineScope): Turbine<T> {
   lateinit var channel: Channel<T>
 
-  val job = scope.launch(Unconfined, start = UNDISPATCHED) {
+  // Use test-specific unconfined if test scheduler is in use to inherit its virtual time.
+  val unconfined = scope.coroutineContext[TestCoroutineScheduler]
+    ?.let(::UnconfinedTestDispatcher)
+    ?: Unconfined
+
+  val job = scope.launch(unconfined, start = UNDISPATCHED) {
     channel = collectIntoChannel(this)
   }
 

src/commonTest/kotlin/app/cash/turbine/FlowTest.kt

@@ -22,12 +22,16 @@ import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
 import kotlin.test.assertSame
 import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+import kotlin.time.ExperimentalTime
+import kotlin.time.measureTime
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
 import kotlinx.coroutines.Dispatchers.Unconfined
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.RENDEZVOUS
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.emitAll
@@ -520,4 +524,16 @@ class FlowTest {
       assertSame(error, actual.cause)
     }
   }
+
+  @OptIn(ExperimentalTime::class)
+  @Test fun turbineSkipsDelaysInRunTest() = runTest {
+    val took = measureTime {
+      flow<Nothing> {
+        delay(5.seconds)
+      }.test {
+        awaitComplete()
+      }
+    }
+    assertTrue(took < 5.seconds, "$took > 5s")
+  }
 }