Turbine Documentation

To test multiple flows concurrently, use testIn to assign each Turbine to a separate variable. Unlike test, testIn does not automatically clean up its coroutine. To prevent tests from hanging, you should:

1. Use runTest's backgroundScope to handle automatic cleanup.
2. Or, manually call cancel(), awaitComplete(), or awaitError() before the scope ends.

ensureAllEventsConsumed() will be invoked when the calling coroutine completes.

runTest {
  turbineScope {
    val turbine1 = flowOf(1).testIn(backgroundScope)
    val turbine2 = flowOf(2).testIn(backgroundScope)
    assertEquals(1, turbine1.awaitItem())
    assertEquals(2, turbine2.awaitItem())
    turbine1.awaitComplete()
    turbine2.awaitComplete()
  }
}

To use Turbine in your Kotlin project, add mavenCentral() to your repositories and include app.cash.turbine:turbine as a testImplementation dependency in your build.gradle file.

repositories {
  mavenCentral()
}
dependencies {
  testImplementation("app.cash.turbine:turbine:1.2.1")
}

You can provide a name to test, testIn, or the Turbine() constructor. This name will be included in any TurbineAssertionError to help identify which turbine failed.

runTest {
  turbineScope {
    val turbine1 = flowOf(1).testIn(backgroundScope, name = "turbine 1")
    turbine1.awaitComplete()
  }
}
// Error: Expected complete for turbine 1 but found Item(1)

The simplest way to test a single Flow is to use the test extension function. This launches a new coroutine, collects the flow, and provides a ReceiveTurbine in a validation block. When the block completes, the coroutine is cancelled and ensureAllEventsConsumed() is automatically called.

Note: Failing to consume all events (including completion or errors) will cause the test to fail with an AssertionError.

flowOf("one").test {
  assertEquals("one", awaitItem())
  awaitComplete()
}

If a flow emits more items than you awaitItem(), Turbine will throw an AssertionError at the end of the validation block. You can handle this in two ways:

1. Ignore remaining events: Use cancelAndIgnoreRemainingEvents() to stop validation without failing due to unconsumed items.
2. Get the most recent item: Use expectMostRecentItem() to retrieve the latest emitted item and ignore all previous ones.

// Option 1: Ignore remaining
flowOf("one", "two").test {
  assertEquals("one", awaitItem())
  cancelAndIgnoreRemainingEvents()
}

// Option 2: Expect most recent
flowOf("one", "two", "three").test {
  delay(250)
  assertEquals("two", expectMostRecentItem())
  cancelAndIgnoreRemainingEvents()
}

Turbine uses a wall-clock timeout (defaulting to 3 seconds) for all await* calls. This ignores runTest virtual time. You can override the timeout in several ways:

1. In test: Pass timeout to the extension function.
2. In testIn: Pass timeout to the extension function.
3. In standalone Turbine: Pass timeout to the constructor.
4. Globally for a block: Wrap code in withTurbineTimeout(duration).

Turbine provides a .test { ... } extension function for kotlinx.coroutines.flow.Flow. Inside the test block, you can use methods like awaitItem() to consume emitted values and awaitComplete() to verify the flow has finished successfully.

flowOf("one", "two").test {
  assertEquals("one", awaitItem())
  assertEquals("two", awaitItem())
  awaitComplete()
}

You can use Turbine as a standalone object to communicate with test code outside of a Flow. This is useful for testing fakes or manual event emitters. Use add() to push items into the Turbine and await* methods to consume them.

class FakeNavigator : Navigator {
  val goTos = Turbine<Screen>()

  override fun goTo(screen: Screen) {
    goTos.add(screen)
  }
}

// Usage in test
runTest {
  val navigator = FakeNavigator()
  // ... setup code ...
  models.test {
    assertEquals(UiModel(title = "Hi there"), awaitItem())
    events.emit(UiEvent.Close)
    assertEquals(Screens.Back, navigator.goTos.awaitItem())
  }
}

For codebases mixing coroutines and non-coroutines code, Turbine provides non-suspending take* methods (e.g., takeItem()). These behave like a simple queue.

Warning: On JVM platforms, these methods will throw an exception if called from a suspending context. They should only be used from non-suspending contexts.

val navigator = FakeNavigator()
// ... setup code ...
val testObserver = models.test()
testObserver.assertValue(UiModel(title = "Hi there"))
events.accept(UiEvent.Close)
assertEquals(Screens.Back, navigator.takeItem())