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 ab53d4699321..a50711540b95 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
@@ -1446,6 +1446,17 @@ def test_{{ service.name|snake_case }}_base_transport_with_adc():
         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.
     with mock.patch.object(google.auth, 'default', autospec=True) as adc:
@@ -1483,6 +1494,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 %}
@@ -1525,6 +1537,7 @@ def test_{{ service.name|snake_case }}_transport_auth_gdch_credentials(transport
             )
 
 
+{% if 'grpc' in opts.transport %}
 @pytest.mark.parametrize(
     "transport_class,grpc_helpers",
     [
@@ -2001,6 +2014,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 %}
@@ -2244,6 +2258,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")
@@ -2696,7 +2711,6 @@ def test_{{ service.name|snake_case }}_base_transport_wrap_method_older_api_core
         assert "is_streaming" not in kwargs
 
 
-{% if 'grpc' in opts.transport %}
 @pytest.mark.parametrize(
     "supports_default_host,is_concrete,expected_kind,expected_has_default_host",
     [
@@ -2745,10 +2759,12 @@ def test_{{ service.name|snake_case }}_base_transport_wrap_async_method_modern_a
             assert "default_host" not in mock_wrap.call_args.kwargs
 
 
+{% 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_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()
         )
 
@@ -2796,7 +2812,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
@@ -2817,6 +2833,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 %}
diff --git a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2 b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
index 436834c20858..b3caea9f7f3b 100644
--- a/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
+++ b/packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2
@@ -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 %}
diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index 50a3c91fc5f8..6f7d127f0656 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/packages/gapic-generator/noxfile.py
@@ -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,12 +646,21 @@ 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"),
+    ],
+    ids=["grpc_rest_no_async", "rest_only", "grpc_rest_async"],
+)
 def showcase_unit(
     session,
     install_async_rest_extra=False,
     templates="DEFAULT",
     other_opts: typing.Iterable[str] = (),
+    transport: str = "grpc+rest",
 ):
     """Run the generated unit tests against the Showcase library."""
     with showcase_library(
@@ -658,6 +668,7 @@ def showcase_unit(
         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(
diff --git a/packages/gapic-generator/requirements.txt b/packages/gapic-generator/requirements.txt
index 6fa4a5f181ab..89af3db0a056 100644
--- a/packages/gapic-generator/requirements.txt
+++ b/packages/gapic-generator/requirements.txt
@@ -1164,9 +1164,7 @@ multidict==7.1.0 \
     # via
     #   aiohttp
     #   yarl
-opentelemetry-api==1.45.2 \
-    --hash=sha256:2cbd737377e710ee740d261c94e93fa0f08e6d037c18f17513cc0a2d94479a34 \
-    --hash=sha256:d7b0008e21e5fd61cbe7a7f34ca3b6905a78ab677d9bc395e299359c75cda6b4
+opentelemetry-api==1.45.1 \
     # via google-api-core
 packaging==26.3 \
     --hash=sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79 \
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 b9dd7e19527a..2a856dff2038 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
@@ -16924,6 +16924,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(),
@@ -17533,6 +17551,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.
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 8c7601cf7d18..bc6998aa0127 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
@@ -3947,6 +3947,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.
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 86fef1613499..ee22af1bd3b6 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
@@ -29582,6 +29582,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_list_locations_rest_bad_request(request_type=locations_pb2.ListLocationsRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29635,6 +29653,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29688,6 +29724,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29741,6 +29795,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29794,6 +29866,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29847,6 +29937,24 @@ def test_cancel_operation_rest(request_type):
     assert response is None
 
 
+def test_cancel_operation_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
+
 def test_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29900,6 +30008,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -29953,6 +30079,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = EventarcClient(
+        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_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -30005,6 +30149,24 @@ def test_list_operations_rest(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
+
+def test_list_operations_empty_call_rest():
+    client = EventarcClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
 def test_initialize_client_w_rest():
     client = EventarcClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -30974,6 +31136,16 @@ def test_eventarc_base_transport_with_adc():
         transport = transports.EventarcTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.EventarcTransport):
+        _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_eventarc_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 f2a5fbe88df5..97697f5ef01a 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
@@ -12891,6 +12891,16 @@ def test_config_service_v2_base_transport_with_adc():
         transport = transports.ConfigServiceV2Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.ConfigServiceV2Transport):
+        _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_config_service_v2_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 6173c39bb9bf..e3d18291f1f6 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
@@ -3483,6 +3483,16 @@ def test_logging_service_v2_base_transport_with_adc():
         transport = transports.LoggingServiceV2Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.LoggingServiceV2Transport):
+        _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_logging_service_v2_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 62b10a0b4b79..fe66a52fd749 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
@@ -3283,6 +3283,16 @@ def test_metrics_service_v2_base_transport_with_adc():
         transport = transports.MetricsServiceV2Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.MetricsServiceV2Transport):
+        _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_metrics_service_v2_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 c6052913a6f3..db1c7260ecd0 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
@@ -12891,6 +12891,16 @@ def test_config_service_v2_base_transport_with_adc():
         transport = transports.ConfigServiceV2Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.ConfigServiceV2Transport):
+        _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_config_service_v2_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 6173c39bb9bf..e3d18291f1f6 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
@@ -3483,6 +3483,16 @@ def test_logging_service_v2_base_transport_with_adc():
         transport = transports.LoggingServiceV2Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.LoggingServiceV2Transport):
