test(generator): fix mTLS tests and template coverage with environment isolation mocks

Author: hebaalazzeh

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/_compat.py.j2

@@ -10,7 +10,7 @@ try:
         get_client_cert_source,
         read_environment_variables,
     )
-except ImportError:
+except ImportError:  # pragma: NO COVER
     from google.auth.transport import mtls  # type: ignore
 
     # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.

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

@@ -102,14 +102,20 @@ from google.iam.v1 import policy_pb2  # type: ignore
 
 
 @pytest.fixture(autouse=True)
-def mock_should_use_client_cert():
+def mock_should_use_client_cert(request):
     if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
-        with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=False):
+        if "mtls" in request.node.name or "client_cert" in request.node.name:
             yield
+        else:
+            with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=False):
+                yield
     else:
         yield
 
 
+
+
+
 CRED_INFO_JSON = {
     "credential_source": "/path/to/file",
     "credential_type": "service account credentials",
@@ -678,18 +684,22 @@ def test_{{ service.client_name|snake_case }}_mtls_env_auto(client_class, transp
         with mock.patch.object(transport_class, '__init__') as patched:
             with mock.patch("google.auth.transport.mtls.has_default_client_cert_source", return_value=False):
                 patched.return_value = None
-                client = client_class(transport=transport_name)
-                patched.assert_called_once_with(
-                    credentials=None,
-                    credentials_file=None,
-                    host=client._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=client._DEFAULT_UNIVERSE),
-                    scopes=None,
-                    client_cert_source_for_mtls=None,
-                    quota_project_id=None,
-                    client_info=transports.base.DEFAULT_CLIENT_INFO,
-                    always_use_jwt_access=True,
-                    api_audience=None,
-                )
+                if use_client_cert_env == "true":
+                    with pytest.raises(ValueError):
+                        client = client_class(transport=transport_name)
+                else:
+                    client = client_class(transport=transport_name)
+                    patched.assert_called_once_with(
+                        credentials=None,
+                        credentials_file=None,
+                        host=client._DEFAULT_ENDPOINT_TEMPLATE.format(UNIVERSE_DOMAIN=client._DEFAULT_UNIVERSE),
+                        scopes=None,
+                        client_cert_source_for_mtls=None,
+                        quota_project_id=None,
+                        client_info=transports.base.DEFAULT_CLIENT_INFO,
+                        always_use_jwt_access=True,
+                        api_audience=None,
+                    )
 
 
 def test_use_client_cert_effective():
@@ -766,15 +776,17 @@ def test_use_client_cert_effective():
     # The method should return False as the environment variable is set to an invalid value.
     if  hasattr(google.auth.transport.mtls, "should_use_client_cert"):
         with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "unsupported"}):
-            assert {{ service.client_name }}._use_client_cert_effective() is False
+            with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=False):
+                assert {{ service.client_name }}._use_client_cert_effective() is False
 
     # Test case 12: Test when `should_use_client_cert` is available and the
     # `GOOGLE_API_USE_CLIENT_CERTIFICATE` environment variable is unset. Also, 
     # the GOOGLE_API_CONFIG environment variable is unset.
     if  hasattr(google.auth.transport.mtls, "should_use_client_cert"):
         with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": ""}):
             with mock.patch.dict(os.environ, {"GOOGLE_API_CERTIFICATE_CONFIG": ""}):
-                assert {{ service.client_name }}._use_client_cert_effective() is False
+                with mock.patch("google.auth.transport.mtls.should_use_client_cert", return_value=False):
+                    assert {{ service.client_name }}._use_client_cert_effective() is False
 
 
 def test__get_client_cert_source():
@@ -807,23 +819,25 @@ def test_{{ service.client_name|snake_case }}_get_mtls_endpoint_and_cert_source(
 
     # Test the case GOOGLE_API_USE_CLIENT_CERTIFICATE is "true".
     with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}):
