From bdad0534e4c13800505c0eed5b46dc9cff03d45f Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 26 Jun 2026 20:00:30 +0000
Subject: [PATCH 01/18] feat: add post-quantum cryptography (PQC) integration
 tests and fixtures

---
 .../gapic-generator/tests/system/conftest.py  | 44 +++++++++++++---
 .../gapic-generator/tests/system/test_pqc.py  | 52 +++++++++++++++++++
 2 files changed, 89 insertions(+), 7 deletions(-)
 create mode 100644 packages/gapic-generator/tests/system/test_pqc.py

diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 180e48b8d59a..5c00c99a5976 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/packages/gapic-generator/tests/system/conftest.py
@@ -18,6 +18,8 @@
 import os
 import pytest
 import pytest_asyncio
+from requests.adapters import HTTPAdapter
+from urllib3.poolmanager import PoolManager
 
 from typing import Sequence, Tuple
 
@@ -328,7 +330,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.trailing_metadata()]
+        metadata = [(k, str(v)) for k, v in response.initial_metadata()] + [(k, str(v)) for k, v in response.trailing_metadata()]
         self.response_metadata = metadata
         return response
 
@@ -453,24 +455,44 @@ async def intercepted_echo_grpc_async():
     return EchoAsyncClient(transport=transport), interceptor
 
 
+class HostNameIgnoringAdapter(HTTPAdapter):
+    """Custom HTTPAdapter that disables hostname verification for local self-signed certs."""
+    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
+        self.poolmanager = PoolManager(
+            num_pools=connections,
+            maxsize=maxsize,
+            block=block,
+            assert_hostname=False,
+            **pool_kwargs
+        )
+
+
 @pytest.fixture
-def intercepted_echo_rest():
+def intercepted_echo_rest(use_mtls):
     transport_name = "rest"
     transport_cls = EchoClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestInterceptor()
 
-    # The custom host explicitly bypasses https.
+    url_scheme = "https" if use_mtls else "http"
     transport = transport_cls(
         credentials=ga_credentials.AnonymousCredentials(),
         host="localhost:7469",
-        url_scheme="http",
+        url_scheme=url_scheme,
         interceptor=interceptor,
     )
+    if use_mtls:
+        dir = os.path.dirname(__file__)
+        cert_path = os.path.join(dir, "../cert/mtls.crt")
+        key_path = os.path.join(dir, "../cert/mtls.key")
+        transport._session.verify = cert_path
+        transport._session.cert = (cert_path, key_path)
+        transport._session.mount("https://", HostNameIgnoringAdapter())
+
     return EchoClient(transport=transport), interceptor
 
 
 @pytest.fixture
-def intercepted_echo_rest_async():
+def intercepted_echo_rest_async(use_mtls):
     if not HAS_ASYNC_REST_ECHO_TRANSPORT:
         pytest.skip("Skipping test with async rest.")
 
@@ -478,11 +500,19 @@ def intercepted_echo_rest_async():
     transport_cls = EchoAsyncClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestAsyncInterceptor()
 
-    # The custom host explicitly bypasses https.
+    url_scheme = "https" if use_mtls else "http"
     transport = transport_cls(
         credentials=async_anonymous_credentials(),
         host="localhost:7469",
-        url_scheme="http",
+        url_scheme=url_scheme,
         interceptor=interceptor,
     )
+    if use_mtls:
+        dir = os.path.dirname(__file__)
+        cert_path = os.path.join(dir, "../cert/mtls.crt")
+        key_path = os.path.join(dir, "../cert/mtls.key")
+        transport._session.verify = cert_path
+        transport._session.cert = (cert_path, key_path)
+        transport._session.mount("https://", HostNameIgnoringAdapter())
+
     return EchoAsyncClient(transport=transport), interceptor
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
new file mode 100644
index 000000000000..127c2f9edc74
--- /dev/null
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -0,0 +1,52 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+from google import showcase
+
+@pytest.fixture
+def run_pqc_test(use_mtls):
+    if not use_mtls:
+        pytest.skip("PQC integration test requires mTLS (--mtls flag) to be enabled.")
+
+@pytest.mark.parametrize(
+    "transport_fixture",
+    ["intercepted_echo_grpc", "intercepted_echo_rest"]
+)
+def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
+    """Verifies that the generated client library negotiates PQC with the Showcase server."""
+    client, interceptor = request.getfixturevalue(transport_fixture)
+    
+    # Make secure call using the standard 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
+    negotiated_group = None
+    supported_groups = None
+    for key, value in interceptor.response_metadata:
+        if key.lower() == "x-showcase-tls-group":
+            negotiated_group = value
+        elif key.lower() == "x-showcase-tls-client-supported-groups":
+            supported_groups = value
+            
+    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_fixture}) Negotiated TLS Group: {negotiated_group}")
+    print(f"[PQC Verification] ({transport_fixture}) Client Advertised Supported Groups: {supported_groups}")
+    
+    # Enforce PQC compliance (this will fail if not using MLKEM/Kyber)
+    assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
+        f"Failed: {transport_fixture} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"

From 6c043e60edea5e5482b01299d6060304153e7410 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Mon, 29 Jun 2026 20:09:11 +0000
Subject: [PATCH 02/18] update

---
 packages/gapic-generator/noxfile.py           |  2 +-
 .../gapic-generator/tests/system/test_pqc.py  | 45 +++++++++++++++++++
 2 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index 6fac5c48853d..343626eab4e4 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/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")
