Flow memancarkan banyak nilai sepanjang waktu — bukan satu. Test tradisional yang memakai toList() mengumpulkan seluruh emissions sebelum membandingkan, tetapi kehilangan informasi urutan dan waktu antar emission. Library Turbine dari Cash App mengatasi ini dengan receiver flow.test { } yang menyediakan awaitItem(), awaitComplete(), dan awaitError() untuk memeriksa setiap peristiwa secara berurutan. Chapter ini memperkenalkan Turbine dan pola manual alternatif yang memakai operator Flow standar.

Mengapa Flow Butuh Pola Khusus #

Tiga karakteristik Flow yang menyulitkan test tradisional:

KarakteristikImplikasi
ColdFlow baru berjalan saat dikoleksi
AsyncEmission terjadi pada waktu berbeda
Dapat gagalPengecualian dapat muncul di tengah aliran

Test yang memakai toList() dapat memverifikasi nilai akhir tetapi tidak dapat memeriksa bahwa error muncul setelah emission ke-3, atau bahwa tidak ada emission setelah complete. Turbine menyediakan API berurutan untuk kasus tersebut.

Turbine pada Produksi #

Bentuk asli test Turbine:

kotlin Contoh: flow.test dengan Turbine 
import app.cash.turbine.test
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.test.runTest
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Assertions.assertEquals

class FlowTurbineTest {
    @Test
    fun `tiga emission lalu complete`() = runTest {
        flowOf(28, 29, 30).test {
            assertEquals(28, awaitItem())
            assertEquals(29, awaitItem())
            assertEquals(30, awaitItem())
            awaitComplete()
        }
    }

    @Test
    fun `error setelah tiga emission`() = runTest {
        flow<Int> {
            emit(0); emit(1); emit(2)
            throw IllegalStateException("aliran gagal pada emit ke-3")
        }.test {
            assertEquals(0, awaitItem())
            assertEquals(1, awaitItem())
            assertEquals(2, awaitItem())
            val err = awaitError()
            assertEquals("aliran gagal pada emit ke-3", err.message)
        }
    }
}

Karena Turbine tidak tersedia pada environment kompilasi ini, berikut versi runnable yang meniru pola flow.test { } dengan koleksi manual dan receiver yang menampung hasil:

kotlin 
// flow_testing.kt
// Mendemonstrasikan pengujian Kotlin Flow secara manual (tanpa Turbine)
// dan pola yang setara dengan Turbine flow.test { }.
// File ini memakai kotlinx.coroutines.flow sehingga classpath
// ditambahkan otomatis.

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.count
import kotlinx.coroutines.runBlocking

// SUT yang memancarkan 3 nilai lalu selesai.
fun suhuHarian(): Flow<Int> = flowOf(28, 29, 30)

// SUT yang memancarkan 0..2 lalu melempar pengecualian.
fun aliranBerbahaya(): Flow<Int> = flow {
    emit(0)
    emit(1)
    emit(2)
    throw IllegalStateException("aliran gagal pada emit ke-3")
}

// Helper mini-Turbine: mengumpulkan emissions dan menyimpannya sebagai list.
class Emissions<T> {
    private val nilai = mutableListOf<T>()
    var error: Throwable? = null
        private set
    var selesai: Boolean = false
        private set

    fun emitNilai(v: T) { nilai += v }
    fun setError(e: Throwable) { error = e }
    fun setSelesai() { selesai = true }

    fun assertEmissions(expected: List<T>, label: String) {
        if (nilai == expected) {
            println("[LULUS] $label: emissions=${nilai}")
        } else {
            println("[GAGAL]  $label: expected=$expected actual=${nilai}")
        }
    }

    fun assertError(label: String, pesanBerisi: String) {
        val e = error
        if (e != null && e.message?.contains(pesanBerisi) == true) {
            println("[LULUS] $label: ${e::class.simpleName} — ${e.message}")
        } else {
            println("[GAGAL]  $label: expected contains '$pesanBerisi', actual=${e?.message}")
        }
    }

    fun assertCompleted(label: String) {
        println("${if (selesai && error == null) "[LULUS]" else "[GAGAL] "} $label: selesai=$selesai error=$error")
    }
}

// Pola Turbine flow.test { awaitItem(); awaitComplete(); awaitError() }
suspend fun <T> Flow<T>.test(blok: suspend Emissions<T>.() -> Unit) {
    val em = Emissions<T>()
    try {
        collect { em.emitNilai(it) }
        em.setSelesai()
    } catch (e: Throwable) {
        em.setError(e)
    }
    em.blok()
}

fun main() = runBlocking {
    println("=== flowOf(28, 29, 30) ===")
    suhuHarian().test {
        assertEmissions(listOf(28, 29, 30), "suhu harian lengkap")
        assertCompleted("aliran selesai tanpa error")
    }

    println()
    println("=== first() hanya emission pertama ===")
    val pertama = sukuHarianFirst()
    check(pertama == 28) { "Expected 28, got $pertama" }
    println("[LULUS] first() = $pertama")

    println()
    println("=== count() menghitung jumlah emission ===")
    val n = suhuHarian().count()
    check(n == 3) { "Expected 3 emissions, got $n" }
    println("[LULUS] count() = $n")

    println()
    println("=== toList() untuk assertion berurutan ===")
    val daftar = suhuHarian().toList()
    check(daftar == listOf(28, 29, 30))
    println("[LULUS] toList() = $daftar")

    println()
    println("=== aliran dengan error ===")
    aliranBerbahaya().test {
        assertEmissions(listOf(0, 1, 2), "emissions sebelum error")
        assertError("error setelah 3 emit", "aliran gagal pada emit ke-3")
    }
}

