testing-flows-with-turbine

Turbine でフローをテストする — toList がハングしないようにエミッションを検証する

Turbine は、ホットフロー（StateFlow、SharedFlow）がテストでハングしないように組み込みのキャンセル機能を備え、Flow のエミッションを 1 つずつ検証する標準的な方法です。このスキルでは、flow.test、ReceiveTurbine API、複数のフローをテストするための turbineScope、そして生の flow.toList() が実際に Turbine より適している場合を説明します。Flow 以外のコルーチン関連の処理については、../testing-coroutines-with-runtest/SKILL.md を参照してください。

このスキルを使用する場合

• テスト対象のクラスが StateFlow<UiState> または SharedFlow<Event> を公開しており、各エミッションを検証する必要がある。
• SharedFlow または StateFlow に対する flow.toList() でテストがハングする。どちらも完了しない。
• 「少しの間、追加のエミッションがないこと」や「次のイベントがエラーであること」を検証したいが、toList() では表現できない。
• 2 つのフロー（例：uiState と一度だけ発生する events フロー）を同時に監視し、交互の順序で検証する必要がある。
• ビルドエラーが TurbineAssertionError: Unconsumed events found または Expected an item but found Complete/Error である。
• 開発者が Turbine API の awaitItem、awaitComplete、awaitError、expectMostRecentItem、skipItems、expectNoEvents、withTurbineTimeout、turbineScope、testIn に言及している。

このスキルを使用しない場合

• テスト対象が runTest、TestScope、Dispatchers.setMain、または仮想時間である。まず ../testing-coroutines-with-runtest/SKILL.md を使用してください。Turbine は runTest の上に構築されています。
• フローがコールドで、決定的に完了し、検証内容が「リスト全体が X と等しい」である。flow.toList() のほうが簡潔です。以下の判断マトリックスを参照してください。
• 「フロー」が実際には Compose のスナップショット状態である。Turbine ではなく、テストルールを介して読み取ってください。

前提条件

• Gradle：testImplementation("app.cash.turbine:turbine:1.x")（docs/CORPUS.md §G.1；R5）。
• Turbine はマルチプラットフォーム対応で、JVM、Android、iOS、JS、Native、wasm で利用できます。Apache 2.0、Square / Cash App。
• テスト全体は runTest { … } ブロック（または任意の suspend コンテキスト）内である必要があります。flow.test { } 自体が suspend です。
• 複数フローのテストでは、turbineScope が TestScope.backgroundScope と連携するため、@OptIn(ExperimentalCoroutinesApi::class) が必要です。

flow.test のシグネチャ

public suspend fun <T> Flow<T>.test(
    timeout: Duration = 3.seconds,
    name: String? = null,
    validate: suspend ReceiveTurbine<T>.() -> Unit,
)

（docs/CORPUS.md §G；R5。）次の 3 点を理解しておきましょう。

1. test { } は新しいコルーチンを起動し、flow.collect { } を呼び出して、すべてのエミッションを ReceiveTurbine<T> チャネルに送ります。コレクターは validate { } の実行前にアタッチされるため、最初のアイテムを取りこぼすことなく、ブロック内からホットフローにエミットできます。
2. デフォルトのタイムアウト 3 秒は実時間です。TestCoroutineScheduler によって進むわけではなく、runTest の仮想時間にも従いません。awaitItem() は実時間で最大 3 秒間待機します。
3. validate { } が戻ると、Turbine は未消費のイベントが残っていないことを検証します。アイテム、完了、エラーのいずれかが保留中の場合、TurbineAssertionError がスローされます。その後、収集コルーチンがキャンセルされるため、コールドフローがリークすることはありません。

name 引数はエラーメッセージに追加されます。複数の Turbine を 1 つのテスト内で実行する場合に非常に役立ちます。

ReceiveTurbine API

public interface ReceiveTurbine<T> {
    // suspending
    public suspend fun awaitItem(): T
    public suspend fun awaitComplete()
    public suspend fun awaitError(): Throwable
    public suspend fun awaitEvent(): Event<T>
    public suspend fun skipItems(count: Int)
    public suspend fun cancel()
    public suspend fun cancelAndIgnoreRemainingEvents()
    public suspend fun cancelAndConsumeRemainingEvents(): List<Event<T>>

    // non-suspending
    public fun expectNoEvents()
    public fun expectMostRecentItem(): T
    public fun ensureAllEventsConsumed()
}

（R5）