+        session.install("pytest", "pytest-asyncio", "pyopenssl")
         test_directory = Path("tests", "system")
         ignore_file = env.get("IGNORE_FILE")
         pytest_command = [
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 127c2f9edc74..0db865f5a39f 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -12,7 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import os
 import pytest
+import requests
 from google import showcase
 
 @pytest.fixture
@@ -50,3 +52,46 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
     # Enforce PQC compliance (this will fail if not using MLKEM/Kyber)
     assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
         f"Failed: {transport_fixture} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
+
+
+def test_google_auth_transport_pqc(run_pqc_test):
+    """Verifies that the google-auth HTTP transport adapter negotiates PQC with the Showcase server."""
+    import google.auth.transport.requests
+    from conftest import HostNameIgnoringAdapter
+
+    # 1. Initialize a standard requests Session with mTLS certs
+    session = requests.Session()
+    cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt"
+    key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key"
+
+    session.verify = cert_path
+    session.cert = (cert_path, key_path)
+    
+    # Bypass localhost hostname mismatch
+    session.mount("https://", HostNameIgnoringAdapter())
+
+    # 2. Wrap it in google-auth's Transport Request adapter
+    # This is the exact object google-auth uses to execute its own HTTP/REST requests.
+    auth_transport = google.auth.transport.requests.Request(session=session)
+
+    # 3. Make secure call using the google-auth transport adapter
+    url = "https://localhost:7469/v1beta1/echo:echo"
+    method = "POST"
+    headers = {"Content-Type": "application/json"}
+    body = b'{"content": "Verify google-auth transport PQC connection."}'
+
+    # Execute the request through google-auth's transport layer
+    response = auth_transport(url=url, method=method, body=body, headers=headers)
+    assert response.status == 200
+
+    # 4. Extract TLS group from response headers returned by Showcase
+    negotiated_group = response.headers.get("x-showcase-tls-group")
+    supported_groups = response.headers.get("x-showcase-tls-client-supported-groups")
+
+    assert negotiated_group is not None, "Failed: Showcase server did not return negotiated TLS group header."
+    print(f"\n[google-auth Transport PQC] Negotiated TLS Group: {negotiated_group}")
+    print(f"[google-auth Transport PQC] Client Advertised Supported Groups: {supported_groups}")
+
+    # Assert PQC compliance
+    assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
+        f"Failed: google-auth Transport is NOT PQC-compliant! Negotiated: {negotiated_group}"

From 449f2e6342fffff99328e2f80ff827f4200d9d73 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Mon, 29 Jun 2026 20:12:29 +0000
Subject: [PATCH 03/18] debug: disable verification in google-auth test

---
 packages/gapic-generator/tests/system/test_pqc.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 0db865f5a39f..59f47a65d946 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -64,7 +64,8 @@ def test_google_auth_transport_pqc(run_pqc_test):
     cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt"
     key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key"
 
-    session.verify = cert_path
+    # Disable verification temporarily to isolate hostname/cert validation errors
+    session.verify = False
     session.cert = (cert_path, key_path)
     
     # Bypass localhost hostname mismatch

From f47628385730a8882293c29d308d782749237e98 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Mon, 29 Jun 2026 20:13:57 +0000
Subject: [PATCH 04/18] feat: use MessageToJson for google-auth test payload

---
 packages/gapic-generator/tests/system/test_pqc.py | 14 ++++++++------
 1 file changed, 8 insertions(+), 6 deletions(-)

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 59f47a65d946..f5b5686bcb86 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -57,6 +57,7 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
 def test_google_auth_transport_pqc(run_pqc_test):
     """Verifies that the google-auth HTTP transport adapter negotiates PQC with the Showcase server."""
     import google.auth.transport.requests
+    from google.protobuf.json_format import MessageToJson
     from conftest import HostNameIgnoringAdapter
 
     # 1. Initialize a standard requests Session with mTLS certs
@@ -64,28 +65,29 @@ def test_google_auth_transport_pqc(run_pqc_test):
     cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt"
     key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key"
 
-    # Disable verification temporarily to isolate hostname/cert validation errors
-    session.verify = False
+    session.verify = cert_path
     session.cert = (cert_path, key_path)
     
     # Bypass localhost hostname mismatch
     session.mount("https://", HostNameIgnoringAdapter())
 
     # 2. Wrap it in google-auth's Transport Request adapter
-    # This is the exact object google-auth uses to execute its own HTTP/REST requests.
     auth_transport = google.auth.transport.requests.Request(session=session)
 
-    # 3. Make secure call using the google-auth transport adapter
+    # 3. Serialize the request body using the official protobuf JSON serializer
+    req = showcase.EchoRequest(content="Verify google-auth transport PQC connection.")
+    body = MessageToJson(req, including_default_value_fields=True).encode("utf-8")
+
+    # 4. Make secure call using the google-auth transport adapter
     url = "https://localhost:7469/v1beta1/echo:echo"
     method = "POST"
     headers = {"Content-Type": "application/json"}
-    body = b'{"content": "Verify google-auth transport PQC connection."}'
 
     # Execute the request through google-auth's transport layer
     response = auth_transport(url=url, method=method, body=body, headers=headers)
     assert response.status == 200
 
-    # 4. Extract TLS group from response headers returned by Showcase
+    # 5. Extract TLS group from response headers returned by Showcase
     negotiated_group = response.headers.get("x-showcase-tls-group")
     supported_groups = response.headers.get("x-showcase-tls-client-supported-groups")
 

From 2d0fe09e354b3898463eb9a84554b6fe2408bde6 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Tue, 7 Jul 2026 06:10:37 +0000
Subject: [PATCH 05/18] wip

---
 .../gapic-generator/tests/system/test_pqc.py  | 48 +++++++++----------
 packages/google-auth/tests/crypt/test_pqc.py  | 44 +++++++++++++++++
 2 files changed, 67 insertions(+), 25 deletions(-)
 create mode 100644 packages/google-auth/tests/crypt/test_pqc.py

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index f5b5686bcb86..6b6876272c65 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -12,28 +12,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 import pytest
 import requests
 from google import showcase
 
+
 @pytest.fixture
 def run_pqc_test(use_mtls):
     if not use_mtls:
         pytest.skip("PQC integration test requires mTLS (--mtls flag) to be enabled.")
 
+
 @pytest.mark.parametrize(
     "transport_fixture",
     ["intercepted_echo_grpc", "intercepted_echo_rest"]
 )
 def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
-    """Verifies that the generated client library negotiates PQC with the Showcase server."""
+    """Verifies that the generated client library negotiates PQC (X25519MLKEM768) with Showcase server."""
     client, interceptor = request.getfixturevalue(transport_fixture)
-    
-    # Make secure call using the standard client library fixture
+
+    # 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
     negotiated_group = None
     supported_groups = None
@@ -42,52 +43,50 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
             negotiated_group = value
         elif key.lower() == "x-showcase-tls-client-supported-groups":
             supported_groups = value
-            
+
     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_fixture}) Negotiated TLS Group: {negotiated_group}")
     print(f"[PQC Verification] ({transport_fixture}) Client Advertised Supported Groups: {supported_groups}")
