Unit Testing Channels & Flows [Android Worldwide]

Unit testing Channels and Flows can be a challenge as they are fairly new. In this talk, I will share with you how to implement and test practical examples from my experience. These examples are testing delays, retries, and errors. I'll also share testing more complex examples such as polling. For each use case, we'll look at how to use features in the coroutines library such as runBlockingTest and TestCoroutineDispatcher. From my journey of using and testing Flows in production, I'll share the challenges I experienced.

Transcript

1. Mohit Sarveiya, Android & Kotlin Google Dev Expert Unit Testing Kotlin Channels & Flows @heyitsmohit
2. Unit Testing Channels & Flows • Testing Flows • Errors • Delays • View Model Test • Common Testing Problems 
 

5. Flow of locations
6. flowOfLocations: Flow<Location> .filter { it.locationType == LocationType.RESTAURANT } .map { it.name }
7. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... }
8. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } How do we test this flow?
9. Kotlin/kotlinx.coroutines Code ! Issues Pull Requests kotlinx-coroutines-core 2,093 commits 4c28f942 13 hours ago Kotlinx-coroutines-debug Kotlinx-coroutines-test reactive js benchmarks
10. Kotlin/kotlinx.coroutines Code ! Issues Pull Requests kotlinx-coroutines-core 2,093 commits 4c28f942 13 hours ago Kotlinx-coroutines-debug Kotlinx-coroutines-test reactive js benchmarks Testing Module
11. Kotlin/kotlinx.coroutines Code ! Issues Pull Requests README.md Module Kotlin-coroutines-test Test utilities for kotlinx.coroutines dependencies { testImplementation ‘org.jetbrains.kotlinx:kotlinx-coroutines-test:x.x.x’ }
12. @Test fun `should get locations`() { }
13. @Test fun `should get locations`() = runBlockingTest { } Creates a coroutine
14. @Test fun `should get locations`() = runBlockingTest { } Test Scope
15. @Test fun `should get locations`() = runBlockingTest { } Test Scope + Test Dispatcher
16. @Test fun `should get locations`() = runBlockingTest { } Test Scope + Test Dispatcher What is executed in this coroutine?
17. @Test fun `should get locations`() = runBlockingTest { } Test Scope + Test Dispatcher
18. @Test fun `should get locations`() = runBlockingTest { } Test Scope + Test Dispatcher
19. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... }
20. @Test fun `should get locations`() = runBlockingTest { val flowOfLocations = mockFlow() val locations: List<Location> = flowOfLocations.toList() } Create Mock Flow
21. @Test fun `should get locations`() = runBlockingTest { val flowOfLocations = mockFlow() val locations: List<Location> = flowOfLocations.toList() }
22. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() }
23. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() }
24. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() }
25. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() }
26. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) }
27. @Test fun `should get locations`() = runBlockingTest { val flowOfLocations = mockFlow() val locations: List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) }
28. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) } How does this work?
29. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) val deferred = scope.async { scope.testBody() } deferred.getCompletionExceptionOrNull() ?. let { throw it } scope.cleanupTestCoroutines() if (hasActiveJobs()) { throw UncompletedCoroutinesError("Test finished with active jobs: $endingJobs") } } Our test to run
30. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Setup
31. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Setup Run Test
32. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Setup Run Test Cleanup
33. @Test fun `should get locations`() = runBlockingTest { val flowOfLocations = mockFlow() val locations: List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) }
34. @Test fun `should get locations`() = runBlockingTest { val flowOfLocations = mockFlow() val locations: List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) } Better way to collect from Flows?
35. cashApp/turbine Code ! Issues Pull Requests Turbine Small testing library for kotlinx.coroutines Flow. testImplementation ‘app.cash.turbine:turbine:x.x.x'
36. Flow to test How Turbine Works
37. Channel (Unlimited) How Turbine Works
38. Channel Send Receive
39. Channel Types • Unlimited • Buffered • Rendezvous • Broadcast Channel [Deprecated] 
 

40. How Turbine Works Channel (Unlimited) Event
41. sealed class Event<out T> { 
 object Complete 
 data class Error( ... ) 
 data class Item(val value: T) 
 } How Turbine Works Channel (Unlimited) Event
