fix: disable always_use_jwt_access (#939)

Some APIs with Cloud Storage integrations have failing
samples when non-default scopes are provided. This PR disables the feature
globally for now while we investigate.

We can no longer rollback da119c7 as some libraries have been
released with the change and removing the kwarg would be a breaking
change.

See internal issue 192297181.

Author: busunkim96

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2

@@ -131,7 +131,7 @@ class {{ service.name }}Transport(abc.ABC):
         # If the credentials is service account credentials, then always try to use self signed JWT.
         if always_use_jwt_access and isinstance(credentials, service_account.Credentials) and hasattr(service_account.Credentials, "with_always_use_jwt_access"):
             credentials = credentials.with_always_use_jwt_access(True)
-        
+
         # Save the credentials.
         self._credentials = credentials
 

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2

@@ -55,6 +55,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
             quota_project_id: Optional[str] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+            always_use_jwt_access: Optional[bool] = False,
             ) -> None:
         """Instantiate the transport.
 
@@ -95,6 +96,8 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
           google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
@@ -150,7 +153,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
-            always_use_jwt_access=True,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:

gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2

@@ -100,6 +100,7 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
             client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
             quota_project_id=None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+            always_use_jwt_access: Optional[bool] = False,
             ) -> None:
         """Instantiate the transport.
 
@@ -141,6 +142,8 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -195,7 +198,7 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
-            always_use_jwt_access=True,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:

gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2

@@ -123,7 +123,22 @@ def test_{{ service.client_name|snake_case }}_service_account_always_use_jwt(cli
     with mock.patch.object(service_account.Credentials, 'with_always_use_jwt_access', create=True) as use_jwt:
         creds = service_account.Credentials(None, None, None)
         client = client_class(credentials=creds)
-        use_jwt.assert_called_with(True)
+        use_jwt.assert_not_called()
+
+
+@pytest.mark.parametrize("transport_class,transport_name", [
+    {% if 'grpc' in opts.transport %}
+    (transports.{{ service.grpc_transport_name }}, "grpc"),
+    (transports.{{ service.grpc_asyncio_transport_name }}, "grpc_asyncio"),
+    {% elif 'rest' in opts.transport %}
+    (transports.{{ service.rest_transport_name }}, "rest"),
+    {% endif %}
+])
+def test_{{ service.client_name|snake_case }}_service_account_always_use_jwt_true(transport_class, transport_name):
+    with mock.patch.object(service_account.Credentials, 'with_always_use_jwt_access', create=True) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        transport = transport_class(credentials=creds, always_use_jwt_access=True)
+        use_jwt.assert_called_once_with(True)
 
 
 @pytest.mark.parametrize("client_class", [

tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc.py

@@ -57,6 +57,7 @@ def __init__(self, *,
             client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
             quota_project_id: Optional[str] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+            always_use_jwt_access: Optional[bool] = False,
             ) -> None:
         """Instantiate the transport.
 
@@ -97,6 +98,8 @@ def __init__(self, *,
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
           google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
@@ -150,7 +153,7 @@ def __init__(self, *,
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
-            always_use_jwt_access=True,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:

tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py

@@ -102,6 +102,7 @@ def __init__(self, *,
             client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
             quota_project_id=None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+            always_use_jwt_access: Optional[bool] = False,
             ) -> None:
         """Instantiate the transport.
 
@@ -143,6 +144,8 @@ def __init__(self, *,
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -195,7 +198,7 @@ def __init__(self, *,
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
-            always_use_jwt_access=True,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:

tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py

@@ -113,7 +113,18 @@ def test_asset_service_client_service_account_always_use_jwt(client_class):
     with mock.patch.object(service_account.Credentials, 'with_always_use_jwt_access', create=True) as use_jwt:
         creds = service_account.Credentials(None, None, None)
         client = client_class(credentials=creds)
-        use_jwt.assert_called_with(True)
+        use_jwt.assert_not_called()
+
+
+@pytest.mark.parametrize("transport_class,transport_name", [
+    (transports.AssetServiceGrpcTransport, "grpc"),
+    (transports.AssetServiceGrpcAsyncIOTransport, "grpc_asyncio"),
+])
+def test_asset_service_client_service_account_always_use_jwt_true(transport_class, transport_name):
+    with mock.patch.object(service_account.Credentials, 'with_always_use_jwt_access', create=True) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        transport = transport_class(credentials=creds, always_use_jwt_access=True)
+        use_jwt.assert_called_once_with(True)
 
 
 @pytest.mark.parametrize("client_class", [

tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc.py

@@ -63,6 +63,7 @@ def __init__(self, *,
             client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
             quota_project_id: Optional[str] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+            always_use_jwt_access: Optional[bool] = False,
             ) -> None:
         """Instantiate the transport.
 
@@ -103,6 +104,8 @@ def __init__(self, *,
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
           google.auth.exceptions.MutualTLSChannelError: If mutual TLS transport
@@ -155,7 +158,7 @@ def __init__(self, *,
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
-            always_use_jwt_access=True,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:

tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py

@@ -108,6 +108,7 @@ def __init__(self, *,
             client_cert_source_for_mtls: Callable[[], Tuple[bytes, bytes]] = None,
             quota_project_id=None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
+            always_use_jwt_access: Optional[bool] = False,
             ) -> None:
         """Instantiate the transport.
 
@@ -149,6 +150,8 @@ def __init__(self, *,
                 API requests. If ``None``, then default info will be used.
                 Generally, you only need to set this if you're developing
                 your own client library.
+            always_use_jwt_access (Optional[bool]): Whether self signed JWT should
+                be used for service account credentials.
 
         Raises:
             google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport
@@ -200,7 +203,7 @@ def __init__(self, *,
             scopes=scopes,
             quota_project_id=quota_project_id,
             client_info=client_info,
-            always_use_jwt_access=True,
+            always_use_jwt_access=always_use_jwt_access,
         )
 
         if not self._grpc_channel:

tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py

@@ -105,7 +105,18 @@ def test_iam_credentials_client_service_account_always_use_jwt(client_class):
     with mock.patch.object(service_account.Credentials, 'with_always_use_jwt_access', create=True) as use_jwt:
         creds = service_account.Credentials(None, None, None)
         client = client_class(credentials=creds)
-        use_jwt.assert_called_with(True)
+        use_jwt.assert_not_called()
+
+
+@pytest.mark.parametrize("transport_class,transport_name", [
+    (transports.IAMCredentialsGrpcTransport, "grpc"),
+    (transports.IAMCredentialsGrpcAsyncIOTransport, "grpc_asyncio"),
+])
+def test_iam_credentials_client_service_account_always_use_jwt_true(transport_class, transport_name):
+    with mock.patch.object(service_account.Credentials, 'with_always_use_jwt_access', create=True) as use_jwt:
+        creds = service_account.Credentials(None, None, None)
+        transport = transport_class(credentials=creds, always_use_jwt_access=True)
+        use_jwt.assert_called_once_with(True)
 
 
 @pytest.mark.parametrize("client_class", [