test_dns.py

# Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0

"""
Tests for async client behavior when connecting with hostnames.
See `DNS Tests <../../DEVELOPER.md#dns-tests>`_ for setup instructions.
"""

import os

import pytest
from glide_shared.config import NodeAddress

from tests.async_tests.conftest import create_client
from tests.test_constants import HOSTNAME_NO_TLS, HOSTNAME_TLS
from tests.utils.utils import assert_connected, get_ca_certificate


async def create_client_with_hostname(cluster_mode: bool, use_tls: bool, hostname: str):
    """Builds and returns a client for the given parameters."""
    if use_tls and cluster_mode:
        cluster = pytest.valkey_tls_cluster  # type: ignore[attr-defined]
    elif use_tls:
        cluster = pytest.standalone_tls_cluster  # type: ignore[attr-defined]
    elif cluster_mode:
        cluster = pytest.valkey_cluster  # type: ignore[attr-defined]
    else:
        cluster = pytest.standalone_cluster  # type: ignore[attr-defined]

    port = cluster.nodes_addr[0].port
    address = NodeAddress(hostname, port)

    return await create_client(
        cluster_mode=cluster_mode,
        use_tls=use_tls,
        root_pem_cacerts=get_ca_certificate() if use_tls else None,
        addresses=[address],
    )


@pytest.mark.anyio
@pytest.mark.skipif(
    not os.getenv("VALKEY_GLIDE_DNS_TESTS_ENABLED"),
    reason="DNS tests disabled. Set VALKEY_GLIDE_DNS_TESTS_ENABLED to enable.",
)
@pytest.mark.parametrize("cluster_mode", [True, False])
class TestDns:
    """Async DNS resolution tests."""

    async def test_connect_with_valid_hostname_succeeds(self, cluster_mode: bool):
        """Test connection with valid hostname (non-TLS)."""
        client = await create_client_with_hostname(
            cluster_mode=cluster_mode,
            use_tls=False,
            hostname=HOSTNAME_NO_TLS,
        )

        await assert_connected(client)
        await client.close()

    async def test_connect_with_invalid_hostname_fails(self, cluster_mode: bool):
        """Test connection with invalid hostname (non-TLS)."""
        with pytest.raises(Exception):
            await create_client_with_hostname(
                cluster_mode=cluster_mode,
                use_tls=False,
                hostname="nonexistent.invalid",
            )

    async def test_tls_with_hostname_in_certificate_succeeds(self, cluster_mode: bool):
        """Test TLS connection with hostname in certificate SAN."""
        client = await create_client_with_hostname(
            cluster_mode=cluster_mode,
            use_tls=True,
            hostname=HOSTNAME_TLS,
        )

        await assert_connected(client)
        await client.close()

    async def test_tls_with_hostname_not_in_certificate_fails(self, cluster_mode: bool):
        """Test TLS connection with hostname not in certificate SAN."""
        with pytest.raises(Exception):
            await create_client_with_hostname(
                cluster_mode=cluster_mode,
                use_tls=True,
                hostname=HOSTNAME_NO_TLS,
            )