42. data class Item(val value: T): Event() How Turbine Works Channel (Unlimited)
43. data class Item(val value: T): Event() How Turbine Works Channel (Unlimited)
44. data class Item(val value: T): Event() How Turbine Works Channel (Unlimited)
45. object Complete: Event() How Turbine Works Channel (Unlimited)
46. Exception data class Error( ... ): Event() How Turbine Works Channel (Unlimited)
47. object Complete: Event() How Turbine Works Channel (Unlimited) How do I query items in channels?
48. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } API to query Channel How Turbine Works
49. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } Verify duration of emission How Turbine Works
50. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } Verify no emissions How Turbine Works
51. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } Verify item emitted How Turbine Works
52. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } Verify error occurred How Turbine Works
53. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } Verify all items emitted How Turbine Works
54. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } How Turbine Works
55. How Turbine Works suspend fun <T> Flow<T>.test( ... ) { coroutineScope { } Create a coroutine
56. How Turbine Works suspend fun <T> Flow<T>.test( ... ) { coroutineScope { val events = Channel<Event<T >> (UNLIMITED) } Channel of events
57. How Turbine Works suspend fun <T> Flow<T>.test( ... ) { coroutineScope { val events = Channel<Event<T >> (UNLIMITED) 
 launch { } }
58. How Turbine Works suspend fun <T> Flow<T>.test( ... ) { coroutineScope { val events = Channel<Event<T >> (UNLIMITED) 
 launch { collect { item ->
 events.send(Event.Item(item)) } } Store emissions
59. How Turbine Works suspend fun <T> Flow<T>.test( validate: suspend FlowTurbine<T>.() -> Unit ) { coroutineScope { val events = Channel<Event<T >> (UNLIMITED) 
 launch { try { collect { ... } } catch { events.send(Event.Error) API
60. @Test fun `should get locations`() = runBlockingTest { val locations: List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) } How do we use Turbine?
61. Channel (Unlimited) Testing With Turbine Location Data
62. Channel (Unlimited) Testing With Turbine Location Data
63. Channel (Unlimited) Testing With Turbine Query Channel in test
64. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Read items into Channel
65. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Get first emitted item
66. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Assertion
67. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Get 1st, 2nd, 3rd item
68. Channel (Unlimited) Testing With Turbine Verify Flow completed
69. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Verify flow completed
70. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 }
71. Channel (Unlimited) Testing With Turbine Miss verifying emission?
72. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectComplete() } 
 } Missed varying location 3
73. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectComplete() } 
 } LocationRepoTest.kt Run: LocationRepoTest.kt should get locations app.cash.turbine.AssertionError: Expected complete but found Item(Location(…))
74. Channel (Unlimited) Testing With Turbine Miss a Flow completed
75. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 } 
 } LocationRepoTests.kt Run: LocationRepoTests should get locations app.cash.turbine.AssertionError: Unconsumed events found: 
 - Complete
76. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test { expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 }
77. Unit Testing Channels & Flows • Testing Flows • Errors • Delays • View Model Test • Common Testing Problems 
 

78. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatMap { ... } Exception
79. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatMap { ... } .onCompletion { ... } Complete on error
80. Exception data class Error( ... ): Event() How Turbine Works Channel (Unlimited)
81. How Turbine Works suspend fun <T> Flow<T>.test( ... ) { coroutineScope { val events = Channel<Event<T >> (UNLIMITED) 
 launch { try { collect { ... } } catch { events.send(Event.Error( ... )) } } Send error to Channel
82. @Test fun `should handle exception`() = runBlockingTest { flowOfLocations.test { expectError() } 
 } Throwable
83. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatMap { ... } Better way to 
 handle Exception?
84. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatMap { ... } .catch { . .. } Catch Exception
85. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatMap { ... } .catch { this: FlowCollector } Allow you to emit from catch
86. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatMap { ... } .catch { throwable -> } Exception
87. Code ! Issues Pull Requests Jetbrains/kotlin stdlib/src/kotlin/util/Result.kt inline class Result(val value: Any?) { val isSuccess: Boolean val isFailure: Boolean }
88. flowOfLocations: Flow<Result> .filter { ... } .map { ... } .flatMap { ... } .catch { throwable -> emit(Result.failure(throwable)) } Emit result
89. flowOfLocations .filter { ... } .map { ... } .flatMap { ... } .catch { . .. } Go to Catch when exception occurs
90. flowOfLocations .collect { result - > result.isFailure }
91. Error stored as item data class Item(throwable): Event() How Turbine Works Channel (Unlimited)
92. @Test fun `should handle exception`() = runBlockingTest { flowOfLocations.test { expectItem().isFailure } 
 } Verify Failure
93. Unit Testing Channels & Flows • Testing Flows • Errors • Delays • View Model Test • Common Testing Problems 
 

94. Testing Delays • Delays in suspending methods • Delays in coroutines • Delays in Flows 
 

95. @Test fun `should handle data updates`() = runBlockingTest { processData() } suspend fun processData() { delay(1000) calculate() }
96. @Test fun `should handle data updates`() = runBlockingTest { processData() } suspend fun processData() { delay(1000) calculate() }
97. @Test fun `should handle data updates`() = runBlockingTest { processData() } suspend fun processData() { delay(1000) calculate() } How does this work in a test?
98. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Setup Run Test Cleanup
99. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Run Test Advance virtual time
100. @Test fun `should handle data updates`() = runBlockingTest { processData() } suspend fun processData() { delay(1000) calculate() } Auto advances time forward
101. Testing Delays • Delays in suspending methods • Delays in coroutines • Delays in Flows 
 

102. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Creating a new coroutine
103. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Delay C
104. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Continue
105. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Outside of coroutine
106. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } How is the delay handled?
107. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } SUSPEND
108. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Continue execution
109. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Run Test Advance virtual time
110. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Proceed forward to calculate
111. @Test fun `should handle data updates`() = runBlockingTest { processData() performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } How do I control this delay?
112. interface DelayController { fun advanceTimeBy(delayTimeMillis: Long) fun advanceUntilIdle(): Long fun pauseDispatcher() fun runCurrent() } Delay Utilities
113. @Test fun `should handle data updates`() = runBlockingTest { processData() advanceTimeBy(1000) performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Advance virtual time
114. @Test fun `should handle data updates`() = runBlockingTest { processData() advanceTimeBy(1000) performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Perform calculation
115. @Test fun `should handle data updates`() = runBlockingTest { processData() advanceTimeBy(1000) performAssertions() } suspend fun CoroutineScope.processData() { launch { delay(1000) calculate() } } Perform assertions
116. Testing Delays • Delays in suspending methods • Delays in coroutines • Delays in Flows 
 

117. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } Exception
118. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } Retry?
119. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry { delay(2000) ... } Retry extension
120. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } # of retries
121. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } Delay before retrying
122. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } Retry flow operators
123. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(dispatcher) Flow on Dispatcher Run on Dispatcher
124. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(dispatcher) How do you test this?
125. @Test fun `should retry twice`() = runBlockingTest { }
126. val testCoroutineDispatcher = TestCoroutineDispatcher() 
 @Test fun `should retry twice`() = runBlockingTest { }
127. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(testCoroutineDispatcher) Run on test dispatcher
128. val testCoroutineDispatcher = TestCoroutineDispatcher() 
 @Test fun `should retry twice`() = testCoroutineDispatcher.runBlockingTest { } Use dispatcher in test
129. fun TestCoroutineDispatcher.runBlockingTest( block: suspend TestCoroutineScope.() -> Unit ) = runBlockingTest(this, block) Extension on dispatcher
130. @Test fun `should retry twice`() = testCoroutineDispatcher.runBlockingTest { } flowOfLocations.test { } Start Collecting from Flow
131. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(dispatcher) Call operators
132. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(dispatcher) Exception
133. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(dispatcher) Delay
134. @Test fun `should retry twice`() = testCoroutineDispatcher.runBlockingTest { } flowOfLocations.test { advanceTimeBy(2000) // perform assertions } Advance virtual time
135. flowOfLocations: Flow<Location> .filter { ... } .map { ... } .flatmap { ... } .retry(1) { delay(2000) ... } .flowOn(dispatcher) Retry Flow
136. @Test fun `should retry twice`() = testCoroutineDispatcher.runBlockingTest { } flowOfLocations.test { advanceTimeBy(2000) // perform assertions } Perform assertion
137. Unit Testing Channels & Flows • Testing Flows • Errors • Delays • View Model Test • Common Testing Problems 
 

138. View View Model State
139. class MyViewModel(): ViewModel() { } Jetpack View Model
140. val ViewModel.viewModelScope: CoroutineScope get() { return setTagIfAbsent(JOB_KEY, CloseableCoroutineScope( 
 SupervisorJob() + Dispatchers.Main.immediate) 
 ) } Main Dispatcher
141. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) } State Flow
142. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) } Default value
143. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) fun getData() { 
 viewModelScope.launchWhenStarted { } } }
144. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) fun getData() { 
 viewModelScope.launchWhenStarted { newsRepository.getData() } } }
145. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) fun getData() { 
 viewModelScope.launchWhenStarted { newsRepository.getData() . collect { } } } }
146. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) fun getData() { 
 viewModelScope.launchWhenStarted { newsRepository.getData() . collect { uiState.value = UiState.Success(it) } } } }
147. class MyViewModel(): ViewModel() { val uiState = MutableStateFlow(UIState()) fun getData() { 
 viewModelScope.launchWhenStarted { newsRepository.getData() . collect { uiState.value = UiState.Success(it) } } } } How do we test this flow?
