How you test units or modules that communicate with streams depends on whether the subject under test uses streams as input or output.
- If the test subject observes a flow, you can generate the flow in a fake dependency that you can control from the test.
- If a unit or module exposes a stream, you can read and verify one or more items emitted by the stream in the test.
Create a fake producer
When the subject under test is a stream consumer, one common way to test it is to replace the producer with a fake implementation. For example, given a class that observes a repository that fetches data from two data sources in production:
To create deterministic tests, you can replace the repository and its dependencies with a fake repository that always emits the same fake data:
To emit a pre-defined set of values in a stream, use the flow builder:
class MyFakeRepository : MyRepository {
fun observeCount() = flow {
emit(ITEM_1)
}
}In testing, this fake repository was injected, replacing the real implementation:
@Test
fun myTest() {
// Given a class with fake dependencies:
val sut = MyUnitUnderTest(MyFakeRepository())
// Trigger and verify
...
}Now that you have control over the output of the test subject, you can verify that it is working properly by examining its output.
Confirming flow emissions in testing
If the subject under test exposes a stream, the test needs to make assertions on the data flow elements.
Let's assume that the previous example repository shows the flow:
With certain tests, you only need to check the first emission or a number of items coming from the stream.
You can consume the first emission to the stream by calling first(). This function waits until the first item is received and then sends a cancellation signal to the producer.
@Test
fun myRepositoryTest() = runTest {
// Given a repository that combines values from two data sources:
val repository = MyRepository(fakeSource1, fakeSource2)
// When the repository emits a value
val firstItem = repository.counter.first() // Returns the first item in the flow
// Then check it's the expected item
assertEquals(ITEM_1, firstItem)
}If the test needs to check multiple values, calling toList() causes the stream to wait for the source to emit all of its values and then returns those values as a list. This only works for finite data streams.
@Test
fun myRepositoryTest() = runTest {
// Given a repository with a fake data source that emits ALL_MESSAGES
val messages = repository.observeChatMessages().toList()
// When all messages are emitted then they should be ALL_MESSAGES
assertEquals(ALL_MESSAGES, messages)
}For data flows that require more complex collections of items or do not return a limited number of items, you can use FlowAPI to select and modify items. Here are some examples:
// Take the second item
outputFlow.drop(1).first()
// Take the first 5 items
outputFlow.take(5).toList()
// Takes the first item verifying that the flow is closed after that
outputFlow.single()
// Finite data streams
// Verify that the flow emits exactly N elements (optional predicate)
outputFlow.count()
outputFlow.count(predicate)Continuous collection during the test
Collecting a stream using toList() as seen in the previous example uses collect() internally, and suspends until the entire list of results is ready to be returned.
To insert actions that cause the stream to emit values and assertions on the emitted values, you can continue to collect values from the stream during testing.
For example, take the following Repository class to test, and an accompanying fake data source implementation that has an emit method to dynamically generate values during testing:
class Repository(private val dataSource: DataSource) {
fun scores(): Flow<Int> {
return dataSource.counts().map { it * 10 }
}
}
class FakeDataSource : DataSource {
private val flow = MutableSharedFlow<Int>()
suspend fun emit(value: Int) = flow.emit(value)
override fun counts(): Flow<Int> = flow
}When using this fake in testing, you can create a collector coroutine that will continually receive values from the Repository. In this example, we collect them into a list and then perform assertions on its contents:
@Test
fun continuouslyCollect() = runTest {
val dataSource = FakeDataSource()
val repository = Repository(dataSource)
val values = mutableListOf<Int>()
val collectJob = launch(UnconfinedTestDispatcher()) {
repository.scores().toList(values)
}
dataSource.emit(1)
assertEquals(10, values[0]) // Assert on the list contents
dataSource.emit(2)
dataSource.emit(3)
assertEquals(30, values[2])
assertEquals(3, values.size) // Assert the number of items collected
collectJob.cancel()
}Since the stream exposed in Repository never completes, the toList call that collects it never returns. Therefore, the collection coroutine needs to be explicitly cancelled before the end of the test. Otherwise, runTest will continue to wait for its completion, causing the test to stop responding and eventually fail.
Notice how UnconfinedTestDispatcher is used to collect the coroutine here. This ensures that the collector coroutine is launched eagerly and is ready to receive a value after launch returns.
Using Turbine
The third-party Turbine library offers a convenient API for creating collector coroutines, as well as other convenience features for testing Flows:
@Test
fun usingTurbine() = runTest {
val dataSource = FakeDataSource()
val repository = Repository(dataSource)
repository.scores().test {
// Make calls that will trigger value changes only within test{}
dataSource.emit(1)
assertEquals(10, awaitItem())
dataSource.emit(2)
awaitItem() // Ignore items if needed, can also use skip(n)
dataSource.emit(3)
assertEquals(30, awaitItem())
}
}Testing StateFlows
StateFlow is an observable data holder, which can be collected to observe the values it holds over time as a stream. Note that this stream of values is coupled, meaning that if a value is set with StateFlow quickly, the StateFlow collector is not guaranteed to receive all intermediate values, only the most recent one.
In testing, if you keep in mind the merge, you can collect StateFlow's values just like you can collect any other flow, including with Turbine. Trying to collect and assert all intermediate values can be desirable in some test scenarios.
However, we usually recommend treating StateFlow as a data holder and asserting its value property instead. This way, the test validates the current state of the object at a given point in time, and does not depend on whether the merge occurred or not.
For example, take this ViewModel that collects values from a Repository and exposes them to the UI in StateFlow:
class MyViewModel(private val myRepository: MyRepository) : ViewModel() {
private val _score = MutableStateFlow(0)
val score: StateFlow<Int> = _score.asStateFlow()
fun initialize() {
viewModelScope.launch {
myRepository.scores().collect { score ->
_score.value = score
}
}
}
}