From 362f9ca5c6ae5b42b9074962ec78a07d3dc98deb Mon Sep 17 00:00:00 2001 From: chalmer lowe Date: Mon, 14 Sep 2026 16:50:37 -0400 Subject: [PATCH] feat(gapic): add async gRPC OpenTelemetry channel tracing Add complete end-to-end support for asynchronous gRPC OpenTelemetry tracing: - packages/google-api-core: - Add coroutine-aware Tier 3 client method tracing in method_async.py - Add async gRPC channel interceptor support in grpc_helpers_async.py - Add tracing_enabled option and async interceptor factory in _observability.py / client_options.py - Add comprehensive unit tests in test_method_async.py and test_grpc_helpers_async.py - packages/gapic-generator: - Wire async OpenTelemetry interceptor into generated GrpcAsyncIOTransport and client - Update unit test template to mock async interceptors - Regenerate and verify all 8 Bazel integration test goldens - Add async gRPC tracing test suite in tests/system/test_tracing.py --- .../%sub/services/%service/client.py.j2 | 12 + .../%service/transports/grpc_asyncio.py.j2 | 10 + .../%name_%version/%sub/test_%service.py.j2 | 90 ++++++ packages/gapic-generator/noxfile.py | 3 + .../asset_v1/services/asset_service/client.py | 12 + .../asset_service/transports/grpc_asyncio.py | 10 + .../unit/gapic/asset_v1/test_asset_service.py | 89 ++++++ .../services/iam_credentials/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../credentials_v1/test_iam_credentials.py | 89 ++++++ .../eventarc_v1/services/eventarc/client.py | 12 + .../eventarc/transports/grpc_asyncio.py | 10 + .../unit/gapic/eventarc_v1/test_eventarc.py | 89 ++++++ .../services/config_service_v2/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../services/logging_service_v2/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../services/metrics_service_v2/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../logging_v2/test_config_service_v2.py | 89 ++++++ .../logging_v2/test_logging_service_v2.py | 89 ++++++ .../logging_v2/test_metrics_service_v2.py | 89 ++++++ .../services/config_service_v2/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../services/logging_service_v2/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../services/metrics_service_v2/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../logging_v2/test_config_service_v2.py | 89 ++++++ .../logging_v2/test_logging_service_v2.py | 89 ++++++ .../logging_v2/test_metrics_service_v2.py | 89 ++++++ .../redis_v1/services/cloud_redis/client.py | 12 + .../cloud_redis/transports/grpc_asyncio.py | 10 + .../unit/gapic/redis_v1/test_cloud_redis.py | 89 ++++++ .../redis_v1/services/cloud_redis/client.py | 12 + .../cloud_redis/transports/grpc_asyncio.py | 10 + .../unit/gapic/redis_v1/test_cloud_redis.py | 89 ++++++ .../storage_batch_operations/client.py | 12 + .../transports/grpc_asyncio.py | 10 + .../test_storage_batch_operations.py | 89 ++++++ .../gapic-generator/tests/system/conftest.py | 26 +- .../tests/system/test_tracing.py | 225 ++++++++++++++- .../google/api_core/_observability.py | 27 +- .../google/api_core/client_options.py | 2 + .../google/api_core/gapic_v1/method_async.py | 185 ++++++++++++- .../google/api_core/grpc_helpers_async.py | 86 +++++- .../tests/asyncio/gapic/test_method_async.py | 261 ++++++++++++++++++ .../tests/asyncio/test_grpc_helpers_async.py | 65 +++++ 48 files changed, 2309 insertions(+), 15 deletions(-) diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 index 157f014508dd..e704f7d07cff 100644 --- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 +++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 @@ -554,6 +554,18 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta): is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, {{ service.grpc_asyncio_transport_name }}) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) {% endif %} # initialize with the provided callable or the passed in class diff --git a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2 b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2 index 7b8a885d227c..10f4fb823623 100644 --- a/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2 +++ b/packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2 @@ -171,6 +171,7 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport): client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -222,6 +223,8 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport): to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -300,6 +303,13 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport): ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 index ec499686a51d..bbc2ac5ea96b 100644 --- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 +++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 @@ -992,6 +992,96 @@ def test_{{ service.name|snake_case }}_grpc_transport_custom_channel_interceptor assert transport.grpc_channel == mock_custom_channel +def test_{{ service.async_client_name|snake_case }}_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join('.') }}.services.{{ service.name|snake_case }}.client._observability", + mock_obs, + ), + mock.patch.object( + transports.{{ service.grpc_asyncio_transport_name }}, "__init__", return_value=None + ) as patched_transport_init, + ): + client = {{ service.async_client_name }}() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_{{ service.async_client_name|snake_case }}_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join('.') }}.services.{{ service.name|snake_case }}.client._observability", + mock_obs, + ), + mock.patch.object( + transports.{{ service.grpc_asyncio_transport_name }}, "__init__", return_value=None + ) as patched_transport_init, + ): + client = {{ service.async_client_name }}() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_{{ service.name|snake_case }}_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.{{ service.grpc_asyncio_transport_name }}, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.{{ service.grpc_asyncio_transport_name }}( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_{{ service.name|snake_case }}_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.{{ service.grpc_asyncio_transport_name }}( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ ({{ service.client_name }}, transports.{{ service.grpc_transport_name }}, "grpc", grpc_helpers), ({{ service.async_client_name }}, transports.{{ service.grpc_asyncio_transport_name }}, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py index 2dc40ae96b05..74ddc242c356 100644 --- a/packages/gapic-generator/noxfile.py +++ b/packages/gapic-generator/noxfile.py @@ -412,6 +412,9 @@ def showcase( "opentelemetry-sdk", "opentelemetry-instrumentation-grpc", ) + local_core = Path(__file__).parent.parent / "google-api-core" + if local_core.exists() and (local_core / "setup.py").exists(): + session.install("-e", str(local_core)) test_directory = Path("tests", "system") ignore_file = env.get("IGNORE_FILE") pytest_command = [ diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py index 492935dd8e5a..9d94ad3c8a44 100755 --- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py +++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/client.py @@ -567,6 +567,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, AssetServiceGrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py index 8fb1179f2fde..c8c560dd7afc 100755 --- a/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/asset/google/cloud/asset_v1/services/asset_service/transports/grpc_asyncio.py @@ -179,6 +179,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -230,6 +231,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -306,6 +309,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py b/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py index 1b833561fbe0..004e9bbcca15 100755 --- a/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py +++ b/packages/gapic-generator/tests/integration/goldens/asset/tests/unit/gapic/asset_v1/test_asset_service.py @@ -848,6 +848,95 @@ def test_asset_service_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_asset_service_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.asset_v1.services.asset_service.client._observability", + mock_obs, + ), + mock.patch.object( + transports.AssetServiceGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = AssetServiceAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_asset_service_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.asset_v1.services.asset_service.client._observability", + mock_obs, + ), + mock.patch.object( + transports.AssetServiceGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = AssetServiceAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_asset_service_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.AssetServiceGrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.AssetServiceGrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_asset_service_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.AssetServiceGrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (AssetServiceClient, transports.AssetServiceGrpcTransport, "grpc", grpc_helpers), (AssetServiceAsyncClient, transports.AssetServiceGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py index 814806e76f0a..ac3bc1854d2d 100755 --- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py +++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/client.py @@ -504,6 +504,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, IAMCredentialsGrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py index d9d401f8d9f1..065a547a1ce7 100755 --- a/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/credentials/google/iam/credentials_v1/services/iam_credentials/transports/grpc_asyncio.py @@ -185,6 +185,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -236,6 +237,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -311,6 +314,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py b/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py index dfc140216554..4681e1186e70 100755 --- a/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py +++ b/packages/gapic-generator/tests/integration/goldens/credentials/tests/unit/gapic/credentials_v1/test_iam_credentials.py @@ -838,6 +838,95 @@ def test_iam_credentials_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_iam_credentials_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.iam.credentials_v1.services.iam_credentials.client._observability", + mock_obs, + ), + mock.patch.object( + transports.IAMCredentialsGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = IAMCredentialsAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_iam_credentials_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.iam.credentials_v1.services.iam_credentials.client._observability", + mock_obs, + ), + mock.patch.object( + transports.IAMCredentialsGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = IAMCredentialsAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_iam_credentials_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.IAMCredentialsGrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.IAMCredentialsGrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_iam_credentials_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.IAMCredentialsGrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (IAMCredentialsClient, transports.IAMCredentialsGrpcTransport, "grpc", grpc_helpers), (IAMCredentialsAsyncClient, transports.IAMCredentialsGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py index 1df357a593f2..3ae2bcbbc19b 100755 --- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py +++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/client.py @@ -687,6 +687,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, EventarcGrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/transports/grpc_asyncio.py index 966a52b3d9dd..f11bd21297d4 100755 --- a/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/eventarc/google/cloud/eventarc_v1/services/eventarc/transports/grpc_asyncio.py @@ -193,6 +193,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -244,6 +245,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -320,6 +323,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py b/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py index d1feb6c06e6b..f02caa1113c0 100755 --- a/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py +++ b/packages/gapic-generator/tests/integration/goldens/eventarc/tests/unit/gapic/eventarc_v1/test_eventarc.py @@ -869,6 +869,95 @@ def test_eventarc_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_eventarc_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.eventarc_v1.services.eventarc.client._observability", + mock_obs, + ), + mock.patch.object( + transports.EventarcGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = EventarcAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_eventarc_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.eventarc_v1.services.eventarc.client._observability", + mock_obs, + ), + mock.patch.object( + transports.EventarcGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = EventarcAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_eventarc_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.EventarcGrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.EventarcGrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_eventarc_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.EventarcGrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (EventarcClient, transports.EventarcGrpcTransport, "grpc", grpc_helpers), (EventarcAsyncClient, transports.EventarcGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py index de57e35ed2ac..673485a36caf 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/client.py @@ -560,6 +560,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, ConfigServiceV2GrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py index e49afb2aa807..b4c161421229 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py @@ -179,6 +179,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -230,6 +231,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -306,6 +309,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py index 9fefe7597513..2e0f345cfbf8 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/client.py @@ -491,6 +491,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, LoggingServiceV2GrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py index 8e816f748369..14dce2af3455 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py @@ -178,6 +178,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -229,6 +230,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -304,6 +307,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py index 0e683c58063f..d65112b84eb3 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/client.py @@ -492,6 +492,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, MetricsServiceV2GrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py index aaa422d2953e..bc2f6aa78efa 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py @@ -178,6 +178,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -229,6 +230,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -304,6 +307,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py index 94a39fe4f05c..d0046cdfcc12 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_config_service_v2.py @@ -824,6 +824,95 @@ def test_config_service_v2_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_config_service_v2_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.logging_v2.services.config_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.ConfigServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = ConfigServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_config_service_v2_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.logging_v2.services.config_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.ConfigServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = ConfigServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_config_service_v2_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.ConfigServiceV2GrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.ConfigServiceV2GrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_config_service_v2_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.ConfigServiceV2GrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (ConfigServiceV2Client, transports.ConfigServiceV2GrpcTransport, "grpc", grpc_helpers), (ConfigServiceV2AsyncClient, transports.ConfigServiceV2GrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py index ec398711b928..4a3997aa91d3 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_logging_service_v2.py @@ -825,6 +825,95 @@ def test_logging_service_v2_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_logging_service_v2_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.logging_v2.services.logging_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.LoggingServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = LoggingServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_logging_service_v2_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.logging_v2.services.logging_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.LoggingServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = LoggingServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_logging_service_v2_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.LoggingServiceV2GrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.LoggingServiceV2GrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_logging_service_v2_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.LoggingServiceV2GrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (LoggingServiceV2Client, transports.LoggingServiceV2GrpcTransport, "grpc", grpc_helpers), (LoggingServiceV2AsyncClient, transports.LoggingServiceV2GrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py index bc55c44d2a43..3319b2fba288 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py +++ b/packages/gapic-generator/tests/integration/goldens/logging/tests/unit/gapic/logging_v2/test_metrics_service_v2.py @@ -823,6 +823,95 @@ def test_metrics_service_v2_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_metrics_service_v2_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.logging_v2.services.metrics_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.MetricsServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = MetricsServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_metrics_service_v2_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.logging_v2.services.metrics_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.MetricsServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = MetricsServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_metrics_service_v2_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.MetricsServiceV2GrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.MetricsServiceV2GrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_metrics_service_v2_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.MetricsServiceV2GrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (MetricsServiceV2Client, transports.MetricsServiceV2GrpcTransport, "grpc", grpc_helpers), (MetricsServiceV2AsyncClient, transports.MetricsServiceV2GrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py index aefac0da88fb..3df782d1c79f 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/client.py @@ -560,6 +560,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, ConfigServiceV2GrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py index e49afb2aa807..b4c161421229 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/config_service_v2/transports/grpc_asyncio.py @@ -179,6 +179,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -230,6 +231,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -306,6 +309,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py index 9fefe7597513..2e0f345cfbf8 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/client.py @@ -491,6 +491,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, LoggingServiceV2GrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py index 8e816f748369..14dce2af3455 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/logging_service_v2/transports/grpc_asyncio.py @@ -178,6 +178,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -229,6 +230,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -304,6 +307,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py index c636aaca7e86..95a32916cc3a 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/client.py @@ -492,6 +492,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, MetricsServiceV2GrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py index aaa422d2953e..bc2f6aa78efa 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/google/cloud/logging_v2/services/metrics_service_v2/transports/grpc_asyncio.py @@ -178,6 +178,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -229,6 +230,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -304,6 +307,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py index e6de5df4ceaf..1f9dda98d1bf 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_config_service_v2.py @@ -824,6 +824,95 @@ def test_config_service_v2_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_base_config_service_v2_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.logging_v2.services.config_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.ConfigServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = BaseConfigServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_base_config_service_v2_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.logging_v2.services.config_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.ConfigServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = BaseConfigServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_config_service_v2_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.ConfigServiceV2GrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.ConfigServiceV2GrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_config_service_v2_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.ConfigServiceV2GrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (BaseConfigServiceV2Client, transports.ConfigServiceV2GrpcTransport, "grpc", grpc_helpers), (BaseConfigServiceV2AsyncClient, transports.ConfigServiceV2GrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py index ec398711b928..4a3997aa91d3 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_logging_service_v2.py @@ -825,6 +825,95 @@ def test_logging_service_v2_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_logging_service_v2_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.logging_v2.services.logging_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.LoggingServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = LoggingServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_logging_service_v2_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.logging_v2.services.logging_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.LoggingServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = LoggingServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_logging_service_v2_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.LoggingServiceV2GrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.LoggingServiceV2GrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_logging_service_v2_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.LoggingServiceV2GrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (LoggingServiceV2Client, transports.LoggingServiceV2GrpcTransport, "grpc", grpc_helpers), (LoggingServiceV2AsyncClient, transports.LoggingServiceV2GrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py index 59ceebba8a28..f61962ee2767 100755 --- a/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py +++ b/packages/gapic-generator/tests/integration/goldens/logging_internal/tests/unit/gapic/logging_v2/test_metrics_service_v2.py @@ -823,6 +823,95 @@ def test_metrics_service_v2_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_base_metrics_service_v2_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.logging_v2.services.metrics_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.MetricsServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = BaseMetricsServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_base_metrics_service_v2_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.logging_v2.services.metrics_service_v2.client._observability", + mock_obs, + ), + mock.patch.object( + transports.MetricsServiceV2GrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = BaseMetricsServiceV2AsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_metrics_service_v2_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.MetricsServiceV2GrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.MetricsServiceV2GrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_metrics_service_v2_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.MetricsServiceV2GrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (BaseMetricsServiceV2Client, transports.MetricsServiceV2GrpcTransport, "grpc", grpc_helpers), (BaseMetricsServiceV2AsyncClient, transports.MetricsServiceV2GrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py index e8f258ff21d3..ce7b13e4aaa0 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py +++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/client.py @@ -554,6 +554,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, CloudRedisGrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py index 110d71537636..5096a8656095 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py @@ -199,6 +199,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -250,6 +251,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -326,6 +329,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py index 632bd64909f4..37940bf96934 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py +++ b/packages/gapic-generator/tests/integration/goldens/redis/tests/unit/gapic/redis_v1/test_cloud_redis.py @@ -856,6 +856,95 @@ def test_cloud_redis_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_cloud_redis_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.redis_v1.services.cloud_redis.client._observability", + mock_obs, + ), + mock.patch.object( + transports.CloudRedisGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = CloudRedisAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_cloud_redis_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.redis_v1.services.cloud_redis.client._observability", + mock_obs, + ), + mock.patch.object( + transports.CloudRedisGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = CloudRedisAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_cloud_redis_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.CloudRedisGrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.CloudRedisGrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_cloud_redis_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.CloudRedisGrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (CloudRedisClient, transports.CloudRedisGrpcTransport, "grpc", grpc_helpers), (CloudRedisAsyncClient, transports.CloudRedisGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py index 828f6d48211e..3f2eb9fce900 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py +++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/client.py @@ -554,6 +554,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, CloudRedisGrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py index c7b03489475f..b86deba5095f 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/grpc_asyncio.py @@ -199,6 +199,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -250,6 +251,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -326,6 +329,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py index 9094b0af41d2..549996b5b11b 100755 --- a/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py +++ b/packages/gapic-generator/tests/integration/goldens/redis_selective/tests/unit/gapic/redis_v1/test_cloud_redis.py @@ -856,6 +856,95 @@ def test_cloud_redis_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_cloud_redis_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.redis_v1.services.cloud_redis.client._observability", + mock_obs, + ), + mock.patch.object( + transports.CloudRedisGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = CloudRedisAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_cloud_redis_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.redis_v1.services.cloud_redis.client._observability", + mock_obs, + ), + mock.patch.object( + transports.CloudRedisGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = CloudRedisAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_cloud_redis_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.CloudRedisGrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.CloudRedisGrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_cloud_redis_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.CloudRedisGrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (CloudRedisClient, transports.CloudRedisGrpcTransport, "grpc", grpc_helpers), (CloudRedisAsyncClient, transports.CloudRedisGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py index 448ac3f79873..99a446765550 100755 --- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py +++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/client.py @@ -528,6 +528,18 @@ def __init__(self, *, is not None ): interceptors.append(otel_interceptor) + elif ( + isinstance(transport_init, type) + and issubclass(transport_init, StorageBatchOperationsGrpcAsyncIOTransport) + and _observability is not None + and ( + otel_async_interceptors := _observability.get_otel_async_interceptor( + self._client_options + ) + ) + is not None + ): + interceptors.extend(otel_async_interceptors) # initialize with the provided callable or the passed in class transport_kwargs = { diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/transports/grpc_asyncio.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/transports/grpc_asyncio.py index 97a7a3213a3c..a68cfdcb1c5e 100755 --- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/transports/grpc_asyncio.py +++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/google/cloud/storagebatchoperations_v1/services/storage_batch_operations/transports/grpc_asyncio.py @@ -185,6 +185,7 @@ def __init__(self, *, client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO, always_use_jwt_access: Optional[bool] = False, api_audience: Optional[str] = None, + interceptors: Optional[Sequence[aio.ClientInterceptor]] = None, ) -> None: """Instantiate the transport. @@ -236,6 +237,8 @@ def __init__(self, *, to the service that will be set when using certain 3rd party authentication flows. Audience is typically a resource identifier. If not set, the host value will be used as a default. + interceptors (Optional[Sequence[aio.ClientInterceptor]]): + Additional interceptors to apply to the gRPC channel. Raises: google.auth.exceptions.MutualTlsChannelError: If mutual TLS transport @@ -312,6 +315,13 @@ def __init__(self, *, ], ) + apply_interceptors = getattr( + grpc_helpers_async, + "apply_channel_interceptors", + lambda channel, interceptors: channel, + ) + self._grpc_channel = apply_interceptors(self._grpc_channel, interceptors) + self._interceptor = _LoggingClientAIOInterceptor() self._grpc_channel._unary_unary_interceptors.append(self._interceptor) self._logged_channel = self._grpc_channel diff --git a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py index 2d51c66c5a73..36867307410c 100755 --- a/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py +++ b/packages/gapic-generator/tests/integration/goldens/storagebatchoperations/tests/unit/gapic/storagebatchoperations_v1/test_storage_batch_operations.py @@ -848,6 +848,95 @@ def test_storage_batch_operations_grpc_transport_custom_channel_interceptors(): assert transport.grpc_channel == mock_custom_channel +def test_storage_batch_operations_async_client_otel_channel_injection_enabled(): + mock_interceptor = mock.Mock() + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = [mock_interceptor] + with ( + mock.patch( + "google.cloud.storagebatchoperations_v1.services.storage_batch_operations.client._observability", + mock_obs, + ), + mock.patch.object( + transports.StorageBatchOperationsGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = StorageBatchOperationsAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert called_kwargs.get("interceptors") == [mock_interceptor] + + +def test_storage_batch_operations_async_client_otel_channel_injection_disabled(): + mock_obs = mock.Mock() + mock_obs.get_otel_async_interceptor.return_value = None + with ( + mock.patch( + "google.cloud.storagebatchoperations_v1.services.storage_batch_operations.client._observability", + mock_obs, + ), + mock.patch.object( + transports.StorageBatchOperationsGrpcAsyncIOTransport, "__init__", return_value=None + ) as patched_transport_init, + ): + client = StorageBatchOperationsAsyncClient() + + mock_obs.get_otel_async_interceptor.assert_called_once_with(client._client._client_options) + called_kwargs = patched_transport_init.call_args.kwargs + assert not called_kwargs.get("interceptors", []) + + +def test_storage_batch_operations_grpc_asyncio_transport_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_channel = mock.Mock() + + with ( + mock.patch.object( + transports.StorageBatchOperationsGrpcAsyncIOTransport, + "create_channel", + return_value=mock_channel, + ), + mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_channel, + create=True, + ) as mock_apply_interceptors, + ): + transport = transports.StorageBatchOperationsGrpcAsyncIOTransport( + credentials=ga_credentials.AnonymousCredentials(), + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_channel + + +def test_storage_batch_operations_grpc_asyncio_transport_custom_channel_interceptors(): + mock_interceptor = mock.Mock() + mock_custom_channel = mock.Mock(spec=aio.Channel) + mock_custom_channel._unary_unary_interceptors = [] + + with mock.patch.object( + grpc_helpers_async, + "apply_channel_interceptors", + return_value=mock_custom_channel, + create=True, + ) as mock_apply_interceptors: + transport = transports.StorageBatchOperationsGrpcAsyncIOTransport( + channel=mock_custom_channel, + interceptors=[mock_interceptor], + ) + + mock_apply_interceptors.assert_called_once_with( + mock_custom_channel, [mock_interceptor] + ) + assert transport.grpc_channel == mock_custom_channel + + @pytest.mark.parametrize("client_class,transport_class,transport_name,grpc_helpers", [ (StorageBatchOperationsClient, transports.StorageBatchOperationsGrpcTransport, "grpc", grpc_helpers), (StorageBatchOperationsAsyncClient, transports.StorageBatchOperationsGrpcAsyncIOTransport, "grpc_asyncio", grpc_helpers_async), diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py index 6d331f7a295e..506d5ac61265 100644 --- a/packages/gapic-generator/tests/system/conftest.py +++ b/packages/gapic-generator/tests/system/conftest.py @@ -154,7 +154,7 @@ def construct_client( client_class, use_mtls, transport_name="grpc", - channel_creator=grpc.insecure_channel, # for grpc,grpc_asyncio only + channel_creator=None, # for grpc,grpc_asyncio only credentials=ga_credentials.AnonymousCredentials(), transport_endpoint="localhost:7469", client_options=None, @@ -177,12 +177,28 @@ def construct_client( transport_cls = client_class.get_transport_class(transport_name) if transport_name in ["grpc", "grpc_asyncio"]: # TODO(gapic-generator-python/issues/1914): Need to test grpc transports without a channel_creator + if channel_creator is None: + if transport_name == "grpc_asyncio": + channel_creator = aio.insecure_channel + else: + channel_creator = grpc.insecure_channel assert channel_creator interceptors = [] - if _observability is not None and transport_name == "grpc": - otel_interceptor = _observability.get_otel_interceptor(client_options) - if otel_interceptor is not None: - interceptors.append(otel_interceptor) + if _observability is not None: + if transport_name == "grpc": + otel_interceptor = _observability.get_otel_interceptor( + client_options + ) + if otel_interceptor is not None: + interceptors.append(otel_interceptor) + elif transport_name == "grpc_asyncio": + get_async = getattr( + _observability, "get_otel_async_interceptor", None + ) + if get_async is not None: + async_interceptors = get_async(client_options) + if async_interceptors: + interceptors.extend(async_interceptors) transport = transport_cls( credentials=credentials, channel=channel_creator(transport_endpoint), diff --git a/packages/gapic-generator/tests/system/test_tracing.py b/packages/gapic-generator/tests/system/test_tracing.py index b187a7ba2015..8113704a6913 100644 --- a/packages/gapic-generator/tests/system/test_tracing.py +++ b/packages/gapic-generator/tests/system/test_tracing.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio import os from unittest import mock @@ -39,9 +40,12 @@ from google.api_core.client_options import ClientOptions from google.auth import credentials as ga_credentials from google.rpc import code_pb2 -from google.showcase import EchoClient +from google.showcase import EchoAsyncClient, EchoClient -from .conftest import construct_client +try: + from .conftest import construct_client +except ImportError: + from conftest import construct_client @pytest.fixture @@ -74,6 +78,24 @@ def otel_echo_client(span_exporter, use_mtls): return client, exporter +@pytest.fixture +def otel_echo_async_client(span_exporter, use_mtls): + """Constructs an EchoAsyncClient wired with an in-memory TracerProvider.""" + exporter, provider = span_exporter + options = ClientOptions( + tracing_enabled=True, + tracer_provider=provider, + ) + client = construct_client( + EchoAsyncClient, + use_mtls, + transport_name="grpc_asyncio", + client_options=options, + credentials=ga_credentials.AnonymousCredentials(), + ) + return client, exporter + + def test_sync_unary_tracing(otel_echo_client): """Verifies that a synchronous unary RPC generates a trace span with expected attributes.""" client, exporter = otel_echo_client @@ -279,3 +301,202 @@ def test_env_var_opt_in(span_exporter, use_mtls): spans = exporter.get_finished_spans() assert len(spans) == 1 assert spans[0].name == "google.showcase.v1beta1.Echo/Echo" + + +@pytest.fixture(autouse=True) +def set_event_loop(): + try: + asyncio.get_running_loop() + yield + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + yield + finally: + loop.close() + asyncio.set_event_loop(None) + + +@pytest.mark.asyncio +async def test_async_unary_tracing(otel_echo_async_client): + """Verifies that an asynchronous unary RPC generates a trace span with expected attributes.""" + client, exporter = otel_echo_async_client + + response = await client.echo(showcase.EchoRequest(content="hello async world")) + assert response.content == "hello async world" + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + + span = spans[0] + assert span.name == "google.showcase.v1beta1.Echo/Echo" + assert span.attributes.get("rpc.system.name") == "grpc" + assert span.attributes.get("rpc.method") == "google.showcase.v1beta1.Echo/Echo" + assert span.attributes.get("rpc.response.status_code") == "OK" + assert span.attributes.get("url.domain") == "googleapis.com" + assert span.kind == trace.SpanKind.CLIENT + + +@pytest.mark.asyncio +async def test_async_unary_retries_tracing(span_exporter, use_mtls): + """Verifies that each attempt of a retried async RPC generates a separate span.""" + from google.api_core import retry_async as retries_async + + exporter, provider = span_exporter + options = ClientOptions( + tracing_enabled=True, + tracer_provider=provider, + ) + client = construct_client( + EchoAsyncClient, + use_mtls, + transport_name="grpc_asyncio", + client_options=options, + credentials=ga_credentials.AnonymousCredentials(), + ) + + custom_retry = retries_async.AsyncRetry( + predicate=retries_async.if_exception_type(exceptions.DeadlineExceeded), + initial=0.05, + maximum=0.1, + multiplier=1.0, + deadline=0.3, + ) + + with pytest.raises((exceptions.DeadlineExceeded, exceptions.RetryError)): + await client.echo( + { + "error": { + "code": code_pb2.Code.Value("DEADLINE_EXCEEDED"), + "message": "Simulated deadline exceeded error for async retry testing.", + }, + }, + retry=custom_retry, + ) + + spans = exporter.get_finished_spans() + assert len(spans) >= 2 + for span in spans: + assert span.name == "google.showcase.v1beta1.Echo/Echo" + assert span.attributes.get("rpc.system.name") == "grpc" + assert span.attributes.get("rpc.method") == "google.showcase.v1beta1.Echo/Echo" + assert span.attributes.get("rpc.response.status_code") != "OK" + + +@pytest.mark.asyncio +async def test_async_tracing_disabled_default(use_mtls): + """Verifies that default async client options emit zero spans.""" + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + + options = ClientOptions( + tracing_enabled=False, + tracer_provider=provider, + ) + client = construct_client( + EchoAsyncClient, + use_mtls, + transport_name="grpc_asyncio", + client_options=options, + credentials=ga_credentials.AnonymousCredentials(), + ) + + response = await client.echo(showcase.EchoRequest(content="no tracing")) + assert response.content == "no tracing" + assert len(exporter.get_finished_spans()) == 0 + + +@pytest.mark.asyncio +async def test_async_custom_tracer_provider(use_mtls): + """Verifies that async spans are emitted exclusively to the injected custom TracerProvider.""" + custom_exporter = InMemorySpanExporter() + custom_provider = TracerProvider() + custom_provider.add_span_processor(SimpleSpanProcessor(custom_exporter)) + + global_exporter = InMemorySpanExporter() + global_provider = TracerProvider() + global_provider.add_span_processor(SimpleSpanProcessor(global_exporter)) + + original_provider = trace.get_tracer_provider() + trace.set_tracer_provider(global_provider) + try: + options = ClientOptions( + tracing_enabled=True, + tracer_provider=custom_provider, + ) + client = construct_client( + EchoAsyncClient, + use_mtls, + transport_name="grpc_asyncio", + client_options=options, + credentials=ga_credentials.AnonymousCredentials(), + ) + + response = await client.echo( + showcase.EchoRequest(content="isolated async trace") + ) + assert response.content == "isolated async trace" + + assert len(custom_exporter.get_finished_spans()) == 1 + assert len(global_exporter.get_finished_spans()) == 0 + finally: + trace.set_tracer_provider(original_provider) + + +@pytest.mark.asyncio +async def test_async_direct_client_initialization_tracing(span_exporter): + """Verifies end-to-end trace injection via direct EchoAsyncClient instantiation.""" + exporter, provider = span_exporter + options = ClientOptions( + tracing_enabled=True, + tracer_provider=provider, + ) + + with mock.patch.object( + EchoAsyncClient.get_transport_class("grpc_asyncio"), + "create_channel", + side_effect=lambda host, **kwargs: grpc.aio.insecure_channel("localhost:7469"), + ): + client = EchoAsyncClient( + client_options=options, + credentials=ga_credentials.AnonymousCredentials(), + ) + response = await client.echo( + showcase.EchoRequest(content="direct async client wiring") + ) + assert response.content == "direct async client wiring" + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "google.showcase.v1beta1.Echo/Echo" + assert spans[0].attributes.get("rpc.system.name") == "grpc" + + +@pytest.mark.asyncio +async def test_async_env_var_opt_in(span_exporter, use_mtls): + """Verifies that setting the environment variable enables async tracing without tracing_enabled=True.""" + exporter, provider = span_exporter + + options = ClientOptions( + tracer_provider=provider, + ) + + env_patch = { + "GOOGLE_SDK_EXPERIMENTAL_PYTHON_TRACING_ENABLED": "true", + } + with mock.patch.dict(os.environ, env_patch): + client = construct_client( + EchoAsyncClient, + use_mtls, + transport_name="grpc_asyncio", + client_options=options, + credentials=ga_credentials.AnonymousCredentials(), + ) + response = await client.echo(showcase.EchoRequest(content="env opt in async")) + assert response.content == "env opt in async" + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "google.showcase.v1beta1.Echo/Echo" diff --git a/packages/google-api-core/google/api_core/_observability.py b/packages/google-api-core/google/api_core/_observability.py index 2d8c50acbfa9..4cae73074bd2 100644 --- a/packages/google-api-core/google/api_core/_observability.py +++ b/packages/google-api-core/google/api_core/_observability.py @@ -48,11 +48,28 @@ def is_otel_capabilities_enabled( Returns: bool: True if enabled and installed, False otherwise. """ - is_tracing_enabled = _feature_gating_helpers.resolve_feature_flags( - env_var=env_var, - feature_key=_TRACER_PROVIDER, - configuration=client_options, - ) + if client_options is not None: + tracing_opt = ( + client_options.get("tracing_enabled") + if isinstance(client_options, dict) + else getattr(client_options, "tracing_enabled", None) + ) + if tracing_opt is False: + return False + if tracing_opt is True: + is_tracing_enabled = True + else: + is_tracing_enabled = _feature_gating_helpers.resolve_feature_flags( + env_var=env_var, + feature_key=_TRACER_PROVIDER, + configuration=client_options, + ) + else: + is_tracing_enabled = _feature_gating_helpers.resolve_feature_flags( + env_var=env_var, + feature_key=_TRACER_PROVIDER, + configuration=client_options, + ) if is_tracing_enabled: try: diff --git a/packages/google-api-core/google/api_core/client_options.py b/packages/google-api-core/google/api_core/client_options.py index 491e9a0bb372..fface6185a80 100644 --- a/packages/google-api-core/google/api_core/client_options.py +++ b/packages/google-api-core/google/api_core/client_options.py @@ -124,6 +124,7 @@ def __init__( api_audience: Optional[str] = None, universe_domain: Optional[str] = None, tracer_provider: Optional["opentelemetry.trace.TracerProvider"] = None, + tracing_enabled: Optional[bool] = None, ): if credentials_file is not None: warnings.warn(general_helpers._CREDENTIALS_FILE_WARNING, DeprecationWarning) @@ -144,6 +145,7 @@ def __init__( self.api_audience = api_audience self.universe_domain = universe_domain self.tracer_provider = tracer_provider + self.tracing_enabled = tracing_enabled def __repr__(self) -> str: return "ClientOptions: " + repr(self.__dict__) diff --git a/packages/google-api-core/google/api_core/gapic_v1/method_async.py b/packages/google-api-core/google/api_core/gapic_v1/method_async.py index d361bf9f961f..5e492d9a7c1a 100644 --- a/packages/google-api-core/google/api_core/gapic_v1/method_async.py +++ b/packages/google-api-core/google/api_core/gapic_v1/method_async.py @@ -17,19 +17,191 @@ compression, pagination, and long-running operations to gRPC methods. """ +import contextlib import functools +import inspect -from google.api_core import grpc_helpers_async +from google.api_core import _observability, grpc_helpers_async from google.api_core.gapic_v1 import client_info +from google.api_core.gapic_v1.client_info import METRICS_METADATA_KEY + +# Retain _GapicCallable import for backward compatibility with external packages from google.api_core.gapic_v1.method import ( # noqa: F401 DEFAULT, USE_DEFAULT_METADATA, + _apply_decorators, + _deduplicate_metadata_tokens, + _extract_error_attributes, + _extract_metrics_header, + _extract_rpc_identity, + _extract_status_code, _GapicCallable, ) +from google.api_core.timeout import TimeToDeadlineTimeout _DEFAULT_ASYNC_TRANSPORT_KIND = "grpc_asyncio" +class _AsyncGapicCallable(object): + """Async callable object that wraps an async RPC method with retry, timeout, metadata, and tracing. + + Args: + target (Callable): The low-level async RPC method. + retry (Optional[google.api_core.retry_async.AsyncRetry]): The default retry for the + callable. If ``None``, this callable will not retry by default. + timeout (Optional[Union[google.api_core.timeout.Timeout, float]]): The default timeout for the + callable. If ``None``, this callable will not specify a timeout argument to the + low-level RPC method. + compression (Optional[grpc.Compression]): The default compression for the callable. + If ``None``, this callable will not specify a compression argument to the low-level + RPC method. + metadata (Optional[Sequence[Tuple[str, str]]]): Additional metadata that is + provided to the RPC method on every invocation. This is merged with + any metadata specified during invocation. If ``None``, no + additional metadata will be passed to the RPC method. + client_options (Optional[google.api_core.client_options.ClientOptions]): + Client options used to configure client-level behavior, such as + custom OpenTelemetry tracer providers. Defaults to None. + method_name (Optional[str]): The optional explicit full RPC method name + (e.g. "/google.cloud.secretmanager.v1.SecretManagerService/AccessSecretVersion"). + is_streaming (bool): Whether the RPC method is streaming. Defaults to False. + Note: Streaming methods do not currently generate Tier 3 observability spans. + client_info (Optional[google.api_core.gapic_v1.client_info.ClientInfo]): + Client information used for metadata headers. Defaults to None. + kind (str): The transport kind for the RPC method. Defaults to "grpc_asyncio". + Allowed values for OpenTelemetry method tracing are "grpc" and "grpc_asyncio". + """ + + def __init__( + self, + target, + retry, + timeout, + compression, + metadata=None, + client_options=None, + method_name=None, + is_streaming=False, + client_info=None, + kind=_DEFAULT_ASYNC_TRANSPORT_KIND, + ): + self._target = target + self._retry = retry + self._timeout = timeout + self._compression = compression + + # Pre-extract the x-goog-api-client header from the initialized metadata. + self._x_goog_api_client, remaining = _extract_metrics_header(metadata) + self._static_metadata = tuple(remaining) + if self._x_goog_api_client: + self._default_metadata = ( + (METRICS_METADATA_KEY, self._x_goog_api_client), + *self._static_metadata, + ) + else: + self._default_metadata = self._static_metadata + + # Configure the OpenTelemetry span factory once at initialization. + # Method tracing is gated to non-streaming gRPC calls where an explicit method_name is provided. + self._start_span_fn = None + if ( + not is_streaming + and kind in ("grpc", "grpc_asyncio") + and method_name is not None + and _observability.is_otel_capabilities_enabled(client_options) + ): + try: + from opentelemetry import trace + + tracer_provider = None + if isinstance(client_options, dict): + tracer_provider = client_options.get("tracer_provider") + elif client_options is not None: + tracer_provider = getattr(client_options, "tracer_provider", None) + if tracer_provider is not None: + tracer = tracer_provider.get_tracer("google.api_core") + else: + tracer = trace.get_tracer("google.api_core") + + span_name, _, _ = _extract_rpc_identity(method_name) + span_attributes = { + "rpc.system.name": "grpc", + "rpc.method": span_name, + } + self._start_span_fn = functools.partial( + tracer.start_as_current_span, + span_name, + kind=trace.SpanKind.CLIENT, + attributes=span_attributes, + ) + except (ImportError, AttributeError, TypeError): + # Gracefully disable tracing if OpenTelemetry or custom provider fails + self._start_span_fn = None + + async def __call__( + self, *args, timeout=DEFAULT, retry=DEFAULT, compression=DEFAULT, **kwargs + ): + """Invoke the low-level async RPC with retry, timeout, compression, and metadata.""" + if retry is DEFAULT: + retry = self._retry + + if timeout is DEFAULT: + timeout = self._timeout + + if compression is DEFAULT: + compression = self._compression + + if isinstance(timeout, (int, float)): + timeout = TimeToDeadlineTimeout(timeout=timeout) + + # Apply all applicable decorators. + wrapped_func = _apply_decorators(self._target, [retry, timeout]) + + if user_metadata := kwargs.get("metadata"): + # Add the user agent metadata to the call. + final_metadata = list(self._static_metadata) + user_x_goog, remaining = _extract_metrics_header(user_metadata) + + merged_header = _deduplicate_metadata_tokens( + self._x_goog_api_client, user_x_goog + ) + if merged_header: + final_metadata.append((METRICS_METADATA_KEY, merged_header)) + final_metadata.extend(remaining) + kwargs["metadata"] = final_metadata + elif self._default_metadata: + kwargs["metadata"] = self._default_metadata + + if self._compression is not None: + kwargs["compression"] = compression + + span_cm = contextlib.nullcontext() + if self._start_span_fn is not None: + try: + span_cm = self._start_span_fn() + except Exception: + span_cm = contextlib.nullcontext() + + with span_cm as span: + try: + res = wrapped_func(*args, **kwargs) + if inspect.isawaitable(res): + result = await res + else: + result = res + if span is not None and hasattr(span, "set_attribute"): + span.set_attribute("rpc.response.status_code", "OK") + return result + except Exception as exc: + if span is not None and hasattr(span, "set_attribute"): + span.set_attribute( + "rpc.response.status_code", _extract_status_code(exc) + ) + for k, v in _extract_error_attributes(exc).items(): + span.set_attribute(k, v) + raise + + def wrap_method( func, default_retry=None, @@ -37,6 +209,10 @@ def wrap_method( default_compression=None, client_info=client_info.DEFAULT_CLIENT_INFO, kind=_DEFAULT_ASYNC_TRANSPORT_KIND, + *, + client_options=None, + method_name=None, + is_streaming=False, ): """Wrap an async RPC method with common behavior. @@ -51,11 +227,16 @@ def wrap_method( metadata = [client_info.to_grpc_metadata()] if client_info is not None else None return functools.wraps(func)( - _GapicCallable( + _AsyncGapicCallable( func, default_retry, default_timeout, default_compression, metadata=metadata, + client_options=client_options, + method_name=method_name, + is_streaming=is_streaming, + client_info=client_info, + kind=kind, ) ) diff --git a/packages/google-api-core/google/api_core/grpc_helpers_async.py b/packages/google-api-core/google/api_core/grpc_helpers_async.py index d1f897901e7a..5df82d42ceb0 100644 --- a/packages/google-api-core/google/api_core/grpc_helpers_async.py +++ b/packages/google-api-core/google/api_core/grpc_helpers_async.py @@ -21,7 +21,17 @@ import asyncio import functools import warnings -from typing import AsyncGenerator, Generic, Iterator, Optional, TypeVar +from typing import ( + AsyncGenerator, + Callable, + Generic, + Iterator, + Optional, + Sequence, + TypeVar, + Union, + cast, +) import grpc from grpc import aio @@ -219,6 +229,7 @@ def create_channel( default_host=None, compression=None, attempt_direct_path: Optional[bool] = False, + interceptors: Optional[Sequence[grpc.aio.ClientInterceptor]] = None, **kwargs, ): """Create an AsyncIO secure channel with credentials. @@ -270,6 +281,8 @@ def create_channel( `False` as the Service may not support Direct Path. - Using `ssl_credentials` with `attempt_direct_path` set to `True` will result in `ValueError` as this combination is not yet supported. + interceptors (Optional[Sequence[grpc.aio.ClientInterceptor]]): An optional + sequence of :class:`grpc.aio.ClientInterceptor` objects to be applied to the channel. kwargs: Additional key-word args passed to :func:`aio.secure_channel`. @@ -303,6 +316,9 @@ def create_channel( if attempt_direct_path: target = grpc_helpers._modify_target_for_direct_path(target) + if interceptors is not None: + kwargs["interceptors"] = interceptors + return aio.secure_channel( target, composite_credentials, compression=compression, **kwargs ) @@ -345,3 +361,71 @@ def __await__(self): async def wait_for_connection(self): pass + + +_ASYNC_CLIENT_INTERCEPTOR_CLASSES = ( + aio.ClientInterceptor, + aio.UnaryUnaryClientInterceptor, + aio.UnaryStreamClientInterceptor, + aio.StreamUnaryClientInterceptor, + aio.StreamStreamClientInterceptor, +) + + +def apply_channel_interceptors( + channel: aio.Channel, + interceptors: Optional[ + Sequence[Union[aio.ClientInterceptor, Callable[[aio.Channel], aio.Channel]]] + ] = None, +) -> aio.Channel: + """Applies async client interceptors or channel-intercepting callables to a gRPC AsyncIO channel. + + Args: + channel (aio.Channel): The channel to intercept. + interceptors (Optional[Sequence[Union[aio.ClientInterceptor, Callable[[aio.Channel], aio.Channel]]]]): + Additional interceptors (or callables that apply interceptors) to apply to the gRPC AsyncIO channel. + + Returns: + aio.Channel: The intercepted channel, or the original channel if no + interceptors were provided. + + Raises: + TypeError: If an item in ``interceptors`` is neither a gRPC aio.ClientInterceptor + nor a Callable[[aio.Channel], aio.Channel]. + """ + if not interceptors: + return channel + + modified_channel = channel + for interceptor in interceptors: + if isinstance(interceptor, _ASYNC_CLIENT_INTERCEPTOR_CLASSES): + matched = False + if isinstance(interceptor, aio.UnaryUnaryClientInterceptor): + if hasattr(modified_channel, "_unary_unary_interceptors"): + modified_channel._unary_unary_interceptors.append(interceptor) + matched = True + if isinstance(interceptor, aio.UnaryStreamClientInterceptor): + if hasattr(modified_channel, "_unary_stream_interceptors"): + modified_channel._unary_stream_interceptors.append(interceptor) + matched = True + if isinstance(interceptor, aio.StreamUnaryClientInterceptor): + if hasattr(modified_channel, "_stream_unary_interceptors"): + modified_channel._stream_unary_interceptors.append(interceptor) + matched = True + if isinstance(interceptor, aio.StreamStreamClientInterceptor): + if hasattr(modified_channel, "_stream_stream_interceptors"): + modified_channel._stream_stream_interceptors.append(interceptor) + matched = True + if not matched and hasattr(modified_channel, "_interceptors"): + modified_channel._interceptors.append(interceptor) + elif callable(interceptor): + interceptor_callable = cast( + Callable[[aio.Channel], aio.Channel], interceptor + ) + modified_channel = interceptor_callable(modified_channel) + else: + raise TypeError( + f"Expected ClientInterceptor or Callable[[aio.Channel], aio.Channel], got {type(interceptor).__name__}" + ) + + return modified_channel diff --git a/packages/google-api-core/tests/asyncio/gapic/test_method_async.py b/packages/google-api-core/tests/asyncio/gapic/test_method_async.py index e410acbdfaab..b10577c300a6 100644 --- a/packages/google-api-core/tests/asyncio/gapic/test_method_async.py +++ b/packages/google-api-core/tests/asyncio/gapic/test_method_async.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio import datetime try: @@ -26,6 +27,9 @@ except ImportError: pytest.skip("No GRPC", allow_module_level=True) +from google.api_core import ( + client_options as client_options_lib, +) from google.api_core import ( exceptions, gapic_v1, @@ -274,3 +278,260 @@ async def test_wrap_method_without_wrap_errors(): await wrapped_method() method.assert_not_called() + + +@pytest.fixture(autouse=True) +def set_event_loop(): + try: + asyncio.get_running_loop() + yield + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + yield + finally: + loop.close() + asyncio.set_event_loop(None) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "kwargs,capabilities_enabled", + [ + ( + { + "method_name": "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets" + }, + False, + ), + ({}, True), + ( + { + "method_name": "/google.cloud.secretmanager.v1.SecretManagerService/StreamingRead", + "is_streaming": True, + }, + True, + ), + ( + { + "method_name": "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + "kind": "rest", + }, + True, + ), + ( + { + "method_name": "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + "kind": "rest_asyncio", + }, + True, + ), + ], + ids=[ + "disabled_by_flag", + "omitted_method_name", + "streaming_skipped", + "rest_kind_skipped", + "rest_asyncio_kind_skipped", + ], +) +async def test_wrap_method_async_otel_tracing_skips_span( + monkeypatch, kwargs, capabilities_enabled +): + """Proves that under various gating conditions, no async Tier 3 span is created.""" + mock_target = mock.AsyncMock(return_value="success") + mock_trace = mock.Mock() + + with ( + mock.patch( + "google.api_core._observability.is_otel_capabilities_enabled", + return_value=capabilities_enabled, + ), + mock.patch.dict( + "sys.modules", + { + "opentelemetry": mock.Mock(trace=mock_trace), + "opentelemetry.trace": mock_trace, + }, + ), + ): + wrapped = gapic_v1.method_async.wrap_method(mock_target, **kwargs) + result = await wrapped() + + assert result == "success" + mock_trace.get_tracer.assert_not_called() + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_enabled_success(mock_otel): + """Proves that when OpenTelemetry tracing is enabled and method_name is passed, a T3 client span is started and awaited.""" + mock_target = mock.AsyncMock(return_value="async_success") + + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + default_timeout=60, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + kind="grpc_asyncio", + ) + result = await wrapped() + + assert result == "async_success" + mock_otel.tracer.start_as_current_span.assert_called_once_with( + "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + kind="CLIENT", + attributes={ + "rpc.system.name": "grpc", + "rpc.method": "google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + }, + ) + mock_otel.span.set_attribute.assert_called_with("rpc.response.status_code", "OK") + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_coroutine_duration(mock_otel): + """Proves that the span remains active across asynchronous awaits and closes only after completion.""" + span_open_during_call = False + + async def delayed_target(*args, **kwargs): + nonlocal span_open_during_call + span_open_during_call = ( + mock_otel.tracer.start_as_current_span.return_value.__enter__.called + and not mock_otel.tracer.start_as_current_span.return_value.__exit__.called + ) + await asyncio.sleep(0.01) + return "delayed_result" + + wrapped = gapic_v1.method_async.wrap_method( + delayed_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + ) + result = await wrapped() + + assert result == "delayed_result" + assert span_open_during_call is True + assert mock_otel.tracer.start_as_current_span.return_value.__exit__.called is True + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_custom_client_options(mock_otel): + """Proves that providing client_options with a custom tracer_provider uses that provider.""" + mock_target = mock.AsyncMock(return_value="success") + + mock_provider = mock.Mock() + mock_provider.get_tracer.return_value = mock_otel.tracer + + client_options = client_options_lib.ClientOptions(tracer_provider=mock_provider) + + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + client_options=client_options, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + ) + result = await wrapped() + + assert result == "success" + mock_provider.get_tracer.assert_called_once_with("google.api_core") + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_dict_client_options(mock_otel): + """Proves that providing a dict with tracer_provider uses that provider.""" + mock_target = mock.AsyncMock(return_value="success") + + mock_provider = mock.Mock() + mock_provider.get_tracer.return_value = mock_otel.tracer + + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + client_options={"tracer_provider": mock_provider}, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + ) + result = await wrapped() + + assert result == "success" + mock_provider.get_tracer.assert_called_once_with("google.api_core") + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_enabled_error(mock_otel): + """Proves that on async error, status code and error attributes are recorded and exception is raised.""" + error = exceptions.NotFound("Secret not found") + mock_target = mock.AsyncMock(side_effect=error) + + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + + with pytest.raises(exceptions.NotFound): + await wrapped() + + mock_otel.span.set_attribute.assert_any_call( + "rpc.response.status_code", "NOT_FOUND" + ) + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_records_gcp_error_attributes(mock_otel): + """Proves that GCP error attributes (domain, reason, metadata) are recorded on the span.""" + error_info = mock.Mock( + domain="googleapis.com", + reason="RESOURCE_NOT_FOUND", + metadata={"service": "secretmanager"}, + ) + error = exceptions.GoogleAPICallError("Resource not found") + error._error_info = error_info + mock_target = mock.AsyncMock(side_effect=error) + + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/GetSecret", + ) + + with pytest.raises(exceptions.GoogleAPICallError): + await wrapped() + + mock_otel.span.set_attribute.assert_any_call("gcp.errors.domain", "googleapis.com") + mock_otel.span.set_attribute.assert_any_call("error.type", "RESOURCE_NOT_FOUND") + mock_otel.span.set_attribute.assert_any_call( + "gcp.errors.metadata.service", "secretmanager" + ) + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_import_error(monkeypatch): + """Proves that if opentelemetry fails to import, method execution proceeds gracefully without tracing.""" + mock_target = mock.AsyncMock(return_value="graceful_success") + + with ( + mock.patch( + "google.api_core._observability.is_otel_capabilities_enabled", + return_value=True, + ), + mock.patch.dict("sys.modules", {"opentelemetry": None}), + ): + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + ) + result = await wrapped() + + assert result == "graceful_success" + + +@pytest.mark.asyncio +async def test_wrap_method_async_otel_tracing_start_span_error_bypasses_tracing( + mock_otel, +): + """Proves that if tracer.start_as_current_span throws an exception, the call executes cleanly.""" + mock_otel.tracer.start_as_current_span.side_effect = RuntimeError("Tracing broken") + mock_target = mock.AsyncMock(return_value="resilient_success") + + wrapped = gapic_v1.method_async.wrap_method( + mock_target, + method_name="/google.cloud.secretmanager.v1.SecretManagerService/ListSecrets", + ) + result = await wrapped() + + assert result == "resilient_success" diff --git a/packages/google-api-core/tests/asyncio/test_grpc_helpers_async.py b/packages/google-api-core/tests/asyncio/test_grpc_helpers_async.py index dcb09f18fea2..ac219497db6b 100644 --- a/packages/google-api-core/tests/asyncio/test_grpc_helpers_async.py +++ b/packages/google-api-core/tests/asyncio/test_grpc_helpers_async.py @@ -737,9 +737,74 @@ def test_create_channel(grpc_secure_channel): credentials.with_scopes.assert_called_once_with(scopes, default_scopes=None) +@mock.patch("grpc.aio.secure_channel") +def test_create_channel_with_interceptors(grpc_secure_channel): + target = "example.com:443" + mock_interceptor = mock.Mock() + credentials = mock.create_autospec(google.auth.credentials.Scoped, instance=True) + credentials.requires_scopes = False + + grpc_helpers_async.create_channel( + target, + credentials=credentials, + interceptors=[mock_interceptor], + ) + grpc_secure_channel.assert_called_once_with( + target, + mock.ANY, + compression=None, + interceptors=[mock_interceptor], + ) + + @pytest.mark.asyncio async def test_fake_stream_unary_call(): fake_call = grpc_helpers_async.FakeStreamUnaryCall() await fake_call.wait_for_connection() response = await fake_call assert fake_call.response == response + + +def test_apply_channel_interceptors_none_or_empty(): + channel = mock.Mock(spec=aio.Channel) + assert grpc_helpers_async.apply_channel_interceptors(channel) is channel + assert grpc_helpers_async.apply_channel_interceptors(channel, []) is channel + + +def test_apply_channel_interceptors_with_classes(): + channel = mock.Mock(spec=aio.Channel) + channel._unary_unary_interceptors = [] + channel._unary_stream_interceptors = [] + channel._stream_unary_interceptors = [] + channel._stream_stream_interceptors = [] + + uu_interceptor = mock.Mock(spec=aio.UnaryUnaryClientInterceptor) + us_interceptor = mock.Mock(spec=aio.UnaryStreamClientInterceptor) + su_interceptor = mock.Mock(spec=aio.StreamUnaryClientInterceptor) + ss_interceptor = mock.Mock(spec=aio.StreamStreamClientInterceptor) + + res = grpc_helpers_async.apply_channel_interceptors( + channel, + [uu_interceptor, us_interceptor, su_interceptor, ss_interceptor], + ) + assert res is channel + assert channel._unary_unary_interceptors == [uu_interceptor] + assert channel._unary_stream_interceptors == [us_interceptor] + assert channel._stream_unary_interceptors == [su_interceptor] + assert channel._stream_stream_interceptors == [ss_interceptor] + + +def test_apply_channel_interceptors_with_callable(): + channel = mock.Mock(spec=aio.Channel) + modified_channel = mock.Mock(spec=aio.Channel) + callable_interceptor = mock.Mock(return_value=modified_channel) + + res = grpc_helpers_async.apply_channel_interceptors(channel, [callable_interceptor]) + assert res is modified_channel + callable_interceptor.assert_called_once_with(channel) + + +def test_apply_channel_interceptors_invalid_type(): + channel = mock.Mock(spec=aio.Channel) + with pytest.raises(TypeError, match="Expected ClientInterceptor or Callable"): + grpc_helpers_async.apply_channel_interceptors(channel, [123])