update tests

Author: ohmayr

packages/gapic-generator/noxfile.py

@@ -457,7 +457,7 @@ def showcase_mtls(
     """Run the Showcase mtls test suite."""
 
     with showcase_library(session, templates=templates, other_opts=other_opts):
-        session.install("pytest", "pytest-asyncio", "pyopenssl")
+        session.install("pytest", "pytest-asyncio")
         test_directory = Path("tests", "system")
         ignore_file = env.get("IGNORE_FILE")
         pytest_command = [
@@ -486,9 +486,9 @@ def showcase_pqc(
 ):
     """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", "pyopenssl")
+        session.install("pytest", "pytest-asyncio")
         session.install("--pre", "--upgrade", "grpcio>=1.83.0rc0", "grpcio-status>=1.83.0rc0")
-        session.run("py.test", "--quiet", "--tls", "-s", *(session.posargs or ["tests/system/test_pqc.py"]), env=env)
+        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):

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

@@ -400,7 +400,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.trailing_metadata()]
+        metadata = [(k, str(v)) for k, v in await response.initial_metadata()] + [(k, str(v)) for k, v in await response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
@@ -449,7 +449,7 @@ def intercepted_echo_grpc(use_mtls, use_tls):
 
 
 @pytest_asyncio.fixture
-async def intercepted_echo_grpc_async():
+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.
@@ -458,8 +458,12 @@ async def intercepted_echo_grpc_async():
         "intercepted",
     )
     host = "localhost:7469"
-    channel = grpc.aio.insecure_channel(host, interceptors=[interceptor])
-    # intercept_channel = grpc.aio.intercept_channel(channel, 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,

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

@@ -24,12 +24,8 @@ def run_pqc_test(use_tls):
         pytest.skip("PQC integration test requires TLS (--tls or --mtls flag) to be enabled.")
 
 
-def _verify_pqc_negotiated_group(client, interceptor, transport_name):
-    # Make secure call using standard GAPIC client library fixture
-    response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
-    assert response.content == "Verify PQC connection."
-
-    # Extract negotiated group and supported groups from response headers
+def _verify_pqc_metadata(interceptor, transport_name):
+    """Extracts and verifies negotiated PQC group and supported groups from interceptor metadata."""
     response_metadata = getattr(interceptor, "response_metadata", []) or []
     headers = {key.lower(): value for key, value in response_metadata}
     negotiated_group = headers.get("x-showcase-tls-group")
@@ -38,12 +34,10 @@ def _verify_pqc_negotiated_group(client, 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."
 
-    print(f"\n[PQC Verification] ({transport_name}) Negotiated TLS Group: {negotiated_group}")
-    print(f"[PQC Verification] ({transport_name}) Client Advertised Supported Groups: {supported_groups}")
-
     # Enforce PQC compliance (X25519MLKEM768 or Kyber)
-    assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
-        f"Failed: {transport_name} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
+    assert (
+        "MLKEM" in negotiated_group or "Kyber" in negotiated_group
+    ), f"Failed: {transport_name} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
 
 
 def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
@@ -53,12 +47,31 @@ def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
         pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
 
     client, interceptor = intercepted_echo_grpc
-    _verify_pqc_negotiated_group(client, interceptor, "grpc")
+    response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
+    assert response.content == "Verify PQC connection."
+    _verify_pqc_metadata(interceptor, "grpc")
 
 
 def test_pqc_rest(run_pqc_test, intercepted_echo_rest):
     """Verifies that the REST client library negotiates PQC (X25519MLKEM768) with Showcase server."""
     client, interceptor = intercepted_echo_rest
-    _verify_pqc_negotiated_group(client, interceptor, "rest")
+    response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
+    assert response.content == "Verify PQC connection."
+    _verify_pqc_metadata(interceptor, "rest")
+
+
+@pytest.mark.asyncio
+async def test_pqc_grpc_async(run_pqc_test, intercepted_echo_grpc_async):
+    """Verifies that the async gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    # TODO(Phase 3): Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
+    if Version(grpc.__version__) < Version("1.83.0rc0"):
+        pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
+
+    client, interceptor = intercepted_echo_grpc_async
+    response = await client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
+    assert response.content == "Verify PQC connection."
+    _verify_pqc_metadata(interceptor, "grpc_asyncio")
+
+