148. @Test fun `should get data`() = runBlockingTest { 
 mockDataFlow() 
 viewModel.states.test { expectItem().data shouldBe state } }
149. @Test fun `should get data`() = runBlockingTest { 
 mockDataFlow() 
 viewModel.getData() viewModel.states.test { expectItem().data shouldBe state } }
150. @Test fun `should get data`() = runBlockingTest { 
 mockDataFlow() 
 viewModel.getData() viewModel.uiState.test { } }
151. @Test fun `should get data`() = runBlockingTest { 
 mockDataFlow() 
 viewModel.getData() viewModel.uiState.test { expectItem() shouldBe UIState.Success( ... ) } }
152. @Test fun `should get data`() = runBlockingTest { 
 mockDataFlow() 
 viewModel.updateState() viewModel.states.test { expectItem().data shouldBe state } } ViewModelTest.kt Run: ViewModelTest should get data Module with the Main dispatcher had failed to initialize. 
 
 For tests Dispatchers.setMain from kotlinx-coroutines-test module can be used
153. val ViewModel.viewModelScope: CoroutineScope get() { return setTagIfAbsent(JOB_KEY, CloseableCoroutineScope( 
 SupervisorJob() + Dispatchers.Main.immediate) 
 ) } Main Dispatcher
154. val dispatcher = TestCoroutineDispatcher() @Before fun setUp() { Dispatchers.setMain(dispatcher) } @After fun tearDown() { Dispatchers.resetMain() } Set Main Dispatcher
155. val dispatcher = TestCoroutineDispatcher() @Before fun setUp() { Dispatchers.setMain(dispatcher) } @After fun tearDown() { Dispatchers.resetMain() } Reset Main Dispatcher
156. class CoroutineTestRule(testDispatcher) : TestWatcher() { 
 override fun starting(description) { Dispatchers.setMain(testDispatcher) } override fun finished(description) { Dispatchers.resetMain() testDispatcher.cleanupTestCoroutines() } }
157. class CoroutineTestRule(testDispatcher) : TestWatcher() { 
 override fun starting(description) { Dispatchers.setMain(testDispatcher) } override fun finished(description) { Dispatchers.resetMain() testDispatcher.cleanupTestCoroutines() } }
158. class CoroutineTestRule(testDispatcher) : TestWatcher() { 
 override fun starting(description) { Dispatchers.setMain(testDispatcher) } override fun finished(description) { Dispatchers.resetMain() testDispatcher.cleanupTestCoroutines() } }
159. @get:Rule val testRule = CoroutineTestRule() @Test fun `should get data`() = runBlockingTest { 
 
 mockDataFlow() 
 viewModel.updateState() viewModel.states.test { expectItem().data shouldBe state } }
160. @get:Rule val testRule = CoroutineTestRule() @Test fun `should get data`() = runBlockingTest { 
 
 mockDataFlow() 
 viewModel.getData() viewModel.uiStates.test { expectItem() shouldBe UIState.Success( ... ) } }
161. @get:Rule val testRule = CoroutineTestRule() @Test fun `should get data`() = runBlockingTest { 
 
 mockDataFlow() 
 viewModel.updateState() viewModel.states.test { expectItem().data shouldBe state } } ViewModelTest.kt Run: ViewModelTest should get data
162. https: // developer.android.com/kotlin/flow/stateflow-and-sharedflow State Flow
163. Unit Testing Channels & Flows • Testing Flows • Errors • Delays • View Model Test • Common Testing Problems 
 

164. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Setup Run Test Cleanup
165. fun runBlockingTest(testBody) { val scope = TestCoroutineScope(safeContext) 
 
 val deferred = scope.async { scope.testBody() } dispatcher.advanceUntilIdle() deferred.getCompletionExceptionOrNull() ?. let { throw it } 
 
 scope.cleanupTestCoroutines() if (activeJobs()).isNotEmpty()) { throw UncompletedCoroutinesError("Test finished with active jobs") } } Cleanup Job is Active
