parthea · Nov 2, 2026 · Nov 2, 2026
@@ -1431,6 +1431,16 @@ def test_{{ service.name|snake_case }}_base_transport_with_adc():
         transport = transports.{{ service.name }}Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.{{ service.name }}Transport):
+        _ignore_credentials = True
+
+    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
+        transport = IgnoreCredsTransport(host="squid.clam.whelk")
+        adc.assert_not_called()
+        assert transport._credentials is None
+        assert transport.host == "squid.clam.whelk:443"
+
 
 def test_{{ service.name|snake_case }}_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
@@ -1469,6 +1479,7 @@ def test_{{ service.name|snake_case }}_transport_auth_adc(transport_class):
                 {%- endfor %}),
             quota_project_id="octopus",
         )
+{% endif %}
 
 
 {% if 'grpc' in opts.transport and 'rest' in opts.transport %}
@@ -1511,6 +1522,7 @@ def test_{{ service.name|snake_case }}_transport_auth_gdch_credentials(transport
             )
 
 
+{% if 'grpc' in opts.transport %}
 @pytest.mark.parametrize(
     "transport_class,grpc_helpers",
     [
@@ -1983,6 +1995,7 @@ async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_a
         ) in str(exc_info.value)
 {% endfor %}
 {% endif %}
+{% endif %}
 
 {% if service.has_resumable_upload_methods and 'rest' in opts.transport %}
 {% for method in service.methods.values() if method.is_resumable_upload %}
@@ -2226,6 +2239,7 @@ async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_rest_a
 {% endif %}
 
 
+{% if 'grpc' in opts.transport %}
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
 @pytest.mark.filterwarnings("ignore::FutureWarning")
@@ -2572,17 +2586,18 @@ def test_{{ service.name|snake_case }}_base_transport_wrap_method_older_api_core
         assert "is_streaming" not in kwargs
 
 
-{% if 'grpc' in opts.transport %}
+{% set async_wrap_transport_class = service.grpc_asyncio_transport_name if 'grpc' in opts.transport else service.name ~ opts.transport[0].capitalize() ~ 'Transport' %}
+{% set async_wrap_transport_kind = 'grpc_asyncio' if 'grpc' in opts.transport else opts.transport[0] %}
 def test_{{ service.name|snake_case }}_base_transport_wrap_async_method_modern_api_core(monkeypatch):
     monkeypatch.setattr(transports.base, "_ASYNC_WRAP_METHOD_SUPPORTS_TRACING", True)
-    # 1. Concrete transport: passes client_options=None and kind="grpc_asyncio"
+    # 1. Concrete transport: passes client_options=None and kind="{{ async_wrap_transport_kind }}"
     with mock.patch.object(transports.base.gapic_v1.method_async, "wrap_method") as mock_wrap:
-        transport = transports.{{ service.grpc_asyncio_transport_name }}(
+        transport = transports.{{ async_wrap_transport_class }}(
             credentials=ga_credentials.AnonymousCredentials()
         )
         func = mock.MagicMock()
         transport._wrap_async_method(func)
-        assert mock_wrap.call_args.kwargs["kind"] == "grpc_asyncio"
+        assert mock_wrap.call_args.kwargs["kind"] == "{{ async_wrap_transport_kind }}"
         assert "client_options" in mock_wrap.call_args.kwargs
 
     # 2. Base transport: kind raises NotImplementedError and is cleanly omitted
@@ -2599,7 +2614,7 @@ def test_{{ service.name|snake_case }}_base_transport_wrap_async_method_modern_a
 def test_{{ service.name|snake_case }}_base_transport_wrap_async_method_older_api_core_fallbacks(monkeypatch):
     monkeypatch.setattr(transports.base, "_ASYNC_WRAP_METHOD_SUPPORTS_TRACING", False)
     with mock.patch.object(transports.base.gapic_v1.method_async, "wrap_method"):
-        transport = transports.{{ service.grpc_asyncio_transport_name }}(
+        transport = transports.{{ async_wrap_transport_class }}(
             credentials=ga_credentials.AnonymousCredentials()
         )
 
@@ -2647,7 +2662,7 @@ def test_{{ service.name|snake_case }}_base_transport_wrap_async_method_older_ap
             is_streaming=False,
         )
         kwargs = mock_wrap.call_args.kwargs
-        assert kwargs.get("kind") == "grpc_asyncio"
+        assert kwargs.get("kind") == "{{ async_wrap_transport_kind }}"
         assert "client_options" not in kwargs
         assert "method_name" not in kwargs
         assert "is_streaming" not in kwargs
@@ -2668,6 +2683,5 @@ def test_{{ service.name|snake_case }}_base_transport_wrap_async_method_older_ap
         assert "client_options" not in kwargs
         assert "method_name" not in kwargs
         assert "is_streaming" not in kwargs
-{% endif %}
 
 {% endblock %}
@@ -1028,9 +1028,7 @@ def test_{{ method_name }}_raw_page_lro():
 {% endmacro %}
 
 {% macro rest_required_tests(api, method, service, numeric_enums=False, full_extended_lro=False) %}
-{% with method_name = method.client_method_name|snake_case + "_unary" if method.extended_lro and not full_extended_lro else method.client_method_name|snake_case, method_output = method.extended_lro.operation_type if method.extended_lro and not full_extended_lro else method.output %}{% if method.http_options %}
-{# TODO(kbandes): remove this if condition when lro and client streaming are supported. #}
-{% if not method.client_streaming %}
+{% with method_name = method.client_method_name|snake_case + "_unary" if method.extended_lro and not full_extended_lro else method.client_method_name|snake_case, method_output = method.extended_lro.operation_type if method.extended_lro and not full_extended_lro else method.output %}
 def test_{{ method_name }}_rest_use_cached_wrapped_rpc():
     # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
     # instead of constructing them on each call
@@ -1080,6 +1078,8 @@ def test_{{ method_name }}_rest_use_cached_wrapped_rpc():
         assert mock_rpc.call_count == 2
 
 
+{% if method.http_options %}
+{% if not method.client_streaming %}
     {% if method.input.required_fields %}
 def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ident }}):
     transport_class = transports.{{ service.rest_transport_name }}
