feat(api-core): add endpoint routing logic to gapic_v1 public helpers

Author: hebaalazzeh

.github/workflows/import-profiler.yml

@@ -26,7 +26,6 @@ jobs:
       with:
         python-version: "3.15"
         allow-prereleases: true
-        cache: 'pip'
     - name: Run import profiler
       env:
         BUILD_TYPE: presubmit

ci/run_single_test.sh

@@ -101,7 +101,6 @@ case ${TEST_TYPE} in
             echo "Creating temporary virtualenv for import profile..."
             python3 -m venv .venv-profiler
             source .venv-profiler/bin/activate
-            export PYO3_USE_ABI3_FORWARD_COMPATIBILITY=1
             python -m pip install --upgrade pip setuptools
 
             PACKAGE_NAME=$(basename $(pwd))

packages/gapic-generator/noxfile.py

@@ -474,24 +474,6 @@ def showcase_mtls(
         )
 
 
-# TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
-# Remove showcase_pqc once grpcio >= 1.83.0 is enforced by default.
-@nox.session(python=NEWEST_PYTHON)
-def showcase_pqc(
-    session,
-    templates="DEFAULT",
-    other_opts: typing.Iterable[str] = (),
-    env: typing.Optional[typing.Dict[str, str]] = {},
-):
-    """Run the Showcase PQC verification test suite against grpcio 1.83+ over standard TLS."""
-    with showcase_library(session, templates=templates, other_opts=other_opts):
-        session.install("pytest", "pytest-asyncio")
-        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751):
-        # Update the version below to `1.83.0` once released, and remove `--pre`.
-        session.install("--pre", "--upgrade", "grpcio>=1.83.0rc0", "grpcio-status>=1.83.0rc0")
-        session.run("py.test", "--quiet", "--tls", *(session.posargs or ["tests/system/test_pqc.py"]), env=env)
-
-
 def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False):
     session.install(
         "coverage",

packages/gapic-generator/tests/system/conftest.py

@@ -18,7 +18,6 @@
 import os
 import pytest
 import pytest_asyncio
-from requests.adapters import HTTPAdapter
 
 from typing import Sequence, Tuple
 
@@ -114,18 +113,15 @@ def async_identity(use_mtls, request, event_loop):
         )
 
 
-base_dir = os.path.dirname(__file__)
-CERT_PATH = os.path.join(base_dir, "../cert/mtls.crt")
-KEY_PATH = os.path.join(base_dir, "../cert/mtls.key")
-with open(CERT_PATH, "rb") as fh:
+dir = os.path.dirname(__file__)
+with open(os.path.join(dir, "../cert/mtls.crt"), "rb") as fh:
     cert = fh.read()
-with open(KEY_PATH, "rb") as fh:
+with open(os.path.join(dir, "../cert/mtls.key"), "rb") as fh:
     key = fh.read()
 
 ssl_credentials = grpc.ssl_channel_credentials(
     root_certificates=cert, certificate_chain=cert, private_key=key
 )
-tls_credentials = grpc.ssl_channel_credentials(root_certificates=cert)
 
 
 def callback():
@@ -140,9 +136,6 @@ def pytest_addoption(parser):
     parser.addoption(
         "--mtls", action="store_true", help="Run system test with mutual TLS channel"
     )
-    parser.addoption(
-        "--tls", action="store_true", help="Run system test with standard one-way TLS channel"
-    )
 
 
 # TODO: Need to test  without passing in a transport class
@@ -196,11 +189,6 @@ def use_mtls(request):
     return request.config.getoption("--mtls")
 
 
