foreign/python/tests/test_connectivity.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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
#
#   http://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.

from datetime import timedelta

import pytest

from apache_iggy import (
    AutoLogin,
    HttpConfig,
    IggyClient,
    QuicConfig,
    QuicReconnectionConfig,
    TcpConfig,
    TcpReconnectionConfig,
    WebSocketConfig,
)

from .utils import (
    get_http_server_config,
    get_quic_server_config,
    get_server_config,
    get_websocket_server_config,
    wait_for_ping,
    wait_for_server,
)


def binary_transport_configs(
    auto_login: AutoLogin | None = None, websocket_marks: tuple = ()
) -> list:
    """Return a config for every transport that holds a connection.

    Auto-login is disabled by default: with credentials to replay, a ping sent
    while disconnected reconnects on its own instead of failing.

    TCP and QUIC wait out `reestablish_after` before they connect again, so the
    configs set it to zero to keep the reconnect cases fast.
    """
    tcp_host, tcp_port = get_server_config()
    ws_host, ws_port = get_websocket_server_config()
    quic_host, quic_port = get_quic_server_config()
    return [
        pytest.param(
            TcpConfig(
                server_address=f"{tcp_host}:{tcp_port}",
                auto_login=auto_login,
                reconnection=TcpReconnectionConfig(reestablish_after=timedelta(0)),
            ),
            id="tcp",
        ),
        pytest.param(
            WebSocketConfig(
                server_address=f"{ws_host}:{ws_port}", auto_login=auto_login
            ),
            id="websocket",
            marks=websocket_marks,
        ),
        pytest.param(
            QuicConfig(
                server_address=f"{quic_host}:{quic_port}",
                auto_login=auto_login,
                reconnection=QuicReconnectionConfig(reestablish_after=timedelta(0)),
            ),
            id="quic",
        ),
    ]


class TestConnectivity:
    """Test basic connectivity and authentication."""

    @pytest.mark.asyncio
    async def test_client_not_none(self, iggy_client: IggyClient):
        """Test that client fixture is properly initialized."""
        assert iggy_client is not None

    @pytest.mark.parametrize(
        "connection_string",
        [
            "iggy://iggy:iggy@127.0.0.1:8090",
            "iggy+tcp://iggy:iggy@127.0.0.1:8090",
            "iggy+tcp://iggy:iggy@127.0.0.1:8090?reconnection_retries=3&reconnection_interval=1s&reestablish_after=5s&heartbeat_interval=5s&nodelay=true&tls_domain=localhost&tls_ca_file=unused.pem&tls=false",
            "iggy+http://iggy:iggy@127.0.0.1:3000",
            "iggy+http://iggy:iggy@127.0.0.1:3000?heartbeat_interval=5s&retries=3",
            "iggy+ws://iggy:iggy@127.0.0.1:8092",
            "iggy+ws://iggy:iggy@127.0.0.1:8092?heartbeat_interval=5s&reconnection_retries=3&reconnection_interval=1s&reestablish_after=5s&read_buffer_size=4096&write_buffer_size=4096&max_write_buffer_size=8192&max_message_size=16384&max_frame_size=16384&accept_unmasked_frames=false&tls_domain=localhost&tls_ca_file=unused.pem&tls_validate_certificate=false&tls=false",
            "iggy+quic://iggy:iggy@127.0.0.1:8080?reconnection_max_retries=0",
        ],
    )
    @pytest.mark.asyncio
    async def test_valid_connection_string(self, connection_string: str):
        """Test that valid connection string formats can connect to the server."""

        client = IggyClient.from_connection_string(connection_string)
        await client.connect()
        await wait_for_ping(client, timeout=5, interval=1)

    @pytest.mark.parametrize(
        ("invalid_value", "expected_error"),
        [
            ("", "Invalid connection string"),
            ("bad address", "Invalid connection string"),
            ("http://{host}:{port}", "Invalid connection string"),
            ("tcp://iggy:iggy@{host}:{port}", "Invalid connection string"),
            ("{host}:", "Invalid connection string"),
            (":{port}", "Invalid connection string"),
            ("{host}:not-a-port", "Invalid connection string"),
            ("{host}:70000", "Invalid connection string"),
            ("iggy+tcp://", "Invalid connection string"),
            ("iggy+tcp://iggy:iggy@", "Invalid connection string"),
            (
                "iggy+tcp://iggy:iggy@{host}:not-a-port",
                "Invalid connection string",
            ),
            ("iggy+tcp://iggy:iggy@{host}:-1", "Invalid connection string"),
            ("iggy+tcp://iggy:iggy@{host}:70000", "Invalid connection string"),
            (
                "iggy+tcp://iggy:bad:format@{host}:{port}",
                "Invalid connection string",
            ),
            (
                "iggy+tcp://iggy:iggy@{host}:{port}?invalid_option=value",
                "Invalid connection string",
            ),
        ],
    )
    def test_invalid_connection_string(self, invalid_value: str, expected_error: str):
        """Test malformed server addresses and connection strings are rejected."""
        host, port = get_server_config()
        value = invalid_value.format(host=host, port=port)

        with pytest.raises(RuntimeError):
            IggyClient(value)

        with pytest.raises(RuntimeError, match=expected_error):
            IggyClient.from_connection_string(value)

    @pytest.mark.asyncio
    async def test_repeated_connect_does_not_error(self):
        """Test calling connect twice on the same client succeeds."""
        host, port = get_server_config()
        wait_for_server(host, port)

        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await client.connect()

    @pytest.mark.asyncio
    async def test_login_before_connect_does_not_error(self):
        """Test login can establish authentication without an explicit connect call."""
        host, port = get_server_config()
        wait_for_server(host, port)
        client = IggyClient(f"{host}:{port}")

        await client.login_user("iggy", "iggy")
        await wait_for_ping(client)

    @pytest.mark.parametrize(
        ("username", "password", "expected_exception"),
        [
            ("invalid-user", "iggy", RuntimeError),
            ("iggy", "invalid-password", RuntimeError),
            ("", "iggy", RuntimeError),
            ("iggy", "", RuntimeError),
            (None, "iggy", TypeError),
            ("iggy", None, TypeError),
        ],
    )
    @pytest.mark.asyncio
    async def test_login_with_invalid_credentials_fails(
        self, username, password, expected_exception
    ):
        """Test login rejects invalid credentials and invalid argument values."""
        host, port = get_server_config()
        wait_for_server(host, port)

        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await wait_for_ping(client)

        with pytest.raises(expected_exception):
            result = client.login_user(username, password)
            await result

    @pytest.mark.asyncio
    async def test_relogin_with_invalid_credentials_fails(self):
        """Test re-login with invalid credentials fails after a successful login."""
        host, port = get_server_config()
        wait_for_server(host, port)

        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await wait_for_ping(client)
        await client.login_user("iggy", "iggy")

        with pytest.raises(RuntimeError):
            await client.login_user("iggy", "invalid-password")

    @pytest.mark.asyncio
    async def test_relogin_with_same_valid_credentials_does_not_error(self):
        """Test repeated login with the same valid credentials is allowed."""
        host, port = get_server_config()
        wait_for_server(host, port)

        client = IggyClient(f"{host}:{port}")
        await client.connect()
        await wait_for_ping(client)
        await client.login_user("iggy", "iggy")
        await client.login_user("iggy", "iggy")

    @pytest.mark.asyncio
    async def test_ping(self, iggy_client: IggyClient):
        """Test server ping functionality."""
        await iggy_client.ping()