@@ -1323,6 +1323,38 @@ def test_{{ method_name }}_rest_flattened():
         {% endwith %}
         {# TODO(kbandes) - reverse-transcode request args to check all request fields #}
 
+    {% if method.flattened_oneof_fields() %}
+    # Verify flattened oneof fields on the constructed request without
+    # conflicting with HTTP path template transcoding.
+    with mock.patch.object(
+        type(client.transport.{{ method.transport_safe_name|snake_case }}),
+        "__call__",
+    ) as call:
+        {% if method.void %}
+        call.return_value = None
+        {% elif method.lro %}
+        call.return_value = operations_pb2.Operation(name="operations/op")
+        {% elif method.server_streaming %}
+        call.return_value = iter([{{ method.output.ident }}()])
+        {% elif method.is_resumable_upload %}
+        call.return_value = {{ method.client_output.ident }}()
+        {% else %}
+        call.return_value = {{ method.output.ident }}()
+        {% endif %}
+        client.{{ method_name }}(
+            {% for field in method.flattened_fields.values() %}
+            {{ field.name }}={{ field.mock_value }},
+            {% endfor %}
+        )
+        assert len(call.mock_calls) == 1
+        _, args, _ = call.mock_calls[0]
+        {% for oneofs in method.flattened_oneof_fields().values() %}
+        {% with field = oneofs[-1] %}
+        assert args[0].{{ method.flattened_field_to_key[field.name] }} == {{ field.mock_value }}
+        {% endwith %}
+        {% endfor %}
+    {% endif %}
+
 
 def test_{{ method_name }}_rest_flattened_error(transport: str = 'rest'):
     client = {{ service.client_name }}(
@@ -1474,6 +1506,11 @@ def test_{{ method_name }}_rest_pager(transport: str = 'rest'):
             assert page_.raw_page.next_page_token == token
 
 
+{% elif method.lro and "next_page_token" in method.lro.response_type.fields.keys() %}
+def test_{{ method_name }}_rest_raw_page_lro():
+    response = {{ method.lro.response_type.ident }}()
+    assert response.raw_page is response
+
 {%- endif %}{# paged_result_field #}
 
 {% else %}{# this is an lro or streaming method #}
@@ -1753,6 +1790,7 @@ def test_unsupported_parameter_rest_asyncio():
 {% if 'rest' in transport %}
 {{ bad_request_mixins_test(service, api, name, sig, transport, is_async) }}
 {{ call_success_mixins_test(service, api, name, sig, transport, is_async) }}
+{{ empty_call_mixins_test(service, api, name, sig, transport, is_async) }}
 {% endif %}{# if 'rest' in transport #}
 {% endfor %}
 {{ initialize_client_with_transport_test(service, transport, is_async) }}
@@ -2254,6 +2292,35 @@ def test_initialize_client_w_{{transport_name}}():
 {% endif %}{# if 'grpc' in transport #}
 {% endmacro %}{# call_success_mixins_test #}
 
+{% macro empty_call_mixins_test(service, api, name, sig, transport, is_async=False) %}
+{% set await_prefix = get_await_prefix(is_async) %}
+{% set async_prefix = get_async_prefix(is_async) %}
+{% set async_decorator = get_async_decorator(is_async) %}
+{% set transport_name = get_transport_name(transport, is_async) %}
+{% set method_name = name|snake_case %}
+{{ async_decorator }}
+{{ async_prefix }}def test_{{ method_name }}_empty_call_{{ transport_name }}():
+    {% if transport_name == 'rest_asyncio' %}
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    {% endif %}
+    client = {{ get_client(service=service, is_async=is_async) }}(
+        credentials={{ get_credentials(is_async=is_async) }},
+        transport="{{ transport_name }}",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.{{ method_name }}), "__call__"
+    ) as call:
+        {{ await_prefix }}client.{{ method_name }}(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == {{ sig.request_type }}()
+{% endmacro %}{# empty_call_mixins_test #}
+
 {% macro empty_call_test(service, api, transport, is_async) %}
 {% for method in service.methods.values() %}{# method #}
 {% if not method.client_streaming %}

@@ -243,6 +243,7 @@ def showcase_library(
     retry_config=True,
     rest_async_io_enabled=False,
     install_async_rest_extra=False,
+    transport="grpc+rest",
 ):
     """Install the generated library into the session for showcase tests.
 
@@ -348,7 +349,7 @@ def showcase_library(
                 other_opts
                 + (
                     f"{template_opt}",
-                    "transport=grpc+rest",
+                    f"transport={transport}",
                     f"service-yaml={tmp_dir}/showcase_v1beta1.yaml",
                     f"retry-config={tmp_dir}/showcase_grpc_service_config.json",
                 )
@@ -645,19 +646,28 @@ def run_showcase_unit_tests(
 
 
 @nox.session(python=ALL_PYTHON)
-@nox.parametrize("install_async_rest_extra", [False, True])
+@nox.parametrize(
+    "install_async_rest_extra,transport",
+    [
+        (False, "grpc+rest"),
+        (False, "rest"),
+        (True, "grpc+rest"),
+    ],
+)
 def showcase_unit(
     session,
     install_async_rest_extra=False,
     templates="DEFAULT",
     other_opts: typing.Iterable[str] = (),
+    transport: str = None,
 ):
     """Run the generated unit tests against the Showcase library."""
     with showcase_library(
         session,
         templates=templates,
         other_opts=other_opts,
         install_async_rest_extra=install_async_rest_extra,
+        transport=transport,
     ) as lib:
         session.chdir(lib)
         run_showcase_unit_tests(

@@ -16910,6 +16910,24 @@ def test_get_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+
+def test_get_operation_empty_call_rest():
+    client = AssetServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_operation), "__call__"
+    ) as call:
+        client.get_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.GetOperationRequest()
+
 def test_initialize_client_w_rest():
     client = AssetServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -17519,6 +17537,16 @@ def test_asset_service_base_transport_with_adc():
         transport = transports.AssetServiceTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.AssetServiceTransport):
+        _ignore_credentials = True
+
+    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
+        transport = IgnoreCredsTransport(host="squid.clam.whelk")
+        adc.assert_not_called()
+        assert transport._credentials is None
+        assert transport.host == "squid.clam.whelk:443"
+
 
 def test_asset_service_auth_adc():
     # If no credentials are provided, we should use ADC credentials.

@@ -3933,6 +3933,16 @@ def test_iam_credentials_base_transport_with_adc():
         transport = transports.IAMCredentialsTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.IAMCredentialsTransport):
+        _ignore_credentials = True
+
+    with mock.patch.object(google.auth, 'default', autospec=True) as adc:
+        transport = IgnoreCredsTransport(host="squid.clam.whelk")
+        adc.assert_not_called()
+        assert transport._credentials is None
+        assert transport.host == "squid.clam.whelk:443"
+
 
 def test_iam_credentials_auth_adc():
     # If no credentials are provided, we should use ADC credentials.