+        _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_logging_service_v2_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 5eca8fde6778..20b3d620c182 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
@@ -3283,6 +3283,16 @@ def test_metrics_service_v2_base_transport_with_adc():
         transport = transports.MetricsServiceV2Transport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.MetricsServiceV2Transport):
+        _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_metrics_service_v2_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 88e200e064a5..edaf87c63bef 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
@@ -8795,6 +8795,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_list_locations_rest_bad_request(request_type=locations_pb2.ListLocationsRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8848,6 +8866,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8901,6 +8937,24 @@ def test_cancel_operation_rest(request_type):
     assert response is None
 
 
+def test_cancel_operation_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
+
 def test_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8954,6 +9008,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9007,6 +9079,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = CloudRedisClient(
+        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_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9060,6 +9150,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_wait_operation_rest_bad_request(request_type=operations_pb2.WaitOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9112,6 +9220,24 @@ def test_wait_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+
+def test_wait_operation_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.wait_operation), "__call__"
+    ) as call:
+        client.wait_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.WaitOperationRequest()
+
 def test_initialize_client_w_rest():
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -10817,6 +10943,26 @@ async def test_get_location_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, locations_pb2.Location)
 
+@pytest.mark.asyncio
+async def test_get_location_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        await client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
 @pytest.mark.asyncio
 async def test_list_locations_rest_asyncio_bad_request(request_type=locations_pb2.ListLocationsRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -10873,6 +11019,26 @@ async def test_list_locations_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
+@pytest.mark.asyncio
+async def test_list_locations_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        await client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
 @pytest.mark.asyncio
 async def test_cancel_operation_rest_asyncio_bad_request(request_type=operations_pb2.CancelOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -10929,6 +11095,26 @@ async def test_cancel_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+@pytest.mark.asyncio
+async def test_cancel_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        await client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 @pytest.mark.asyncio
 async def test_delete_operation_rest_asyncio_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -10985,6 +11171,26 @@ async def test_delete_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+@pytest.mark.asyncio
+async def test_delete_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        await client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
 @pytest.mark.asyncio
 async def test_get_operation_rest_asyncio_bad_request(request_type=operations_pb2.GetOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -11041,6 +11247,26 @@ async def test_get_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+@pytest.mark.asyncio
+async def test_get_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_operation), "__call__"
+    ) as call:
+        await 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()
+
 @pytest.mark.asyncio
 async def test_list_operations_rest_asyncio_bad_request(request_type=operations_pb2.ListOperationsRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -11097,6 +11323,26 @@ async def test_list_operations_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
+@pytest.mark.asyncio
+async def test_list_operations_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        await client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
 @pytest.mark.asyncio
 async def test_wait_operation_rest_asyncio_bad_request(request_type=operations_pb2.WaitOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -11153,6 +11399,26 @@ async def test_wait_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+@pytest.mark.asyncio
+async def test_wait_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.wait_operation), "__call__"
+    ) as call:
+        await client.wait_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.WaitOperationRequest()
+
 def test_initialize_client_w_rest_asyncio():
     if not HAS_ASYNC_REST_EXTRA:
         pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
@@ -11553,6 +11819,16 @@ def test_cloud_redis_base_transport_with_adc():
         transport = transports.CloudRedisTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.CloudRedisTransport):
