Unit Testing Kotlin Channels & Flows

Mohit will share with you best practices for testing Channels and Flows. He will show practical use cases where we have to handle retries, errors and delays. For each of these scenarios, he will show you patterns to use for testing. We will look at features of the coroutine testing library such as advancing time forward. We will also look at ways to do assertions on a Flow.

Transcript

2. • Use case with ViewModel & Repo • Testing Repo & ViewModel • Best practices for unit testing
3. ViewModel Repository
5. interface ApiService { @GET("/users/{id}") suspend fun userDetails(@Path("id") id: Int): UserDetails }
6. class UserRepository(val apiService: ApiService) { fun userDetails(id: Int): Flow<Result<UserDetails.> { } }
7. fun userDetails(id: Int): Flow<Result<UserDetails.> { } Coroutines Library fun <T> flow( block: suspend FlowCollector<T>.() .> Unit ): Flow<T>
8. fun userDetails(id: Int): Flow<Result<UserDetails.> { } Coroutines Library interface FlowCollector<in T> { suspend fun emit(value: T) }
9. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) } }
10. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) } }
11. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) } } Coroutines Library Dispatcher - Control scheduling threads - Default, IO, Main, etc...
12. class UserRepository( val apiService: ApiService, val dispatcher: CoroutineDispatcher ) { fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val users = apiService.userDetails(id) emit(Result.success(users)) }.flowOn(dispatcher) }
13. class UserRepository( val apiService: ApiService, val dispatcher: CoroutineDispatcher ) { fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val users = apiService.userDetails(id) emit(Result.success(users)) }.flowOn(dispatcher) }
14. ViewModel Repository
15. class UserDetailViewModel : ViewModel { }
16. class UserDetailViewModel : ViewModel { val scope = CoroutineScope(Dispatchers.Main) } Coroutines Library fun CoroutineScope(context: CoroutineContext)
17. class UserDetailViewModel : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { } }
18. class UserDetailViewModel(val repository: UserRepository) : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { } }
19. class UserDetailViewModel(val repository: UserRepository) : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { val flow = repository.getUserDetails(id = 1) } }
20. class UserDetailViewModel(val repository: UserRepository) : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { val flow = repository.getUserDetails(id = 1) } } Coroutines Library fun <T> Flow<T>.collect( action: suspend (value: T) .> Unit )
21. class UserDetailViewModel(val repository: UserRepository) : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> } } }
22. class UserDetailViewModel( val repository: UserRepository, val stateManager: StateManager ) : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> stateManager.dispatchState(result) } } }
23. class UserDetailViewModel( val repository: UserRepository, val stateManager: StateManager ) : ViewModel { val scope = CoroutineScope(Dispatchers.Main) scope.launch { val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> stateManager.dispatchState(result) } } }
24. class UserDetailViewModel( val repository: UserRepository, val stateManager: StateManager ): ViewModel { { val scope = CoroutineScope(Dispatchers.Main) } Architecture Components val ViewModel.viewModelScope: CoroutineScope get() { val scope: CoroutineScope? = this.getTag(JOB_KEY) return scope }
25. ViewModel Repository
26. 1. Flow Collection 2. Error handling 3. Retries with Delays (Advanced)
27. testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.3.3' • Experimental • runBlockingTest • Test Coroutine Scope • Test Coroutine Dispatcher
28. 1. Flow emits Success 2. Flow emits Error 3. Retries with delay (Advanced)
29. class UserRepository( val apiService: ApiService, val dispatcher: CoroutineDispatcher ) { fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.flowOn(dispatcher) } }
30. @Test fun `should get users details on success`() = runBlocking { } Coroutines Library RunBlocking - Blocks untils coroutine completes
31. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() }
32. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() } API Service suspend fun userDetails(id: Int): UserDetails
33. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() { on { userDetails(1) } doReturn userDetails } }
34. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() { on { userDetails(1) } doReturn userDetails } } Mockito-Kotlin fun <R> on(methodCall: T.() .> R)
35. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() } Mockito-Kotlin fun onBlocking(m: suspend T.() .> R) { return runBlocking { Mockito.`when`(mock.m()) } }
36. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() { onBlocking { userDetails(1) } doReturn userDetails } }
37. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() { onBlocking { userDetails(1) } doReturn userDetails } val dispatcher = TestCoroutineDispatcher() val repository = UserRepository(userService, dispatcher) val flow = repository.getUserDetails(id = 1) }
38. @Test fun `should get users details on success`() = runBlocking { val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService>() { onBlocking { userDetails(1) } doReturn userDetails } val dispatcher = TestCoroutineDispatcher() val repository = UserRepository(userService, dispatcher) val flow = repository.getUserDetails(id = 1) val result = flow.single() result: Result<UserDetails> result.isSuccess.assertTrue() }
39. 1. Flow emits Success 2. Flow emits Error 3. Retries with delay (Advanced)
40. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.flowOn(dispatcher) }
41. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.flowOn(dispatcher) } ViewModel scope.launch { val flow: = userRepository.getUserDetails(id = 1) flow.collect { } }
42. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.flowOn(dispatcher) } Coroutines-Library fun <T> Flow<T>.catch( action: suspend FlowCollector<T>.(t: Throwable) .> Unit ): Flow<T>
43. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
44. @Test fun `should get error for user details`() = runBlocking { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { throw IOException() } } }
45. @Test fun `should get error for user details`() = runBlocking { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { throw IOException() } } val repository = UserRepository(apiService, dispatcher) val flow = repository.getUserDetails(id = 1) val result = flow.single() result.isFailure.assertTrue() }
46. 1. Flow emits Success 2. Flow emits Error 3. Retries with delay (Advanced)
49. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.catch { emit(Result.failure(it)) } } Coroutines-Library fun retry( retries: Long, block: suspend (Throwable) .>Boolean ): Flow<T>
50. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
51. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
52. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
53. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(users)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
54. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
55. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
56. 1. All Retries fail with Error 2. Retry succeeds
57. 1. Launch a coroutine with a test scope and test dispatcher. 2. Advance virtual time forward.
58. @Test fun `should retry with error`() { } Coroutine-Test-Library fun runBlockingTest( block: suspend TestCoroutineScope.() .> Unit ) fun TestCoroutineDispatcher.runBlockingTest( block: suspend TestCoroutineScope.() .> Unit )
59. @Test fun `should retry with error`() = dispatcher.runBlockingTest { }
60. @Test fun `should retry with error`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { throw IOException() } } val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> result.isFailure.assertTrue() } }
61. @Test fun `should retry with error`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { throw IOException() } } val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> result.isFailure.assertTrue() } }
62. @Test fun `should retry with error`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { throw IOException() } } val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> result.isFailure.assertTrue() } }
63. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
64. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
65. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val users = userService.userDetails(id) emit(Result.success(users)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
66. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val users = userService.userDetails(id) emit(Result.success(users)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) } Coroutine-Test-Library fun runBlockingTest { dispatcher.advanceUntilIdle() }
67. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
68. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
69. @Test fun `should retry with error`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { throw IOException() } } val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> result.isFailure.assertTrue() } }
70. 1. All Retries fail with Error 2. Retry succeeds
71. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
72. 1. Pause Dispatcher 2. Advance virtual time forward by certain milliseconds.
73. @Test fun `should retry with success`() = dispatcher.runBlockingTest { pauseDispatcher { } } Coroutine-Test-Library pausedDispatcher - Do not start coroutines eagerly
74. @Test fun `should retry with success`() = dispatcher.runBlockingTest { var throwError = true val userDetails = UserDetails(1, "User 1", "avatar_url") val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } }
75. @Test fun `should retry with success`() = dispatcher.runBlockingTest { var throwError = true val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } pauseDispatcher { val flow = repository.getUserDetails(id = 1) launch { flow.collect { it.isSuccess.assertTrue() } } } }
76. @Test fun `should retry with success`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } pauseDispatcher { val flow = repository.getUserDetails(id = 1) launch { flow.collect { it.isSuccess.assertTrue() } } advanceTimeBy(DELAY_ONE_SECOND) throwError = false advanceTimeBy(DELAY_ONE_SECOND) } }
77. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
78. @Test fun `should retry with success`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } pauseDispatcher { val flow = repository.getUserDetails(id = 1) launch { flow.collect { it.isSuccess.assertTrue() } } advanceTimeBy(DELAY_ONE_SECOND) throwError = false advanceTimeBy(DELAY_ONE_SECOND) } }
79. @Test fun `should retry with success`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } pauseDispatcher { val flow = repository.getUserDetails(id = 1) launch { flow.collect { it.isSuccess.assertTrue() } } advanceTimeBy(DELAY_ONE_SECOND) throwError = false advanceTimeBy(DELAY_ONE_SECOND) } }
80. @Test fun `should retry with success`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } pauseDispatcher { val flow = repository.getUserDetails(id = 1) launch { flow.collect { it.isSuccess.assertTrue() } } advanceTimeBy(DELAY_ONE_SECOND) throwError = false advanceTimeBy(DELAY_ONE_SECOND) } }
81. fun userDetails(id: Int): Flow<Result<UserDetails.> { return flow { val userDetails = apiService.userDetails(id) emit(Result.success(userDetails)) }.retry(retries = 2) { t -> (t is Exception).also { if (it) delay(DELAY_ONE_SECOND) } } .catch { emit(Result.failure(it)) } .flowOn(dispatcher) }
82. @Test fun `should retry with success`() = dispatcher.runBlockingTest { val apiService = mock<ApiService> { onBlocking { userDetails(1) } doAnswer { if (throwError) throw IOException() else userDetails } } pauseDispatcher { val flow = repository.getUserDetails(id = 1) launch { flow.collect { it.isSuccess.assertTrue() } } advanceTimeBy(DELAY_ONE_SECOND) throwError = false advanceTimeBy(DELAY_ONE_SECOND) } }
83. 1. Flow emits Success 2. Flow emits Error 3. Retries with delay
84. ViewModel Repository
85. class UserDetailViewModel( val repository: UserRepository, val stateManager: StateManager ) : ViewModel { viewModelScope.launch { val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> stateManager.dispatchState(result) } } }
86. 1. Dispatchers.setMain() 2. Dispatchers.resetMain()
87. class CoroutineTestRule( val dispatcher= TestCoroutineDispatcher() ) : TestWatcher() { override fun starting(description: Description?) { super.starting(description) Dispatchers.setMain(dispatcher) } }
88. class CoroutineTestRule( val dispatcher= TestCoroutineDispatcher() ) : TestWatcher() { override fun finished(description: Description?) { super.finished(description) Dispatchers.resetMain() dispatcher.cleanupTestCoroutines() } }
89. @get:Rule val rule = CoroutineTestRule() val repository = mock<UserRepository>() val stateManager = mock<StateManager>() val viewModel = UserDetailsViewModel(repository, stateManager)
90. @Test fun `should dispatch details`() = rule.dispatcher.runBlockingTest { }
91. class UserDetailViewModel( val repository: UserRepository, val stateManager: StateManager ) : ViewModel { viewModelScope.launch { val flow = repository.getUserDetails(id = 1) flow.collect { result: Result<UserDetails> -> stateManager.dispatchState(result) } } }
92. Flow Channel
93. @Test fun `should dispatch details`() = rule.dispatcher.runBlockingTest { val userDetails = UserDetails(1, "User 1", "avatar") val result = Result.success(userDetails) }
94. @Test fun `should dispatch details`() = rule.dispatcher.runBlockingTest { val userDetails = UserDetails(1, "User 1", "avatar") val result = Result.success(userDetails) val channel = Channel<Result<UserDetails.>() val flow = channel.consumeAsFlow() whenever(repository.getUserDetails(id = 1)) doReturn flow } Flow Channel Convert to Flow
95. @Test fun `should dispatch details`() = rule.dispatcher.runBlockingTest { val userDetails = UserDetails(1, "User 1", "avatar") val result = Result.success(userDetails) val channel = Channel<Result<UserDetails.>() val flow = channel.consumeAsFlow() whenever(repository.getUserDetails(id = 1)) doReturn flow launch { channel.send(result) } } Flow Channel Send Producer to send values
96. @Test fun `should dispatch details`() = rule.dispatcher.runBlockingTest { val userDetails = UserDetails(1, "User 1", "avatar") val result = Result.success(userDetails) val channel = Channel<Result<UserDetails.>() val flow = channel.consumeAsFlow() whenever(repository.getUserDetails(id = 1)) doReturn flow launch { channel.send(result) } userDetailsViewModel.getUserDetails() verify(stateManager).dispatch(result) }
97. ViewModel Repository
98. @Test fun `should get users details on success`() = runBlocking { ... flow.collect { } } RxJava Observable.test()
99. Flow Channel API fun expectItem(): T fun expectNoMoreEvents() fun expectComplete() fun expectError(): Throwable
100. @Test fun `should get users details on success`() = runBlocking { ... flow.test { expectItem() assertEquals userDetails expectComplete() } }
101. • Channels & Flow in Practice https://speakerdeck.com/heyitsmohit/channels-and-flows-in-practice • Dissecting Coroutines https://speakerdeck.com/heyitsmohit/dissecting-coroutines • Koltin Channels Under the Hood https://medium.com/@heyitsmohit/kotlin-coroutine-channels-under-the-hood-part-1