suspend fun sukuHarianFirst(): Int = suhuHarian().first()

Output yang Dihasilkan:

text 
=== flowOf(28, 29, 30) ===
[LULUS] suhu harian lengkap: emissions=[28, 29, 30]
[LULUS] aliran selesai tanpa error: selesai=true error=null

=== first() hanya emission pertama ===
[LULUS] first() = 28

=== count() menghitung jumlah emission ===
[LULUS] count() = 3

=== toList() untuk assertion berurutan ===
[LULUS] toList() = [28, 29, 30]

=== aliran dengan error ===
[LULUS] emissions sebelum error: emissions=[0, 1, 2]
[LULUS] error setelah 3 emit: IllegalStateException — aliran gagal pada emit ke-3

Mari kita bedah beberapa poin penting pada file di atas:

  1. Emissions<T> class: menampung daftar nilai yang dipancarkan, optional error, dan flag selesai. Method assertEmissions, assertError, dan assertCompleted meniru awaitItem(), awaitError(), dan awaitComplete() dari Turbine.
  2. Flow<T>.test(blok): extension function yang mengkoleksi Flow, merekam setiap emission dan error, lalu memanggil blok receiver dengan Emissions yang sudah terisi. Pola ini sama dengan Turbine flow.test { }.
  3. flowOf(28, 29, 30): Flow cold yang memancarkan tiga nilai lalu complete. assertCompleted memastikan selesai=true dan error=null.
  4. aliranBerbahaya(): Flow yang melempar IllegalStateException setelah emit 3 nilai. assertError memeriksa tipe dan pesan pengecualian.
  5. Operator first, count, toList: alternatif tanpa Turbine untuk test sederhana. first() mengambil emission pertama, count() menghitung total, toList() mengumpulkan semuanya.

Pola Lanjutan Turbine #

kotlin Contoh: awaitItem dengan timeout kustom 
import app.cash.turbine.test
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.test.runTest

class PolaLanjut {
    suspend fun produceFlow() = flow {
        emit("A"); delay(10); emit("B")
    }

    fun `emit dengan delay`() = runTest {
        produceFlow().test {
            assertEquals("A", awaitItem())
            assertEquals("B", awaitItem())
            awaitComplete()
        }
    }
}

Kapan Tidak Pakai Turbine #

Untuk Flow sederhana yang tidak melibatkan error atau delay kompleks, operator standar toList(), first(), dan count() sering sudah cukup. Turbine berlebih jika test hanya ingin memeriksa hasil akhir koleksi. Pakai Turbine ketika urutan emission, keberadaan error di tengah aliran, atau kontrak completion menjadi hal yang diuji.

Konfigurasi Dependensi Turbine #

kotlin Contoh: dependensi Turbine 
dependencies {
    testImplementation("app.cash.turbine:turbine:1.1.0")
    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.8.1")
}

Turbine dirancang khusus untuk Kotlin dan memanfaatkan coroutine sepenuhnya. Library ini menyediakan tiga API utama: flow.test { } untuk skenario normal, flow.testIn(scope) untuk kontrol TestScope manual, dan TurbineTestContext receiver yang dipakai di dalam blok test. Kombinasi Turbine dengan runTest menghasilkan test yang sepenuhnya deterministik dan cepat.

Pola Test State Flow #

StateFlow adalah varian khusus Flow yang selalu memancarkan nilai terbaru ke kolektor baru. Test untuk StateFlow sering memakai first() untuk mengambil snapshot awal, lalu value property untuk membaca state terkini. Turbine tidak terlalu dibutuhkan karena StateFlow tidak memancarkan banyak nilai sekaligus — ia hanya menyimpan satu nilai yang dapat dibaca kapan saja.

kotlin Contoh: test StateFlow dengan value 
import kotlinx.coroutines.flow.MutableStateFlow
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Assertions.assertEquals

class CounterViewModelTest {
    private val counter = MutableStateFlow(0)

    @Test
    fun `increment updates state`() {
        counter.value = 1
        assertEquals(1, counter.value)
    }
}

Referensi Lanjutan #

  • Turbine — Flow testing — pustaka Turbine dari Cash App untuk menguji Flow Kotlin dengan awaitItem() dan awaitComplete()
  • kotlinx-coroutines-test — rujukan resmi kotlinx-coroutines-test untuk runTest, TestScope, dan UnconfinedTestDispatcher

Selanjutnya #

Setelah menguasai Flow, langkah berikutnya adalah menggabungkan beberapa komponen nyata dalam satu test — fokus pada integrasi antar layer aplikasi. Chapter integration-testing akan membahas pola @SpringBootTest, Ktor testApplication, dan kapan memakai real database.