0Pricing
Kotlin Academy · Lektion

StateFlow und SharedFlow mit Turbine testen

Schreiben Sie Unit-Tests für Flow-basierten Zustand mit der Testbibliothek Turbine.

StateFlow und SharedFlow mit Turbine testen ist eine kostenlose Kotlin Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des Kotlin Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der Kotlin Academy-Kurs umfasst insgesamt 4 Lektionen.

Warum Turbine?

Turbine ist eine Testbibliothek für Kotlin Flow, die das Überprüfen von Emissionen erleichtert. Ohne sie erfordert das Testen von Flows das manuelle Sammeln über einen Channel mit Timeouts.

// 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") }

Grundlegendes Flow-Testen mit test()

flow.test { } sammelt den Flow und ermöglicht es Ihnen, jede Emission mit awaitItem(), awaitComplete() und awaitError() zu überprüfen.

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 testen

StateFlow gibt bei der Sammlung immer seinen aktuellen Wert aus. Turbines awaitItem() erfasst diese anfängliche Emission.

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-Ereignisse testen

Ein SharedFlow mit replay=0 gibt nur zukünftige Ereignisse aus. Beginnen Sie zuerst mit dem Sammeln und emittieren Sie dann.

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 für virtuelle Zeit

runTest verwendet virtuelle Zeit – Aufrufe von delay() lassen die virtuelle Zeit sofort voranschreiten, wodurch Tests mit Verzögerungen schnell ausgeführt werden.

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 zum Testen von Ausnahmen

awaitError() stellt sicher, dass der Flow mit einer Ausnahme beendet wird. Verwenden Sie es zusammen mit dem erwarteten Ausnahmetyp.

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 und cancelAndIgnoreRemainingEvents

expectNoEvents() stellt sicher, dass nichts emittiert wurde. cancelAndIgnoreRemainingEvents() bricht das Sammeln ab und ignoriert ausstehende Emissionen.

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 mit Turbine testen

Testen Sie den StateFlow eines ViewModels, indem Sie ein Fake-Repository injizieren und Zustandsübergänge überprüfen.

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()
    }
}

Mehrere Collector mit turbineScope

turbineScope { } ermöglicht es Ihnen, mehrere Flows gleichzeitig zu testen, jeweils mit einer eigenen Turbine-Instanz.

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()
    }
}

Flow-Operatoren testen

Testen Sie das Verhalten eigener Flow-Operatoren und Transformationen isoliert mit Turbine.

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()
        }
}

Testen mit simuliertem Zeitvorschub

Verwenden Sie testScheduler.advanceTimeBy(), um die virtuelle Zeit in Tests mit delay präzise zu steuern.

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()
    }
}

Häufige Fehler

Beenden Sie Turbine am Ende von Tests mit Hot Flows immer. Wenn Sie cancelAndIgnoreRemainingEvents() vergessen, hängen Tests beim Warten auf weitere Ereignisse.

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()
    }
}

Kurzer Check

Welche Turbine-Funktion sammelt das nächste emittierte Element und löst eine Ausnahme aus, wenn der Flow vorher abgeschlossen wird oder ein Fehler auftritt?

Zusammenfassung

Turbine vereinfacht das Testen von Flows mit awaitItem(), awaitComplete() und awaitError(). Verwenden Sie runTest für virtuelle Zeit. Beenden Sie Tests mit Hot Flows immer mit cancelAndIgnoreRemainingEvents().

Häufig gestellte Fragen

Ist die Lektion „StateFlow und SharedFlow mit Turbine testen“ kostenlos?

Ja — der vollständige Text von „StateFlow und SharedFlow mit Turbine testen“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des Kotlin Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der Kotlin Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „StateFlow und SharedFlow mit Turbine testen“?

Schreiben Sie Unit-Tests für Flow-basierten Zustand mit der Testbibliothek Turbine. Du übst Kotlin Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um Kotlin Academy zu starten?

Keine Vorkenntnisse erforderlich. Kotlin Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „StateFlow und SharedFlow mit Turbine testen“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser Kotlin Academy-Lektion Code schreiben und ausführen?

Ja. Jede Kotlin Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. StateFlow: reaktiver Zustandscontainer für die UI
  2. SharedFlow: Event-Busse und einmalige Events
  3. Cold Flow mit shareIn und stateIn in Hot Flow umwandeln
  4. StateFlow und SharedFlow mit Turbine testen
← Zurück zu Kotlin Academy