Turbine is a small testing library for kotlinx.coroutines
Flow.
flowOf("one", "two").test {
assertEquals("one", awaitItem())
assertEquals("two", awaitItem())
awaitComplete()
}A turbine is a rotary mechanical device that extracts energy from a fluid flow and converts it into useful work.
repositories {
mavenCentral()
}
dependencies {
testImplementation 'app.cash.turbine:turbine:0.7.0'
}Snapshots of the development version are available in Sonatype's snapshots repository.
repositories {
maven {
url 'https://oss.sonatype.org/content/repositories/snapshots/'
}
}
dependencies {
testImplementation 'app.cash.turbine:turbine:0.8.0-SNAPSHOT'
}The entrypoint for the library is the test extension for Flow<T> which accepts a validation
block. Like collect, test is a suspending function that will not return until the flow is
complete or canceled.
someFlow.test {
// Validation code here!
}Inside the test block you must consume all received events from the flow. Failing to consume all
events will fail your test.
flowOf("one", "two").test {
assertEquals("one", awaitItem())
}Exception in thread "main" AssertionError:
Unconsumed events found:
- Item(two)
- Complete
As the exception indicates, consuming the "two" item is not enough. The complete event must
also be consumed.
flowOf("one", "two").test {
assertEquals("one", awaitItem())
assertEquals("two", awaitItem())
awaitComplete()
}Received events can be explicitly ignored, however.
flowOf("one", "two").test {
assertEquals("one", awaitItem())
cancelAndIgnoreRemainingEvents()
}Additionally, we can receive the most recent emitted item and ignore the previous ones.
flowOf("one", "two", "three")
.map {
delay(100)
it
}
.test {
// 0 - 100ms -> no emission yet
// 100ms - 200ms -> "one" is emitted
// 200ms - 300ms -> "two" is emitted
// 300ms - 400ms -> "three" is emitted
delay(250)
assertEquals("two", expectMostRecentItem())
cancelAndIgnoreRemainingEvents()
}Unlike collect, a flow which causes an exception will still be exposed as an event that you
must consume.
flow { throw RuntimeException("broken!") }.test {
assertEquals("broken!", awaitError().message)
}Failure to consume an error will result in the same unconsumed event exception as above, but with the exception added as the cause so that the full stacktrace is available.
flow { throw RuntimeException("broken!") }.test { }java.lang.AssertionError: Unconsumed events found:
- Error(RuntimeException)
at app.cash.turbine.ChannelBasedFlowTurbine.ensureAllEventsConsumed(FlowTurbine.kt:240)
... 53 more
Caused by: java.lang.RuntimeException: broken!
at example.MainKt$main$1.invokeSuspend(Main.kt:7)
... 32 more
Calls to awaitItem(), awaitComplete(), and awaitError() are suspending and will wait
for events from asynchronous flows.
channelFlow {
withContext(IO) {
Thread.sleep(100)
send("item")
}
}.test {
assertEquals("item", awaitItem())
awaitComplete()
}By default, when one of the "await" methods suspends waiting for an event it will timeout after one second.
channelFlow {
withContext(IO) {
Thread.sleep(2_000)
send("item")
}
}.test {
assertEquals("item", awaitItem())
awaitComplete()
}Exception in thread "main" TimeoutCancellationException: Timed out waiting for 1000 ms
A longer timeout can be specified as an argument to test.
channelFlow {
withContext(IO) {
Thread.sleep(2_000)
send("item")
}
}.test(timeout = 3_000) {
assertEquals("item", awaitItem())
awaitComplete()
}Asynchronous flows can be canceled at any time and will not require consuming a complete or error event.
channelFlow {
withContext(IO) {
repeat(10) {
Thread.sleep(200)
send("item $it")
}
}
}.test {
assertEquals("item 0", awaitItem())
assertEquals("item 1", awaitItem())
assertEquals("item 2", awaitItem())
cancel()
}Emissions to hot flows that don't have active consumers are dropped. It's important to call test
(and therefore have an active collector) on a flow before emissions to a flow are made. For example:
val mutableSharedFlow = MutableSharedFlow<Int>(replay = 0)
mutableSharedFlow.emit(1)
mutableSharedFlow.test {
assertEquals(awaitItem(), 1)
cancelAndConsumeRemainingEvents()
}will fail with a timeout exception.
kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms
(Coroutine boundary)
at app.cash.turbine.ChannelBasedFlowTurbine$awaitEvent$2.invokeSuspend(FlowTurbine.kt:238)
at app.cash.turbine.ChannelBasedFlowTurbine$withTimeout$2.invokeSuspend(FlowTurbine.kt:206)
at app.cash.turbine.ChannelBasedFlowTurbine.awaitItem(FlowTurbine.kt:243)
Proper usage of Turbine with hot flows looks like the following.
val mutableSharedFlow = MutableSharedFlow<Int>(replay = 0)
mutableSharedFlow.test {
mutableSharedFlow.emit(1)
assertEquals(awaitItem(), 1)
cancelAndConsumeRemainingEvents()
}The hot flow types Kotlin currently provides are:
MutableStateFlowStateFlowMutableSharedFlowSharedFlow- Channels converted to flow with
Channel.consumeAsFlow
Copyright 2018 Square, Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.