dataconnect(test): ConnectRPCIntegrationTest.kt added with its first test (#8097)

Author: dconeybe

firebase-dataconnect/androidTestutil/src/main/kotlin/com/google/firebase/dataconnect/testutil/DataConnectIntegrationTestBase.kt

@@ -39,6 +39,8 @@ abstract class DataConnectIntegrationTestBase {
 
   @get:Rule val dataConnectFactory = TestDataConnectFactory(firebaseAppFactory)
 
+  @get:Rule val cleanups = CleanupsRule()
+
   @get:Rule(order = Int.MIN_VALUE) val randomSeedTestRule = RandomSeedTestRule()
 
   val rs: RandomSource by randomSeedTestRule.rs

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/ConnectRPCIntegrationTest.kt

@@ -0,0 +1,225 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.dataconnect
+
+import app.cash.turbine.test
+import com.google.firebase.dataconnect.testutil.DataConnectIntegrationTestBase
+import com.google.firebase.dataconnect.testutil.createGrpcManagedChannel
+import com.google.firebase.dataconnect.testutil.schemas.RealtimeConnector
+import com.google.firebase.dataconnect.testutil.schemas.RealtimeConnector.GetStringByKeyQuery
+import com.google.firebase.dataconnect.testutil.schemas.RealtimeConnector.InsertStringMutation
+import com.google.firebase.dataconnect.util.ProtoUtil.decodeFromStruct
+import com.google.firebase.dataconnect.util.ProtoUtil.encodeToStruct
+import google.firebase.dataconnect.proto.ConnectorStreamServiceGrpcKt.ConnectorStreamServiceCoroutineStub
+import google.firebase.dataconnect.proto.ExecuteRequest
+import google.firebase.dataconnect.proto.StreamRequest
+import google.firebase.dataconnect.proto.StreamResponse
+import io.kotest.assertions.assertSoftly
+import io.kotest.assertions.withClue
+import io.kotest.matchers.nulls.shouldNotBeNull
+import io.kotest.matchers.result.shouldBeSuccess
+import io.kotest.matchers.shouldBe
+import io.kotest.property.Arb
+import io.kotest.property.arbitrary.Codepoint
+import io.kotest.property.arbitrary.az
+import io.kotest.property.arbitrary.map
+import io.kotest.property.arbitrary.string
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+
+class ConnectRPCIntegrationTest : DataConnectIntegrationTestBase() {
+
+  @Test
+  fun responseIsReceivedForExecuteQueryRequest() = runTest {
+    val requestId = requestIdArb().sample()
+    val name = nameArb().sample()
+    val (connector, outgoingRequests, incomingResponses) = connect()
+    val key = connector.insertString(name)
+    outgoingRequests.sendInitRequest(requestId = "init", connector.resourceName)
+
+    incomingResponses.test {
+      outgoingRequests.sendNonBlockingOrThrow(
+        StreamRequest.newBuilder().let { streamRequest ->
+          streamRequest.setRequestId(requestId)
+          streamRequest.setExecute(
+            ExecuteRequest.newBuilder().let { executeRequest ->
+              executeRequest.setOperationName(GetStringByKeyQuery.OPERATION_NAME)
+              executeRequest.setVariables(encodeToStruct(GetStringByKeyQuery.Variables(key)))
+              executeRequest.build()
+            }
+          )
+          streamRequest.build()
+        }
+      )
+
+      val streamResponse = awaitItem()
+
+      streamResponse.shouldBeGetStringByKeyDataWithName(requestId = requestId, name = name)
+    }
+  }
+
+  @Test
+  fun responseIsReceivedForExecuteMutationRequest() = runTest {
+    val requestId = requestIdArb().sample()
+    val name = nameArb().sample()
+    val (connector, outgoingRequests, incomingResponses) = connect()
+    outgoingRequests.sendInitRequest(requestId = "init", connector.resourceName)
+
+    incomingResponses.test {
+      outgoingRequests.sendNonBlockingOrThrow(
+        StreamRequest.newBuilder().let { streamRequest ->
+          streamRequest.setRequestId(requestId)
+          streamRequest.setExecute(
+            ExecuteRequest.newBuilder().let { executeRequest ->
+              executeRequest.setOperationName(InsertStringMutation.OPERATION_NAME)
+              executeRequest.setVariables(encodeToStruct(InsertStringMutation.Variables(name)))
+              executeRequest.build()
+            }
+          )
+          streamRequest.build()
+        }
+      )
+
+      val streamResponse = awaitItem()
+
+      val key = streamResponse.shouldBeInsertStringData(requestId = requestId)
+      connector.getString(key).shouldNotBeNull().name shouldBe name
+    }
+  }
+
+  /**
+   * Establishes a connection to the Data Connect `Connect` RPC and returns the streams for
+   * communication.
+   */
+  private fun connect(): ConnectionStreams {
+    val connector = RealtimeConnector.getInstance(dataConnectFactory)
+    val grpcManagedChannel = createGrpcManagedChannel(connector.dataConnect)
+    val outgoingRequests = Channel<StreamRequest>(Channel.UNLIMITED)
+    val stub = ConnectorStreamServiceCoroutineStub(grpcManagedChannel)
+    val incomingResponses = stub.connect(outgoingRequests.consumeAsFlow())
+    return ConnectionStreams(connector, outgoingRequests, incomingResponses)
+  }
+
+  /**
+   * Convenience extension function on [Arb] that gets a non-edge-case value using the [rs] property
+   * of the [DataConnectIntegrationTestBase] superclass for the randomness source.
+   */
+  private fun <T> Arb<T>.sample(): T = sample(rs).value
+}
+
+/**
+ * Creates and returns an [Arb] that generates strings that are suitable for, and recognizable as,
+ * "request IDs" in [StreamRequest] messages.
+ */
+private fun requestIdArb(): Arb<String> = Arb.string(size = 4, Codepoint.az()).map { "rid$it" }
+
+/**
+ * Creates and returns an [Arb] that generates strings that are suitable for, and recognizable as,
+ * "name" values for [RealtimeConnector.InsertStringMutation.Variables.name] and
+ * [RealtimeConnector.UpdateStringMutation.Variables.name].
+ */
+private fun nameArb(): Arb<String> = Arb.string(size = 4, Codepoint.az()).map { "nam$it" }
+
+/** Exception thrown by [sendNonBlockingOrThrow] if sending fails. */
+private class SendChannelTrySendUnsuccessfulException(message: String, cause: Throwable?) :
+  Exception(message, cause)
+
+/**
+ * Sends an element to receiver [SendChannel] in a non-blocking way, throwing an exception if
+ * sending was unsuccessful.
+ *
+ * @throws SendChannelTrySendUnsuccessfulException if [SendChannel.trySend] returns a failed result.
+ */
+private fun <T> SendChannel<T>.sendNonBlockingOrThrow(element: T) {
+  val result = trySend(element)
+  if (!result.isSuccess) {
+    throw SendChannelTrySendUnsuccessfulException(
+      "trySend() failed with isClosed=${result.isClosed}",
+      result.exceptionOrNull()
+    )
+  }
+}
+
+private data class ConnectionStreams(
+  val connector: RealtimeConnector,
+  val outgoingRequests: SendChannel<StreamRequest>,
+  val incomingResponses: Flow<StreamResponse>,
+)
+
+/**
+ * Sends an initialization request to the receiver [ConnectionStreams.outgoingRequests].
+ *
+ * @param requestId The "request ID" to send in the outgoing message.
+ * @param connectorResourceName The "connector resource name" to send in the outgoing message.
+ */
+private fun SendChannel<StreamRequest>.sendInitRequest(
+  requestId: String,
+  connectorResourceName: String
+) {
+  val streamRequest =
+    StreamRequest.newBuilder().let {
+      it.setRequestId(requestId)
+      it.setName(connectorResourceName)
+      it.build()
+    }
+  sendNonBlockingOrThrow(streamRequest)
+}
+
+/**
+ * Assertion to verify that a [StreamResponse] matches the expected "RealtimeString_GetByKey" data.
+ *
+ * @param requestId The expected request ID.
+ * @param name The expected name in the data.
+ */
+private fun StreamResponse.shouldBeGetStringByKeyDataWithName(requestId: String, name: String) {
+  withClue("StreamResponse.shouldBeGetStringByKeyDataWithName") {
+    assertSoftly {
+      val data = shouldBeSuccess<GetStringByKeyQuery.Data>(requestId = requestId)
+      withClue("data") { data.item.shouldNotBeNull().name shouldBe name }
+    }
+  }
+}
+
+/**
+ * Assertion to verify that a [StreamResponse] matches the expected "RealtimeString_Insert" data.
+ *
+ * @param requestId The expected request ID.
+ */
+private fun StreamResponse.shouldBeInsertStringData(requestId: String): RealtimeConnector.Key =
+  withClue("StreamResponse.shouldBeInsertStringData") {
+    assertSoftly { shouldBeSuccess<InsertStringMutation.Data>(requestId = requestId).key }
+  }
+
+/**
+ * Assertion to verify that a [StreamResponse] matches the expected "RealtimeString_Insert" data.
+ *
+ * @param requestId The expected request ID.
+ * @return the [Data] decoded from the receiver.
+ */
+private inline fun <reified Data> StreamResponse.shouldBeSuccess(requestId: String): Data {
+  withClue("requestId") { this.requestId shouldBe requestId }
+  withClue("errorsCount") { this.errorsCount shouldBe 0 }
+
+  val dataDecodeResult = runCatching { decodeFromStruct<Data>(this.data) }
+  val data = withClue("decodeFromStruct(data)") { dataDecodeResult.shouldBeSuccess() }
+
+  return data
+}

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/testutil/GrpcTestUtil.kt

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.dataconnect.testutil
+
+import com.google.firebase.dataconnect.FirebaseDataConnect
+import com.google.firebase.dataconnect.core.DataConnectGrpcRPCs
+import com.google.firebase.dataconnect.core.FirebaseDataConnectInternal
+import com.google.firebase.dataconnect.core.Logger
+import io.grpc.ManagedChannel
+import io.mockk.mockk
+import java.util.concurrent.TimeUnit
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.asExecutor
+
+/**
+ * Information about the gRPC configuration used by this [FirebaseDataConnect] instance.
+ *
+ * This is primarily intended for use in tests to inspect or replicate the gRPC connection
+ * parameters, such as the host and whether SSL is enabled.
+ */
+internal val FirebaseDataConnect.grpcTestInfo: DataConnectGrpcRPCs.TestInfo
+  get() = (this as FirebaseDataConnectInternal).grpcRPCs.testInfo
+
+/**
+ * Creates a gRPC [ManagedChannel] that connects to the same remote host as the given
+ * [FirebaseDataConnect] instance, and registers the [ManagedChannel] for automatic cleanup after
+ * the test completes.
+ *
+ * This is useful for integration tests that need to interact with the gRPC layer directly, for
+ * example to verify interceptors or low-level communication.
+ *
+ * @param dataConnect The [FirebaseDataConnect] instance whose gRPC configuration (host, SSL, etc.)
+ * to use to configure the new [ManagedChannel].
+ * @return A new [ManagedChannel] configured to connect to the same backend as [dataConnect].
+ */
+internal fun DataConnectIntegrationTestBase.createGrpcManagedChannel(
+  dataConnect: FirebaseDataConnect
+): ManagedChannel {
+  val (context, host, sslEnabled) = dataConnect.grpcTestInfo
+
+  val managedChannel =
+    DataConnectGrpcRPCs.createGrpcManagedChannel(
+      context,
+      host,
+      sslEnabled,
+      Dispatchers.IO.asExecutor(),
+      mockk<Logger>(name = "TestGrpcManagedChannel", relaxed = true),
+    )
+
+  cleanups.register {
+    managedChannel.shutdownNow()
+    managedChannel.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS)
+  }
+
+  return managedChannel
+}

firebase-dataconnect/src/androidTest/kotlin/com/google/firebase/dataconnect/testutil/schemas/RealtimeConnector.kt

@@ -18,13 +18,18 @@ package com.google.firebase.dataconnect.testutil.schemas
 
 import com.google.firebase.dataconnect.ConnectorConfig
 import com.google.firebase.dataconnect.FirebaseDataConnect
+import com.google.firebase.dataconnect.core.FirebaseDataConnectInternal
 import com.google.firebase.dataconnect.serializers.UUIDSerializer
 import com.google.firebase.dataconnect.testutil.TestDataConnectFactory
 import java.util.UUID
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.serializer
 
-class RealtimeConnector private constructor(val dataConnect: FirebaseDataConnect) {
+class RealtimeConnector private constructor(dataConnectInternal: FirebaseDataConnectInternal) {
+
+  val dataConnect: FirebaseDataConnect = dataConnectInternal
+
+  val resourceName: String = dataConnectInternal.connectorResourceName
 
   val getStringByKey = GetStringByKeyQuery(this)
 
@@ -34,19 +39,19 @@ class RealtimeConnector private constructor(val dataConnect: FirebaseDataConnect
 
   val deleteString = DeleteStringMutation(this)
 
-  suspend fun getString(key: Key) = getStringByKey.execute(key)
+  suspend fun getString(key: Key): GetStringByKeyQuery.Data.Item? = getStringByKey.execute(key)
 
-  suspend fun insertString(name: String) = insertString.execute(name)
+  suspend fun insertString(name: String): Key = insertString.execute(name)
 
-  suspend fun updateString(key: Key, name: String) = updateString.execute(key, name)
+  suspend fun updateString(key: Key, name: String): Unit = updateString.execute(key, name)
 
-  suspend fun deleteString(key: Key) = deleteString.execute(key)
+  suspend fun deleteString(key: Key): Unit = deleteString.execute(key)
 
   class GetStringByKeyQuery(val connector: RealtimeConnector) {
 
-    suspend fun execute(key: Key) = execute(Variables(key))
+    suspend fun execute(key: Key): Data.Item? = execute(Variables(key))
 
-    suspend fun execute(variables: Variables) = queryRef(variables).execute().data.item
+    suspend fun execute(variables: Variables): Data.Item? = queryRef(variables).execute().data.item
 
     fun queryRef(variables: Variables) =
       connector.dataConnect.query(OPERATION_NAME, variables, serializer<Data>(), serializer())
@@ -67,9 +72,9 @@ class RealtimeConnector private constructor(val dataConnect: FirebaseDataConnect
     @Serializable data class Variables(val name: String)
     @Serializable data class Data(val key: Key)
 
-    suspend fun execute(name: String) = execute(Variables(name = name))
+    suspend fun execute(name: String): Key = execute(Variables(name = name))
 
-    suspend fun execute(variables: Variables) = mutationRef(variables).execute().data.key
+    suspend fun execute(variables: Variables): Key = mutationRef(variables).execute().data.key
 
     fun mutationRef(variables: Variables) =
       connector.dataConnect.mutation(OPERATION_NAME, variables, serializer<Data>(), serializer())
@@ -82,7 +87,7 @@ class RealtimeConnector private constructor(val dataConnect: FirebaseDataConnect
   class UpdateStringMutation(val connector: RealtimeConnector) {
     @Serializable data class Variables(val key: Key, val name: String)
 
-    suspend fun execute(key: Key, name: String) = execute(Variables(key = key, name = name))
+    suspend fun execute(key: Key, name: String): Unit = execute(Variables(key = key, name = name))
 
     suspend fun execute(variables: Variables) {
       mutationRef(variables).execute()
@@ -99,7 +104,7 @@ class RealtimeConnector private constructor(val dataConnect: FirebaseDataConnect
   class DeleteStringMutation(val connector: RealtimeConnector) {
     @Serializable data class Variables(val key: Key)
 
-    suspend fun execute(key: Key) = execute(Variables(key = key))
+    suspend fun execute(key: Key): Unit = execute(Variables(key = key))
 
     suspend fun execute(variables: Variables) {
       mutationRef(variables).execute()
@@ -125,7 +130,7 @@ class RealtimeConnector private constructor(val dataConnect: FirebaseDataConnect
 
     fun getInstance(dataConnectFactory: TestDataConnectFactory): RealtimeConnector {
       val dataConnect = dataConnectFactory.newInstance(config)
-      return RealtimeConnector(dataConnect)
+      return RealtimeConnector(dataConnect as FirebaseDataConnectInternal)
     }
   }
 }