+        _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_cloud_redis_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 e91d07951808..6ce13aee9d1c 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
@@ -4962,6 +4962,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_list_locations_rest_bad_request(request_type=locations_pb2.ListLocationsRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5015,6 +5033,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5068,6 +5104,24 @@ def test_cancel_operation_rest(request_type):
     assert response is None
 
 
+def test_cancel_operation_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
+
 def test_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5121,6 +5175,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5174,6 +5246,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = CloudRedisClient(
+        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_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5227,6 +5317,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_wait_operation_rest_bad_request(request_type=operations_pb2.WaitOperationRequest):
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5279,6 +5387,24 @@ def test_wait_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+
+def test_wait_operation_empty_call_rest():
+    client = CloudRedisClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.wait_operation), "__call__"
+    ) as call:
+        client.wait_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.WaitOperationRequest()
+
 def test_initialize_client_w_rest():
     client = CloudRedisClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6205,6 +6331,26 @@ async def test_get_location_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, locations_pb2.Location)
 
+@pytest.mark.asyncio
+async def test_get_location_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        await client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
 @pytest.mark.asyncio
 async def test_list_locations_rest_asyncio_bad_request(request_type=locations_pb2.ListLocationsRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -6261,6 +6407,26 @@ async def test_list_locations_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
+@pytest.mark.asyncio
+async def test_list_locations_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        await client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
 @pytest.mark.asyncio
 async def test_cancel_operation_rest_asyncio_bad_request(request_type=operations_pb2.CancelOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -6317,6 +6483,26 @@ async def test_cancel_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+@pytest.mark.asyncio
+async def test_cancel_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        await client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 @pytest.mark.asyncio
 async def test_delete_operation_rest_asyncio_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -6373,6 +6559,26 @@ async def test_delete_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+@pytest.mark.asyncio
+async def test_delete_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        await client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
 @pytest.mark.asyncio
 async def test_get_operation_rest_asyncio_bad_request(request_type=operations_pb2.GetOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -6429,6 +6635,26 @@ async def test_get_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+@pytest.mark.asyncio
+async def test_get_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_operation), "__call__"
+    ) as call:
+        await 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()
+
 @pytest.mark.asyncio
 async def test_list_operations_rest_asyncio_bad_request(request_type=operations_pb2.ListOperationsRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -6485,6 +6711,26 @@ async def test_list_operations_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
+@pytest.mark.asyncio
+async def test_list_operations_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        await client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
 @pytest.mark.asyncio
 async def test_wait_operation_rest_asyncio_bad_request(request_type=operations_pb2.WaitOperationRequest):
     if not HAS_ASYNC_REST_EXTRA:
@@ -6541,6 +6787,26 @@ async def test_wait_operation_rest_asyncio(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.Operation)
 
+@pytest.mark.asyncio
+async def test_wait_operation_empty_call_rest_asyncio():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
+    client = CloudRedisAsyncClient(
+        credentials=async_anonymous_credentials(),
+        transport="rest_asyncio",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.wait_operation), "__call__"
+    ) as call:
+        await client.wait_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.WaitOperationRequest()
+
 def test_initialize_client_w_rest_asyncio():
     if not HAS_ASYNC_REST_EXTRA:
         pytest.skip("the library must be installed with the `async_rest` extra to test this feature.")
@@ -6791,6 +7057,16 @@ def test_cloud_redis_base_transport_with_adc():
         transport = transports.CloudRedisTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.CloudRedisTransport):
+        _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_cloud_redis_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_echo.py b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_echo.py
index 11c8bc0ee587..b72839b0dc2d 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_echo.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_echo.py
@@ -3938,6 +3938,40 @@ def test_expand_rest_flattened_error(transport: str = 'rest'):
         )
 
 