-    
-    # Enforce PQC compliance (this will fail if not using MLKEM/Kyber)
+
+    # Enforce PQC compliance (X25519MLKEM768 or Kyber)
     assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
         f"Failed: {transport_fixture} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
 
 
 def test_google_auth_transport_pqc(run_pqc_test):
-    """Verifies that the google-auth HTTP transport adapter negotiates PQC with the Showcase server."""
+    """Verifies that google-auth transport adapter negotiates PQC (X25519MLKEM768) with Showcase server."""
     import google.auth.transport.requests
-    from google.protobuf.json_format import MessageToJson
     from conftest import HostNameIgnoringAdapter
 
-    # 1. Initialize a standard requests Session with mTLS certs
+    # 1. Initialize requests Session with mTLS certs
     session = requests.Session()
     cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt"
     key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key"
 
     session.verify = cert_path
     session.cert = (cert_path, key_path)
-    
-    # Bypass localhost hostname mismatch
     session.mount("https://", HostNameIgnoringAdapter())
 
-    # 2. Wrap it in google-auth's Transport Request adapter
+    # 2. Wrap session in google-auth transport adapter
     auth_transport = google.auth.transport.requests.Request(session=session)
 
-    # 3. Serialize the request body using the official protobuf JSON serializer
+    # 3. Serialize request body
     req = showcase.EchoRequest(content="Verify google-auth transport PQC connection.")
-    body = MessageToJson(req, including_default_value_fields=True).encode("utf-8")
+    body = showcase.EchoRequest.to_json(req, including_default_value_fields=False).encode("utf-8")
 
-    # 4. Make secure call using the google-auth transport adapter
+    # 4. Execute request through google-auth's transport layer
     url = "https://localhost:7469/v1beta1/echo:echo"
-    method = "POST"
-    headers = {"Content-Type": "application/json"}
+    headers = {
+        "Content-Type": "application/json",
+        "x-goog-api-client": "gapic/1.0 rest/1.0",
+    }
 
-    # Execute the request through google-auth's transport layer
-    response = auth_transport(url=url, method=method, body=body, headers=headers)
-    assert response.status == 200
+    response = auth_transport(url=url, method="POST", body=body, headers=headers)
+    assert response.status == 200, f"Failed: status={response.status}, body={response.data}"
 
-    # 5. Extract TLS group from response headers returned by Showcase
+    # 5. Extract and verify negotiated TLS group returned by Showcase
     negotiated_group = response.headers.get("x-showcase-tls-group")
     supported_groups = response.headers.get("x-showcase-tls-client-supported-groups")
 
@@ -95,6 +94,5 @@ def test_google_auth_transport_pqc(run_pqc_test):
     print(f"\n[google-auth Transport PQC] Negotiated TLS Group: {negotiated_group}")
     print(f"[google-auth Transport PQC] Client Advertised Supported Groups: {supported_groups}")
 
-    # Assert PQC compliance
     assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
         f"Failed: google-auth Transport is NOT PQC-compliant! Negotiated: {negotiated_group}"
diff --git a/packages/google-auth/tests/crypt/test_pqc.py b/packages/google-auth/tests/crypt/test_pqc.py
new file mode 100644
index 000000000000..5486f47967f1
--- /dev/null
+++ b/packages/google-auth/tests/crypt/test_pqc.py
@@ -0,0 +1,44 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import cryptography
+import pytest
+from packaging.version import Version
+
+
+def test_cryptography_pqc_mlkem_support():
+    """Verifies that the cryptography package installed in google-auth supports Post-Quantum Cryptography (ML-KEM / FIPS 203)."""
+    # Attempt to import ML-KEM (FIPS 203) introduced in cryptography 44.0.0
+    try:
+        from cryptography.hazmat.primitives.asymmetric import mlkem
+    except ImportError:
+        pytest.fail(
+            f"PQC Compliance Failure: cryptography version {cryptography.__version__} does not support ML-KEM (FIPS 203). "
+            f"google-auth requires cryptography >= 44.0.0 for PQC support."
+        )
+
+    # 1. Generate ML-KEM-768 (FIPS 203) private key and derive public key
+    private_key = mlkem.MLKEM768PrivateKey.generate()
+    public_key = private_key.public_key()
+    assert isinstance(private_key, mlkem.MLKEM768PrivateKey)
+    assert isinstance(public_key, mlkem.MLKEM768PublicKey)
+
+    # 2. Perform Key Encapsulation (sender generates shared secret + ciphertext)
+    shared_secret_sender, ciphertext = public_key.encapsulate()
+    assert len(ciphertext) > 0
+    assert len(shared_secret_sender) > 0
+
+    # 3. Perform Key Decapsulation (receiver recovers shared secret from ciphertext)
+    shared_secret_receiver = private_key.decapsulate(ciphertext)
+    assert shared_secret_receiver == shared_secret_sender

From 30c6eccaedfb1cbc6be30981fa3497d875d32224 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Wed, 15 Jul 2026 22:17:21 +0000
Subject: [PATCH 06/18] update tests

