diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index a36df8b39599..b1b71b056658 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -26,7 +26,7 @@ def is_otel_capabilities_enabled( client_options: Optional[ClientOptions | dict[str, Any]] = None, - env_var: str = "GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", + env_var: str = "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", ) -> bool: """Checks if OTel capabilities are enabled and installed. diff --git a/packages/google-api-core/google/api_core/grpc_helpers.py b/packages/google-api-core/google/api_core/grpc_helpers.py index 263079e7d1f7..13f89c64a13f 100644 --- a/packages/google-api-core/google/api_core/grpc_helpers.py +++ b/packages/google-api-core/google/api_core/grpc_helpers.py @@ -17,7 +17,17 @@ import collections import functools import warnings -from typing import Generic, Iterator, Optional, TypeVar +from typing import ( + Callable, + Generic, + Iterator, + Optional, + Sequence, + TypeAlias, + TypeVar, + cast, + get_args, +) import google.auth import google.auth.credentials @@ -34,6 +44,23 @@ # denotes the proto response type for grpc calls P = TypeVar("P") +# Type alias representing any client-side gRPC interceptor +ClientInterceptor: TypeAlias = ( + grpc.UnaryUnaryClientInterceptor + | grpc.UnaryStreamClientInterceptor + | grpc.StreamUnaryClientInterceptor + | grpc.StreamStreamClientInterceptor +) + +# Runtime tuple of gRPC client interceptor base classes for isinstance checks +_CLIENT_INTERCEPTOR_CLASSES = get_args(ClientInterceptor) + +# Type alias representing a channel-wrapping callable +ChannelWrapperCallable: TypeAlias = Callable[[grpc.Channel], grpc.Channel] + +# Generic type alias representing any channel wrapper (interceptor or callable) +ChannelWrapper: TypeAlias = ClientInterceptor | ChannelWrapperCallable + def _patch_callable_name(callable_): """Fix-up gRPC callable attributes. @@ -419,6 +446,48 @@ def _modify_target_for_direct_path(target: str) -> str: return target +def apply_channel_wrappers( + channel: grpc.Channel, + wrappers: Sequence[ChannelWrapper] | None = None, +) -> grpc.Channel: + """Applies channel wrappers (client interceptors or channel-wrapping callables) to a gRPC channel. + + Executes in reverse order so the first wrapper in the sequence becomes the + outermost layer on outbound requests and the innermost layer on inbound responses. + + Args: + channel (grpc.Channel): The channel to wrap. + wrappers (Optional[Sequence[ChannelWrapper]]): + An optional sequence of client interceptors or channel-wrapping + callables to apply. + + Returns: + grpc.Channel: The wrapped channel, or the original channel if no + wrappers were provided. + + Raises: + TypeError: If an item in ``wrappers`` is neither a gRPC ClientInterceptor + nor a Callable[[Channel], Channel]. + """ + if not wrappers: + return channel + + modified_channel = channel + # Reverse the inputs to align with the behavior of grpc.create_channel(*interceptors) + for wrapper in reversed(list(wrappers)): + if isinstance(wrapper, _CLIENT_INTERCEPTOR_CLASSES): + modified_channel = grpc.intercept_channel(modified_channel, wrapper) + elif callable(wrapper): + wrapper_callable = cast(ChannelWrapperCallable, wrapper) + modified_channel = wrapper_callable(modified_channel) + else: + raise TypeError( + f"Expected ChannelWrapper (ClientInterceptor or Callable[[Channel], Channel]), got {type(wrapper).__name__}" + ) + + return modified_channel + + _MethodCall = collections.namedtuple( "_MethodCall", ("request", "timeout", "metadata", "credentials", "compression") ) diff --git a/packages/google-api-core/tests/unit/test_client_options.py b/packages/google-api-core/tests/unit/test_client_options.py index c15e83174ed4..49fb35e8ba2f 100644 --- a/packages/google-api-core/tests/unit/test_client_options.py +++ b/packages/google-api-core/tests/unit/test_client_options.py @@ -15,6 +15,7 @@ from re import match import pytest + from google.api_core import client_options from ..helpers import warn_deprecated_credentials_file diff --git a/packages/google-api-core/tests/unit/test_grpc_helpers.py b/packages/google-api-core/tests/unit/test_grpc_helpers.py index 69281d58109b..4367552651d7 100644 --- a/packages/google-api-core/tests/unit/test_grpc_helpers.py +++ b/packages/google-api-core/tests/unit/test_grpc_helpers.py @@ -932,3 +932,102 @@ def test_subscribe_unsubscribe(self): def test_close(self): channel = grpc_helpers.ChannelStub() assert channel.close() is None + + +@pytest.mark.parametrize("falsy_wrappers", [None, [], ()]) +def test_apply_channel_wrappers_passthrough(falsy_wrappers): + """Verify that falsy or empty wrapper sequences return the channel unmodified.""" + mock_base_channel = mock.Mock(name="base_channel") + result = grpc_helpers.apply_channel_wrappers(mock_base_channel, falsy_wrappers) + assert result is mock_base_channel + + +def test_apply_channel_wrappers_grpc_client_interceptors(): + """Verify that standard gRPC ClientInterceptor instances are applied via grpc.intercept_channel.""" + + class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor): + def intercept_unary_unary(self, continuation, client_call_details, request): + return continuation(client_call_details, request) + + class DummyStreamInterceptor(grpc.StreamStreamClientInterceptor): + def intercept_stream_stream( + self, continuation, client_call_details, request_iterator + ): + return continuation(client_call_details, request_iterator) + + interceptor1 = DummyUnaryInterceptor() + interceptor2 = DummyStreamInterceptor() + + mock_base_channel = mock.Mock(name="base_channel") + mock_chan_after_i2 = mock.Mock(name="chan_after_i2") + mock_chan_after_i1 = mock.Mock(name="chan_after_i1") + + with mock.patch( + "grpc.intercept_channel", + side_effect=[mock_chan_after_i2, mock_chan_after_i1], + ) as mock_intercept: + result = grpc_helpers.apply_channel_wrappers( + mock_base_channel, [interceptor1, interceptor2] + ) + + assert result is mock_chan_after_i1 + assert mock_intercept.call_count == 2 + # Executed in reverse order so interceptor1 is outermost + mock_intercept.assert_has_calls( + [ + mock.call(mock_base_channel, interceptor2), + mock.call(mock_chan_after_i2, interceptor1), + ] + ) + + +def test_apply_channel_wrappers_callables(): + """Verify that channel-wrapping callables Callable[[Channel], Channel] are invoked in sequence.""" + mock_base_channel = mock.Mock(name="base_channel") + mock_chan_1 = mock.Mock(name="chan_1") + mock_chan_2 = mock.Mock(name="chan_2") + + wrapper1 = mock.Mock(side_effect=lambda ch: mock_chan_2) + wrapper2 = mock.Mock(side_effect=lambda ch: mock_chan_1) + + result = grpc_helpers.apply_channel_wrappers( + mock_base_channel, [wrapper1, wrapper2] + ) + + assert result is mock_chan_2 + # Executed in reverse order: wrapper2 runs first on base channel, then wrapper1 + wrapper2.assert_called_once_with(mock_base_channel) + wrapper1.assert_called_once_with(mock_chan_1) + + +def test_apply_channel_wrappers_interspersed(): + """Verify that a mixed sequence of gRPC interceptors and channel wrapper callables are applied.""" + + class DummyUnaryInterceptor(grpc.UnaryUnaryClientInterceptor): + def intercept_unary_unary(self, continuation, client_call_details, request): + return continuation(client_call_details, request) + + interceptor = DummyUnaryInterceptor() + mock_base_channel = mock.Mock(name="base_channel") + mock_chan_after_wrapper = mock.Mock(name="chan_after_wrapper") + mock_chan_after_interceptor = mock.Mock(name="chan_after_interceptor") + + wrapper = mock.Mock(return_value=mock_chan_after_wrapper) + + with mock.patch( + "grpc.intercept_channel", return_value=mock_chan_after_interceptor + ) as mock_intercept: + result = grpc_helpers.apply_channel_wrappers( + mock_base_channel, [interceptor, wrapper] + ) + + assert result is mock_chan_after_interceptor + wrapper.assert_called_once_with(mock_base_channel) + mock_intercept.assert_called_once_with(mock_chan_after_wrapper, interceptor) + + +def test_apply_channel_wrappers_invalid_type_raises(): + """Verify that passing an invalid object that is neither an interceptor nor callable raises TypeError.""" + mock_base_channel = mock.Mock(name="base_channel") + with pytest.raises(TypeError, match="Expected ChannelWrapper"): + grpc_helpers.apply_channel_wrappers(mock_base_channel, [12345]) diff --git a/packages/google-api-core/tests/unit/test_observability.py b/packages/google-api-core/tests/unit/test_observability.py index fc63023aadcd..f0ebe0afc14d 100644 --- a/packages/google-api-core/tests/unit/test_observability.py +++ b/packages/google-api-core/tests/unit/test_observability.py @@ -15,17 +15,20 @@ import sys from unittest import mock +import pytest + from google.api_core import _observability +from google.api_core._feature_gating_helpers import FeatureGatingError from google.api_core.client_options import ClientOptions def test_is_otel_capabilities_enabled_disabled(monkeypatch): - monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "false") + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "false") assert not _observability.is_otel_capabilities_enabled() def test_is_otel_capabilities_enabled_otel_missing(monkeypatch): - monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true") + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true") # Simulate OTel not being installed by blocking imports monkeypatch.setitem(sys.modules, "opentelemetry.instrumentation.grpc", None) @@ -33,7 +36,7 @@ def test_is_otel_capabilities_enabled_otel_missing(monkeypatch): def test_is_otel_capabilities_enabled_otel_installed(monkeypatch): - monkeypatch.setenv("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED", "true") + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true") mock_otel = mock.Mock() mock_otel_grpc = mock_otel.instrumentation.grpc @@ -49,6 +52,41 @@ def test_is_otel_capabilities_enabled_otel_installed(monkeypatch): assert _observability.is_otel_capabilities_enabled() +def test_is_otel_capabilities_enabled_experimental_requires_env_var(monkeypatch): + """Proves that passing client_options with tracer_provider without the experimental + env var set to 'true' raises FeatureGatingError (Fail Fast). + """ + monkeypatch.delenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", raising=False) + options = ClientOptions(tracer_provider=object()) + + with pytest.raises( + FeatureGatingError, + match="requires GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", + ): + _observability.is_otel_capabilities_enabled(options) + + +def test_is_otel_capabilities_enabled_experimental_enabled_with_config(monkeypatch): + """Proves that when GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED=true and tracer_provider + is supplied via client_options, is_otel_capabilities_enabled returns True. + """ + monkeypatch.setenv("GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED", "true") + + mock_otel = mock.Mock() + mock_otel_grpc = mock_otel.instrumentation.grpc + + monkeypatch.setitem(sys.modules, "opentelemetry", mock_otel) + monkeypatch.setitem( + sys.modules, "opentelemetry.instrumentation", mock_otel.instrumentation + ) + monkeypatch.setitem( + sys.modules, "opentelemetry.instrumentation.grpc", mock_otel_grpc + ) + + options = ClientOptions(tracer_provider=object()) + assert _observability.is_otel_capabilities_enabled(options) + + def test_apply_otel_capabilities_to_channel_enabled_otel_installed(monkeypatch): mock_channel = mock.Mock() mock_intercepted_channel = mock.Mock() diff --git a/packages/google-api-core/tests/unit/test_path_template.py b/packages/google-api-core/tests/unit/test_path_template.py index fb67973549f7..f053fc952176 100644 --- a/packages/google-api-core/tests/unit/test_path_template.py +++ b/packages/google-api-core/tests/unit/test_path_template.py @@ -18,6 +18,7 @@ import pytest from google.api import auth_pb2 + from google.api_core import path_template