+def test_collect_rest_use_cached_wrapped_rpc():
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method.wrap_method") as wrapper_fn:
+        client = EchoClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport="rest",
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._transport.collect in client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.Mock()
+        mock_rpc.return_value.name = "foo" # operation_request.operation in compute client(s) expect a string.
+        client._transport._wrapped_methods[client._transport.collect] = mock_rpc
+
+        request = [{}]
+        client.collect(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        client.collect(request)
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+        assert mock_rpc.call_count == 2
+
+
 def test_collect_rest_unimplemented():
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -3949,6 +3983,40 @@ def test_collect_rest_unimplemented():
         client.collect(requests)
 
 
+def test_chat_rest_use_cached_wrapped_rpc():
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method.wrap_method") as wrapper_fn:
+        client = EchoClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport="rest",
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._transport.chat in client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.Mock()
+        mock_rpc.return_value.name = "foo" # operation_request.operation in compute client(s) expect a string.
+        client._transport._wrapped_methods[client._transport.chat] = mock_rpc
+
+        request = [{}]
+        client.chat(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        client.chat(request)
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+        assert mock_rpc.call_count == 2
+
+
 def test_chat_rest_no_http_options():
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6697,6 +6765,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_location_rest_bad_request(request_type=locations_pb2.GetLocationRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6750,6 +6836,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6803,6 +6907,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6856,6 +6978,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6909,6 +7049,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6962,6 +7120,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -7015,6 +7191,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = EchoClient(
+        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_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -7068,6 +7262,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -7120,6 +7332,24 @@ def test_cancel_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+
+def test_cancel_operation_empty_call_rest():
+    client = EchoClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 def test_initialize_client_w_rest():
     client = EchoClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -7653,6 +7883,16 @@ def test_echo_base_transport_with_adc():
         transport = transports.EchoTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.EchoTransport):
+        _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_echo_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_identity.py b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_identity.py
index 556c1228c6cb..d5b38c0765ac 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_identity.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_identity.py
@@ -4173,6 +4173,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_location_rest_bad_request(request_type=locations_pb2.GetLocationRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4226,6 +4244,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4279,6 +4315,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4332,6 +4386,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4385,6 +4457,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4438,6 +4528,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4491,6 +4599,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = IdentityClient(
+        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_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4544,6 +4670,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4596,6 +4740,24 @@ def test_cancel_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+
+def test_cancel_operation_empty_call_rest():
+    client = IdentityClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 def test_initialize_client_w_rest():
     client = IdentityClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4795,6 +4957,16 @@ def test_identity_base_transport_with_adc():
         transport = transports.IdentityTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.IdentityTransport):
+        _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_identity_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_messaging.py b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_messaging.py
index 0dc1fa615d4d..41a42d0f4add 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_messaging.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_messaging.py
@@ -5704,6 +5704,23 @@ def test_create_blurb_rest_flattened():
         _, args, _ = req.mock_calls[0]
         assert path_template.validate("%s/v1beta1/{parent=rooms/*}/blurbs" % client.transport._host, args[1])
 
+    # Verify flattened oneof fields on the constructed request without
+    # conflicting with HTTP path template transcoding.
+    with mock.patch.object(
+        type(client.transport.create_blurb),
+        "__call__",
+    ) as call:
+        call.return_value = messaging.Blurb()
+        client.create_blurb(
+            parent='parent_value',
+            user='user_value',
+            text='text_value',
+            image=b'image_blob',
+        )
+        assert len(call.mock_calls) == 1
+        _, args, _ = call.mock_calls[0]
+        assert args[0].blurb.image == b'image_blob'
+
 
 def test_create_blurb_rest_flattened_error(transport: str = 'rest'):
     client = MessagingClient(
@@ -6470,6 +6487,11 @@ def test_search_blurbs_rest_flattened_error(transport: str = 'rest'):
         )
 
 
+def test_search_blurbs_rest_raw_page_lro():
+    response = messaging.SearchBlurbsResponse()
+    assert response.raw_page is response
+
+
 def test_stream_blurbs_rest_use_cached_wrapped_rpc():
     # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
     # instead of constructing them on each call
@@ -6582,6 +6604,40 @@ def test_stream_blurbs_rest_required_fields(request_type=messaging.StreamBlurbsR
             assert sorted(expected_params) == sorted(actual_params)
 
 
+def test_send_blurbs_rest_use_cached_wrapped_rpc():
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method.wrap_method") as wrapper_fn:
+        client = MessagingClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport="rest",
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._transport.send_blurbs in client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.Mock()
+        mock_rpc.return_value.name = "foo" # operation_request.operation in compute client(s) expect a string.
+        client._transport._wrapped_methods[client._transport.send_blurbs] = mock_rpc
+
+        request = [{}]
+        client.send_blurbs(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        client.send_blurbs(request)
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+        assert mock_rpc.call_count == 2
+
+
 def test_send_blurbs_rest_unimplemented():
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6593,6 +6649,40 @@ def test_send_blurbs_rest_unimplemented():
         client.send_blurbs(requests)
 
 
+def test_connect_rest_use_cached_wrapped_rpc():
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method.wrap_method") as wrapper_fn:
+        client = MessagingClient(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport="rest",
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._transport.connect in client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.Mock()
+        mock_rpc.return_value.name = "foo" # operation_request.operation in compute client(s) expect a string.
+        client._transport._wrapped_methods[client._transport.connect] = mock_rpc
+
+        request = [{}]
+        client.connect(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        client.connect(request)
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+        assert mock_rpc.call_count == 2
+
+
 def test_connect_rest_no_http_options():
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8797,6 +8887,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_location_rest_bad_request(request_type=locations_pb2.GetLocationRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8850,6 +8958,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8903,6 +9029,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -8956,6 +9100,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9009,6 +9171,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9062,6 +9242,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9115,6 +9313,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = MessagingClient(
+        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_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9168,6 +9384,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9220,6 +9454,24 @@ def test_cancel_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+
+def test_cancel_operation_empty_call_rest():
+    client = MessagingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 def test_initialize_client_w_rest():
     client = MessagingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -9596,6 +9848,16 @@ def test_messaging_base_transport_with_adc():
         transport = transports.MessagingTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.MessagingTransport):
+        _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_messaging_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_resumable_upload_service.py b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_resumable_upload_service.py
index f354e167b850..571ec8827130 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_resumable_upload_service.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_resumable_upload_service.py
@@ -1546,6 +1546,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_location_rest_bad_request(request_type=locations_pb2.GetLocationRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1599,6 +1617,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1652,6 +1688,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1705,6 +1759,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1758,6 +1830,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1811,6 +1901,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1864,6 +1972,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        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_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1917,6 +2043,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -1969,6 +2113,24 @@ def test_cancel_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+
+def test_cancel_operation_empty_call_rest():
+    client = ResumableUploadServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 def test_initialize_client_w_rest():
     client = ResumableUploadServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -2080,6 +2242,16 @@ def test_resumable_upload_service_base_transport_with_adc():
         transport = transports.ResumableUploadServiceTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.ResumableUploadServiceTransport):
+        _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_resumable_upload_service_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_sequence_service.py b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_sequence_service.py
index 1b64c7e1e271..5465092f872b 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_sequence_service.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_sequence_service.py
@@ -4779,6 +4779,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_location_rest_bad_request(request_type=locations_pb2.GetLocationRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4832,6 +4850,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4885,6 +4921,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4938,6 +4992,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4991,6 +5063,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5044,6 +5134,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5097,6 +5205,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = SequenceServiceClient(
+        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_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5150,6 +5276,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5202,6 +5346,24 @@ def test_cancel_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+
+def test_cancel_operation_empty_call_rest():
+    client = SequenceServiceClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 def test_initialize_client_w_rest():
     client = SequenceServiceClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5423,6 +5585,16 @@ def test_sequence_service_base_transport_with_adc():
         transport = transports.SequenceServiceTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.SequenceServiceTransport):
+        _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_sequence_service_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
diff --git a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_testing.py b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_testing.py
index e453f8c98db9..363b0e063f95 100755
--- a/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_testing.py
+++ b/packages/gapic-generator/tests/integration/goldens/showcase/tests/unit/gapic/showcase_v1beta1/test_testing.py
@@ -4929,6 +4929,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_get_location_rest_bad_request(request_type=locations_pb2.GetLocationRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -4982,6 +5000,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_set_iam_policy_rest_bad_request(request_type=iam_policy_pb2.SetIamPolicyRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5035,6 +5071,24 @@ def test_set_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_set_iam_policy_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.set_iam_policy), "__call__"
+    ) as call:
+        client.set_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.SetIamPolicyRequest()
+
+
 def test_get_iam_policy_rest_bad_request(request_type=iam_policy_pb2.GetIamPolicyRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5088,6 +5142,24 @@ def test_get_iam_policy_rest(request_type):
     assert isinstance(response, policy_pb2.Policy)
 
 
+def test_get_iam_policy_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_iam_policy), "__call__"
+    ) as call:
+        client.get_iam_policy(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.GetIamPolicyRequest()
+
+
 def test_test_iam_permissions_rest_bad_request(request_type=iam_policy_pb2.TestIamPermissionsRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5141,6 +5213,24 @@ def test_test_iam_permissions_rest(request_type):
     assert isinstance(response, iam_policy_pb2.TestIamPermissionsResponse)
 
 
+def test_test_iam_permissions_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.test_iam_permissions), "__call__"
+    ) as call:
+        client.test_iam_permissions(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == iam_policy_pb2.TestIamPermissionsRequest()
+
+
 def test_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5194,6 +5284,24 @@ def test_list_operations_rest(request_type):
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
 
+def test_list_operations_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5247,6 +5355,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = TestingClient(
+        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_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5300,6 +5426,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5352,6 +5496,24 @@ def test_cancel_operation_rest(request_type):
     # Establish that the response is the type that we expect.
     assert response is None
 
+
+def test_cancel_operation_empty_call_rest():
+    client = TestingClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
 def test_initialize_client_w_rest():
     client = TestingClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -5617,6 +5779,16 @@ def test_testing_base_transport_with_adc():
         transport = transports.TestingTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.TestingTransport):
+        _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_testing_auth_adc():
     # If no credentials are provided, we should use ADC credentials.
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 522152f1acbc..e2317f7760b1 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
@@ -6321,6 +6321,24 @@ def test_get_location_rest(request_type):
     assert isinstance(response, locations_pb2.Location)
 
 
+def test_get_location_empty_call_rest():
+    client = StorageBatchOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.get_location), "__call__"
+    ) as call:
+        client.get_location(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.GetLocationRequest()
+
+
 def test_list_locations_rest_bad_request(request_type=locations_pb2.ListLocationsRequest):
     client = StorageBatchOperationsClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6374,6 +6392,24 @@ def test_list_locations_rest(request_type):
     assert isinstance(response, locations_pb2.ListLocationsResponse)
 
 
+def test_list_locations_empty_call_rest():
+    client = StorageBatchOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_locations), "__call__"
+    ) as call:
+        client.list_locations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == locations_pb2.ListLocationsRequest()
+
+
 def test_cancel_operation_rest_bad_request(request_type=operations_pb2.CancelOperationRequest):
     client = StorageBatchOperationsClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6427,6 +6463,24 @@ def test_cancel_operation_rest(request_type):
     assert response is None
 
 
+def test_cancel_operation_empty_call_rest():
+    client = StorageBatchOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.cancel_operation), "__call__"
+    ) as call:
+        client.cancel_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.CancelOperationRequest()
+
+
 def test_delete_operation_rest_bad_request(request_type=operations_pb2.DeleteOperationRequest):
     client = StorageBatchOperationsClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6480,6 +6534,24 @@ def test_delete_operation_rest(request_type):
     assert response is None
 
 
+def test_delete_operation_empty_call_rest():
+    client = StorageBatchOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.delete_operation), "__call__"
+    ) as call:
+        client.delete_operation(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.DeleteOperationRequest()
+
+
 def test_get_operation_rest_bad_request(request_type=operations_pb2.GetOperationRequest):
     client = StorageBatchOperationsClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6533,6 +6605,24 @@ def test_get_operation_rest(request_type):
     assert isinstance(response, operations_pb2.Operation)
 
 
+def test_get_operation_empty_call_rest():
+    client = StorageBatchOperationsClient(
+        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_list_operations_rest_bad_request(request_type=operations_pb2.ListOperationsRequest):
     client = StorageBatchOperationsClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6585,6 +6675,24 @@ def test_list_operations_rest(request_type):
     # Establish that the response is the type that we expect.
     assert isinstance(response, operations_pb2.ListOperationsResponse)
 
+
+def test_list_operations_empty_call_rest():
+    client = StorageBatchOperationsClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+        transport="rest",
+    )
+
+    # Mock the actual call, and fake the request.
+    with mock.patch.object(
+        type(client.transport.list_operations), "__call__"
+    ) as call:
+        client.list_operations(request=None)
+
+        # Establish that the underlying stub method was called.
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == operations_pb2.ListOperationsRequest()
+
 def test_initialize_client_w_rest():
     client = StorageBatchOperationsClient(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -6856,6 +6964,16 @@ def test_storage_batch_operations_base_transport_with_adc():
         transport = transports.StorageBatchOperationsTransport()
         adc.assert_called_once()
 
+    # Test _ignore_credentials=True skips ADC and host without port defaults to :443.
+    class IgnoreCredsTransport(transports.StorageBatchOperationsTransport):
+        _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_storage_batch_operations_auth_adc():
     # If no credentials are provided, we should use ADC credentials.