test(generator): sandbox verification of spec-driven telemetry compliance tests

[chalmerlowe]

Description

Temporary sandbox pull request to validate clean-slate spec-driven telemetry compliance tests in GitHub Actions CI without impacting PR #18433.

Scope

• Replaces legacy boilerplate with table-driven tests parametrized over all 22 feature rows in telemetry_requirements_matrix.csv.
• All 22 features pass locally in Docker Nox (showcase-3.10).

[gemini-code-assist]

Code Review

This pull request integrates OpenTelemetry tracing and observability capabilities into the generated Python GAPIC clients across REST and gRPC transports (both synchronous and asynchronous). Key updates include wrapping HTTP requests with tracing context, injecting OpenTelemetry interceptors into gRPC channels, and adding corresponding unit tests. Feedback on the changes highlights a fragility in the newly added async gRPC transport unit tests, where grpc_helpers_async.apply_channel_interceptors is not mocked, potentially causing test failures in environments where this helper is available. Mocking this helper is recommended to ensure test robustness.

[gemini-code-assist]

The test test_{{ service.name|snake_case }}_grpc_asyncio_transport_channel_interceptors does not mock grpc_helpers_async.apply_channel_interceptors. In environments where apply_channel_interceptors is available in google-api-core, it will be called and will return a wrapped InterceptedChannel instance instead of mock_channel. This will cause the assertion assert transport.grpc_channel == mock_channel to fail, making the test fragile.

To make the test robust and consistent with the synchronous gRPC transport tests, we should mock grpc_helpers_async.apply_channel_interceptors to return mock_channel and assert that it was called with the expected 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,
        ) as mock_create_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],
        )

        assert mock_create_channel.call_count == 1
        mock_apply_interceptors.assert_called_once_with(
            mock_channel, [mock_interceptor, transport._interceptor]
        )
        assert transport.grpc_channel == mock_channel

[gemini-code-assist]

The test test_{{ service.name|snake_case }}_grpc_asyncio_transport_otel_channel_interceptor does not mock grpc_helpers_async.apply_channel_interceptors. In environments where apply_channel_interceptors is available in google-api-core, it will be called and will return a wrapped InterceptedChannel instance instead of mock_channel. This will cause the assertion assert transport.grpc_channel == mock_channel to fail, making the test fragile.

To make the test robust and consistent with the synchronous gRPC transport tests, we should mock grpc_helpers_async.apply_channel_interceptors to return mock_channel and assert that it was called with the expected interceptors.

def test_{{ service.name|snake_case }}_grpc_asyncio_transport_otel_channel_interceptor():
    mock_otel_interceptor = mock.Mock()
    mock_obs = mock.Mock()
    mock_obs.get_otel_async_interceptor.return_value = mock_otel_interceptor
    mock_channel = mock.Mock()

    with (
        mock.patch(
            "{{ (api.naming.module_namespace + (api.naming.versioned_module_name,) + service.meta.address.subpackage)|join('.') }}.services.{{ service.name|snake_case }}.transports.grpc_asyncio._observability",
            mock_obs,
        ),
        mock.patch.object(
            transports.{{ service.grpc_asyncio_transport_name }},
            "create_channel",
            return_value=mock_channel,
        ) as mock_create_channel,
        mock.patch.object(
            grpc_helpers_async,
            "apply_channel_interceptors",
            return_value=mock_channel,
            create=True,
        ) as mock_apply_interceptors,
    ):
        options = client_options.ClientOptions()
        transport = transports.{{ service.grpc_asyncio_transport_name }}(
            credentials=ga_credentials.AnonymousCredentials(),
            client_options=options,
        )

        mock_obs.get_otel_async_interceptor.assert_called_once_with(options)
        assert mock_create_channel.call_count == 1
        mock_apply_interceptors.assert_called_once_with(
            mock_channel, [transport._interceptor, mock_otel_interceptor]
        )
        assert transport.grpc_channel == mock_channel