feat(templates): remove pragmas from _compat.py and add test_compat.py unit test template for 100% coverage

Author: hebaalazzeh

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

@@ -19,7 +19,7 @@ try:
         get_api_endpoint,
         get_universe_domain,
     )
-except ImportError:  # pragma: NO COVER
+except ImportError:
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Universe domain support was introduced in google-api-core 2.18.0.
     # Remove these fallback definitions when google-api-core >= 2.18.0 becomes the minimum required version in generated client setup dependencies.
     def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
@@ -91,7 +91,7 @@ try:
         get_client_cert_source,
         read_environment_variables,
     )
-except ImportError:  # pragma: NO COVER
+except ImportError:
     from google.auth.transport import mtls  # type: ignore
 
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
@@ -144,7 +144,7 @@ except ImportError:  # pragma: NO COVER
 
 try:
     from google.api_core.gapic_v1.request import setup_request_id  # type: ignore
-except ImportError:  # pragma: NO COVER
+except ImportError:
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Request ID setup helper was introduced in google-api-core 2.26.0.
     # Remove this fallback definition when google-api-core >= 2.26.0 becomes the minimum required version in generated client setup dependencies.
     def setup_request_id(request, field_name: str, is_proto3_optional: bool):
@@ -194,17 +194,17 @@ try:
         flatten_query_params,
         transcode_request,
     )
-except ImportError:  # pragma: NO COVER
+except ImportError:
     # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
     from google.protobuf import json_format  # type: ignore
     from google.api_core import path_template  # type: ignore
 
-    def flatten_query_params(obj, strict=False):  # pragma: NO COVER
+    def flatten_query_params(obj, strict=False):
         if obj is not None and not isinstance(obj, dict):
             raise TypeError("flatten_query_params must be called with dict object")
         return _flatten(obj, key_path=[], strict=strict)
 
-    def _flatten(obj, key_path, strict=False):  # pragma: NO COVER
+    def _flatten(obj, key_path, strict=False):
         if obj is None:
             return []
         if isinstance(obj, dict):
@@ -213,40 +213,40 @@ except ImportError:  # pragma: NO COVER
             return _flatten_list(obj, key_path=key_path, strict=strict)
         return _flatten_value(obj, key_path=key_path, strict=strict)
 
-    def _is_primitive_value(obj):  # pragma: NO COVER
+    def _is_primitive_value(obj):
         if obj is None:
             return False
         if isinstance(obj, (list, dict)):
             raise ValueError("query params may not contain repeated dicts or lists")
         return True
 
-    def _flatten_value(obj, key_path, strict=False):  # pragma: NO COVER
+    def _flatten_value(obj, key_path, strict=False):
         return [(".".join(key_path), _canonicalize(obj, strict=strict))]
 
-    def _flatten_dict(obj, key_path, strict=False):  # pragma: NO COVER
+    def _flatten_dict(obj, key_path, strict=False):
         items = (
             _flatten(value, key_path=key_path + [key], strict=strict)
             for key, value in obj.items()
         )
         return functools.reduce(operator.concat, items, [])
 
-    def _flatten_list(elems, key_path, strict=False):  # pragma: NO COVER
+    def _flatten_list(elems, key_path, strict=False):
         items = (
             _flatten_value(elem, key_path=key_path, strict=strict)
             for elem in elems
             if _is_primitive_value(elem)
         )
         return functools.reduce(operator.concat, items, [])
 
-    def _canonicalize(obj, strict=False):  # pragma: NO COVER
+    def _canonicalize(obj, strict=False):
         if strict:
             value = str(obj)
             if isinstance(obj, bool):
                 value = value.lower()
             return value
         return obj
 