---
 packages/gapic-generator/noxfile.py           | 10 +++
 .../gapic-generator/tests/system/test_pqc.py  | 70 ++++++-------------
 packages/google-auth/tests/crypt/test_pqc.py  | 44 ------------
 3 files changed, 31 insertions(+), 93 deletions(-)
 delete mode 100644 packages/google-auth/tests/crypt/test_pqc.py

diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index 343626eab4e4..ad3382c3d61b 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/packages/gapic-generator/noxfile.py
@@ -476,6 +476,16 @@ def showcase_mtls(
         )
 
 
+# TODO(Phase 3): Remove showcase_pqc once grpcio >= 1.83.0 is enforced by default in google-api-core.
+@nox.session(python=NEWEST_PYTHON)
+def showcase_pqc(session):
+    """Run the Showcase PQC verification test suite against grpcio 1.83+."""
+    with showcase_library(session):
+        session.install("pytest", "pytest-asyncio", "pyopenssl")
+        session.install("--pre", "--upgrade", "grpcio>=1.83.0rc0", "grpcio-status>=1.83.0rc0")
+        session.run("py.test", "--quiet", "--mtls", "-s", *(session.posargs or ["tests/system/test_pqc.py"]))
+
+
 def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False):
     session.install(
         "coverage",
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 6b6876272c65..ac037b02c834 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import pytest
-import requests
 from google import showcase
 
 
@@ -23,14 +22,17 @@ def run_pqc_test(use_mtls):
         pytest.skip("PQC integration test requires mTLS (--mtls flag) to be enabled.")
 
 
-@pytest.mark.parametrize(
-    "transport_fixture",
-    ["intercepted_echo_grpc", "intercepted_echo_rest"]
-)
-def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
-    """Verifies that the generated client library negotiates PQC (X25519MLKEM768) with Showcase server."""
-    client, interceptor = request.getfixturevalue(transport_fixture)
+def _require_pqc_support(transport_name):
+    # TODO(Phase 3): Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
+    if transport_name == "grpc":
+        import grpc
+        from packaging.version import Version
+        if Version(grpc.__version__) < Version("1.83.0rc0"):
+            pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
 
+
+def _verify_pqc_negotiated_group(client, interceptor, transport_name):
+    _require_pqc_support(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."
@@ -47,52 +49,22 @@ def test_pqc_negotiated_group(run_pqc_test, request, transport_fixture):
     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_fixture}) Negotiated TLS Group: {negotiated_group}")
-    print(f"[PQC Verification] ({transport_fixture}) Client Advertised Supported Groups: {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_fixture} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
-
-
-def test_google_auth_transport_pqc(run_pqc_test):
-    """Verifies that google-auth transport adapter negotiates PQC (X25519MLKEM768) with Showcase server."""
-    import google.auth.transport.requests
-    from conftest import HostNameIgnoringAdapter
-
-    # 1. Initialize requests Session with mTLS certs
-    session = requests.Session()
-    cert_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.crt"
-    key_path = "/usr/local/google/home/omairn/git/googleapis/google-cloud-python-dev2/packages/gapic-generator/tests/cert/mtls.key"
-
-    session.verify = cert_path
-    session.cert = (cert_path, key_path)
-    session.mount("https://", HostNameIgnoringAdapter())
+        f"Failed: {transport_name} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
 
-    # 2. Wrap session in google-auth transport adapter
-    auth_transport = google.auth.transport.requests.Request(session=session)
 
-    # 3. Serialize request body
-    req = showcase.EchoRequest(content="Verify google-auth transport PQC connection.")
-    body = showcase.EchoRequest.to_json(req, including_default_value_fields=False).encode("utf-8")
+def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
+    """Verifies that the gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    client, interceptor = intercepted_echo_grpc
+    _verify_pqc_negotiated_group(client, interceptor, "grpc")
 
-    # 4. Execute request through google-auth's transport layer
-    url = "https://localhost:7469/v1beta1/echo:echo"
-    headers = {
-        "Content-Type": "application/json",
-        "x-goog-api-client": "gapic/1.0 rest/1.0",
-    }
 
-    response = auth_transport(url=url, method="POST", body=body, headers=headers)
-    assert response.status == 200, f"Failed: status={response.status}, body={response.data}"
+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")
 
-    # 5. Extract and verify negotiated TLS group returned by Showcase
-    negotiated_group = response.headers.get("x-showcase-tls-group")
-    supported_groups = response.headers.get("x-showcase-tls-client-supported-groups")
-
-    assert negotiated_group is not None, "Failed: Showcase server did not return negotiated TLS group header."
-    print(f"\n[google-auth Transport PQC] Negotiated TLS Group: {negotiated_group}")
-    print(f"[google-auth Transport PQC] Client Advertised Supported Groups: {supported_groups}")
-
-    assert "MLKEM" in negotiated_group or "Kyber" in negotiated_group, \
-        f"Failed: google-auth Transport is NOT PQC-compliant! Negotiated: {negotiated_group}"
diff --git a/packages/google-auth/tests/crypt/test_pqc.py b/packages/google-auth/tests/crypt/test_pqc.py
deleted file mode 100644
index 5486f47967f1..000000000000
--- a/packages/google-auth/tests/crypt/test_pqc.py
+++ /dev/null
@@ -1,44 +0,0 @@
-# Copyright 2026 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     https://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import cryptography
-import pytest
-from packaging.version import Version
-
-
-def test_cryptography_pqc_mlkem_support():
-    """Verifies that the cryptography package installed in google-auth supports Post-Quantum Cryptography (ML-KEM / FIPS 203)."""
-    # Attempt to import ML-KEM (FIPS 203) introduced in cryptography 44.0.0
-    try:
-        from cryptography.hazmat.primitives.asymmetric import mlkem
-    except ImportError:
-        pytest.fail(
-            f"PQC Compliance Failure: cryptography version {cryptography.__version__} does not support ML-KEM (FIPS 203). "
-            f"google-auth requires cryptography >= 44.0.0 for PQC support."
-        )
-
-    # 1. Generate ML-KEM-768 (FIPS 203) private key and derive public key
-    private_key = mlkem.MLKEM768PrivateKey.generate()
-    public_key = private_key.public_key()
-    assert isinstance(private_key, mlkem.MLKEM768PrivateKey)
-    assert isinstance(public_key, mlkem.MLKEM768PublicKey)
-
-    # 2. Perform Key Encapsulation (sender generates shared secret + ciphertext)
-    shared_secret_sender, ciphertext = public_key.encapsulate()
-    assert len(ciphertext) > 0
-    assert len(shared_secret_sender) > 0
-
-    # 3. Perform Key Decapsulation (receiver recovers shared secret from ciphertext)
-    shared_secret_receiver = private_key.decapsulate(ciphertext)
-    assert shared_secret_receiver == shared_secret_sender

From 4e3be178a0acf02c59551df4ec8e07a2c2d9dd82 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Wed, 15 Jul 2026 23:11:41 +0000
Subject: [PATCH 07/18] cleanup

---
 .../gapic-generator/tests/system/test_pqc.py    | 17 +++++++----------
 1 file changed, 7 insertions(+), 10 deletions(-)

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index ac037b02c834..729a313b1537 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import grpc
+from packaging.version import Version
 import pytest
 from google import showcase
 
@@ -22,17 +24,7 @@ def run_pqc_test(use_mtls):
         pytest.skip("PQC integration test requires mTLS (--mtls flag) to be enabled.")
 
 
-def _require_pqc_support(transport_name):
-    # TODO(Phase 3): Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
-    if transport_name == "grpc":
-        import grpc
-        from packaging.version import Version
-        if Version(grpc.__version__) < Version("1.83.0rc0"):
-            pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
-
-
 def _verify_pqc_negotiated_group(client, interceptor, transport_name):
-    _require_pqc_support(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."
@@ -59,6 +51,10 @@ def _verify_pqc_negotiated_group(client, interceptor, transport_name):
 
 def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
     """Verifies that the 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
     _verify_pqc_negotiated_group(client, interceptor, "grpc")
 
@@ -68,3 +64,4 @@ def test_pqc_rest(run_pqc_test, intercepted_echo_rest):
     client, interceptor = intercepted_echo_rest
     _verify_pqc_negotiated_group(client, interceptor, "rest")
 
+

From 03ac316a2f35a5865f39c87ff3846456305cb9a1 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Thu, 16 Jul 2026 19:21:19 +0000
Subject: [PATCH 08/18] address feedback

---
 .../gapic-generator/tests/system/conftest.py   | 18 +++++++++---------
 .../gapic-generator/tests/system/test_pqc.py   | 11 ++++-------
 2 files changed, 13 insertions(+), 16 deletions(-)

diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 5c00c99a5976..522d6c2aa191 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/packages/gapic-generator/tests/system/conftest.py
@@ -115,10 +115,10 @@ def async_identity(use_mtls, request, event_loop):
         )
 
 
