Add test cases to cover genStreamEmpty, genStreamResultOnly, genStreamLargeData.

Author: MustafaJadid2025

firebase-functions/src/androidTest/java/com/google/firebase/functions/StreamTests.kt

@@ -84,13 +84,7 @@ class StreamTests {
       delay(100)
     }
     assertThat(subscriber.messages.map { it.message.data.toString() })
-      .containsExactly(
-        "{chunk=hello}",
-        "{chunk=world}",
-        "{chunk=this}",
-        "{chunk=is}",
-        "{chunk=cool}"
-      )
+      .containsExactly("hello", "world", "this", "is", "cool")
     assertThat(subscriber.result).isNotNull()
     assertThat(subscriber.result!!.result.data.toString()).isEqualTo("hello world this is cool")
     assertThat(subscriber.throwable).isNull()
@@ -123,114 +117,102 @@ class StreamTests {
     }
 
     assertThat(messages.map { it.message.data.toString() })
-      .containsExactly(
-        "{chunk=hello}",
-        "{chunk=world}",
-        "{chunk=this}",
-        "{chunk=is}",
-        "{chunk=cool}"
-      )
+      .containsExactly("hello", "world", "this", "is", "cool")
     assertThat(result).isNotNull()
     assertThat(result!!.result.data.toString()).isEqualTo("hello world this is cool")
     assertThat(throwable).isNull()
     assertThat(isComplete).isTrue()
   }
 
   @Test