166. Active Job Causes • Coroutine is Suspended • Delays
167. val channel = Channel<Event>() Active Jobs Rendezvous Channel
168. val channel = Channel<Event>() Active Jobs Send Receive
169. val channel = Channel<Event>() Active Jobs fun CoroutineScope.channelSend() { launch { channel.send(Event()) } .. . } Send events
170. val channel = Channel<Event>() Active Jobs suspend fun processEvents() { channel.consumeAsFlow().collect { event -> ... } } Process event
171. Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() } Coroutine
172. Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() } channel.send(LocationEvent()) Coroutine
173. Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() } channel.send(LocationEvent()) Coroutine Rendezvous Channel
174. Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() } channel.send(LocationEvent()) Coroutine SUSPENDED
175. Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() } Tests.kt Run: Tests should process event Test finished with active jobs: ["Coroutine#2":StandaloneCoroutine{Active}]
176. Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() processEvents() }
177. Delay Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() } suspend fun CoroutineScope.sendEvent() { launch(testCoroutineDispatcher) { delay(1000) ... } } Delay
178. @Test fun `should process event`() = runBlockingTest { sendEvent() } suspend fun CoroutineScope.sendEvent() { launch(testCoroutineDispatcher) { delay(1000) ... } } Delay Active Jobs Tests.kt Run: Tests should process event Test finished with active jobs: ["Coroutine#2":StandaloneCoroutine{Active}]
179. Delay Active Jobs @Test fun `should process event`() = runBlockingTest { sendEvent() testCoroutineDispatcher.advanceTimeBy(1000) } suspend fun CoroutineScope.sendEvent() { launch(testCoroutineDispatcher) { delay(1000) ... } } Advance virtual time
180. Testing inside Coroutines Lib • Learn Coroutines from tests inside lib • Multiplatform testing - JS, JVM and Native 
 

181. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) }
182. class JobBasicCancellationTest : TestBase() { @Test fun testCancelJobImpl() { } } Defines utilities
183. class JobBasicCancellationTest : TestBase() { @Test fun testCancelJobImpl() { } } Test parent job is not cancelled
184. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) }
185. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) } Launch coroutine
186. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) } Child coroutine Job
187. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) } Cancel the job
188. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) } Verify order of execution Verify completion
189. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) }
190. expect open class TestBase { fun error(message, cause) fun expect(index: Int) fun finish(index: Int) fun runTest( block: suspend CoroutineScope.() -> Unit ) }
191. JVM JS Native Common TestBase.kt TestBase.kt TestBase.kt TestBase.kt
192. actual fun runTest( block: suspend CoroutineScope.() - > Unit ) { runBlocking( block = block, context = CoroutineExceptionHandler { ... }) 
 }
193. expect open class TestBase { fun error(message, cause) fun expect(index: Int) fun finish(index: Int) fun runTest( block: suspend CoroutineScope.() -> Unit ) } How does this work?
194. private var actionIndex = AtomicInteger() actual fun expect(index: Int) { 
 val wasIndex = actionIndex.incrementAndGet() check(index == wasIndex) { “Expecting action index $index but it is actually $wasIndex" } }
195. private var actionIndex = AtomicInteger() actual fun expect(index: Int) { 
 val wasIndex = actionIndex.incrementAndGet() check(index == wasIndex) { “Expecting action index $index but it is actually $wasIndex" } }
196. private var actionIndex = AtomicInteger() actual fun expect(index: Int) { 
 val wasIndex = actionIndex.incrementAndGet() check(index == wasIndex) { “Expecting action index $index but it is actually $wasIndex" } }
197. expect open class TestBase { fun error(message, cause) fun expect(index: Int) fun finish(index: Int) fun runTest( block: suspend CoroutineScope.() -> Unit ) } How does this work?
198. private var finished = AtomicBoolean() actual fun finish(index: Int) { expect(index) check(!finished.getAndSet(true)) { "Should call 'finish( ... )' at most once" } }
199. private var finished = AtomicBoolean() actual fun finish(index: Int) { expect(index) check(!finished.getAndSet(true)) { "Should call 'finish( ... )' at most once" } }
200. private var finished = AtomicBoolean() actual fun finish(index: Int) { expect(index) check(!finished.getAndSet(true)) { "Should call 'finish( ... )' at most once" } }
201. @Test fun testCancelJobImpl() = runTest { val parent = launch { expect(1) val child = Job(coroutineContext[Job]) expect(2) child.cancel() child.join() expect(3) } parent.join() finish(4) }
202. https: // codingwithmohit.com/coroutines/unit-testing-delays-errors-retries-with-kotlin-flows/ Unit Testing
203. Kotlin Coroutines Training • Groups or Individual Sessions • Contact for booking kotlin.coroutine.training@gmail.com
204. Resources • Unit Testing Delays, Errors & Retries with Kotlin Flows 
 
 https: // codingwithmohit.com/coroutines/unit-testing-delays- 
 errors-retries-with-kotlin-flows/ • Multiplatform Testing pattern inside Coroutine Lib 
 
 https: // codingwithmohit.com/coroutines/kotlin-coroutine-lib-testing/
205. Thank You! @heyitsmohit www.codingwithmohit.com