-dir = os.path.dirname(__file__)
-with open(os.path.join(dir, "../cert/mtls.crt"), "rb") as fh:
+base_dir = os.path.dirname(__file__)
+with open(os.path.join(base_dir, "../cert/mtls.crt"), "rb") as fh:
     cert = fh.read()
-with open(os.path.join(dir, "../cert/mtls.key"), "rb") as fh:
+with open(os.path.join(base_dir, "../cert/mtls.key"), "rb") as fh:
     key = fh.read()
 
 ssl_credentials = grpc.ssl_channel_credentials(
@@ -481,9 +481,9 @@ def intercepted_echo_rest(use_mtls):
         interceptor=interceptor,
     )
     if use_mtls:
-        dir = os.path.dirname(__file__)
-        cert_path = os.path.join(dir, "../cert/mtls.crt")
-        key_path = os.path.join(dir, "../cert/mtls.key")
+        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")
         transport._session.verify = cert_path
         transport._session.cert = (cert_path, key_path)
         transport._session.mount("https://", HostNameIgnoringAdapter())
@@ -508,9 +508,9 @@ def intercepted_echo_rest_async(use_mtls):
         interceptor=interceptor,
     )
     if use_mtls:
-        dir = os.path.dirname(__file__)
-        cert_path = os.path.join(dir, "../cert/mtls.crt")
-        key_path = os.path.join(dir, "../cert/mtls.key")
+        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")
         transport._session.verify = cert_path
         transport._session.cert = (cert_path, key_path)
         transport._session.mount("https://", HostNameIgnoringAdapter())
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 729a313b1537..ed65200cfb81 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -30,13 +30,10 @@ def _verify_pqc_negotiated_group(client, interceptor, transport_name):
     assert response.content == "Verify PQC connection."
 
     # Extract negotiated group and supported groups from response headers
-    negotiated_group = None
-    supported_groups = None
-    for key, value in interceptor.response_metadata:
-        if key.lower() == "x-showcase-tls-group":
-            negotiated_group = value
-        elif key.lower() == "x-showcase-tls-client-supported-groups":
-            supported_groups = value
+    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")
+    supported_groups = headers.get("x-showcase-tls-client-supported-groups")
 
     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."

From ee8b1c82af54cdc1e8950a92dc046dde28263beb Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Thu, 16 Jul 2026 19:55:32 +0000
Subject: [PATCH 09/18] update tests

---
 packages/gapic-generator/noxfile.py           | 13 +++--
 .../gapic-generator/tests/system/conftest.py  | 54 ++++++++++---------
 .../gapic-generator/tests/system/test_pqc.py  |  6 +--
 3 files changed, 40 insertions(+), 33 deletions(-)

diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index ad3382c3d61b..0a67167b8800 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/packages/gapic-generator/noxfile.py
@@ -478,12 +478,17 @@ def showcase_mtls(
 
 # TODO(Phase 3): Remove showcase_pqc once grpcio >= 1.83.0 is enforced by default in google-api-core.
 @nox.session(python=NEWEST_PYTHON)
-def showcase_pqc(session):
-    """Run the Showcase PQC verification test suite against grpcio 1.83+."""
-    with showcase_library(session):
+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", "pyopenssl")
         session.install("--pre", "--upgrade", "grpcio>=1.83.0rc0", "grpcio-status>=1.83.0rc0")
-        session.run("py.test", "--quiet", "--mtls", "-s", *(session.posargs or ["tests/system/test_pqc.py"]))
+        session.run("py.test", "--quiet", "--tls", "-s", *(session.posargs or ["tests/system/test_pqc.py"]), env=env)
 
 
 def run_showcase_unit_tests(session, fail_under=100, rest_async_io_enabled=False):
diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 522d6c2aa191..326833cbff6f 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/packages/gapic-generator/tests/system/conftest.py
@@ -116,14 +116,17 @@ def async_identity(use_mtls, request, event_loop):
 
 
 base_dir = os.path.dirname(__file__)
-with open(os.path.join(base_dir, "../cert/mtls.crt"), "rb") as fh:
+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:
     cert = fh.read()
-with open(os.path.join(base_dir, "../cert/mtls.key"), "rb") as fh:
+with open(KEY_PATH, "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():
@@ -138,6 +141,9 @@ 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
@@ -191,6 +197,11 @@ def use_mtls(request):
     return request.config.getoption("--mtls")
 
 
+@pytest.fixture
+def use_tls(request):
+    return request.config.getoption("--tls")
+
+
 @pytest.fixture
 def parametrized_echo(
     use_mtls,
@@ -414,7 +425,7 @@ async def intercept_stream_stream(
 
 
 @pytest.fixture
-def intercepted_echo_grpc(use_mtls):
+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.
@@ -423,11 +434,12 @@ def intercepted_echo_grpc(use_mtls):
         "intercepted",
     )
     host = "localhost:7469"
-    channel = (
-        grpc.secure_channel(host, ssl_credentials)
-        if use_mtls
-        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(),
@@ -468,31 +480,29 @@ def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
 
 
 @pytest.fixture
-def intercepted_echo_rest(use_mtls):
+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_mtls 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_mtls:
-        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")
-        transport._session.verify = cert_path
-        transport._session.cert = (cert_path, key_path)
+    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
 
 
 @pytest.fixture
-def intercepted_echo_rest_async(use_mtls):
+def intercepted_echo_rest_async():
     if not HAS_ASYNC_REST_ECHO_TRANSPORT:
         pytest.skip("Skipping test with async rest.")
 
@@ -500,19 +510,11 @@ def intercepted_echo_rest_async(use_mtls):
     transport_cls = EchoAsyncClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestAsyncInterceptor()
 
-    url_scheme = "https" if use_mtls else "http"
     transport = transport_cls(
         credentials=async_anonymous_credentials(),
         host="localhost:7469",
-        url_scheme=url_scheme,
+        url_scheme="http",
         interceptor=interceptor,
     )
-    if use_mtls:
-        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")
-        transport._session.verify = cert_path
-        transport._session.cert = (cert_path, key_path)
-        transport._session.mount("https://", HostNameIgnoringAdapter())
 
     return EchoAsyncClient(transport=transport), interceptor
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index ed65200cfb81..998c327c3418 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -19,9 +19,9 @@
 
 
 @pytest.fixture
-def run_pqc_test(use_mtls):
-    if not use_mtls:
-        pytest.skip("PQC integration test requires mTLS (--mtls flag) to be enabled.")
+def run_pqc_test(use_tls):
+    if not 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):

From 2620f09488a930c98135366c988385e28d4cac6d Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Thu, 16 Jul 2026 21:48:39 +0000
Subject: [PATCH 10/18] update tests

---
 packages/gapic-generator/noxfile.py           |  6 +--
 .../gapic-generator/tests/system/conftest.py  | 12 ++++--
 .../gapic-generator/tests/system/test_pqc.py  | 39 ++++++++++++-------
 3 files changed, 37 insertions(+), 20 deletions(-)

diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index 0a67167b8800..978c4e3c43c3 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/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):
diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 326833cbff6f..af3a0b516184 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/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,
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 998c327c3418..e2702d23c5f6 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/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")
+
+
 
 

From 63f33f28c73416124e815d3d8623574038f50b66 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Thu, 16 Jul 2026 22:01:45 +0000
Subject: [PATCH 11/18] link bugs

---
 packages/gapic-generator/noxfile.py               |  5 ++++-
 packages/gapic-generator/tests/system/test_pqc.py | 10 ++++++++--
 2 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/packages/gapic-generator/noxfile.py b/packages/gapic-generator/noxfile.py
index 978c4e3c43c3..3a79d3299b70 100644
--- a/packages/gapic-generator/noxfile.py
+++ b/packages/gapic-generator/noxfile.py
@@ -476,7 +476,8 @@ def showcase_mtls(
         )
 
 
-# TODO(Phase 3): Remove showcase_pqc once grpcio >= 1.83.0 is enforced by default in google-api-core.
+# 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,
@@ -487,6 +488,8 @@ 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")
+        # 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)
 
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index e2702d23c5f6..672100f300f0 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -42,8 +42,11 @@ def _verify_pqc_metadata(interceptor, transport_name):
 
 def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
     """Verifies that the 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.
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
+    # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
     if Version(grpc.__version__) < Version("1.83.0rc0"):
+        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751): 
+        # Update the version in the check above to `1.83.0` once released.
         pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
 
     client, interceptor = intercepted_echo_grpc
@@ -63,8 +66,11 @@ def test_pqc_rest(run_pqc_test, intercepted_echo_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.
+    # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
+    # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
     if Version(grpc.__version__) < Version("1.83.0rc0"):
+        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751): 
+        # Update the version in the check above to `1.83.0` once released.
         pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
 
     client, interceptor = intercepted_echo_grpc_async

From 7b52775c11d15c427cf06a9801e87d4a6b37fb09 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Thu, 16 Jul 2026 22:08:53 +0000
Subject: [PATCH 12/18] address feedback

---
 packages/gapic-generator/tests/system/conftest.py | 12 +++---------
 1 file changed, 3 insertions(+), 9 deletions(-)

diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index af3a0b516184..0c8e95e061b5 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/packages/gapic-generator/tests/system/conftest.py
@@ -19,7 +19,6 @@
 import pytest
 import pytest_asyncio
 from requests.adapters import HTTPAdapter
-from urllib3.poolmanager import PoolManager
 
 from typing import Sequence, Tuple
 
@@ -473,14 +472,9 @@ async def intercepted_echo_grpc_async(use_mtls, use_tls):
 
 class HostNameIgnoringAdapter(HTTPAdapter):
     """Custom HTTPAdapter that disables hostname verification for local self-signed certs."""
-    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
-        self.poolmanager = PoolManager(
-            num_pools=connections,
-            maxsize=maxsize,
-            block=block,
-            assert_hostname=False,
-            **pool_kwargs
-        )
+    def cert_verify(self, conn, url, verify, cert):
+        super().cert_verify(conn, url, verify, cert)
+        conn.assert_hostname = False
 
 
 @pytest.fixture

From 7283c3ba8c2dbcbe8d21f8b408208c2fe8466b68 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 17 Jul 2026 03:52:00 +0000
Subject: [PATCH 13/18] cleanup

---
 .../gapic-generator/tests/system/conftest.py  | 34 ++++++++-----------
 .../gapic-generator/tests/system/test_pqc.py  | 26 +++++++-------
 2 files changed, 28 insertions(+), 32 deletions(-)

diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 0c8e95e061b5..0ed270adf993 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/packages/gapic-generator/tests/system/conftest.py
@@ -424,7 +424,7 @@ async def intercept_stream_stream(
 
 
 @pytest.fixture
-def intercepted_echo_grpc(use_mtls, use_tls):
+def intercepted_echo_grpc(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,12 +433,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, tls_credentials)
+        if use_tls
+        else grpc.insecure_channel(host)
+    )
     intercept_channel = grpc.intercept_channel(channel, interceptor)
     transport = EchoClient.get_transport_class("grpc")(
         credentials=ga_credentials.AnonymousCredentials(),
@@ -448,7 +447,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(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.
@@ -457,12 +456,11 @@ 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.secure_channel(host, tls_credentials, interceptors=[interceptor])
+        if use_tls
+        else grpc.aio.insecure_channel(host, interceptors=[interceptor])
+    )
     transport = EchoAsyncClient.get_transport_class("grpc_asyncio")(
         credentials=ga_credentials.AnonymousCredentials(),
         channel=channel,
@@ -478,23 +476,21 @@ def cert_verify(self, conn, url, verify, cert):
 
 
 @pytest.fixture
-def intercepted_echo_rest(use_mtls, use_tls):
+def intercepted_echo_rest(use_tls):
     transport_name = "rest"
     transport_cls = EchoClient.get_transport_class(transport_name)
     interceptor = EchoMetadataClientRestInterceptor()
 
-    url_scheme = "https" if (use_mtls or use_tls) else "http"
+    url_scheme = "https" if use_tls else "http"
     transport = transport_cls(
         credentials=ga_credentials.AnonymousCredentials(),
         host="localhost:7469",
         url_scheme=url_scheme,
         interceptor=interceptor,
     )
-    if use_mtls or use_tls:
+    if 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
 
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 672100f300f0..a632506dbfa4 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -18,10 +18,10 @@
 from google import showcase
 
 
-@pytest.fixture
-def run_pqc_test(use_tls):
+@pytest.fixture(autouse=True)
+def require_tls(use_tls):
     if not use_tls:
-        pytest.skip("PQC integration test requires TLS (--tls or --mtls flag) to be enabled.")
+        pytest.skip("PQC integration test requires standard TLS (--tls flag) to be enabled.")
 
 
 def _verify_pqc_metadata(interceptor, transport_name):
@@ -40,7 +40,7 @@ def _verify_pqc_metadata(interceptor, transport_name):
     ), f"Failed: {transport_name} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
 
 
-def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
+def test_pqc_grpc(intercepted_echo_grpc):
     """Verifies that the gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
     # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
@@ -55,7 +55,7 @@ def test_pqc_grpc(run_pqc_test, intercepted_echo_grpc):
     _verify_pqc_metadata(interceptor, "grpc")
 
 
-def test_pqc_rest(run_pqc_test, intercepted_echo_rest):
+def test_pqc_rest(intercepted_echo_rest):
     """Verifies that the REST client library negotiates PQC (X25519MLKEM768) with Showcase server."""
     client, interceptor = intercepted_echo_rest
     response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
@@ -64,20 +64,20 @@ def test_pqc_rest(run_pqc_test, intercepted_echo_rest):
 
 
 @pytest.mark.asyncio
-async def test_pqc_grpc_async(run_pqc_test, intercepted_echo_grpc_async):
+async def test_pqc_grpc_async(intercepted_echo_grpc_async):
     """Verifies that the async gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
     # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
     if Version(grpc.__version__) < Version("1.83.0rc0"):
-        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751): 
+        # TODO(https://github.com/googleapis/google-cloud-python/issues/17751):
         # Update the version in the check above to `1.83.0` once released.
-        pytest.skip(f"gRPC PQC negotiation requires grpcio >= 1.83.0 (current: {grpc.__version__})")
+        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."))
+    response = await client.echo(
+        request=showcase.EchoRequest(content="Verify PQC connection.")
+    )
     assert response.content == "Verify PQC connection."
     _verify_pqc_metadata(interceptor, "grpc_asyncio")
-
-
-
-

From a3d4c58aa9a65aec327c878bfecc430a684a5c7b Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 17 Jul 2026 03:57:36 +0000
Subject: [PATCH 14/18] remove Kyber

---
 packages/gapic-generator/tests/system/test_pqc.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index a632506dbfa4..87d3b1d8659a 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -34,10 +34,10 @@ 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 or Kyber)
+    # Enforce PQC compliance (X25519MLKEM768)
     assert (
-        "MLKEM" in negotiated_group or "Kyber" in negotiated_group
-    ), f"Failed: {transport_name} Connection is NOT PQC-compliant! Negotiated: {negotiated_group}"
+        "MLKEM" in negotiated_group
+    ), f"Failed: {transport_name} Connection did not negotiate X25519MLKEM768! Negotiated: {negotiated_group}"
 
 
 def test_pqc_grpc(intercepted_echo_grpc):

From bd15409b36cdbfd5a730f6790eff5871d9f0639b Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 17 Jul 2026 19:51:39 +0000
Subject: [PATCH 15/18] address feedback

---
 .../gapic-generator/tests/system/conftest.py  | 34 +++++++++++--------
 .../gapic-generator/tests/system/test_pqc.py  |  5 +--
 2 files changed, 22 insertions(+), 17 deletions(-)

diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 0ed270adf993..0c8e95e061b5 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/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
 
diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 87d3b1d8659a..3318a14e0e50 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/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):

From fbc88309880e49729d739a356c040db53077e637 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 17 Jul 2026 19:59:58 +0000
Subject: [PATCH 16/18] address feedback

---
 packages/gapic-generator/tests/system/conftest.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/gapic-generator/tests/system/conftest.py b/packages/gapic-generator/tests/system/conftest.py
index 0c8e95e061b5..73169dd8a79f 100644
--- a/packages/gapic-generator/tests/system/conftest.py
+++ b/packages/gapic-generator/tests/system/conftest.py
@@ -198,7 +198,7 @@ def use_mtls(request):
 
 @pytest.fixture
 def use_tls(request):
-    return request.config.getoption("--tls")
+    return request.config.getoption("--tls") or request.config.getoption("--mtls")
 
 
 @pytest.fixture

From ce88e4a4d87995f3e3f6a87b954647331c30fd75 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 17 Jul 2026 20:43:52 +0000
Subject: [PATCH 17/18] address feedback

---
 packages/gapic-generator/tests/system/test_pqc.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index 3318a14e0e50..cd10c99d8df6 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -34,11 +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 by verifying an MLKEM group (e.g. X25519MLKEM768) was negotiated.
-    # Substring check ("MLKEM" in ...) allows compatibility across gRPC and REST implementation strings.
+    # Enforce PQC compliance by verifying a post-quantum MLKEM hybrid group was negotiated.
+    # Substring check ("MLKEM" in ...) ensures compatibility across different transport and library group strings.
     assert (
         "MLKEM" in negotiated_group
-    ), f"Failed: {transport_name} Connection did not negotiate an MLKEM group (e.g. X25519MLKEM768)! Negotiated: {negotiated_group}"
+    ), f"Failed: {transport_name} Connection did not negotiate a post-quantum MLKEM group! Negotiated: {negotiated_group}"
 
 
 def test_pqc_grpc(intercepted_echo_grpc):

From c19087de4279ecb1b1f5c850ea37d71d3357b108 Mon Sep 17 00:00:00 2001
From: ohmayr <omairn@google.com>
Date: Fri, 17 Jul 2026 20:44:57 +0000
Subject: [PATCH 18/18] address feedback

---
 packages/gapic-generator/tests/system/test_pqc.py | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/packages/gapic-generator/tests/system/test_pqc.py b/packages/gapic-generator/tests/system/test_pqc.py
index cd10c99d8df6..2d694b6528f0 100644
--- a/packages/gapic-generator/tests/system/test_pqc.py
+++ b/packages/gapic-generator/tests/system/test_pqc.py
@@ -34,7 +34,7 @@ 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 by verifying a post-quantum MLKEM hybrid group was negotiated.
+    # Enforce PQC compliance by verifying a post-quantum MLKEM group was negotiated.
     # Substring check ("MLKEM" in ...) ensures compatibility across different transport and library group strings.
     assert (
         "MLKEM" in negotiated_group
@@ -42,7 +42,7 @@ def _verify_pqc_metadata(interceptor, transport_name):
 
 
 def test_pqc_grpc(intercepted_echo_grpc):
-    """Verifies that the gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    """Verifies that the gRPC client library negotiates post-quantum MLKEM with Showcase server."""
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
     # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
     if Version(grpc.__version__) < Version("1.83.0rc0"):
@@ -57,7 +57,7 @@ def test_pqc_grpc(intercepted_echo_grpc):
 
 
 def test_pqc_rest(intercepted_echo_rest):
-    """Verifies that the REST client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    """Verifies that the REST client library negotiates post-quantum MLKEM with Showcase server."""
     client, interceptor = intercepted_echo_rest
     response = client.echo(request=showcase.EchoRequest(content="Verify PQC connection."))
     assert response.content == "Verify PQC connection."
@@ -66,7 +66,7 @@ def test_pqc_rest(intercepted_echo_rest):
 
 @pytest.mark.asyncio
 async def test_pqc_grpc_async(intercepted_echo_grpc_async):
-    """Verifies that the async gRPC client library negotiates PQC (X25519MLKEM768) with Showcase server."""
+    """Verifies that the async gRPC client library negotiates post-quantum MLKEM with Showcase server."""
     # TODO(https://github.com/googleapis/google-cloud-python/issues/17752):
     # Remove this check once grpcio >= 1.83.0 is enforced across all client libraries.
     if Version(grpc.__version__) < Version("1.83.0rc0"):