-        mock_api_endpoint = "foo"
-        options = client_options.ClientOptions(client_cert_source=mock_client_cert_source, api_endpoint=mock_api_endpoint)
-        api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source(options)
-        assert api_endpoint == mock_api_endpoint
-        assert cert_source == mock_client_cert_source
+        with mock.patch("google.auth.transport.mtls.should_use_client_cert", create=True, return_value=True):
+            mock_api_endpoint = "foo"
+            options = client_options.ClientOptions(client_cert_source=mock_client_cert_source, api_endpoint=mock_api_endpoint)
+            api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source(options)
+            assert api_endpoint == mock_api_endpoint
+            assert cert_source == mock_client_cert_source
 
     # Test the case GOOGLE_API_USE_CLIENT_CERTIFICATE is "false".
-    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false"}):
-        mock_client_cert_source = mock.Mock()
-        mock_api_endpoint = "foo"
-        options = client_options.ClientOptions(client_cert_source=mock_client_cert_source, api_endpoint=mock_api_endpoint)
-        api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source(options)
-        assert api_endpoint == mock_api_endpoint
-        assert cert_source is None
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "false", "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false"}):
+        with mock.patch("google.auth.transport.mtls.should_use_client_cert", create=True, return_value=False):
+            mock_client_cert_source = mock.Mock()
+            mock_api_endpoint = "foo"
+            options = client_options.ClientOptions(client_cert_source=mock_client_cert_source, api_endpoint=mock_api_endpoint)
+            api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source(options)
+            assert api_endpoint == mock_api_endpoint
+            assert cert_source is None
 
     # Test the case GOOGLE_API_USE_CLIENT_CERTIFICATE is "Unsupported".
-    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "Unsupported"}):
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "Unsupported", "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false"}):
         if hasattr(google.auth.transport.mtls, "should_use_client_cert"):
             mock_client_cert_source = mock.Mock()
             mock_api_endpoint = "foo"
@@ -864,6 +878,7 @@ def test_{{ service.client_name|snake_case }}_get_mtls_endpoint_and_cert_source(
         for config_data, expected_cert_source in test_cases:
             env = os.environ.copy()
             env.pop("GOOGLE_API_USE_CLIENT_CERTIFICATE", None)
+            env.pop("CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE", None)
             with mock.patch.dict(os.environ, env, clear=True):
                     config_filename = "mock_certificate_config.json"
                     config_file_content = json.dumps(config_data)
@@ -911,6 +926,7 @@ def test_{{ service.client_name|snake_case }}_get_mtls_endpoint_and_cert_source(
         for config_data, expected_cert_source in test_cases:
             env = os.environ.copy()
             env.pop("GOOGLE_API_USE_CLIENT_CERTIFICATE", "")
+            env.pop("CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE", "")
             with mock.patch.dict(os.environ, env, clear=True):
                     config_filename = "mock_certificate_config.json"
                     config_file_content = json.dumps(config_data)
@@ -931,26 +947,26 @@ def test_{{ service.client_name|snake_case }}_get_mtls_endpoint_and_cert_source(
                             assert cert_source is expected_cert_source
 
     # Test the case GOOGLE_API_USE_MTLS_ENDPOINT is "never".
-    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}):
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never", "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false", "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false"}):
         api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source()
         assert api_endpoint == client_class.DEFAULT_ENDPOINT
         assert cert_source is None
 
     # Test the case GOOGLE_API_USE_MTLS_ENDPOINT is "always".
-    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}):
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always", "GOOGLE_API_USE_CLIENT_CERTIFICATE": "false", "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false"}):
         api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source()
         assert api_endpoint == client_class.DEFAULT_MTLS_ENDPOINT
         assert cert_source is None
 
     # Test the case GOOGLE_API_USE_MTLS_ENDPOINT is "auto" and default cert doesn't exist.
-    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}):
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true", "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false"}):
         with mock.patch('google.auth.transport.mtls.has_default_client_cert_source', return_value=False):
             api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source()
             assert api_endpoint == client_class.DEFAULT_ENDPOINT
             assert cert_source is None
 
     # Test the case GOOGLE_API_USE_MTLS_ENDPOINT is "auto" and default cert exists.
-    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true"}):
+    with mock.patch.dict(os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "true", "CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "false"}):
         with mock.patch('google.auth.transport.mtls.has_default_client_cert_source', return_value=True):
             with mock.patch('google.auth.transport.mtls.default_client_cert_source', return_value=mock_client_cert_source):
                 api_endpoint, cert_source = client_class.get_mtls_endpoint_and_cert_source()