KTLN-575 Testing Flow in Kotlin (#1137)

* Added unit tests

* changes to code

* code changes

* code fixes

* code fixes

* code fixes

* Added code

* fixes

* fixes

* fixes

* fixes

---------

Co-authored-by: nappy29 <[email protected]>

Author: naoussi

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

@@ -1,11 +1,11 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
-    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
     <modelVersion>4.0.0</modelVersion>
     <artifactId>core-kotlin-flows</artifactId>
     <name>core-kotlin-flows</name>
     <packaging>jar</packaging>
-
     <parent>
         <groupId>com.baeldung</groupId>
         <artifactId>core-kotlin-modules</artifactId>
@@ -21,17 +21,30 @@
         <dependency>
             <groupId>org.jetbrains.kotlinx</groupId>
             <artifactId>kotlinx-coroutines-reactor</artifactId>
-            <version>1.9.0</version>
+            <version>${kotlinx.coroutines.version}</version>
+
         </dependency>
         <dependency>
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-core</artifactId>
-            <version>3.6.10</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,144 @@
+package com.baeldung.flowTest
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.onFailure
+import kotlinx.coroutines.flow.*
+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 `implicitCancellationFlow stops on cancellation`() = runTest {
+        val emittedValues = mutableListOf<Int>()
+        val job = launch {
+            implicitCancellationFlow().collect { emittedValues.add(it) }
+        }
+
+        advanceTimeBy(600)
+        job.cancelAndJoin()
+
+        assertEquals(listOf(1,2), 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 `uncancellableFlow ensures cleanup occurs`() = runTest {
+        val emittedValues = mutableListOf<Int>()
+        val job = launch {
+            uncancellableFlow().collect { emittedValues.add(it) }
+        }
+
+        advanceTimeBy(400)
+        job.cancelAndJoin()
+
+        assertEquals(listOf(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)
+}
+
+fun implicitCancellationFlow(): Flow<Int> = flow {
+    emit(1)
+    delay(500)
+    emit(2)
+    delay(500)
+    emit(3)
+}
+
+fun uncancellableFlow(): Flow<Int> = flow {
+    try {
+        emit(1)
+        delay(500)
+        emit(2)
+    } finally {
+        withContext(NonCancellable) {
+            println("Releasing resources")
+        }
+    }
+}