feat: support using client credentials with emulator

google/cloud/firestore_v1/base_client.py

@@ -148,7 +148,7 @@ def _firestore_api_helper(self, transport, client_class, client_module) -> Any:
             # We need this in order to set appropriate keepalive options.
 
             if self._emulator_host is not None:
-                channel = grpc.insecure_channel(self._emulator_host)
+                channel = self._emulator_channel()
             else:
                 channel = transport.create_channel(
                     self._target,
@@ -165,6 +165,48 @@ def _firestore_api_helper(self, transport, client_class, client_module) -> Any:
 
         return self._firestore_api_internal
 
+    def _emulator_channel(self):
+        """
+        Creates a channel using self._credentials in a similar way to grpc.secure_channel but
+        using grpc.local_channel_credentials() rather than grpc.ssh_channel_credentials() to allow easy connection
+        to a local firestore emulator. This allows local testing of firestore rules if the credentials have been
+        created from a signed custom token.
+
+        :return: grcp.Channel
+        """
+        return grpc._channel.Channel(
+            self._emulator_host,
+            (),
+            self._local_composite_credentials()._credentials,
+            None,
+        )
+
+    def _local_composite_credentials(self):
+        """
+        Ceates the credentials for the local emulator channel
+        :return: grpc.ChannelCredentials
+        """
+        credentials = google.auth.credentials.with_scopes_if_required(
+            self._credentials, None
+        )
+        request = google.auth.transport.requests.Request()
+
+        # Create the metadata plugin for inserting the authorization header.
+        metadata_plugin = google.auth.transport.grpc.AuthMetadataPlugin(
+            credentials, request
+        )
+
+        # Create a set of grpc.CallCredentials using the metadata plugin.
+        google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
+
+        # Using the local_credentials to allow connection to emulator
+        local_credentials = grpc.local_channel_credentials()
+
+        # Combine the local credentials and the authorization credentials.
+        return grpc.composite_channel_credentials(
+            local_credentials, google_auth_credentials
+        )
+
     def _target_helper(self, client_class) -> str:
         """Return the target (where the API is).
         Eg. "firestore.googleapis.com"

tests/unit/v1/test_base_client.py

@@ -14,6 +14,7 @@
 
 import datetime
 import unittest
+import grpc
 
 import mock
 
@@ -67,10 +68,11 @@ def test__firestore_api_property(self, mock_channel, mock_client):
         return_value=mock.sentinel.firestore_api,
     )
     @mock.patch(
-        "grpc.insecure_channel", autospec=True,
+        "google.cloud.firestore_v1.base_client.BaseClient._emulator_channel",
+        autospec=True,
     )
     def test__firestore_api_property_with_emulator(
-        self, mock_insecure_channel, mock_client
+        self, mock_emulator_channel, mock_client
     ):
         emulator_host = "localhost:8081"
         with mock.patch("os.getenv") as getenv:
@@ -82,7 +84,7 @@ def test__firestore_api_property_with_emulator(
         self.assertIs(firestore_api, mock_client.return_value)
         self.assertIs(firestore_api, client._firestore_api_internal)
 
-        mock_insecure_channel.assert_called_once_with(emulator_host)
+        mock_emulator_channel.assert_called_once()
 
         # Call again to show that it is cached, but call count is still 1.
         self.assertIs(client._firestore_api, mock_client.return_value)
@@ -135,6 +137,36 @@ def test__rpc_metadata_property_with_emulator(self):
             ],
         )
 
+    def test_emulator_channel(self):
+        emulator_host = "localhost:8081"
+        with mock.patch("os.getenv") as getenv:
+            getenv.return_value = emulator_host
+
+            credentials = _make_credentials()
+            database = "quanta"
+            client = self._make_one(
+                project=self.PROJECT, credentials=credentials, database=database
+            )
+
+        # checks that a channel is created
+        channel = client._emulator_channel()
+        self.assertTrue(isinstance(channel, grpc._channel.Channel))
+        # checks that the credentials are composite ones using a local channel from grpc
+        composite_credentials = client._local_composite_credentials()
+        self.assertTrue(isinstance(composite_credentials, grpc.ChannelCredentials))
+        self.assertTrue(
+            isinstance(
+                composite_credentials._credentials._call_credentialses[0],
+                grpc._cython.cygrpc.MetadataPluginCallCredentials,
+            )
+        )
+        self.assertTrue(
+            isinstance(
+                composite_credentials._credentials._channel_credentials,
+                grpc._cython.cygrpc.LocalChannelCredentials,
+            )
+        )
+
     def test_field_path(self):
         klass = self._get_target_class()
         self.assertEqual(klass.field_path("a", "b", "c"), "a.b.c")