呼び出し	使用する場面
awaitItem(): T	次のイベントがアイテムである必要があります。Complete／Error の場合は失敗します。
awaitComplete()	次のイベントがフローの完了である必要があります。アイテム／エラーの場合は失敗します。
awaitError(): Throwable	次のイベントがエラーである必要があります。後続のアサーションに使う Throwable を返します。
awaitEvent(): Event<T>	シールドクラスのエントリーです。Event.Item(value)、Event.Complete、Event.Error(throwable) があります。
skipItems(n)	n 個のアイテムイベントを破棄します。次のイベントがアイテム以外の場合は失敗します。StateFlow の初期値をスキップするのに便利です。
expectMostRecentItem()	サスペンドしません。バッファー内のアイテムをすべて取り出し、最新のものを返します。中間値が失われる StateFlow 駆動の UI で「最終状態」をアサートするためのものです。
expectNoEvents()	サスペンドしません。バッファー内に何かあれば例外をスローします。runCurrent()／advanceTimeBy(...) と組み合わせて、不要なイベントが発生しないことをアサートします。
cancel()	コレクターを停止します。残りのイベントがあれば、スコープ終了時にアサーションが失敗します。
cancelAndIgnoreRemainingEvents()	停止して、残りを通知なしで破棄します。ホットフローのテストの末尾では、これをデフォルトとして使います。
cancelAndConsumeRemainingEvents()	任意に調べるための List<Event<T>> を返します。

ホットフローとコールドフローの契約

フローの種類	自然に完了するか	最初の awaitItem() が返すもの	cancel() が必要か
flowOf(...)／flow { }（コールド）	はい	最初のエミッション	いいえ — awaitComplete() で終了します
MutableSharedFlow<T>（replay = 0）	いいえ	次のエミッション（またはハングします）	常に必要
MutableStateFlow<T>	いいえ	初期値（replay = 1）	常に必要

（docs/CORPUS.md §G；R5）

重大な注意：SharedFlow と StateFlow は onCompletion を呼び出しません。どちらかに対する awaitComplete() は、実時間での 3 秒のタイムアウトまでハングします。必ず cancel()（または cancelAndIgnoreRemainingEvents()）で終了してください。

StateFlow.test は StateFlow の replay = 1 により、必ず最初に初期値をエミットします。初期値が不要な場合は、skipItems(1) でスキップしてください。

turbineScope と testIn — 1 つのテストで複数のフローを扱う

public suspend fun <R> turbineScope(
    timeout: Duration? = null,
    validate: suspend TurbineTestContext.() -> R,
): R

public fun <T> Flow<T>.testIn(
    scope: CoroutineScope,
    timeout: Duration? = null,
    name: String? = null,
): ReceiveTurbine<T>

2 つのフロー間でアサーションを交互に行う場合に使います。

runTest(mainRule.dispatcher) {
    turbineScope {
        val items = repo.items.testIn(backgroundScope, name = "items")
        val errors = repo.errors.testIn(backgroundScope, name = "errors")
        assertEquals(0, items.awaitItem())
        assertEquals(emptyList<Throwable>(), errors.expectMostRecentItem())
        items.cancel()
        errors.cancel()
    }
}

turbineScope が終了する前に、testIn に対して明示的に cancel()（または awaitComplete()／awaitError()）を呼び出す必要があります。呼び出さないと、スコープのアサーションが例外をスローします。

単独の Turbine<T> — コールバックと単発シグナルの捕捉

Turbine は Flow アダプターだけではありません。単独の Turbine<T> は、ReceiveTurbine と同じアサーション API を持つ送受信チャネルです。テスト対象が Flow ではない場合に使います。たとえば、(T) -> Unit コールバック、リスナーインターフェース、自分で管理する Channel、呼び出しを記録する必要がある手作りのフェイクなどです。

public fun <T> Turbine(timeout: Duration? = null, name: String? = null): Turbine<T>

public interface Turbine<T> : ReceiveTurbine<T> {
    public val isClosed: Boolean
    public fun add(item: T)                       // producer side — push an item
    public fun close(cause: Throwable? = null)    // signal Complete (or Error if cause != null)
    public fun asChannel(): Channel<T>            // hand the producer a plain SendChannel
}

（docs/CORPUS.md §G；R5）これにより、「このコールバックは X を受け取って呼び出されたか、それだけか」という問いを、同じ awaitItem() の語彙で扱えます。

@Test fun emitsClickedId() = runTest(mainRule.dispatcher) {
    val clicks = Turbine<Long>()
    val adapter = ItemAdapter(onClick = clicks::add)        // production code calls onClick(id)

    adapter.bind(item)
    adapter.simulateClick()

    assertEquals(42L, clicks.awaitItem())
    clicks.ensureAllEventsConsumed()                        // no extra invocations slipped through
}