-    def transcode_request(  # pragma: NO COVER
+    def transcode_request(
         http_options: List[Dict[str, str]],
         request: Any,
         required_fields_default_values: Optional[Dict[str, Any]] = None,

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

@@ -0,0 +1,102 @@
+# {% include '_license.j2' %}
+
+import pytest
+import os
+from google.auth.exceptions import MutualTLSChannelError
+
+{% set package_path = api.naming.module_namespace|join('.') + "." + api.naming.versioned_module_name %}
+from {{package_path}} import _compat
+
+
+def test_get_default_mtls_endpoint():
+    assert _compat.get_default_mtls_endpoint(None) is None
+    assert _compat.get_default_mtls_endpoint("foo.googleapis.com") == "foo.mtls.googleapis.com"
+    assert _compat.get_default_mtls_endpoint("foo.sandbox.googleapis.com") == "foo.mtls.sandbox.googleapis.com"
+    assert _compat.get_default_mtls_endpoint("foo.mtls.googleapis.com") == "foo.mtls.googleapis.com"
+    assert _compat.get_default_mtls_endpoint("invalid") == "invalid"
+
+
+def test_get_api_endpoint():
+    assert _compat.get_api_endpoint("https://override.com", None, "googleapis.com", "auto", "googleapis.com", "mtls.com", "https://{UNIVERSE_DOMAIN}") == "https://override.com"
+    assert _compat.get_api_endpoint(None, lambda: (b"", b""), "googleapis.com", "always", "googleapis.com", "mtls.com", "https://{UNIVERSE_DOMAIN}") == "mtls.com"
+    with pytest.raises(MutualTLSChannelError):
+        _compat.get_api_endpoint(None, None, "otheruniverse.com", "always", "googleapis.com", "mtls.com", "https://{UNIVERSE_DOMAIN}")
+    assert _compat.get_api_endpoint(None, None, "googleapis.com", "never", "googleapis.com", "mtls.com", "https://{UNIVERSE_DOMAIN}") == "https://googleapis.com"
+
+
+def test_get_universe_domain():
+    assert _compat.get_universe_domain("custom.com", None, "googleapis.com") == "custom.com"
+    assert _compat.get_universe_domain(None, "env.com", "googleapis.com") == "env.com"
+    assert _compat.get_universe_domain(None, None, "googleapis.com") == "googleapis.com"
+    with pytest.raises(ValueError):
+        _compat.get_universe_domain("   ", None, "googleapis.com")
+
+
+def test_use_client_cert_effective(monkeypatch):
+    monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "true")
+    assert _compat.use_client_cert_effective() is True
+    monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "false")
+    assert _compat.use_client_cert_effective() is False
+    monkeypatch.setenv("GOOGLE_API_USE_CLIENT_CERTIFICATE", "invalid")
+    with pytest.raises(ValueError):
+        _compat.use_client_cert_effective()
+
+
+def test_get_client_cert_source():
+    cert_fn = lambda: (b"cert", b"key")
+    assert _compat.get_client_cert_source(cert_fn, True) == cert_fn
+    assert _compat.get_client_cert_source(None, False) is None
+    with pytest.raises(ValueError):
+        _compat.get_client_cert_source(None, True)
+
+
+def test_read_environment_variables(monkeypatch):
+    monkeypatch.setenv("GOOGLE_API_USE_MTLS_ENDPOINT", "always")
+    monkeypatch.setenv("GOOGLE_CLOUD_UNIVERSE_DOMAIN", "myuniverse.com")
+    use_cert, use_mtls, universe = _compat.read_environment_variables()
+    assert use_mtls == "always"
+    assert universe == "myuniverse.com"
+
+    monkeypatch.setenv("GOOGLE_API_USE_MTLS_ENDPOINT", "invalid")
+    with pytest.raises(MutualTLSChannelError):
+        _compat.read_environment_variables()
+
+
+def test_setup_request_id():
+    # Test with None request
+    _compat.setup_request_id(None, "request_id", False)
+
+    # Test with dict request (proto3 optional and non-optional)
+    d1 = {}
+    _compat.setup_request_id(d1, "request_id", is_proto3_optional=True)
+    assert "request_id" in d1
+
+    d2 = {}
+    _compat.setup_request_id(d2, "request_id", is_proto3_optional=False)
+    assert "request_id" in d2
+
+    # Test when field already present
+    d3 = {"request_id": "existing"}
+    _compat.setup_request_id(d3, "request_id", is_proto3_optional=False)
+    assert d3["request_id"] == "existing"
+
+    # Test with object
+    class Dummy:
+        request_id = None
+    obj = Dummy()
+    _compat.setup_request_id(obj, "request_id", is_proto3_optional=False)
+    assert obj.request_id is not None
+
+
+def test_flatten_query_params():
+    assert _compat.flatten_query_params(None) == []
+    res = _compat.flatten_query_params({"a": "val1", "b": [1, 2]})
+    assert ("a", "val1") in res
+    assert ("b", 1) in res
+    assert ("b", 2) in res
+
+    with pytest.raises(TypeError):
+        _compat.flatten_query_params("invalid")
+
+    with pytest.raises(ValueError):
+        _compat.flatten_query_params({"a": [{"nested": "dict"}]})

packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/_compat.py

@@ -32,7 +32,7 @@
         get_api_endpoint,
         get_universe_domain,
     )
-except ImportError:  # pragma: NO COVER
+except ImportError:
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Universe domain support was introduced in google-api-core 2.18.0.
     # Remove these fallback definitions when google-api-core >= 2.18.0 becomes the minimum required version in generated client setup dependencies.
     def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]:
@@ -104,7 +104,7 @@ def get_universe_domain(
         get_client_cert_source,
         read_environment_variables,
     )
-except ImportError:  # pragma: NO COVER
+except ImportError:
     from google.auth.transport import mtls  # type: ignore
 
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
@@ -157,7 +157,7 @@ def read_environment_variables() -> Tuple[bool, str, Optional[str]]:
 
 try:
     from google.api_core.gapic_v1.request import setup_request_id  # type: ignore
-except ImportError:  # pragma: NO COVER
+except ImportError:
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17813): Request ID setup helper was introduced in google-api-core 2.26.0.
     # Remove this fallback definition when google-api-core >= 2.26.0 becomes the minimum required version in generated client setup dependencies.
     def setup_request_id(request, field_name: str, is_proto3_optional: bool):
@@ -207,17 +207,17 @@ def setup_request_id(request, field_name: str, is_proto3_optional: bool):
         flatten_query_params,
         transcode_request,
     )
-except ImportError:  # pragma: NO COVER
+except ImportError:
     # TODO: Remove these fallbacks when google-api-core >= 2.18.0 is the minimum required version.
     from google.protobuf import json_format  # type: ignore
     from google.api_core import path_template  # type: ignore
 
-    def flatten_query_params(obj, strict=False):  # pragma: NO COVER
+    def flatten_query_params(obj, strict=False):
         if obj is not None and not isinstance(obj, dict):
             raise TypeError("flatten_query_params must be called with dict object")
         return _flatten(obj, key_path=[], strict=strict)
 
-    def _flatten(obj, key_path, strict=False):  # pragma: NO COVER
+    def _flatten(obj, key_path, strict=False):
         if obj is None:
             return []
         if isinstance(obj, dict):
@@ -226,40 +226,40 @@ def _flatten(obj, key_path, strict=False):  # pragma: NO COVER
             return _flatten_list(obj, key_path=key_path, strict=strict)
         return _flatten_value(obj, key_path=key_path, strict=strict)
 
-    def _is_primitive_value(obj):  # pragma: NO COVER
+    def _is_primitive_value(obj):
         if obj is None:
             return False
         if isinstance(obj, (list, dict)):
             raise ValueError("query params may not contain repeated dicts or lists")
         return True
 
-    def _flatten_value(obj, key_path, strict=False):  # pragma: NO COVER
+    def _flatten_value(obj, key_path, strict=False):
         return [(".".join(key_path), _canonicalize(obj, strict=strict))]
 
-    def _flatten_dict(obj, key_path, strict=False):  # pragma: NO COVER
+    def _flatten_dict(obj, key_path, strict=False):
         items = (
             _flatten(value, key_path=key_path + [key], strict=strict)
             for key, value in obj.items()
         )
         return functools.reduce(operator.concat, items, [])
 
-    def _flatten_list(elems, key_path, strict=False):  # pragma: NO COVER
+    def _flatten_list(elems, key_path, strict=False):
         items = (
             _flatten_value(elem, key_path=key_path, strict=strict)
             for elem in elems
             if _is_primitive_value(elem)
         )
         return functools.reduce(operator.concat, items, [])
 
-    def _canonicalize(obj, strict=False):  # pragma: NO COVER
+    def _canonicalize(obj, strict=False):
         if strict:
             value = str(obj)
             if isinstance(obj, bool):
                 value = value.lower()
             return value
         return obj
 
-    def transcode_request(  # pragma: NO COVER
+    def transcode_request(
         http_options: List[Dict[str, str]],
         request: Any,
         required_fields_default_values: Optional[Dict[str, Any]] = None,