اختبار StateFlow وSharedFlow باستخدام Turbine

لماذا Turbine؟

// Add to test dependencies:
// testImplementation("app.cash.turbine:turbine:1.0.0")
// testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.7.3")
import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun main() { println("Turbine + runTest = clean flow testing") }

اختبار التدفق الأساسي باستخدام test()

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testBasicFlow() = runTest {
    val flow = flowOf(1, 2, 3)
    flow.test {
        assertEquals(1, awaitItem())
        assertEquals(2, awaitItem())
        assertEquals(3, awaitItem())
        awaitComplete()
    }
}

اختبار StateFlow

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testStateFlow() = runTest {
    val state = MutableStateFlow("loading")
    state.test {
        assertEquals("loading", awaitItem())  // initial value
        state.value = "success"
        assertEquals("success", awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

اختبار أحداث SharedFlow

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testSharedFlow() = runTest {
    val events = MutableSharedFlow<String>()
    events.test {
        events.emit("LoginSuccess")
        assertEquals("LoginSuccess", awaitItem())
        events.emit("Navigate")
        assertEquals("Navigate", awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

runTest للوقت الافتراضي

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
import app.cash.turbine.*
fun testDelayedFlow() = runTest {
    val flow = flow {
        delay(1000)  // virtual — runs instantly
        emit("done")
    }
    flow.test {
        assertEquals("done", awaitItem())
        awaitComplete()
    }
}

استخدام awaitError لاختبار الاستثناءات

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testFlowError() = runTest {
    val flow = flow {
        emit(1)
        throw RuntimeException("stream error")
    }
    flow.test {
        assertEquals(1, awaitItem())
        val error = awaitError()
        assertEquals("stream error", error.message)
    }
}

expectNoEvents وcancelAndIgnoreRemainingEvents

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testNoEvents() = runTest {
    val flow = MutableSharedFlow<Int>()
    flow.test {
        expectNoEvents()  // nothing emitted yet
        flow.emit(1)
        assertEquals(1, awaitItem())
        cancelAndIgnoreRemainingEvents()
    }
}

اختبار ViewModel باستخدام Turbine

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
sealed class State { object Loading : State(); data class Data(val v: String) : State() }
class FakeRepo { suspend fun fetch() = "result" }
class VM(repo: FakeRepo, scope: kotlinx.coroutines.CoroutineScope) {
    private val _s = MutableStateFlow<State>(State.Loading)
    val state = _s.asStateFlow()
    init { scope.launch { _s.value = State.Data(repo.fetch()) } }
}
fun testVM() = runTest {
    val vm = VM(FakeRepo(), backgroundScope)
    vm.state.test {
        assertEquals(State.Loading, awaitItem())
        val data = awaitItem() as State.Data
        assertEquals("result", data.v)
        cancelAndIgnoreRemainingEvents()
    }
}

عدة جامعين باستخدام turbineScope

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testMultipleFlows() = runTest {
    val a = MutableStateFlow(1)
    val b = MutableStateFlow("x")
    turbineScope {
        val aTurbine = a.testIn(backgroundScope)
        val bTurbine = b.testIn(backgroundScope)
        assertEquals(1, aTurbine.awaitItem())
        assertEquals("x", bTurbine.awaitItem())
        a.value = 2
        assertEquals(2, aTurbine.awaitItem())
        aTurbine.cancelAndIgnoreRemainingEvents()
        bTurbine.cancelAndIgnoreRemainingEvents()
    }
}

اختبار معاملات التدفق

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testOperators() = runTest {
    (1..5).asFlow()
        .filter { it % 2 == 0 }
        .map { it * 10 }
        .test {
            assertEquals(20, awaitItem())
            assertEquals(40, awaitItem())
            awaitComplete()
        }
}

الاختبار باستخدام تقديم وقت وهمي

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testTimedFlow() = runTest {
    val flow = flow {
        emit("start")
        delay(500)
        emit("end")
    }
    flow.test {
        assertEquals("start", awaitItem())
        testScheduler.advanceTimeBy(500)
        assertEquals("end", awaitItem())
        awaitComplete()
    }
}

الأخطاء الشائعة

import app.cash.turbine.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*
fun testGoodPractice() = runTest {
    val state = MutableStateFlow(0)
    state.test {
        awaitItem()  // initial 0
        state.value = 1
        awaitItem()  // 1
        // Always cancel hot flows:
        cancelAndIgnoreRemainingEvents()
    }
}

تحقق سريع

مراجعة