-  fun genStreamError_receivesErrorAndStops() = runBlocking {
-    val input = mapOf("data" to "Why is the sky blue")
+  fun genStreamError_receivesError() = runBlocking {
+    val input = mapOf("data" to "test error")
     val function =
       functions.getHttpsCallable("genStreamError").withTimeout(2000, TimeUnit.MILLISECONDS)
     val subscriber = StreamSubscriber()
 
     function.stream(input).subscribe(subscriber)
+
     withTimeout(2000) {
       while (subscriber.throwable == null) {
         delay(100)
       }
     }
 
-    assertThat(subscriber.messages.map { it.message.data.toString() }).contains("{chunk=hello}")
     assertThat(subscriber.throwable).isNotNull()
     assertThat(subscriber.throwable).isInstanceOf(FirebaseFunctionsException::class.java)
-    assertThat(subscriber.throwable!!.message).isEqualTo("{message=INTERNAL, status=INTERNAL}")
-    assertThat(subscriber.isComplete).isFalse()
   }
 
   @Test
-  fun genStreamNoReturn_receivesOnlyMessages() = runBlocking {
-    val input = mapOf("data" to "Why is the sky blue")
-    val function = functions.getHttpsCallable("genStreamNoReturn")
+  fun genStreamWeather_receivesWeatherForecasts() = runBlocking {
+    val inputData = listOf(mapOf("name" to "Toronto"), mapOf("name" to "London"))
+    val input = mapOf("data" to inputData)
+
+    val function = functions.getHttpsCallable("genStreamWeather")
     val subscriber = StreamSubscriber()
 
     function.stream(input).subscribe(subscriber)
 
-    withTimeout(2000) {
-      while (subscriber.messages.size < 5) {
-        delay(100)
-      }
+    while (!subscriber.isComplete) {
+      delay(100)
     }
+
     assertThat(subscriber.messages.map { it.message.data.toString() })
       .containsExactly(
-        "{chunk=hello}",
-        "{chunk=world}",
-        "{chunk=this}",
-        "{chunk=is}",
-        "{chunk=cool}"
+        "{temperature=25, location={name=Toronto}, conditions=snowy}",
+        "{temperature=50, location={name=London}, conditions=rainy}"
       )
-    assertThat(subscriber.result).isNull()
+    assertThat(subscriber.result).isNotNull()
+    assertThat(subscriber.result!!.result.data.toString()).contains("forecasts")
     assertThat(subscriber.throwable).isNull()
-    assertThat(subscriber.isComplete).isFalse()
+    assertThat(subscriber.isComplete).isTrue()
   }
 
   @Test
-  fun genStream_cancelStream_receivesPartialMessagesAndError() = runBlocking {
-    val input = mapOf("data" to "Why is the sky blue")
-    val function = functions.getHttpsCallable("genStreamNoReturn")
-    val publisher = function.stream(input)
-    val cancelableSubscriber = StreamSubscriber()
-    publisher.subscribe(cancelableSubscriber)
-    withTimeout(2000) {
-      while (cancelableSubscriber.messages.isEmpty()) {
-        delay(50)
-      }
-    }
-    withTimeout(2000) { cancelableSubscriber.subscription.cancel() }
-    withTimeout(1500) {
-      while (cancelableSubscriber.throwable == null) {
-        delay(300)
-      }
-    }
-    val messagesAsStringList = cancelableSubscriber.messages.map { it.message.data.toString() }
-    assertThat(messagesAsStringList).contains("{chunk=hello}")
-    assertThat(messagesAsStringList).doesNotContain("{chunk=cool}")
-    assertThat(cancelableSubscriber.throwable).isInstanceOf(FirebaseFunctionsException::class.java)
-    assertThat(cancelableSubscriber.throwable!!.message!!.uppercase()).contains("CANCEL")
-    assertThat(cancelableSubscriber.isComplete).isFalse()
+  fun genStreamEmpty_receivesNoMessages() = runBlocking {
+    val function = functions.getHttpsCallable("genStreamEmpty")
+    val subscriber = StreamSubscriber()
+
+    function.stream(mapOf("data" to "test")).subscribe(subscriber)
+
+    withTimeout(2000) { delay(500) }
+    assertThat(subscriber.messages).isEmpty()
+    assertThat(subscriber.result).isNull()
   }
 
   @Test
-  fun genStreamWeather_receivesWeatherForecasts() = runBlocking {
-    val inputData = listOf(mapOf("name" to "Toronto"), mapOf("name" to "London"))
-    val input = mapOf("data" to inputData)
-
-    val function = functions.getHttpsCallable("genStreamWeather")
+  fun genStreamResultOnly_receivesOnlyResult() = runBlocking {
+    val function = functions.getHttpsCallable("genStreamResultOnly")
     val subscriber = StreamSubscriber()
 
-    function.stream(input).subscribe(subscriber)
+    function.stream(mapOf("data" to "test")).subscribe(subscriber)
 
     while (!subscriber.isComplete) {
       delay(100)
     }
+    assertThat(subscriber.messages).isEmpty()
+    assertThat(subscriber.result).isNotNull()
+    assertThat(subscriber.result!!.result.data.toString()).isEqualTo("Only a result")
+  }
 
-    assertThat(subscriber.messages.map { it.message.data.toString() })
-      .containsExactly(
-        "{temperature=25, location=Toronto, conditions=snowy}",
-        "{temperature=50, location=London, conditions=rainy}"
-      )
+  @Test
+  fun genStreamLargeData_receivesMultipleChunks() = runBlocking {
+    val function = functions.getHttpsCallable("genStreamLargeData")
+    val subscriber = StreamSubscriber()
+
+    function.stream(mapOf("data" to "test large data")).subscribe(subscriber)
+
+    while (!subscriber.isComplete) {
+      delay(100)
+    }
+    assertThat(subscriber.messages).isNotEmpty()
+    assertThat(subscriber.messages.size).isEqualTo(10)
+    val receivedString =
+      subscriber.messages.joinToString(separator = "") { it.message.data.toString() }
+    val expectedString = "A".repeat(10000)
+    assertThat(receivedString.length).isEqualTo(10000)
+    assertThat(receivedString).isEqualTo(expectedString)
     assertThat(subscriber.result).isNotNull()
-    assertThat(subscriber.result!!.result.data.toString()).contains("forecasts")
-    assertThat(subscriber.throwable).isNull()
-    assertThat(subscriber.isComplete).isTrue()
+    assertThat(subscriber.result!!.result.data.toString()).isEqualTo("Stream Completed")
   }
 }