Kotlin FlowのUnitTestが上手く書けない・意図したテストにならないなと悩んでいたところ、同じ職場の@iwata-nさんにturbineというライブラリを紹介していただきました。（名前がとてもカッチョイイ✨）

導入してみたところ、「今まで悩んでいたのはなんだったのか」と思うほど簡単にテストを書くことができて興奮したので記事を書きました。

この記事で紹介すること

turbineを使ったKotlin FlowのUnitTestの作成方法を紹介していきます。

最初にやること（dependencies）

build.gradleに以下を追加してください。（2022年3月現在）

repositories {
  mavenCentral()
}
dependencies {
  testImplementation 'app.cash.turbine:turbine:0.7.0'
}

なお、開発中のスナップショットはSonatypeのスナップショットリポジトリから入手できます。

repositories {
  maven {
    url 'https://oss.sonatype.org/content/repositories/snapshots/'
  }
}
dependencies {
  testImplementation 'app.cash.turbine:turbine:0.8.0-SNAPSHOT'
}

turbineの基本形

testという検証ブロックを受け取るFlowの拡張関数を使います。
testはFlowが完了するかキャンセルされるまでreturnしないsuspend関数です。

検証ブロックの中でawaitItem()やawaitComplete()などを使い、期待した動作をしているかテストできます。
以下の例ではflowが"apple"、"orange"の順番で流れてきて、最後にflowが完了しているかの確認をしています。

@Test
fun test() {
    runBlocking {
        flowOf("apple", "orange").test {
            assertThat(awaitItem()).isEqualTo("apple")
            assertThat(awaitItem()).isEqualTo("orange")
            awaitComplete()
        }
    }
}

解説

turbineにはEventというsealed classが定義されています。
turbineは流れてくるflowを、このEventに変換して検証ブロックの中に流してくれます。
※コードは以下の実際のライブラリを参照ください。

基本的にflowが流れている間はEvent.Itemが流れてきます。
flowが完了するとEvent.Completeが、エラーが起きるとEvent.Errorが流れてきます。
このEventを検証してテストを構築していくことになるのですが、検証自体はturbineが用意しているsuspend関数で簡単にできます。

では、turbineが用意しているsuspend関数を、それぞれ解説していきます。

awaitItem()

awaitItem()は、次に流れてくるflowがEvent.Itemであることをassertし、
問題なければそれを返すsuspend関数です。

turbineでテストする際、主に使用するsuspend関数となります。

返り値であるEvent.Item、すなわち流れてくるflowを、上記基本形の例で示したようなassertThat(awaitItem()).isEqualTo(hogehohe)という形で検証できます。

「流れてくるflowがEvent.Itemであることをassertし」と書きましたが、ここでEvent.Itemが流れてこなかった場合はAssertionErrorをthrowします。

@Test
fun assertionErrorFailtTest() {
    runBlocking {
        flowOf("apple").test {
            assertThat(awaitItem()).isEqualTo("apple")
            
            // flowが完了してEvent.Completeが流れてくるのでAssertionError!
            assertThat(awaitItem()).isEqualTo("orange")
            awaitComplete()
        }
    }
}

また、時間内に Eventを受信しなかった場合はkotlinx.coroutines.TimeoutCancellationExceptionをthrowします。

ちなみにタイムアウトの時間はtimeoutMsで指定できます。デフォルト値は1秒です。（後述するawaitEvent()、awaitComplete()、awaitError()のタイムアウトもtimeoutMsで指定します。）

@Test
fun timeoutFailTest() {
    runBlocking {
        flowOf("one", "two")
            .map{
                // 3秒遅れてflowを流す
                delay(3000)
                it
            }
             // timeoutを２秒に設定
            .test(timeoutMs = 2_000) {
                awaitItem()
                assertThat(awaitItem()).isEqualTo("two")
                awaitComplete()
            }
    }
}

Timed out waiting for 2000 ms
kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 2000 ms

つまり、awaitItem()を呼び出したタイミングで、タイムアウトでもエラーでもflowの完了でもなく、期待したflowが流れてくることを検証することができます。

awaitComplete()

awaitComplete()は、次に流れてくるflowがEvent.Completeであることをassertし、
問題なければそれを返すsuspend関数です。