-@pytest.fixture
-def use_tls(request):
-    return request.config.getoption("--tls") or request.config.getoption("--mtls")
-
-
 @pytest.fixture
 def parametrized_echo(
     use_mtls,
@@ -340,7 +328,7 @@ def _read_response_metadata_stream(self):
     def intercept_unary_unary(self, continuation, client_call_details, request):
         self._add_request_metadata(client_call_details)
         response = continuation(client_call_details, request)
-        metadata = [(k, str(v)) for k, v in response.initial_metadata()] + [(k, str(v)) for k, v in response.trailing_metadata()]
+        metadata = [(k, str(v)) for k, v in response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
@@ -399,7 +387,7 @@ async def _add_request_metadata(self, client_call_details):
     async def intercept_unary_unary(self, continuation, client_call_details, request):
         await self._add_request_metadata(client_call_details)
         response = await continuation(client_call_details, request)
-        metadata = [(k, str(v)) for k, v in await response.initial_metadata()] + [(k, str(v)) for k, v in await response.trailing_metadata()]
+        metadata = [(k, str(v)) for k, v in await response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
@@ -424,7 +412,7 @@ async def intercept_stream_stream(
 
 
 @pytest.fixture
-def intercepted_echo_grpc(use_mtls, use_tls):
+def intercepted_echo_grpc(use_mtls):
     # The interceptor adds 'showcase-trailer' client metadata. Showcase server
     # echoes any metadata with key 'showcase-trailer', so the same metadata
     # should appear as trailing metadata in the response.
@@ -433,12 +421,11 @@ def intercepted_echo_grpc(use_mtls, use_tls):
         "intercepted",
     )
     host = "localhost:7469"
-    if use_mtls:
-        channel = grpc.secure_channel(host, ssl_credentials)
-    elif use_tls:
-        channel = grpc.secure_channel(host, tls_credentials)
-    else:
-        channel = grpc.insecure_channel(host)
+    channel = (
+        grpc.secure_channel(host, ssl_credentials)
+        if use_mtls
+        else grpc.insecure_channel(host)
+    )
     intercept_channel = grpc.intercept_channel(channel, interceptor)
     transport = EchoClient.get_transport_class("grpc")(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -448,7 +435,7 @@ def intercepted_echo_grpc(use_mtls, use_tls):
 
 
 @pytest_asyncio.fixture
-async def intercepted_echo_grpc_async(use_mtls, use_tls):
+async def intercepted_echo_grpc_async():
     # The interceptor adds 'showcase-trailer' client metadata. Showcase server
     # echoes any metadata with key 'showcase-trailer', so the same metadata
     # should appear as trailing metadata in the response.
@@ -457,45 +444,28 @@ async def intercepted_echo_grpc_async(use_mtls, use_tls):
         "intercepted",
     )
     host = "localhost:7469"
-    if use_mtls:
-        channel = grpc.aio.secure_channel(host, ssl_credentials, interceptors=[interceptor])
-    elif use_tls:
-        channel = grpc.aio.secure_channel(host, tls_credentials, interceptors=[interceptor])
-    else:
-        channel = grpc.aio.insecure_channel(host, interceptors=[interceptor])
+    channel = grpc.aio.insecure_channel(host, interceptors=[interceptor])
+    # intercept_channel = grpc.aio.intercept_channel(channel, interceptor)
     transport = EchoAsyncClient.get_transport_class("grpc_asyncio")(
         credentials=ga_credentials.AnonymousCredentials(),
         channel=channel,
     )
     return EchoAsyncClient(transport=transport), interceptor
 
 
-class HostNameIgnoringAdapter(HTTPAdapter):
-    """Custom HTTPAdapter that disables hostname verification for local self-signed certs."""
-    def cert_verify(self, conn, url, verify, cert):
-        super().cert_verify(conn, url, verify, cert)
-        conn.assert_hostname = False
-
-
 @pytest.fixture
-def intercepted_echo_rest(use_mtls, use_tls):
+def intercepted_echo_rest():
     transport_name = "rest"
     transport_cls = EchoClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestInterceptor()
 
-    url_scheme = "https" if (use_mtls or use_tls) else "http"
+    # The custom host explicitly bypasses https.
     transport = transport_cls(
         credentials=ga_credentials.AnonymousCredentials(),
         host="localhost:7469",
-        url_scheme=url_scheme,
+        url_scheme="http",
         interceptor=interceptor,
     )
-    if use_mtls or use_tls:
-        transport._session.verify = CERT_PATH
-        transport._session.mount("https://", HostNameIgnoringAdapter())
-    if use_mtls:
-        transport._session.cert = (CERT_PATH, KEY_PATH)
-
     return EchoClient(transport=transport), interceptor
 
 
@@ -508,11 +478,11 @@ def intercepted_echo_rest_async():
     transport_cls = EchoAsyncClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestAsyncInterceptor()
 
+    # The custom host explicitly bypasses https.
     transport = transport_cls(
         credentials=async_anonymous_credentials(),
         host="localhost:7469",
         url_scheme="http",
         interceptor=interceptor,
     )
-
     return EchoAsyncClient(transport=transport), interceptor

packages/gapic-generator/tests/system/test_pqc.py

@@ -2,11 +2,8 @@
 branch = True
 
 [report]
-fail_under = 100
+fail_under = 99
 show_missing = True
-omit =
-    tests/*
-    */tests/*
 exclude_lines =
     # Re-enable the standard pragma
     pragma: NO COVER

packages/google-api-core/.coveragerc

@@ -25,10 +25,12 @@
 # Older Python versions safely ignore this variable.
 __lazy_modules__: Set[str] = {
     "google.api_core.gapic_v1.client_info",
+    "google.api_core.gapic_v1.client_utils",
     "google.api_core.gapic_v1.requests",
     "google.api_core.gapic_v1.routing_header",
 }
-__all__ = ["client_info", "requests", "routing_header"]
+__all__ = ["client_info", "client_utils", "requests", "routing_header"]
+
 
 if _has_grpc:
     __lazy_modules__.update(
@@ -42,6 +44,7 @@
 
 from google.api_core.gapic_v1 import (  # noqa: E402
     client_info,
+    client_utils,
     requests,
     routing_header,
 )