test: add unit tests for default_scopes

Author: busunkim96

tests/asyncio/test_grpc_helpers_async.py

@@ -317,6 +317,25 @@ def test_create_channel_implicit_with_scopes(
     grpc_secure_channel.assert_called_once_with(target, composite_creds)
 
 
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch(
+    "google.auth.default",
+    return_value=(mock.sentinel.credentials, mock.sentinel.projet),
+)
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_implicit_with_default_scopes(
+    grpc_secure_channel, default, composite_creds_call
+):
+    target = "example.com:443"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers_async.create_channel(target, scopes=["one", "two"], default_scopes=["three", "four"])
+
+    assert channel is grpc_secure_channel.return_value
+    default.assert_called_once_with(scopes=["one", "two"], default_scopes=["three", "four"])
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
 def test_create_channel_explicit_with_duplicate_credentials():
     target = "example:443"
 
@@ -363,6 +382,27 @@ def test_create_channel_explicit_scoped(grpc_secure_channel, composite_creds_cal
     grpc_secure_channel.assert_called_once_with(target, composite_creds)
 
 
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch("grpc.experimental.aio.secure_channel")
+def test_create_channel_explicit_default_scopes(grpc_secure_channel, composite_creds_call):
+    target = "example.com:443"
+    scopes = ["1", "2"]
+    default_scopes = ["3", "4"]
+    composite_creds = composite_creds_call.return_value
+
+    credentials = mock.create_autospec(google.auth.credentials.Scoped, instance=True)
+    credentials.requires_scopes = True
+
+    channel = grpc_helpers_async.create_channel(
+        target, credentials=credentials, scopes=scopes, default_scopes=default_scopes
+    )
+
+    credentials.with_scopes.assert_called_once_with(scopes, default_scopes=default_scopes)
+    assert channel is grpc_secure_channel.return_value
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+
 @mock.patch("grpc.composite_channel_credentials")
 @mock.patch("grpc.experimental.aio.secure_channel")
 def test_create_channel_explicit_with_quota_project(grpc_secure_channel, composite_creds_call):
@@ -423,6 +463,29 @@ def test_create_channel_with_credentials_file_and_scopes(load_credentials_from_f
     grpc_secure_channel.assert_called_once_with(target, composite_creds)
 
 
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch("grpc.experimental.aio.secure_channel")
+@mock.patch(
+    "google.auth.load_credentials_from_file",
+    return_value=(mock.sentinel.credentials, mock.sentinel.project)
+)
+def test_create_channel_with_credentials_file_and_default_scopes(load_credentials_from_file, grpc_secure_channel, composite_creds_call):
+    target = "example.com:443"
+    scopes = ["1", "2"]
+    default_scopes = ["3", "4"]
+
+    credentials_file = "/path/to/credentials/file.json"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers_async.create_channel(
+        target, credentials_file=credentials_file, scopes=scopes, default_scopes=default_scopes
+    )
+
+    google.auth.load_credentials_from_file.assert_called_once_with(credentials_file, scopes=scopes, default_scopes=default_scopes)
+    assert channel is grpc_secure_channel.return_value
+    grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
 @pytest.mark.skipif(grpc_helpers_async.HAS_GRPC_GCP, reason="grpc_gcp module not available")
 @mock.patch("grpc.experimental.aio.secure_channel")
 def test_create_channel_without_grpc_gcp(grpc_secure_channel):

tests/unit/test_grpc_helpers.py

@@ -285,6 +285,28 @@ def test_create_channel_implicit_with_scopes(
         grpc_secure_channel.assert_called_once_with(target, composite_creds)
 
 
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch(
+    "google.auth.default",
+    return_value=(mock.sentinel.credentials, mock.sentinel.projet),
+)
+@mock.patch("grpc.secure_channel")
+def test_create_channel_implicit_with_default_scopes(
+    grpc_secure_channel, default, composite_creds_call
+):
+    target = "example.com:443"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers.create_channel(target, scopes=["one", "two"], default_scopes=["three", "four"])
+
+    assert channel is grpc_secure_channel.return_value
+    default.assert_called_once_with(scopes=["one", "two"], default_scopes=["three", "four"])
+    if grpc_helpers.HAS_GRPC_GCP:
+        grpc_secure_channel.assert_called_once_with(target, composite_creds, None)
+    else:
+        grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
 def test_create_channel_explicit_with_duplicate_credentials():
     target = "example.com:443"
 
@@ -333,6 +355,29 @@ def test_create_channel_explicit_scoped(grpc_secure_channel, composite_creds_cal
         grpc_secure_channel.assert_called_once_with(target, composite_creds, None)
     else:
         grpc_secure_channel.assert_called_once_with(target, composite_creds)
+   
+
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch("grpc.secure_channel")
+def test_create_channel_explicit_default_scopes(grpc_secure_channel, composite_creds_call):
+    target = "example.com:443"
+    scopes = ["1", "2"]
+    default_scopes = ["3", "4"]
+    composite_creds = composite_creds_call.return_value
+
+    credentials = mock.create_autospec(google.auth.credentials.Scoped, instance=True)
+    credentials.requires_scopes = True
+
+    channel = grpc_helpers.create_channel(
+        target, credentials=credentials, scopes=scopes, default_scopes=default_scopes
+    )
+
+    credentials.with_scopes.assert_called_once_with(scopes, default_scopes=default_scopes)
+    assert channel is grpc_secure_channel.return_value
+    if grpc_helpers.HAS_GRPC_GCP:
+        grpc_secure_channel.assert_called_once_with(target, composite_creds, None)
+    else:
+        grpc_secure_channel.assert_called_once_with(target, composite_creds)
 
 
 @mock.patch("grpc.composite_channel_credentials")
@@ -408,6 +453,33 @@ def test_create_channel_with_credentials_file_and_scopes(load_credentials_from_f
         grpc_secure_channel.assert_called_once_with(target, composite_creds)
 
 
+@mock.patch("grpc.composite_channel_credentials")
+@mock.patch("grpc.secure_channel")
+@mock.patch(
+    "google.auth.load_credentials_from_file",
+    return_value=(mock.sentinel.credentials, mock.sentinel.project)
+)
+def test_create_channel_with_credentials_file_and_default_scopes(load_credentials_from_file, grpc_secure_channel, composite_creds_call):
+    target = "example.com:443"
+    scopes = ["1", "2"]
+    default_scopes = ["3", "4"]
+
+    credentials_file = "/path/to/credentials/file.json"
+    composite_creds = composite_creds_call.return_value
+
+    channel = grpc_helpers.create_channel(
+        target, credentials_file=credentials_file, scopes=scopes, default_scopes=default_scopes
+    )
+
+    load_credentials_from_file.assert_called_once_with(credentials_file, scopes=scopes, default_scopes=default_scopes)
+    assert channel is grpc_secure_channel.return_value
+    if grpc_helpers.HAS_GRPC_GCP:
+        grpc_secure_channel.assert_called_once_with(target, composite_creds, None)
+    else:
+        grpc_secure_channel.assert_called_once_with(target, composite_creds)
+
+
+
 @pytest.mark.skipif(
     not grpc_helpers.HAS_GRPC_GCP, reason="grpc_gcp module not available"
 )