handle http:// while creating storage control client

Author: Mahalaxmibejugam

gcsfs/extended_gcsfs.py

@@ -8,6 +8,7 @@
 from enum import Enum
 from glob import has_magic
 
+import grpc
 from fsspec import asyn
 from fsspec.callbacks import NoOpCallback
 from google.api_core import exceptions as api_exceptions
@@ -188,7 +189,14 @@ async def _get_control_plane_client(self):
                 endpoint = self._location.split("://")[-1]
                 channel_kwargs["host"] = endpoint
 
-            channel = transport_cls.create_channel(**channel_kwargs)
+            if self._location and self._location.startswith("http://"):
+                host = channel_kwargs["host"]
+                channel = grpc.aio.insecure_channel(
+                    host, options=channel_kwargs.get("options")
+                )
+            else:
+                channel = transport_cls.create_channel(**channel_kwargs)
+
             transport = transport_cls(channel=channel)
             self._storage_control_client = storage_control_v2.StorageControlAsyncClient(
                 transport=transport

gcsfs/tests/test_extended_hns_gcsfs.py

@@ -2175,39 +2175,49 @@ async def test_get_control_plane_client_quota_project_id(
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    "endpoint_url, env_updates, expected_host",
+    "endpoint_url, env_updates, expected_host, expected_insecure",
     [
-        ("https://my-endpoint.com", {}, "my-endpoint.com"),
+        ("https://my-endpoint.com", {}, "my-endpoint.com", False),
         (
             None,
             {
                 "GOOGLE_CLOUD_UNIVERSE_DOMAIN": "apis-tpczero.goog",
                 "STORAGE_EMULATOR_HOST": "",
             },
             "storage.apis-tpczero.goog",
+            False,
         ),
         (
             None,
             {
                 "GOOGLE_CLOUD_UNIVERSE_DOMAIN": "apis-tpczero.goog",
-                "STORAGE_EMULATOR_HOST": "my-emulator.com",
+                "STORAGE_EMULATOR_HOST": "http://my-emulator.com",
             },
             "my-emulator.com",
+            True,
         ),
-        (None, {"STORAGE_EMULATOR_HOST": "my-emulator.com"}, "my-emulator.com"),
-        (None, {"STORAGE_EMULATOR_HOST": "http://my-emulator.com"}, "my-emulator.com"),
-        (None, {"STORAGE_EMULATOR_HOST": ""}, "storage.googleapis.com"),
+        (
+            None,
+            {"STORAGE_EMULATOR_HOST": "http://my-emulator.com"},
+            "my-emulator.com",
+            True,
+        ),
+        (
+            None,
+            {"STORAGE_EMULATOR_HOST": "https://my-emulator.com"},
+            "my-emulator.com",
+            False,
+        ),
+        (None, {"STORAGE_EMULATOR_HOST": ""}, "storage.googleapis.com", False),
     ],
 )
 async def test_get_control_plane_client_endpoint(
-    endpoint_url, env_updates, expected_host
+    endpoint_url, env_updates, expected_host, expected_insecure
 ):
     fs_kwargs = {"token": "anon"}
     if endpoint_url:
         fs_kwargs["endpoint_url"] = endpoint_url
 
-    fs = ExtendedGcsFileSystem(**fs_kwargs)
-
     mock_transport_cls = mock.Mock()
     mock_channel = mock.Mock()
     mock_transport_cls.create_channel.return_value = mock_channel
@@ -2220,16 +2230,22 @@ async def test_get_control_plane_client_endpoint(
             "get_transport_class",
             return_value=mock_transport_cls,
         ),
+        mock.patch("grpc.aio.insecure_channel") as mock_insecure_channel,
         mock.patch.dict(os.environ, env_updates),
     ):
-        # Clear cached client to force re-initialization
-        fs._storage_control_client = None
+        ExtendedGcsFileSystem.clear_instance_cache()
+        fs = ExtendedGcsFileSystem(**fs_kwargs)
 
         await fs._get_control_plane_client()
 
-        mock_transport_cls.create_channel.assert_called_once()
-        kwargs = mock_transport_cls.create_channel.call_args.kwargs
-        assert kwargs.get("host") == expected_host
+        if expected_insecure:
+            mock_insecure_channel.assert_called_once_with(
+                expected_host, options=mock.ANY
+            )
+        else:
+            mock_transport_cls.create_channel.assert_called_once()
+            kwargs = mock_transport_cls.create_channel.call_args.kwargs
+            assert kwargs.get("host") == expected_host
 
 
 def test_extended_gcsfs_retry_init():

pyproject.toml

@@ -73,4 +73,4 @@ log_cli_level = "DEBUG"
 
 [tool.isort]
 profile = "black"
-known_third_party = ["aiohttp", "click", "decorator", "fsspec", "fuse", "google", "google_auth_oauthlib", "numpy", "prettytable", "psutil", "pytest", "pytest_asyncio", "requests", "resource_monitor", "yaml"]
+known_third_party = ["aiohttp", "click", "decorator", "fsspec", "fuse", "google", "google_auth_oauthlib", "grpc", "numpy", "prettytable", "psutil", "pytest", "pytest_asyncio", "requests", "resource_monitor", "yaml"]