address feedback

Author: ohmayr

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

@@ -424,7 +424,7 @@ async def intercept_stream_stream(
 
 
 @pytest.fixture
-def intercepted_echo_grpc(use_tls):
+def intercepted_echo_grpc(use_mtls, use_tls):
     # 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,11 +433,12 @@ def intercepted_echo_grpc(use_tls):
         "intercepted",
     )
     host = "localhost:7469"
-    channel = (
-        grpc.secure_channel(host, tls_credentials)
-        if use_tls
-        else grpc.insecure_channel(host)
-    )
+    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)
     intercept_channel = grpc.intercept_channel(channel, interceptor)
     transport = EchoClient.get_transport_class("grpc")(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -447,7 +448,7 @@ def intercepted_echo_grpc(use_tls):
 
 
 @pytest_asyncio.fixture
-async def intercepted_echo_grpc_async(use_tls):
+async def intercepted_echo_grpc_async(use_mtls, use_tls):
     # 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.
@@ -456,11 +457,12 @@ async def intercepted_echo_grpc_async(use_tls):
         "intercepted",
     )
     host = "localhost:7469"
-    channel = (
-        grpc.aio.secure_channel(host, tls_credentials, interceptors=[interceptor])
-        if use_tls
-        else grpc.aio.insecure_channel(host, interceptors=[interceptor])
-    )
+    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])
     transport = EchoAsyncClient.get_transport_class("grpc_asyncio")(
         credentials=ga_credentials.AnonymousCredentials(),
         channel=channel,
@@ -476,21 +478,23 @@ def cert_verify(self, conn, url, verify, cert):
 
 
 @pytest.fixture
-def intercepted_echo_rest(use_tls):
+def intercepted_echo_rest(use_mtls, use_tls):
     transport_name = "rest"
     transport_cls = EchoClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestInterceptor()
 
-    url_scheme = "https" if use_tls else "http"
+    url_scheme = "https" if (use_mtls or use_tls) else "http"
     transport = transport_cls(
         credentials=ga_credentials.AnonymousCredentials(),
         host="localhost:7469",
         url_scheme=url_scheme,
         interceptor=interceptor,
     )
-    if use_tls:
+    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
 

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

@@ -34,10 +34,11 @@ def _verify_pqc_metadata(interceptor, transport_name):
     assert negotiated_group is not None, "Failed: Showcase server did not return negotiated TLS group header."
     assert supported_groups is not None, "Failed: Showcase server did not return client advertised supported groups."
 
-    # Enforce PQC compliance (X25519MLKEM768)
+    # Enforce PQC compliance by verifying an MLKEM group (e.g. X25519MLKEM768) was negotiated.
+    # Substring check ("MLKEM" in ...) allows compatibility across gRPC and REST implementation strings.
     assert (
         "MLKEM" in negotiated_group
-    ), f"Failed: {transport_name} Connection did not negotiate X25519MLKEM768! Negotiated: {negotiated_group}"
+    ), f"Failed: {transport_name} Connection did not negotiate an MLKEM group (e.g. X25519MLKEM768)! Negotiated: {negotiated_group}"
 
 
 def test_pqc_grpc(intercepted_echo_grpc):