NO-SNOW: Add more tests for async crl

Author: sfc-gh-turbaszek

src/snowflake/connector/aio/_session_manager.py

@@ -16,7 +16,6 @@
 from ..ssl_wrap_socket import (
     FEATURE_CRL_CONFIG,
     FEATURE_OCSP_RESPONSE_CACHE_FILE_NAME,
-    get_current_session_manager,
     load_trusted_certificates,
     resolve_cafile,
 )
@@ -117,10 +116,11 @@ def validate_crl(self, protocol: ResponseHandler, req: ClientRequest):
         cafile_for_ctx = resolve_cafile({"ca_certs": certifi.where()})
         crl_validator = CRLValidator.from_config(
             FEATURE_CRL_CONFIG,
-            get_current_session_manager(),
+            self._session_manager,
             trusted_certificates=load_trusted_certificates(cafile_for_ctx),
         )
         sll_object = protocol.transport.get_extra_info("ssl_object")
+        # TODO(asyncio): SNOW-2681061 Add sync support for validate_connection
         if not crl_validator.validate_connection(sll_object):
             raise OperationalError(
                 msg=(

test/integ/aio_it/test_crl_async.py

@@ -0,0 +1,179 @@
+#!/usr/bin/env python
+"""
+CRL (Certificate Revocation List) Validation Integration Tests - Async version
+
+These tests verify that CRL validation works correctly with real Snowflake connections
+in different modes: DISABLED, ADVISORY, and ENABLED.
+"""
+from __future__ import annotations
+
+import tempfile
+
+import pytest
+
+
+@pytest.mark.skipolddriver
+async def test_crl_validation_enabled_mode(conn_cnx):
+    """Test that connection works with CRL validation in ENABLED mode."""
+    # ENABLED mode should work for normal Snowflake connections since they typically
+    # have valid certificates with proper CRL distribution points
+    async with conn_cnx(
+        cert_revocation_check_mode="ENABLED",
+        allow_certificates_without_crl_url=True,  # Allow certs without CRL URLs
+        crl_connection_timeout_ms=5000,  # 5 second timeout
+        crl_read_timeout_ms=5000,  # 5 second timeout
+        disable_ocsp_checks=True,
+    ) as cnx:
+        assert cnx, "Connection should succeed with CRL validation in ENABLED mode"
+
+        # Verify we can execute a simple query
+        cur = cnx.cursor()
+        await cur.execute("SELECT 1")
+        result = await cur.fetchone()
+        assert result[0] == 1, "Query should execute successfully"
+        await cur.close()
+
+        # Verify CRL settings were applied
+        assert cnx.cert_revocation_check_mode == "ENABLED"
+        assert cnx.allow_certificates_without_crl_url is True
+
+
+@pytest.mark.skipolddriver
+async def test_crl_validation_advisory_mode(conn_cnx):
+    """Test that connection works with CRL validation in ADVISORY mode."""
+    # ADVISORY mode should be more lenient and allow connections even if CRL checks fail
+    async with conn_cnx(
+        cert_revocation_check_mode="ADVISORY",
+        allow_certificates_without_crl_url=False,  # Don't allow certs without CRL URLs
+        crl_connection_timeout_ms=3000,  # 3 second timeout
+        crl_read_timeout_ms=3000,  # 3 second timeout
+        enable_crl_cache=True,  # Enable caching
+        crl_cache_validity_hours=1,  # Cache for 1 hour
+    ) as cnx:
+        assert cnx, "Connection should succeed with CRL validation in ADVISORY mode"
+
+        # Verify we can execute a simple query
+        cur = cnx.cursor()
+        await cur.execute("SELECT CURRENT_VERSION()")
+        result = await cur.fetchone()
+        assert result[0], "Query should return a version string"
+        await cur.close()
+
+        # Verify CRL settings were applied
+        assert cnx.cert_revocation_check_mode == "ADVISORY"
+        assert cnx.allow_certificates_without_crl_url is False
+        assert cnx.enable_crl_cache is True
+        assert cnx.crl_connection_timeout_ms == 3000
+        assert cnx.crl_read_timeout_ms == 3000
+        assert cnx.crl_cache_validity_hours == 1
+        assert cnx.crl_cache_dir is None
+
+
+@pytest.mark.skipolddriver
+async def test_crl_validation_disabled_mode(conn_cnx):
+    """Test that connection works with CRL validation in DISABLED mode (default)."""
+    # DISABLED mode should work without any CRL checks
+    async with conn_cnx(
+        cert_revocation_check_mode="DISABLED",
+    ) as cnx:
+        assert cnx, "Connection should succeed with CRL validation in DISABLED mode"
+
+        # Verify we can execute a simple query
+        cur = cnx.cursor()
+        await cur.execute("SELECT 'CRL_DISABLED' as test_value")
+        result = await cur.fetchone()
+        assert result[0] == "CRL_DISABLED", "Query should execute successfully"
+        await cur.close()
+
+        # Verify CRL settings were applied
+        assert cnx.cert_revocation_check_mode == "DISABLED"
+
+
+@pytest.mark.skipolddriver
+@pytest.mark.parametrize(
+    "crl_mode,allow_without_crl,should_succeed",
+    [
+        ("DISABLED", True, True),  # DISABLED mode always succeeds
+        ("DISABLED", False, True),  # DISABLED mode always succeeds
+        ("ADVISORY", True, True),  # ADVISORY mode is lenient
+        ("ADVISORY", False, True),  # ADVISORY mode is lenient
+        ("ENABLED", True, True),  # ENABLED with allow_without_crl should succeed
+        ("ENABLED", False, True),  # ENABLED might succeed if certs have valid CRL URLs
+    ],
+)
+async def test_crl_validation_modes_parametrized(
+    conn_cnx, crl_mode, allow_without_crl, should_succeed
+):
+    """Parametrized test for different CRL validation modes and settings."""
+    try:
+        async with conn_cnx(
+            cert_revocation_check_mode=crl_mode,
+            allow_certificates_without_crl_url=allow_without_crl,
+            crl_connection_timeout_ms=5000,
+            crl_read_timeout_ms=5000,
+        ) as cnx:
+            if should_succeed:
+                assert (
+                    cnx
+                ), f"Connection should succeed with mode={crl_mode}, allow_without_crl={allow_without_crl}"
+
+                # Test basic functionality
+                cur = cnx.cursor()
+                await cur.execute("SELECT 1")
+                result = await cur.fetchone()
+                assert result[0] == 1, "Basic query should work"
+                await cur.close()
+
+                # Verify settings
+                assert cnx.cert_revocation_check_mode == crl_mode
+                assert cnx.allow_certificates_without_crl_url == allow_without_crl
+            else:
+                pytest.fail(
+                    f"Connection should have failed with mode={crl_mode}, allow_without_crl={allow_without_crl}"
+                )
+
+    except Exception as e:
+        if should_succeed:
+            pytest.fail(
+                f"Connection unexpectedly failed with mode={crl_mode}, allow_without_crl={allow_without_crl}: {e}"
+            )
+        else:
+            # Expected failure - verify it's a connection-related error
+            assert (
+                "revoked" in str(e).lower() or "crl" in str(e).lower()
+            ), f"Expected CRL-related error, got: {e}"
+
+
+@pytest.mark.skipolddriver
+async def test_crl_cache_configuration(conn_cnx):
+    """Test CRL cache configuration options."""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        async with conn_cnx(
+            cert_revocation_check_mode="ADVISORY",  # Use advisory to avoid strict failures
+            enable_crl_cache=True,
+            enable_crl_file_cache=True,
+            crl_cache_dir=temp_dir,
+            crl_cache_validity_hours=2,
+            crl_cache_removal_delay_days=1,
+            crl_cache_cleanup_interval_hours=1,
+            crl_cache_start_cleanup=False,  # Don't start background cleanup in tests
+        ) as cnx:
+            assert cnx, "Connection should succeed with CRL cache configuration"
+
+            # Verify cache settings were applied
+            assert cnx.enable_crl_cache is True
+            assert cnx.enable_crl_file_cache is True
+            assert cnx.crl_cache_dir == temp_dir
+            assert cnx.crl_cache_validity_hours == 2
+            assert cnx.crl_cache_removal_delay_days == 1
+            assert cnx.crl_cache_cleanup_interval_hours == 1
+            assert cnx.crl_cache_start_cleanup is False
+
+            # Test basic functionality
+            cur = cnx.cursor()
+            await cur.execute("SELECT 'cache_test' as result")
+            result = await cur.fetchone()
+            assert (
+                result[0] == "cache_test"
+            ), "Query should work with cache configuration"
+            await cur.close()