ensureAllEventsConsumed() はバッファーに何も残っていないことをアサートします。これは、flow.test { } がブロックの終了時に暗黙的に実行するチェックに相当します。単独の Turbine は自動的に完了しないため、close() しない限り、awaitComplete() はタイムアウトまでハングします。ホットフローと同じように扱ってください。

同期（非 suspend）での消費：takeItem／takeComplete／takeError

アサーションを行う箇所でサスペンドできない場合 — 生成側のコルーチンがすでに実行された後の通常の @Test 本文や、非 suspend コールバック内の assert など — ReceiveTurbine はサスペンドしない取得メソッドを提供します。これらはイベントがすでにバッファー内にあることを前提とし、なければ例外をスローします。

public fun takeItem(): T
public fun takeComplete()
public fun takeError(): Throwable

@Test fun synchronousDrain() = runTest(mainRule.dispatcher) {
    val turbine = vm.uiState.testIn(backgroundScope)
    advanceUntilIdle()                              // let the producer run to quiescence (virtual time)
    assertEquals(Idle, turbine.takeItem())          // no suspension — the value is already buffered
    assertEquals(Loaded, turbine.takeItem())
    turbine.cancel()
}

runTest／flow.test ブロック内では、サスペンドする awaitItem() を優先してください。呼び出し箇所で本当にサスペンドできない場合に限り、takeItem() を使います。

Turbine のタイムアウトと runTest の仮想時間

Turbine のデフォルトの 3 秒は実時間であり、仮想時間ではありません。runTest のデフォルトの 60 秒も実時間ですが、テスト本体内の advanceTimeBy(...) は TestCoroutineScheduler の仮想時間を操作します。Turbine はこれを認識しません。

そのため、delay(10.minutes) を使うテストは、runTest が遅延をスキップするため、マイクロ秒で完了します。しかし、実際のネットワーク I/O を行うソースからのイベントを待つ flow.test { awaitItem() } では、実際の経過時間が測定されます。実時間で本当に 3 秒を超えて待つ場合は、次のように設定を上書きしてください。

flow.test(timeout = 10.seconds) { … }            // per-call

withTurbineTimeout(10.seconds) {                 // scoped
    flow.test { … }
}

判断基準 — Turbine と生の flow.toList()

シナリオ	選択
完了するコールドフロー。「リスト全体が X と等しい」	flow.toList()（より簡潔）
ホットな StateFlow / SharedFlow	Turbine（キャンセルを処理）
中間状態を検証してから、さらにイベントを発生させる必要がある	Turbine
「しばらくの間、これ以上イベントが発生しない」ことを検証する必要がある	Turbine（expectNoEvents）
エラーの型または完了を検証する必要がある	Turbine（awaitError、awaitComplete）
1 つのテスト内で複数のフローが交互に発生する	Turbine（turbineScope/testIn）
JS/Native でのマルチプラットフォームテスト	どちらも動作しますが、Turbine のほうが簡単です
追加の依存関係をゼロにしたい	flow.toList()

（R5。）

経験則として、UI 状態や ViewModel のテスト（ほとんどの場合、StateFlow を使用）には Turbine を、決定的に完了するコールドフローのリポジトリ／変換テストには flow.toList() を使用してください。

パターン

パターン：誤り — SharedFlow に対する flow.toList()

// WRONG
@Test fun events() = runTest {
    val events: SharedFlow<Event> = vm.events
    val collected = events.toList()                 // never returns
    assertEquals(listOf(LoggedIn), collected)
}
// WRONG because: SharedFlow never completes. toList() suspends until the upstream
// completes, which it never does. The test hangs to the 60s runTest timeout.

// RIGHT
@Test fun events() = runTest(mainRule.dispatcher) {
    vm.events.test {
        vm.login()
        assertEquals(LoggedIn, awaitItem())
        cancelAndIgnoreRemainingEvents()             // hot flow — must cancel
    }
}

パターン：誤り — StateFlow の初期値を忘れる

// WRONG
@Test fun stateFlowEmits() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        vm.refresh()
        assertEquals(Loading, awaitItem())           // FAIL: actually Idle (seed)
        cancel()
    }
}
// WRONG because: StateFlow has replay = 1. The collector immediately receives the
// current value (Idle) before any new emission. The first awaitItem() returns
// the seed, not the post-refresh state.