先述の通り、flowが完了するとEvent.Completeが流れてくるので、awaitComplete()を呼び出したタイミングでflowが完了した検証することができます。

// 失敗するテスト
@Test
fun awaitCompleteFailTest() {
    runBlocking {
        flowOf("apple", "orange").test {
            assertThat(awaitItem()).isEqualTo("apple")
            awaitComplete() // まだflowが続いていて、Event.Itemであるoragneが流れてきたので失敗する
        }
    }
}

// 成功するテスト
@Test
fun awaitCompleteSuccessTest() {
    runBlocking {
        flowOf("apple", "orange").test {
            assertThat(awaitItem()).isEqualTo("apple")
            assertThat(awaitItem()).isEqualTo("orange")
            awaitComplete() // flowが完了してEvent.Completeが流れてくるので成功する
        }
    }
}

awaitError()

awaitError()は、次に流れてくるflowがEvent.Errorであることをassertし、
問題なければそれを返すsuspend関数です。

先述の通り、エラーが発生するとEvent.Errorが流れてくるので、「狙ったタイミングでflowがエラーをthrowした」という検証することができます。

@Test
fun errorTest() {
    runBlocking {
      flow { throw RuntimeException("error!") }.test {
        assertEquals("error!", awaitError().message)
      }
    }
}

awaitEvent()

awaitEvent()は、次に流れてくるflowがEventであることをassertし、
問題なければそれを返すsuspend関数です。

何かしらflowが流れてくることを検証できますが、あまり使うことはないかもしれません。（公式のUsageにも登場していません。）

こちらを使用したとして、Event.ItemやEvent.Errorを呼び出して検証することになると思うのですが、それならばawaitItem()やawaitError()があるので、awaitEvent()をあえて使用する理由はなさそうです。

ハマりがちなポイント

直感的に操作できるturbineですが、ちょっとハマりがちなポイントがいくつかありました。
筆者が体験した限りですが、最後に挙げていこうと思います。

awaitItem()を呼び出す回数が足りない

以下のようなテストコードはawaitItem()を呼び出す回数が足りていないため、失敗します。

@Test
fun test() {
    runBlocking {
        flowOf("apple", "orange").test {
            assertThat(awaitItem()).isEqualTo("orange") // ここでoragneが来てしまう
            awaitComplete()
        }
    }
}

テスト観点的に２番めの"orange"だけ評価したい場合でも最初に流れてくるのは"apple"なので、その分awaitItem()を呼び出して上げる必要があります。

@Test
fun test() {
    runBlocking {
        flowOf("apple", "orange").test {
            awaitItem() // ここでappleを受け取ってから
            assertThat(awaitItem()).isEqualTo("orange") // oragneを評価
            awaitComplete()
        }
    }
}

ここまでわかりやすい例だと間違えることはあまりないですが、大量のflowが流れてくるケースにおいてループで評価しているときに回数が足りなくてFailするというミスに私はハマりましたTT

runBlockingTestを使うとTimeoutCancellationExceptionで失敗することがある

@Test
fun test() {
    // これだと失敗することがある
    runBlockingTest {
        flowOf("apple", "orange").test {
            awaitItem()
            assertThat(awaitItem()).isEqualTo("orange")
            awaitComplete()
        }
    }
}

suspend関数のテスト時には実行時間短縮のためにrunBlockingTestをよく使うと思うのですが、turbineで使用するとテストが20msで終わっているにも関わらずTimeoutCancellationExceptionで失敗することがあります。

ローカルで成功してはCIで失敗して、なぜだ〜！と悩んでいたんですが、幸いにも解決済みのissueが上がっておりました

Jakeさんによると、runBlockingTestは可能な限り速く時間を回しているので、シミュレート上で経過した1000ms以上のタイムアウトが、実際の時間ではわずか20msで発生してしまっているとのことでした。なので、runBlockingを使ってね！ということのようです。
（今までのテストコードでrunBlockingを使用していたのも実はこの件でハマったからですw）

参考文献（というよりライブラリ本体）

全部英語ですが、READMEにもっと詳しい解説が載っているので、この記事で駆け出しできた方は是非ご一読ください！

Kotlin FlowのUnitTestをturbineでサクッと書く