Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Unit Testing Kotlin Flows (Chicago Roboto 2021)

Mohit S
September 27, 2021

Unit Testing Kotlin Flows (Chicago Roboto 2021)

Chicago Roboto 2021

Mohit S

September 27, 2021
Tweet

More Decks by Mohit S

Other Decks in Programming

Transcript

  1. flowOfData: Flow<Data> .filter { ... } .map { ... }

    .flatmap { ... } How do we test this flow?
  2. @Test fun `should get data`() = runBlockingTest { } Test

    Scope + Test Dispatcher What is executed in this coroutine?
  3. @Test fun `should get data`() = runBlockingTest { val flowOfData

    = mockFlow() val locations: List<Location> = flowOfLocations.toList() } Create Mock Flow
  4. @Test fun `should get data`() = runBlockingTest { val flowOfData

    = mockFlow() val items: List<Data> = flowOfData.toList() }
  5. @Test fun `should get locations`() = runBlockingTest { val items:

    List<Data> = flowOfData.toList() items shouldContainAll listOf(item1, item2, item3) }
  6. @Test fun `should get data`() = runBlockingTest { val flowOfData

    = mockFlow() val items: List<Data> = flowOfData.toList() items shouldContainAll listOf(item1, item2, item3) }
  7. @Test fun `should get data`() = runBlockingTest { val locations:

    List<Location> = flowOfLocations.toList() locations shouldContainAll listOf(location1, location2, location3) } How does this work?
  8. 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
  9. 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
  10. 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
  11. 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
  12. @Test fun `should get data`() = runBlockingTest { val flowOfData

    = mockFlow() val items: List<Data> = flowOfData.toList() items shouldContainAll listOf(items1, items2, items3) }
  13. Run Blocking Test Different Approaches 1. Use run blocking test

    for all tests 2. Testing Delays -> Use run blocking test 3. Non Delay Uses Cases -> Use run blocking
  14. @Test fun `should get data`() = runBlockingTest { val flowOfLocations

    = mockFlow() val items: List<Data> = flowOfData.toList() locations shouldContainAll listOf(location1, location2, location3) } Better way to collect from Flows?
  15. cashApp/turbine Code ! Issues Pull Requests Turbine Small testing library

    for kotlinx.coroutines Flow. testImplementation ‘app.cash.turbine:turbine:x.x.x'
  16. What Turbine Provides? suspend fun <T> Flow<T>.test( ... ) coroutineScope

    { val events = Channel<Event<T >> (UNLIMITED) 
 launch { try { collect { ... } } catch { events.send(Event.Error) }
  17. interface FlowTurbine { suspend fun expectItem(): T fun expectNoEvents() fun

    expectError(): Throwable suspend fun expectComplete() } What Turbine Provides?
  18. interface FlowTurbine { suspend fun expectItem(): T fun expectNoEvents() fun

    expectError(): Throwable suspend fun expectComplete() } What Turbine Provides?
  19. interface FlowTurbine { suspend fun expectItem(): T fun expectNoEvents() fun

    expectError(): Throwable suspend fun expectComplete() } What Turbine Provides?
  20. interface FlowTurbine { suspend fun expectItem(): T fun expectNoEvents() fun

    expectError(): Throwable suspend fun expectComplete() } What Turbine Provides?
  21. @Test fun `should get data`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 }
  22. @Test fun `should get data`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Get first emitted item
  23. @Test fun `should get data`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 } Assertion
  24. @Test fun `should get data`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo item2 expectItem() shouldBeEqualTo item3 expectComplete() } 
 } Get 1st, 2nd, 3rd item
  25. @Test fun `should get data`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo item2 } 
 } Missed collecting item
  26. @Test fun `should get data`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo item2 } 
 } RepoTest.kt Run: app.cash.turbine.AssertionError: Expected complete but found Item(Data(…)) DataRepoTests should get data
  27. @Test fun `should get locations`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo item2 expectItem() shouldBeEqualTo item3 expectComplete() } 
 }
  28. @Test fun `should get locations`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo item2 expectItem() shouldBeEqualTo item3 } 
 } RepoTests.kt Run: DataRepoTests should get data app.cash.turbine.AssertionError: Unconsumed events found: 
 - Complete
  29. @Test fun `should get locations`() = runBlockingTest { flowOfData.test {

    expectItem() shouldBeEqualTo item1 expectItem() shouldBeEqualTo item2 expectItem() shouldBeEqualTo item3 expectComplete() } 
 }
  30. sealed class Event<out T> { 
 object Complete 
 data

    class Error( ... ) 
 data class Item(val value: T) 
 } How Turbine Works Channel (Unlimited) Event
  31. 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
  32. interface FlowTurbine<T> { val timeout: Duration fun expectNoEvents() suspend fun

    expectItem(): T fun expectError(): Throwable suspend fun expectComplete() } How Turbine Works
  33. How Turbine Works suspend fun <T> Flow<T>.test( ... ) {

    coroutineScope { } Create a coroutine
  34. How Turbine Works suspend fun <T> Flow<T>.test( ... ) {

    coroutineScope { val events = Channel<Event<T >> (UNLIMITED) } Channel of events
  35. How Turbine Works suspend fun <T> Flow<T>.test( ... ) {

    coroutineScope { val events = Channel<Event<T >> (UNLIMITED) 
 launch { } }
  36. 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
  37. 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
  38. @Test fun `should get locations`() = runBlockingTest { flowOfLocations.test {

    expectItem() shouldBeEqualTo location1 expectItem() shouldBeEqualTo location2 expectItem() shouldBeEqualTo location3 expectComplete() } 
 }
  39. State Flow sealed class UiState { 
 data class Error(

    
 val exception: Throwable 
 ): UiState() }
  40. State Flow sealed class UiState { 
 data class Success(

    val data: Data ): UiState() data class Error( 
 val exception: Throwable 
 ): UiState() }
  41. State Flow sealed class UiState { 
 data class Success(

    val data: Data ): UiState() data class Error( 
 val exception: Throwable 
 ): UiState() }
  42. State Flow val stateFlow = MutableStateFlow(UiState.Success(Data())) @Test fun `should emit

    default value`() = runBlockingTest { stateFlow.test { expectItem() shouldBe UIState.Success } }
  43. State Flow @Test fun `should emit default value`() = runBlockingTest

    { stateFlow.test { expectItem() shouldBe UIState.Success 
 expectComplete() } }
  44. State Flow @Test fun `should emit default value`() = runBlockingTest

    { stateFlow.test { expectItem() shouldBe UIState.Success 
 expectComplete() } } ! Timed out waiting for 1000 ms kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 1000 ms
  45. State Flow suspend fun <T> Flow<T>.test( ... ) { coroutineScope

    { val events = Channel<Event<T >> (UNLIMITED) 
 launch { collect { item ->
 events.send(Event.Item(item)) } events.send(Event.Complete) } }
  46. @Test fun `should emit default value`() = runBlockingTest { stateFlow

    .onCompletion { println("ON COMPLETE") } .test { expectItem() shouldBe UIState.Success } } State Flow
  47. @Test fun `should emit default value`() = runBlockingTest { stateFlow

    .onCompletion { println("ON COMPLETE") } .test { expectItem() shouldBe UIState.Success } } State Flow Will log
  48. How Turbine Works suspend fun <T> Flow<T>.test( ... ) {

    coroutineScope { 
 val collectJob = launch { collect { item -> ... } ... } ... collectJob.cancel() }
  49. @Test fun `should emit default value`() = runBlockingTest { stateFlow

    .onCompletion { println("ON COMPLETE") } .test { expectItem() shouldBe UIState.Success } } State Flow Will log
  50. val stateFlow = MutableStateFlow<UIState>(UIState.Success) @Test fun `should emit default value`()

    = runBlockingTest { stateFlow.emit(UIState.Error) stateFlow.test { expectItem() shouldBe UIState.Success expectItem() shouldBe UIState.Error } } State Flow
  51. val stateFlow = MutableStateFlow<UIState>(UIState.Success) @Test fun `should emit default value`()

    = runBlockingTest { stateFlow.emit(UIState.Error) stateFlow.test { expectItem() shouldBe UIState.Success expectItem() shouldBe UIState.Error } } State Flow
  52. val stateFlow = MutableStateFlow<UIState>(UIState.Success) @Test fun `should emit default value`()

    = runBlockingTest { stateFlow.emit(UIState.Error) stateFlow.test { expectItem() shouldBe UIState.Success expectItem() shouldBe UIState.Error } } State Flow Fail
  53. val stateFlow = MutableStateFlow<UIState>(UIState.Success) @Test fun `should emit default value`()

    = runBlockingTest { stateFlow.emit(UIState.Error) stateFlow.test { expectItem() shouldBe UIState.Success expectItem() shouldBe UIState.Error } } State Flow Most Recent Emission
  54. val stateFlow = MutableStateFlow<UIState>(UIState.Success) @Test fun `should emit default value`()

    = runBlockingTest { stateFlow.emit(UIState.Error) stateFlow.test { expectItem() shouldBe UIState.Error } } State Flow Success
  55. Mocking Flows class FakeRepository: Repository { private val channel =

    Channel<Result>() 
 override fun getData(): Flow<Result> { return channel.consumeAsFlow() } }
  56. Mocking Flows class FakeRepository: Repository { private val channel =

    Channel<Result>() 
 suspend fun emitResult(result: Result) { channel.send(result) } }
  57. Mocking Flows class FakeRepository: Repository { private val channel =

    Channel<Result>() 
 suspend fun emitFailure(result: Result.Failure) { channel.send(result) } }
  58. Mocking Flows class FakeRepository: Repository { private val channel =

    Channel<Result>() 
 fun closeChannel() = apply { channel.close() } }
  59. val stateFlow = MutableStateFlow<UIState>(UIState.Success) 
 flow1.combine(flow2) { a, b ->

    combineItems(a, b) }.collect { stateFlow.emit(it) } State Flow
  60. @Test fun `should combine flows`() = runBlockingTest { 
 val

    mockFlow1 = mockFlow() 
 val mockFlow2 = mockFlow() } State Flow
  61. @Test fun `should combine flows`() = runBlockingTest { 
 val

    mockFlow1 = mockFlow() 
 val mockFlow2 = mockFlow() 
 val stateHolder = StateHolder(mockFlow1, mockFlow2) } State Flow
  62. @Test fun `should combine flows`() = runBlockingTest { 
 


    val stateHolder = StateHolder(mockFlow1, mockFlow2) 
 stateFlow.test { 
 ... 
 } } State Flow
  63. State Flow Challenges • Why didn’t the flow emit an

    item? Conflated? • Too many flow combines makes it harder to follow the 
 
 data flow
  64. Shared Flow val flow = MutableSharedFlow<String>() @Test fun `collect from

    shared flow`() = runBlockingTest { flow.emit("Event 1") }
  65. Shared Flow val flow = MutableSharedFlow<String>() @Test fun `collect from

    shared flow`() = runBlockingTest { flow.emit("Event 1") flow.test { expectItem() shouldBeEqualTo "Event 1" } } Failed
  66. Shared Flow val flow = MutableSharedFlow<String>() @Test fun `collect from

    shared flow`() = runBlockingTest { flow.emit("Event 1") flow.test { expectItem() shouldBeEqualTo "Event 1" } } Subscriber count 0
  67. Shared Flow val flow = MutableSharedFlow<String>() @Test fun `collect from

    shared flow`() = runBlockingTest { flow.emit("Event 1") flow.test { expectItem() shouldBeEqualTo "Event 1" } } Flow doesn’t replay
  68. Shared Flow @Test fun `collect from shared flow`() = runBlockingTest

    { 
 val job = launch(start = CoroutineStart.LAZY) { flow.emit("Event 1") } }
  69. Shared Flow @Test fun `collect from shared flow`() = runBlockingTest

    { 
 val job = launch(start = CoroutineStart.LAZY) { flow.emit("Event 1") } }
  70. Shared Flow @Test fun `collect from shared flow`() = runBlockingTest

    { 
 val job = launch(start = CoroutineStart.LAZY) { flow.emit("Event 1") } flow.test { job.start() expectItem() shouldBeEqualTo "Event 1" } } Subscriber count 1
  71. Shared Flow @Test fun `collect from shared flow`() = runBlockingTest

    { 
 val job = launch(start = CoroutineStart.LAZY) { flow.emit("Event 1") } flow.test { job.start() expectItem() shouldBeEqualTo "Event 1" } } Success
  72. @Test fun `collect from shared flow`() = runBlockingTest { flow.emit("Event

    1") flow.test { expectItem() shouldBeEqualTo "Event 1" } } Shared Flow val flow = MutableSharedFlow<String>(replay = 1) Replay
  73. Mocking Flows class FakeRepository: Repository { val flow = MutableSharedFlow<Result>(

    
 replay = 1, 
 BufferOverFlow.DROP_OLDEST 
 ) 
 }
  74. Mocking Flows class FakeRepository: Repository { val flow = MutableSharedFlow<Result>(

    
 replay = 1, 
 BufferOverFlow.DROP_OLDEST 
 ) 
 suspend fun emitResult(result: Result) { flow.emit(result) } }
  75. While Subscribed @Test fun `collect with while subscribed strategy`() =

    runBlockingTest { 
 val sharingScope = TestCoroutineScope() val sharedFlow = flow .onStart { println("ON START") } .shareIn( sharingScope, SharingStarted.WhileSubscribed(), 1 ) }
  76. While Subscribed @Test fun `collect with while subscribed strategy`() =

    runBlockingTest { 
 val sharingScope = TestCoroutineScope() val sharedFlow = flow .shareIn( sharingScope, SharingStarted.WhileSubscribed(), 1 ) }
  77. While Subscribed @Test fun `collect with while subscribed strategy`() =

    runBlockingTest { 
 val sharingScope = TestCoroutineScope() val sharedFlow = flow .onStart { println("ON START") } .shareIn( sharingScope, SharingStarted.WhileSubscribed(), 1 ) } Will not log
  78. While Subscribed @Test fun `collect with while subscribed strategy`() =

    runBlockingTest { 
 val sharedFlow = flow .onStart { println("ON START") } .shareIn( ... ) sharedFlow.test { ... } }
  79. While Subscribed @Test fun `collect with while subscribed strategy`() =

    runBlockingTest { 
 val sharedFlow = flow .onStart { println("ON START") } .shareIn( ... ) sharedFlow.test { ... } }
  80. Eagerly @Test fun `collect with eagerly strategy`() = runBlockingTest {

    
 val sharedFlow = flow .onStart { println("ON START") } .shareIn( sharingScope, SharingStarted.Eagerly, 1 ) }
  81. Eagerly @Test fun `collect with eagerly strategy`() = runBlockingTest {

    
 val sharedFlow = flow .onStart { println("ON START") } .shareIn( sharingScope, SharingStarted.Eagerly, 1 ) } Will start
  82. Lazily @Test fun `collect with lazily strategy`() = runBlockingTest {

    
 val sharedFlow = flow .onStart { println("ON START") } .shareIn( sharingScope, SharingStarted.Lazily, 1 ) } Will start
  83. Lazily @Test fun `collect with lazily strategy`() = runBlockingTest {

    
 val sharedFlow = flow .onComplete { println("ON Complete”) } .shareIn( sharingScope, SharingStarted.Lazily, 1 ) }