I have a callback based API like this:
class CallbackApi {
fun addListener(callback: Callback) {
// todo
}
fun removeListener(callback: Callback) {
// todo
}
interface Callback {
fun onResult(result: Int)
}
}
and an extension function which converts the API into a hot cold flow:
fun CallbackApi.toFlow() = callbackFlow<Int> {
val callback = object : CallbackApi.Callback {
override fun onResult(result: Int) {
trySendBlocking(result)
}
}
addListener(callback)
awaitClose { removeListener(callback) }
}
Would you mind suggesting how to write a unit test which ensures that the API is correctly converted to a hot flow?
Here is my attempt. By trial and error, I came up with this solution.
@Test
fun callbackFlowTest() = runBlocking {
val callbackApi = mockk<CallbackApi>()
val callbackSlot = slot<CallbackApi.Callback>()
every { callbackApi.addListener(capture(callbackSlot)) } just Runs
every { callbackApi.removeListener(any()) } just Runs
val list = mutableListOf<Int>()
val flow: Flow<Int> = callbackApi.toFlow().onEach { list.add(it) }
val coroutineScope = CoroutineScope(this.coroutineContext + SupervisorJob())
flow.launchIn(coroutineScope)
yield()
launch {
callbackSlot.captured.onResult(10)
callbackApi.removeListener(mockk()) // this was a misunderstanding
}.join()
assert(list.single() == 10)
}
But I don't understand two pieces of this solution.
1- In the absence of that SupervisorJob(), it appears that test will never end. Maybe collecting the flow never ends for some reason, which I don't understand. I'm feeding captured callback in a separate coroutine.
2- If I remove the launch body which callbackSlot.captured.onResult(10) is inside it, test will fail with this error UninitializedPropertyAccessException: lateinit property captured has not been initialized. I would think that yield should start the flow.
Here is the solution I found based on @Joffrey useful guides:
@Test
fun callbackFlowTestSolution() = runTest {
val callbackApi = mockk<CallbackApi>()
val callbackSlot = slot<CallbackApi.Callback>()
every { callbackApi.addListener(capture(callbackSlot)) } just Runs
every { callbackApi.removeListener(any()) } just Runs
val itemsToSend = Array(9) { index -> index } // [0, 1, 2, 3, 4, 5, 6, 7, 8]
val collectedItems = mutableListOf<Int>()
val flow: Flow<Int> = callbackApi.toFlow().onEach { collectedItems.add(it) }
val collectJob = launch { flow.collect() }
advanceUntilIdle() // wait for callbackFlow builder to call addListener
itemsToSend.forEach { callbackSlot.captured.onResult(it) }
advanceUntilIdle() // wait for flow collection
collectJob.cancel()
advanceUntilIdle() // wait for awaitClose
verify { callbackApi.removeListener(callbackSlot.captured) }
assertArrayEquals(itemsToSend.toIntArray(), collectedItems.toIntArray())
}
Please consider upgrading to kotlinx-coroutines-test:1.6.0 to use runTest. In older versions we can use runBlockingTest, but it's deprecated.
UPDATE: If you want to check that the flow can collect items one by one
@Test
fun callbackFlowTestSolution() = runTest {
val callbackApi = mockk<CallbackApi>()
val callbackSlot = slot<CallbackApi.Callback>()
every { callbackApi.addListener(capture(callbackSlot)) } just Runs
every { callbackApi.removeListener(any()) } just Runs
val itemsToSend = Array(9) { index -> index } // [0, 1, 2, 3, 4, 5, 6, 7, 8]
var lastCollectedItem: Int? = null
val flow: Flow<Int> = callbackApi.toFlow().onEach { lastCollectedItem = it }
val collectJob = launch { flow.collect() }
advanceUntilIdle() // wait for callbackFlow builder to call addListener
itemsToSend.forEach {
callbackSlot.captured.onResult(it)
advanceUntilIdle() // wait for flow collection
assertEquals(it, lastCollectedItem)
}
collectJob.cancel()
advanceUntilIdle() // wait for awaitClose
verify { callbackApi.removeListener(callbackSlot.captured) }
}