KTLN-575 Testing Flow in Kotlin

core-kotlin-modules/core-kotlin-flows/pom.xml

@@ -18,24 +18,36 @@
             <version>${junit.version}</version>
         </dependency>
 
-
-
         <dependency>
             <groupId>org.jetbrains.kotlinx</groupId>
             <artifactId>kotlinx-coroutines-reactor</artifactId>
-            <version>1.6.0</version>
+            <version>${kotlinx.coroutines.version}</version>
         </dependency>
 
         <dependency>
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-core</artifactId>
-            <version>3.4.11</version>
+            <version>${reactor.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-core</artifactId>
+            <version>${kotlinx.coroutines.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-test</artifactId>
+            <version>${kotlinx.coroutines.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <properties>
         <junit.version>4.13.2</junit.version>
+        <kotlinx.coroutines.version>1.7.3</kotlinx.coroutines.version>
+        <reactor.version>3.4.11</reactor.version>
     </properties>
-    
+
 </project>

core-kotlin-modules/core-kotlin-flows/src/test/kotlin/com/baeldung/flowTest/FlowTestUnitTest.kt

@@ -0,0 +1,102 @@
+package com.baeldung.flowTest
+
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.runTest
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import java.util.concurrent.CancellationException
+
+class FlowTestUnitTest {
+
+    @Test
+    fun `simpleFlow should emit 1 2 3`() = runTest {
+        val flow = simpleFlow().toList()
+
+        assertEquals(listOf(1, 2, 3), flow)
+    }
+
+    @Test
+    fun `transformedFlow should multiply values by 2`() = runTest {
+        val result = transformedFlow().toList()
+        assertEquals(listOf(2, 4, 6), result)
+    }
+
+    @Test
+    fun `errorFlow should emit values and recover from exception`() = runTest {
+        val emittedValues = errorFlow().toList()
+
+        assertEquals(listOf(1, 2, -1), emittedValues)
+    }
+
+    @Test
+    fun `cancellableFlow stops emitting after external cancellation`() = runTest {
+        val emittedValues = mutableListOf<Int>()
+        val job = launch {
+            cancellableFlow().collect { emittedValues.add(it) }
+        }
+
+        advanceTimeBy(2000)
+        job.cancelAndJoin()
+
+        assertEquals(listOf(0, 1), emittedValues)
+    }
+
+    @Test
+    fun `delayedFlow should handle delayed emissions`() = runTest {
+        val result = delayedFlow().toList()
+        assertEquals(listOf(1, 2), result)
+    }
+}
+
+fun simpleFlow(): Flow<Int> = flow {
+    emit(1)
+    emit(2)
+    emit(3)
+}
+
+fun transformedFlow(): Flow<Int> = flow {
+    emit(1)
+    emit(2)
+    emit(3)
+}.map { it * 2 }
+
+fun errorFlow(): Flow<Int> = flow {
+    emit(1)
+    emit(2)
+    throw Exception("Test Exception")
+}.catch { e ->
+    emit(-1)
+}
+
+fun cancellableFlow(): Flow<Int> = flow {
+    try {
+        repeat(5) {
+            emit(it)
+            delay(1000)
+        }
+    } finally {
+        println("Cleanup: Emitting -1")
+        emit(-1)
+    }
+}.onEach { value ->
+    if (value == 2) throw CancellationException("Flow was canceled at value 2")
+}.onCompletion { cause ->
+    if (cause is CancellationException) {
+        println("Flow canceled: Releasing resources")
+    }
+}
+
+fun delayedFlow(): Flow<Int> = flow {
+    delay(500)
+    emit(1)
+    delay(500)
+    emit(2)
+}
+
+