class TestLifecycle:
    """Test the disconnect and shutdown lifecycle of the client."""

    @pytest.mark.parametrize("config", binary_transport_configs())
    @pytest.mark.asyncio
    async def test_disconnect_is_idempotent_and_rejects_requests(
        self, config: TcpConfig | WebSocketConfig | QuicConfig
    ):
        """Test repeated disconnects succeed, requests fail, and connect works again."""
        client = IggyClient(config)
        await client.connect()
        await wait_for_ping(client)

        await client.disconnect()
        await client.disconnect()

        with pytest.raises(RuntimeError):
            await client.ping()

        await client.connect()
        await client.ping()

    @pytest.mark.parametrize("config", binary_transport_configs())
    @pytest.mark.asyncio
    async def test_disconnected_client_reconnects_and_logs_in_again(
        self, config: TcpConfig | WebSocketConfig | QuicConfig
    ):
        """Test a disconnected client needs a new login after it reconnects."""
        client = IggyClient(config)
        await client.connect()
        await wait_for_ping(client)
        await client.login_user("iggy", "iggy")
        await client.get_streams()

        await client.disconnect()
        await client.connect()
        await wait_for_ping(client)

        with pytest.raises(RuntimeError):
            await client.get_streams()

        await client.login_user("iggy", "iggy")
        await client.get_streams()

    @pytest.mark.parametrize(
        "config",
        binary_transport_configs(AutoLogin.username_password("iggy", "iggy")),
    )
    @pytest.mark.asyncio
    async def test_disconnected_client_with_auto_login_signs_in_on_reconnect(
        self, config: TcpConfig | WebSocketConfig | QuicConfig
    ):
        """Test a client with auto-login credentials signs in again on connect."""
        client = IggyClient(config)
        await client.connect()
        await wait_for_ping(client)

        await client.disconnect()
        await client.connect()

        await client.get_streams()

    @pytest.mark.parametrize("config", binary_transport_configs())
    @pytest.mark.asyncio
    async def test_shutdown_is_idempotent_and_rejects_requests(
        self, config: TcpConfig | WebSocketConfig | QuicConfig
    ):
        """Test repeated shutdowns succeed and later requests report the shutdown."""
        client = IggyClient(config)
        await client.connect()
        await wait_for_ping(client)

        await client.shutdown()
        await client.shutdown()

        with pytest.raises(RuntimeError, match="Client shutdown"):
            await client.ping()

    @pytest.mark.parametrize(
        "config",
        binary_transport_configs(
            websocket_marks=(
                pytest.mark.xfail(
                    reason="WebSocketClient connects again after shutdown: #4287",
                    strict=True,
                ),
            )
        ),
    )
    @pytest.mark.asyncio
    async def test_shutdown_client_cannot_connect_again(
        self, config: TcpConfig | WebSocketConfig | QuicConfig
    ):
        """Test repeated shutdowns leave the client unable to connect again."""
        client = IggyClient(config)
        await client.connect()
        await wait_for_ping(client)

        await client.shutdown()
        await client.shutdown()

        with pytest.raises(RuntimeError):
            await client.connect()

    @pytest.mark.asyncio
    async def test_http_disconnect_and_shutdown_keep_the_session(self):
        """Test HTTP has no connection to close, so the session keeps working."""
        host, port = get_http_server_config()
        client = IggyClient(HttpConfig(api_url=f"http://{host}:{port}"))
        await client.connect()
        await wait_for_ping(client)
        await client.login_user("iggy", "iggy")

        await client.disconnect()
        await client.get_streams()

        await client.shutdown()
        await client.get_streams()