update tests

Author: ohmayr

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):

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,51 +480,41 @@ 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.")
 
     transport_name = "rest_asyncio"
     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

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):