// RIGHT — assert the seed
@Test fun stateFlowEmits() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        assertEquals(Idle, awaitItem())              // seed
        vm.refresh()
        assertEquals(Loading, awaitItem())
        assertEquals(Success(items), awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

// RIGHT — skip the seed if you do not care
@Test fun stateFlowEmits_skipSeed() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        skipItems(1)
        vm.refresh()
        assertEquals(Loading, awaitItem())
        assertEquals(Success(items), awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

パターン：誤り — ホットフローに対する awaitComplete()

// WRONG
@Test fun events() = runTest {
    vm.events.test {
        vm.login()
        assertEquals(LoggedIn, awaitItem())
        awaitComplete()                              // hangs 3s, then TurbineAssertionError
    }
}
// WRONG because: SharedFlow never completes. awaitComplete() waits 3 wall-clock
// seconds (Turbine default) and then throws. The test is slow AND fails.

// RIGHT
@Test fun events() = runTest(mainRule.dispatcher) {
    vm.events.test {
        vm.login()
        assertEquals(LoggedIn, awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

パターン：StateFlow の合流 — expectMostRecentItem を使用する

@Test fun finalState() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        skipItems(1)                                 // seed
        vm.flipFast()                                // posts 5 transitions in <1ms
        advanceUntilIdle()                           // virtual; let coroutines settle
        // StateFlow conflates — only the latest value is kept.
        // awaitItem() may see anything from the burst; expectMostRecentItem is deterministic.
        assertEquals(Final, expectMostRecentItem())
        cancelAndIgnoreRemainingEvents()
    }
}

パターン：「これ以上イベントがない」ことを検証する

@Test fun noSpuriousEmissions() = runTest(mainRule.dispatcher) {
    vm.uiState.test {
        skipItems(1)
        vm.refresh()
        assertEquals(Loading, awaitItem())
        assertEquals(Success(items), awaitItem())
        advanceTimeBy(5.seconds)                     // virtual — flush any timers
        runCurrent()
        expectNoEvents()                             // confirms quiescence
        cancelAndIgnoreRemainingEvents()
    }
}

パターン：完了するコールドフロー — Turbine と toList の比較

// FINE — toList() is shorter for terminal cold flows
@Test fun coldList() = runTest {
    val collected = flowOf(1, 2, 3).toList()
    assertEquals(listOf(1, 2, 3), collected)
}

// ALSO FINE — Turbine if you need intermediate-state assertions
@Test fun coldTurbine() = runTest {
    flowOf(1, 2, 3).test {
        assertEquals(1, awaitItem())
        assertEquals(2, awaitItem())
        assertEquals(3, awaitItem())
        awaitComplete()                              // cold flow — completes naturally
    }
}

パターン：turbineScope で 2 つのフローを交互に扱う

@Test fun interleaved() = runTest(mainRule.dispatcher) {
    turbineScope {
        val state = vm.uiState.testIn(backgroundScope, name = "state")
        val events = vm.events.testIn(backgroundScope, name = "events")
        assertEquals(Idle, state.awaitItem())
        vm.login()
        assertEquals(Loading, state.awaitItem())
        assertEquals(LoggedIn, events.awaitItem())
        assertEquals(Success, state.awaitItem())
        state.cancel(); events.cancel()
    }
}

必須ルール

• 必須: StateFlow / SharedFlow のアサーションは、必ず flow.test { … cancelAndIgnoreRemainingEvents() } でラップしてください。ホットフローに対して flow.toList() を呼び出してはいけません。テストがハングします。
• 必須: ホットフローに対するすべての test { } ブロックの末尾で、必ず cancel() / cancelAndIgnoreRemainingEvents() を呼び出してください。ホットフローは完了しないため、awaitComplete() は不適切です。
• 必須: StateFlow の初期値を考慮してください。最初の awaitItem() は現在の値を返します（リプレイ = 1）。これをアサートするか、skipItems(1) を呼び出してください。
• 必須: Turbine は testImplementation のみに依存させてください。app.cash.turbine を implementation に含めてはいけません。
• 必須: 待機対象のエミッションに実際に 3 秒を超える時間がかかる場合は、withTurbineTimeout(...)（または呼び出しごとの timeout =）をオーバーライドしてください。デフォルトの 3 秒は実時間であり、runTest の仮想時間とは独立しています。
• 必須: testIn ブロック内では、turbineScope が終了する前に cancel()（awaitComplete() ではなく）を使用してください。そうしないと、スコープのアサーションが例外をスローします。
• 必須: Turbine のタイムアウトと runTest の仮想クロックのセマンティクスを混在させてはいけません。advanceTimeBy(10.minutes) は仮想時間ですが、Turbine の 3 秒の待機は実時間です。
• 推奨: 中間値がまとめられる可能性のある、畳み込み StateFlow の決定論的な「最終状態」のアサーションには、expectMostRecentItem() を使用してください。
• 推奨: 決定論的に完了するコールドフローでは、Turbine を使うとオーバーヘッドになるため、flow.toList() をそのまま使用してください。
• 推奨: 1 つのテストで複数の Turbine を実行する場合は、Turbine に名前を付けてください（flow.test(name = "uiState")、flow.testIn(scope, name = "events")）。エラーメッセージに名前が表示されます。
• 推奨: (T) -> Unit コールバックやリスナーの呼び出しを検証する際は、手動で値を蓄積する MutableList よりも、単独の Turbine<T>()（add / close / ensureAllEventsConsumed を使用）を優先してください。同じ awaitItem() の診断機能と、未消費イベントのチェックが得られます。
• 推奨: runTest / flow.test ブロック内では、非サスペンド関数の takeItem() / takeComplete() / takeError() よりも、サスペンド関数の awaitItem() を使用してください。呼び出し箇所で実際にサスペンドできない場合に限り、取り出し関数を使用してください。

検証

• いずれの StateFlow または SharedFlow に対しても flow.toList() を呼び出していないこと（grep -rn 'toList()' src/test の結果をフローの型と照合してください）。
• ホットフローに対するすべての flow.test { … } が、cancel() または cancelAndIgnoreRemainingEvents() で終了していること。
• StateFlow / SharedFlow に対して awaitComplete() を呼び出していないこと。
• StateFlow のテストで、awaitItem() を使って初期値をアサートするか、後続のアサーションの前に skipItems(1) を呼び出していること。
• 3 秒を超えて待機する可能性がある Turbine では、withTurbineTimeout(...) または呼び出しごとの timeout = N.seconds を使用していること。
• 複数フローのテストで、ネストした test { } ブロックではなく、turbineScope { … testIn(backgroundScope) … } を使用していること。
• コールバック / リスナーのアサーションで、場当たり的な MutableList ではなく、単独の Turbine<T>()（ensureAllEventsConsumed() を使用）を使っていること。
• CI の実行で 50 回の反復が完了し、TurbineAssertionError による不安定な失敗が発生しないこと。

参考資料

• Cash App Turbine — README と API：https://github.com/cashapp/turbine
• Cash App Turbine — リリース：https://github.com/cashapp/turbine/releases
• Android Developers — Android で Kotlin Flow をテストする：https://developer.android.com/kotlin/flow/test
• Android Developers — StateFlow と SharedFlow：https://developer.android.com/kotlin/flow/stateflow-and-sharedflow
• JetBrains — 非同期 Flow：https://kotlinlang.org/docs/flow.html
• 調査：tasks/research/R5-coroutines-test-turbine.md — Turbine API の概要、ホット／コールドの契約、toList() との比較。
• docs/CORPUS.md §G.1 — Gradle の座標とバージョンの固定。
• 関連項目：../testing-coroutines-with-runtest/SKILL.md — Turbine が前提とする runTest、MainDispatcherRule、仮想時間の基礎。
• 関連項目：../../mocking/mocking-with-mockito/SKILL.md — Flow を生成する suspend 依存関係のモック。
• 関連項目：../../mocking/mocking-with-mockk/SKILL.md — coEvery { repo.items() } returns flowOf(...)。
• 関連項目：../../runner/configuring-junit4-on-android/SKILL.md — JUnit4 の設定。
• 関連項目：../../robolectric/using-robolectric-correctly/SKILL.md — SUT が JVM 上で Android フレームワークを必要とする場合。
• 横断項目：../../../fundamentals/strategies/applying-testing-strategies/SKILL.md — Flow テストの小・中・大の規模設定。
• 横断項目：../../../fundamentals/strategies/organizing-test-source-sets/SKILL.md — Flow テストの配置場所（src/test/）。
• 横断項目：../../../compose/synchronization/controlling-the-test-clock/SKILL.md — Compose の MainTestClock は Turbine の実時間クロックとは独立しています。
• 横断項目：../../../instrumentation/runner/running-instrumented-tests-with-androidjunit4/SKILL.md — Turbine は同じ API で実機上でも動作します。
• 横断項目：../../../kotlin/kotlin-test/writing-tests-with-kotlin-test/SKILL.md — awaitItem() と kotlin.test の assertEquals／assertIs の組み合わせ、および flow.test { } 内で使うアサーションの語彙。