diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml index f97462f34d..ce21915e47 100644 --- a/.github/actions/python-maturin/pre-merge/action.yml +++ b/.github/actions/python-maturin/pre-merge/action.yml @@ -153,6 +153,7 @@ runs: # overwrite the coverage-instrumented .so with a non-instrumented one IGGY_SERVER_HOST=127.0.0.1 \ IGGY_SERVER_TCP_PORT=8090 \ + IGGY_SERVER_WS_PORT=8092 \ IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \ uv run --no-sync pytest tests/ -v \ --junitxml=../../reports/python-junit.xml \ diff --git a/.github/workflows/coverage-baseline.yml b/.github/workflows/coverage-baseline.yml index baa1ae160c..e98af77a13 100644 --- a/.github/workflows/coverage-baseline.yml +++ b/.github/workflows/coverage-baseline.yml @@ -349,6 +349,7 @@ jobs: cd foreign/python IGGY_SERVER_HOST=127.0.0.1 \ IGGY_SERVER_TCP_PORT=8090 \ + IGGY_SERVER_WS_PORT=8092 \ IGGY_SERVER_DOCKER_IMAGE=iggy-server:local \ uv run --no-sync pytest tests/ -v \ --junitxml=../../reports/python-junit.xml \ diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index 81f7d8c16d..e9d46f78f9 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -65,8 +65,8 @@ pub use iggy_common::{ TcpClientConfigBuilder, TcpClientReconnectionConfig, Topic, TopicCreateOptions, TopicDetails, TopicPermissions, TopicUpdateOptions, TransportEndpoints, TransportProtocol, UserId, UserInfo, UserInfoDetails, UserStatus, UserUpdateOptions, Validatable, WebSocketClientConfig, - WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig, defaults, locking, - topic_option_keys, + WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig, WebSocketConfig, defaults, + locking, topic_option_keys, }; pub use iggy_common::{ Client, ClusterClient, ConsumerGroupClient, ConsumerOffsetClient, MessageClient, diff --git a/examples/python/README.md b/examples/python/README.md index 7e5da180d4..72e2047e70 100644 --- a/examples/python/README.md +++ b/examples/python/README.md @@ -94,6 +94,19 @@ python message-headers/typed-headers/producer.py python message-headers/typed-headers/consumer.py ``` +## Transport Protocol Examples + +### WebSocket + +Uses the explicit `IggyClient.websocket()` constructor. Assumes a server started with defaults, +which enables all four transports (`cargo run --bin iggy-server`, or the `docker run` command +above). + +```bash +uv run websocket/producer.py +uv run websocket/consumer.py +``` + ## TLS Examples To test with a TLS-enabled server, start the server with TLS configured (see main README), then run: diff --git a/examples/python/websocket/consumer.py b/examples/python/websocket/consumer.py new file mode 100644 index 0000000000..20e58bc35a --- /dev/null +++ b/examples/python/websocket/consumer.py @@ -0,0 +1,166 @@ +# 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. + +import argparse +import asyncio +import typing +import urllib.parse +from datetime import timedelta + +from apache_iggy import ( + AutoLogin, + Consumer, + IggyClient, + PollingStrategy, + ReceiveMessage, + WebSocketConfig, + WebSocketReconnectionConfig, +) +from loguru import logger + +STREAM_NAME = "sample-stream" +TOPIC_NAME = "sample-topic" +STREAM_ID = 0 +TOPIC_ID = 0 +PARTITION_ID = 0 +CONSUMER_NAME = "sample-consumer" +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + server_address: str + username: str + password: str + + +class ValidateUrl(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str, + _option_string: str | None = None, + ): + parsed_url: urllib.parse.ParseResult = urllib.parse.urlparse("//" + values) + if parsed_url.netloc == "" or parsed_url.path != "": + parser.error(f"Invalid server address: {values}") + setattr(namespace, self.dest, values) + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--server-address", + help="Iggy WebSocket server address (host:port)", + action=ValidateUrl, + default="127.0.0.1:8092", + ) + parser.add_argument( + "--username", + default="iggy", + help="Username for authentication", + ) + parser.add_argument( + "--password", + default="iggy", + help="Password for authentication", + ) + args = parser.parse_args() + return ArgNamespace(**vars(args)) + + +def build_config(args: ArgNamespace) -> WebSocketConfig: + """Build a WebSocket client configuration with auto-login and reconnection.""" + + return WebSocketConfig( + server_address=args.server_address, + auto_login=AutoLogin.username_password(args.username, args.password), + reconnection=WebSocketReconnectionConfig( + enabled=True, + interval=timedelta(seconds=1), + ), + ) + + +async def main(): + args: ArgNamespace = parse_args() + try: + config = build_config(args) + except ValueError as error: + logger.error(f"Invalid client configuration: {error}") + return + logger.info(f"Connecting to {args.server_address}") + + client = IggyClient.websocket(config) + try: + logger.info("Connecting to IggyClient...") + # No login_user() call: auto_login replays the credentials on every connect. + await client.connect() + logger.info("Connected.") + await consume_messages(client) + except Exception as error: + logger.exception(f"Exception occurred in main function: {error}") + + +async def consume_messages(client: IggyClient): + interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep + logger.info( + f"Messages will be consumed from stream: {STREAM_NAME}, " + f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} " + f"with interval {interval * 1000} ms." + ) + offset = 0 + messages_per_batch = 10 + n_consumed_batches = 0 + while n_consumed_batches < BATCHES_LIMIT: + try: + logger.debug("Polling for messages...") + polled_messages = await client.poll_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + consumer=Consumer.Single(CONSUMER_NAME), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.Next(), + count=messages_per_batch, + auto_commit=True, + ) + if not polled_messages: + logger.info("No messages found in current poll") + await asyncio.sleep(interval) + continue + + offset += len(polled_messages) + for message in polled_messages: + handle_message(message) + n_consumed_batches += 1 + await asyncio.sleep(interval) + except Exception as error: + logger.exception(f"Exception occurred while consuming messages: {error}") + break + + logger.info(f"Consumed {n_consumed_batches} batches of messages, exiting.") + + +def handle_message(message: ReceiveMessage): + payload = message.payload().decode("utf-8") + logger.info( + f"Handling message at offset: {message.offset()} with payload: {payload}..." + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/python/websocket/producer.py b/examples/python/websocket/producer.py new file mode 100644 index 0000000000..f851b47055 --- /dev/null +++ b/examples/python/websocket/producer.py @@ -0,0 +1,181 @@ +# 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. + +import argparse +import asyncio +import typing +import urllib.parse +from datetime import timedelta + +from apache_iggy import ( + AutoLogin, + IggyClient, + StreamDetails, + TopicDetails, + WebSocketConfig, + WebSocketReconnectionConfig, +) +from apache_iggy import SendMessage as Message +from loguru import logger + +STREAM_NAME = "sample-stream" +TOPIC_NAME = "sample-topic" +STREAM_ID = 0 +TOPIC_ID = 0 +PARTITION_ID = 0 +BATCHES_LIMIT = 5 + + +class ArgNamespace(typing.NamedTuple): + server_address: str + username: str + password: str + + +class ValidateUrl(argparse.Action): + def __call__( + self, + parser: argparse.ArgumentParser, + namespace: argparse.Namespace, + values: str, + _option_string: str | None = None, + ): + parsed_url: urllib.parse.ParseResult = urllib.parse.urlparse("//" + values) + if parsed_url.netloc == "" or parsed_url.path != "": + parser.error(f"Invalid server address: {values}") + setattr(namespace, self.dest, values) + + +def parse_args() -> ArgNamespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "--server-address", + help="Iggy WebSocket server address (host:port)", + action=ValidateUrl, + default="127.0.0.1:8092", + ) + parser.add_argument( + "--username", + default="iggy", + help="Username for authentication", + ) + parser.add_argument( + "--password", + default="iggy", + help="Password for authentication", + ) + args = parser.parse_args() + return ArgNamespace(**vars(args)) + + +def build_config(args: ArgNamespace) -> WebSocketConfig: + """Build a WebSocket client configuration with auto-login and reconnection.""" + + return WebSocketConfig( + server_address=args.server_address, + auto_login=AutoLogin.username_password(args.username, args.password), + reconnection=WebSocketReconnectionConfig( + enabled=True, + interval=timedelta(seconds=1), + ), + ) + + +async def main(): + args: ArgNamespace = parse_args() + try: + config = build_config(args) + except ValueError as error: + logger.error(f"Invalid client configuration: {error}") + return + logger.info(f"Connecting to {args.server_address}") + + client = IggyClient.websocket(config) + logger.info("Connecting to IggyClient") + # No login_user() call: auto_login replays the credentials on every connect. + await client.connect() + logger.info("Connected.") + await init_system(client) + await produce_messages(client) + + +async def init_system(client: IggyClient): + logger.info(f"Creating stream with name {STREAM_NAME}...") + stream: StreamDetails | None = await client.get_stream(STREAM_NAME) + if stream is None: + await client.create_stream(name=STREAM_NAME) + logger.info("Stream was created successfully.") + else: + logger.warning(f"Stream {stream.name} already exists with ID {stream.id}") + + logger.info(f"Creating topic {TOPIC_NAME} in stream {STREAM_NAME}") + topic: TopicDetails | None = await client.get_topic(STREAM_NAME, TOPIC_NAME) + if topic is None: + await client.create_topic( + stream=STREAM_NAME, + partitions_count=1, + name=TOPIC_NAME, + ) + logger.info("Topic was created successfully.") + else: + logger.warning(f"Topic {topic.name} already exists with ID {topic.id}") + + +async def produce_messages(client: IggyClient): + interval = 0.5 # 500 milliseconds in seconds for asyncio.sleep + logger.info( + f"Messages will be sent to stream: {STREAM_NAME}, " + f"topic: {TOPIC_NAME}, partition: {PARTITION_ID} " + f"with interval {interval * 1000} ms." + ) + current_id = 0 + messages_per_batch = 10 + n_sent_batches = 0 + while n_sent_batches < BATCHES_LIMIT: + messages = [] + for _ in range(messages_per_batch): + current_id += 1 + payload = f"message-{current_id}" + message = Message(payload) + messages.append(message) + logger.info( + f"Attempting to send batch of {messages_per_batch} messages. " + f"Batch ID: {current_id // messages_per_batch}" + ) + try: + await client.send_messages( + stream=STREAM_NAME, + topic=TOPIC_NAME, + partitioning=PARTITION_ID, + messages=messages, + ) + n_sent_batches += 1 + logger.info( + f"Successfully sent batch of {messages_per_batch} messages. " + f"Batch ID: {current_id // messages_per_batch}" + ) + except Exception as error: + logger.error(f"Exception type: {type(error).__name__}, message: {error}") + logger.exception(error) + break + + await asyncio.sleep(interval) + logger.info(f"Sent {n_sent_batches} batches of messages, exiting.") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/foreign/python/README.md b/foreign/python/README.md index dbbdbe02ca..d16a464089 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -166,6 +166,50 @@ async def main(): await client.connect() +asyncio.run(main()) +``` + +`IggyClient.websocket(...)` takes a `WebSocketConfig` the same way, built from +`IggyClient.websocket()`'s own config type rather than passed to `IggyClient(...)`: + +```python +import asyncio +from datetime import timedelta + +from apache_iggy import ( + AutoLogin, + IggyClient, + WebSocketConfig, + WebSocketFramingConfig, + WebSocketReconnectionConfig, +) + + +async def main(): + client = IggyClient.websocket( + WebSocketConfig( + server_address="127.0.0.1:8092", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=WebSocketReconnectionConfig( + enabled=True, + max_retries=10, + interval=timedelta(seconds=2), + reestablish_after=timedelta(seconds=30), + ), + heartbeat_interval=timedelta(seconds=5), + framing=WebSocketFramingConfig( + max_message_size=64 * 1024 * 1024, + max_frame_size=16 * 1024 * 1024, + ), + # tls_enabled=True, + # tls_domain="localhost", + # tls_ca_file="../../core/certs/iggy_ca_cert.pem", + # tls_validate_certificate=True, + ) + ) + await client.connect() + + asyncio.run(main()) ``` diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 254b23f41c..7b98a6c4f6 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -60,6 +60,9 @@ __all__ = [ "UserInfo", "UserInfoDetails", "UserStatus", + "WebSocketConfig", + "WebSocketFramingConfig", + "WebSocketReconnectionConfig", ] class AutoCommit: @@ -867,6 +870,21 @@ class IggyClient: Constructs a new IggyClient from a connection string. Returns an error if the connection string provided is invalid. """ + @classmethod + def websocket(cls, config: WebSocketConfig | None = None) -> IggyClient: + r""" + Constructs a new IggyClient configured for the WebSocket transport. + + `server_address` is already validated when `config` is built, so this + does not currently fail; the exception is documented for interface + consistency with the other transport constructors. + + Args: + config: WebSocket transport configuration. Defaults to `WebSocketConfig()`. + + Raises: + RuntimeError: If the client cannot be constructed. + """ def ping(self) -> collections.abc.Awaitable[None]: r""" Sends a ping request to the server to check connectivity. @@ -2331,3 +2349,165 @@ class UserStatus(enum.Enum): r""" The user account is inactive and cannot be used. """ + +@typing.final +class WebSocketConfig: + r""" + Configuration for the WebSocket transport, accepted by `IggyClient.websocket(...)`. + + Every field is keyword-only and optional. + """ + @property + def server_address(self) -> builtins.str: ... + @property + def auto_login(self) -> AutoLogin: ... + @property + def reconnection(self) -> WebSocketReconnectionConfig: ... + @property + def heartbeat_interval(self) -> datetime.timedelta: ... + @property + def framing(self) -> WebSocketFramingConfig: ... + @property + def tls_enabled(self) -> builtins.bool: ... + @property + def tls_domain(self) -> builtins.str: ... + @property + def tls_ca_file(self) -> builtins.str | None: ... + @property + def tls_validate_certificate(self) -> builtins.bool: ... + def __new__( + cls, + *, + server_address: builtins.str | None = None, + auto_login: AutoLogin | None = None, + reconnection: WebSocketReconnectionConfig | None = None, + heartbeat_interval: datetime.timedelta | None = None, + framing: WebSocketFramingConfig | None = None, + tls_enabled: builtins.bool | None = None, + tls_domain: builtins.str | None = None, + tls_ca_file: builtins.str | None = None, + tls_validate_certificate: builtins.bool | None = None, + ) -> WebSocketConfig: + r""" + Constructs a WebSocket configuration. + + Args: + server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8092`. + auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + reconnection: Reconnection policy. Defaults to `WebSocketReconnectionConfig()`. + heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + framing: Frame- and buffer-level options. Defaults to `WebSocketFramingConfig()`. + tls_enabled: Whether to connect over TLS. Defaults to disabled. + tls_domain: Domain to validate the certificate against. Defaults to `localhost`. + tls_ca_file: Path to the CA file for TLS. Read only when `tls_enabled` + and `tls_validate_certificate` are both on; with either one off it + is kept but never consulted, so pairing it with + `tls_validate_certificate=False` pins nothing. + tls_validate_certificate: Whether to validate the server certificate. + Defaults to `False`, unlike the TCP and QUIC transports. Disabling + this accepts any certificate the server presents, including + self-signed and mismatched ones, and takes precedence over + `tls_ca_file`. + + Raises: + ValueError: If `server_address` is not a valid `host:port` pair, if a + duration is negative, or if `heartbeat_interval` is zero. + """ + def __repr__(self) -> builtins.str: ... + +@typing.final +class WebSocketFramingConfig: + r""" + Frame- and buffer-level options passed through to the underlying WebSocket + implementation, accepted by `WebSocketConfig`'s `framing` argument. + + Every field is keyword-only and optional; unset fields fall back to the + underlying WebSocket library's own defaults. + """ + @property + def read_buffer_size(self) -> builtins.int | None: ... + @property + def write_buffer_size(self) -> builtins.int | None: ... + @property + def max_write_buffer_size(self) -> builtins.int | None: ... + @property + def max_message_size(self) -> builtins.int | None: ... + @property + def max_frame_size(self) -> builtins.int | None: ... + @property + def accept_unmasked_frames(self) -> builtins.bool: ... + def __new__( + cls, + *, + read_buffer_size: builtins.int | None = None, + write_buffer_size: builtins.int | None = None, + max_write_buffer_size: builtins.int | None = None, + max_message_size: builtins.int | None = None, + max_frame_size: builtins.int | None = None, + accept_unmasked_frames: builtins.bool | None = None, + ) -> WebSocketFramingConfig: + r""" + Constructs a WebSocket framing configuration. + + Args: + read_buffer_size: Read buffer size in bytes. + write_buffer_size: Write buffer size in bytes. + max_write_buffer_size: Maximum write buffer size in bytes. + max_message_size: Maximum message size in bytes, or `None` for no limit. + max_frame_size: Maximum frame size in bytes, or `None` for no limit. + accept_unmasked_frames: Whether to accept unmasked frames. Defaults to + `False`; clients should typically keep this off for RFC compliance. + + Raises: + ValueError: If a numeric field is outside the range of a pointer-sized + unsigned integer. + """ + def __repr__(self) -> builtins.str: ... + +@typing.final +class WebSocketReconnectionConfig: + r""" + How the WebSocket client reconnects after the connection to the server is lost. + """ + @property + def enabled(self) -> builtins.bool: ... + @property + def max_retries(self) -> builtins.int | None: ... + @property + def interval(self) -> datetime.timedelta: ... + @property + def reestablish_after(self) -> datetime.timedelta: ... + def __new__( + cls, + *, + enabled: builtins.bool | None = None, + max_retries: builtins.int | None = None, + interval: datetime.timedelta | None = None, + reestablish_after: datetime.timedelta | None = None, + ) -> WebSocketReconnectionConfig: + r""" + Constructs a reconnection policy. + + Args: + enabled: Whether to reconnect at all. Defaults to enabled. + max_retries: Passes over the known endpoints after the first, or + `None` for unlimited; `0` still makes that first pass. One pass + tries the endpoint the client is on, the address it was + configured with, and every node the roster named, so this counts + passes rather than dials. Defaults + to unlimited, which means a call awaited while the server is + down never returns: `connect()`, `send_messages()` and + `poll_messages()` all wait inside the retry loop. Set a finite + number for request/reply style usage, so a call fails instead. + interval: Delay between passes. Defaults to 1 second. The first pass + runs at once when more than one endpoint is known. + reestablish_after: Cooldown before redialing the endpoint of the last + successful connection, measured from when it was established, so + a session that outlived the interval is redialed at once. Owed to + that endpoint alone. Defaults to 5 seconds. + + Raises: + ValueError: If a duration is negative, if `max_retries` is outside the + range of an unsigned 32-bit integer, or if `interval` is zero. + """ + def __repr__(self) -> builtins.str: ... diff --git a/foreign/python/docker-compose.test.yml b/foreign/python/docker-compose.test.yml index 78caed9929..069b6d3ec5 100644 --- a/foreign/python/docker-compose.test.yml +++ b/foreign/python/docker-compose.test.yml @@ -33,6 +33,7 @@ services: - "3000:3000" - "8080:8080" - "8090:8090" + - "8092:8092" environment: - IGGY_HTTP_ADDRESS=0.0.0.0:3000 - IGGY_TCP_ADDRESS=0.0.0.0:8090 @@ -64,6 +65,7 @@ services: - IGGY_SERVER_TCP_PORT=8090 - IGGY_SERVER_HTTP_PORT=3000 - IGGY_SERVER_QUIC_PORT=8080 + - IGGY_SERVER_WS_PORT=8092 - PYTHONPATH=/workspace/foreign/python - PYTEST_ARGS=-v --tb=short volumes: diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index f669a87484..b067796925 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -31,7 +31,7 @@ use std::collections::BTreeMap; use std::str::FromStr; use std::sync::Arc; -use crate::config::PyClientConfig; +use crate::config::{PyClientConfig, WebSocketConfig}; use crate::consumer::{ AutoCommit, Consumer as PyConsumer, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer, @@ -144,6 +144,32 @@ impl IggyClient { }) } + /// Constructs a new IggyClient configured for the WebSocket transport. + /// + /// `server_address` is already validated when `config` is built, so this + /// does not currently fail; the exception is documented for interface + /// consistency with the other transport constructors. + /// + /// Args: + /// config: WebSocket transport configuration. Defaults to `WebSocketConfig()`. + /// + /// Raises: + /// RuntimeError: If the client cannot be constructed. + #[classmethod] + #[pyo3(signature = (config=None))] + fn websocket(_cls: &Bound<'_, PyType>, config: Option) -> PyResult { + let config = config + .map(|config| config.client_config()) + .unwrap_or_else(|| Arc::new(WebSocketClientConfig::default())); + let websocket_client = WebSocketClient::create(config) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Self { + inner: Arc::new(RustIggyClient::new(ClientWrapper::WebSocket( + websocket_client, + ))), + }) + } + /// Sends a ping request to the server to check connectivity. /// Raises `RuntimeError` if the connection fails. #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs index ed6e2a14f0..fe8cd759be 100644 --- a/foreign/python/src/config.rs +++ b/foreign/python/src/config.rs @@ -19,6 +19,9 @@ use iggy::prelude::{ AutoLogin as RustAutoLogin, Credentials as RustCredentials, TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, + WebSocketClientConfig as RustWebSocketClientConfig, WebSocketClientConfigBuilder, + WebSocketClientReconnectionConfig as RustWebSocketClientReconnectionConfig, + WebSocketConfig as RustWebSocketFramingConfig, }; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; @@ -408,10 +411,465 @@ impl TcpConfig { } } +/// How the WebSocket client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct WebSocketReconnectionConfig { + pub(crate) inner: RustWebSocketClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl WebSocketReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Passes over the known endpoints after the first, or + /// `None` for unlimited; `0` still makes that first pass. One pass + /// tries the endpoint the client is on, the address it was + /// configured with, and every node the roster named, so this counts + /// passes rather than dials. Defaults + /// to unlimited, which means a call awaited while the server is + /// down never returns: `connect()`, `send_messages()` and + /// `poll_messages()` all wait inside the retry loop. Set a finite + /// number for request/reply style usage, so a call fails instead. + /// interval: Delay between passes. Defaults to 1 second. The first pass + /// runs at once when more than one endpoint is known. + /// reestablish_after: Cooldown before redialing the endpoint of the last + /// successful connection, measured from when it was established, so + /// a session that outlived the interval is redialed at once. Owed to + /// that endpoint alone. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative, if `max_retries` is outside the + /// range of an unsigned 32-bit integer, or if `interval` is zero. + #[new] + #[pyo3(signature = (*, enabled=None, max_retries=None, interval=None, reestablish_after=None))] + fn new( + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] enabled: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option>, + ) -> PyResult { + let defaults = RustWebSocketClientReconnectionConfig::default(); + let enabled = enabled.unwrap_or(defaults.enabled); + let max_retries = max_retries + .map(|max_retries| { + u32::try_from(max_retries).map_err(|_| { + PyValueError::new_err(format!( + "'max_retries' must be between 0 and {}", + u32::MAX + )) + }) + }) + .transpose()?; + let interval = interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .map(|interval| reject_zero(interval, "interval")) + .transpose()? + .unwrap_or(defaults.interval); + Ok(Self { + inner: RustWebSocketClientReconnectionConfig { + enabled, + max_retries, + interval, + reestablish_after: reestablish_after + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.reestablish_after), + }, + }) + } + + #[getter] + fn enabled(&self) -> bool { + self.inner.enabled + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn max_retries(&self) -> Option { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.interval.get()) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.reestablish_after) + } + + fn __repr__(&self) -> String { + let max_retries = match self.inner.max_retries { + Some(max_retries) => max_retries.to_string(), + None => "None".to_owned(), + }; + format!( + "WebSocketReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + python_bool(self.inner.enabled), + duration_repr(self.inner.interval.get()), + duration_repr(self.inner.reestablish_after), + ) + } +} + +/// Frame- and buffer-level options passed through to the underlying WebSocket +/// implementation, accepted by `WebSocketConfig`'s `framing` argument. +/// +/// Every field is keyword-only and optional; unset fields fall back to the +/// underlying WebSocket library's own defaults. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct WebSocketFramingConfig { + pub(crate) inner: RustWebSocketFramingConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl WebSocketFramingConfig { + /// Constructs a WebSocket framing configuration. + /// + /// Args: + /// read_buffer_size: Read buffer size in bytes. + /// write_buffer_size: Write buffer size in bytes. + /// max_write_buffer_size: Maximum write buffer size in bytes. + /// max_message_size: Maximum message size in bytes, or `None` for no limit. + /// max_frame_size: Maximum frame size in bytes, or `None` for no limit. + /// accept_unmasked_frames: Whether to accept unmasked frames. Defaults to + /// `False`; clients should typically keep this off for RFC compliance. + /// + /// Raises: + /// ValueError: If a numeric field is outside the range of a pointer-sized + /// unsigned integer, or if `max_write_buffer_size` does not come out + /// greater than `write_buffer_size`. tungstenite enforces the same + /// invariant with an `assert!` at connect time, which would otherwise + /// surface as an unrecoverable Rust panic instead of a `ValueError`. + #[new] + #[pyo3(signature = ( + *, + read_buffer_size=None, + write_buffer_size=None, + max_write_buffer_size=None, + max_message_size=None, + max_frame_size=None, + accept_unmasked_frames=None, + ))] + fn new( + #[gen_stub(override_type(type_repr = "builtins.int | None"))] read_buffer_size: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] write_buffer_size: Option< + i64, + >, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_write_buffer_size: Option< + i64, + >, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_message_size: Option, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_frame_size: Option, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + accept_unmasked_frames: Option, + ) -> PyResult { + let mut inner = RustWebSocketFramingConfig::default(); + if let Some(read_buffer_size) = read_buffer_size { + inner.read_buffer_size = Some(usize_param(read_buffer_size, "read_buffer_size")?); + } + if let Some(write_buffer_size) = write_buffer_size { + inner.write_buffer_size = Some(usize_param(write_buffer_size, "write_buffer_size")?); + } + if let Some(max_write_buffer_size) = max_write_buffer_size { + inner.max_write_buffer_size = + Some(usize_param(max_write_buffer_size, "max_write_buffer_size")?); + } + if let Some(max_message_size) = max_message_size { + inner.max_message_size = Some(usize_param(max_message_size, "max_message_size")?); + } + if let Some(max_frame_size) = max_frame_size { + inner.max_frame_size = Some(usize_param(max_frame_size, "max_frame_size")?); + } + if let Some(accept_unmasked_frames) = accept_unmasked_frames { + inner.accept_unmasked_frames = accept_unmasked_frames; + } + if let (Some(write_buffer_size), Some(max_write_buffer_size)) = + (inner.write_buffer_size, inner.max_write_buffer_size) + && max_write_buffer_size <= write_buffer_size + { + return Err(PyValueError::new_err(format!( + "'max_write_buffer_size' ({max_write_buffer_size}) must be greater than \ + 'write_buffer_size' ({write_buffer_size})" + ))); + } + + Ok(Self { inner }) + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn read_buffer_size(&self) -> Option { + self.inner.read_buffer_size + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn write_buffer_size(&self) -> Option { + self.inner.write_buffer_size + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn max_write_buffer_size(&self) -> Option { + self.inner.max_write_buffer_size + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn max_message_size(&self) -> Option { + self.inner.max_message_size + } + + #[gen_stub(override_return_type(type_repr = "builtins.int | None"))] + #[getter] + fn max_frame_size(&self) -> Option { + self.inner.max_frame_size + } + + #[getter] + fn accept_unmasked_frames(&self) -> bool { + self.inner.accept_unmasked_frames + } + + fn __repr__(&self) -> String { + let optional_usize = |value: Option| match value { + Some(value) => value.to_string(), + None => "None".to_owned(), + }; + format!( + "WebSocketFramingConfig(read_buffer_size={}, write_buffer_size={}, max_write_buffer_size={}, max_message_size={}, max_frame_size={}, accept_unmasked_frames={})", + optional_usize(self.inner.read_buffer_size), + optional_usize(self.inner.write_buffer_size), + optional_usize(self.inner.max_write_buffer_size), + optional_usize(self.inner.max_message_size), + optional_usize(self.inner.max_frame_size), + python_bool(self.inner.accept_unmasked_frames), + ) + } +} + +/// Configuration for the WebSocket transport, accepted by `IggyClient.websocket(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct WebSocketConfig { + inner: Arc, +} + +impl WebSocketConfig { + /// The configuration in the shape `WebSocketClient::create` expects. + pub(crate) fn client_config(&self) -> Arc { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl WebSocketConfig { + /// Constructs a WebSocket configuration. + /// + /// Args: + /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8092`. + /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + /// reconnection: Reconnection policy. Defaults to `WebSocketReconnectionConfig()`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// framing: Frame- and buffer-level options. Defaults to `WebSocketFramingConfig()`. + /// tls_enabled: Whether to connect over TLS. Defaults to disabled. + /// tls_domain: Domain to validate the certificate against. Defaults to `localhost`. + /// tls_ca_file: Path to the CA file for TLS. Read only when `tls_enabled` + /// and `tls_validate_certificate` are both on; with either one off it + /// is kept but never consulted, so pairing it with + /// `tls_validate_certificate=False` pins nothing. + /// tls_validate_certificate: Whether to validate the server certificate. + /// Defaults to `False`, unlike the TCP and QUIC transports. Disabling + /// this accepts any certificate the server presents, including + /// self-signed and mismatched ones, and takes precedence over + /// `tls_ca_file`. + /// + /// Raises: + /// ValueError: If `server_address` is not a valid `host:port` pair, if a + /// duration is negative, or if `heartbeat_interval` is zero. + #[new] + #[pyo3(signature = ( + *, + server_address=None, + auto_login=None, + reconnection=None, + heartbeat_interval=None, + framing=None, + tls_enabled=None, + tls_domain=None, + tls_ca_file=None, + tls_validate_certificate=None, + ))] + #[allow(clippy::too_many_arguments)] + fn new( + #[gen_stub(override_type(type_repr = "builtins.str | None"))] server_address: Option< + String, + >, + #[gen_stub(override_type(type_repr = "AutoLogin | None"))] auto_login: Option, + #[gen_stub(override_type(type_repr = "WebSocketReconnectionConfig | None"))] + reconnection: Option, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option>, + #[gen_stub(override_type(type_repr = "WebSocketFramingConfig | None"))] framing: Option< + WebSocketFramingConfig, + >, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + tls_validate_certificate: Option, + ) -> PyResult { + // The builder starts from `WebSocketClientConfig::default()`, and its + // `build()` trims and validates the address whether or not one was set here. + let mut builder = WebSocketClientConfigBuilder::new(); + if let Some(server_address) = server_address { + builder = builder.with_server_address(server_address); + } + let mut inner = builder + .build() + .map_err(|e| PyValueError::new_err(e.to_string()))?; + if let Some(auto_login) = auto_login { + inner.auto_login = auto_login.inner; + } + if let Some(reconnection) = reconnection { + inner.reconnection = reconnection.inner; + } + if let Some(heartbeat_interval) = heartbeat_interval { + inner.heartbeat_interval = reject_zero( + py_delta_to_iggy_duration(&heartbeat_interval)?, + "heartbeat_interval", + )?; + } + if let Some(framing) = framing { + inner.ws_config = framing.inner; + } + if let Some(tls_enabled) = tls_enabled { + inner.tls_enabled = tls_enabled; + } + if let Some(tls_domain) = tls_domain { + inner.tls_domain = tls_domain; + } + if tls_ca_file.is_some() { + inner.tls_ca_file = tls_ca_file; + } + if let Some(tls_validate_certificate) = tls_validate_certificate { + inner.tls_validate_certificate = tls_validate_certificate; + } + + Ok(Self { + inner: Arc::new(inner), + }) + } + + #[getter] + fn server_address(&self) -> String { + self.inner.server_address.clone() + } + + #[getter] + fn auto_login(&self) -> AutoLogin { + AutoLogin { + inner: self.inner.auto_login.clone(), + } + } + + #[getter] + fn reconnection(&self) -> WebSocketReconnectionConfig { + WebSocketReconnectionConfig { + inner: self.inner.reconnection.clone(), + } + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult> { + iggy_duration_to_py_delta(py, self.inner.heartbeat_interval.get()) + } + + #[getter] + fn framing(&self) -> WebSocketFramingConfig { + WebSocketFramingConfig { + inner: self.inner.ws_config.clone(), + } + } + + #[getter] + fn tls_enabled(&self) -> bool { + self.inner.tls_enabled + } + + #[getter] + fn tls_domain(&self) -> String { + self.inner.tls_domain.clone() + } + + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn tls_ca_file(&self) -> Option { + self.inner.tls_ca_file.clone() + } + + #[getter] + fn tls_validate_certificate(&self) -> bool { + self.inner.tls_validate_certificate + } + + fn __repr__(&self) -> String { + let tls_ca_file = match &self.inner.tls_ca_file { + Some(tls_ca_file) => format!("{tls_ca_file:?}"), + None => "None".to_owned(), + }; + format!( + "WebSocketConfig(server_address={:?}, auto_login={}, reconnection={}, heartbeat_interval={}, framing={}, tls_enabled={}, tls_domain={:?}, tls_ca_file={tls_ca_file}, tls_validate_certificate={})", + self.inner.server_address, + self.auto_login().__repr__(), + self.reconnection().__repr__(), + duration_repr(self.inner.heartbeat_interval.get()), + self.framing().__repr__(), + python_bool(self.inner.tls_enabled), + self.inner.tls_domain, + python_bool(self.inner.tls_validate_certificate), + ) + } +} + fn python_bool(value: bool) -> &'static str { if value { "True" } else { "False" } } +/// Converts a Python int to the unsigned pointer-sized integer a WebSocket +/// framing field expects, naming the parameter in the error so a caller can +/// tell which argument was out of range. +fn usize_param(value: i64, parameter: &str) -> PyResult { + usize::try_from(value).map_err(|_| { + PyValueError::new_err(format!( + "'{parameter}' must be between 0 and {}", + usize::MAX + )) + }) +} + /// What `IggyClient(...)` accepts: a bare `host:port` or a full `TcpConfig`. #[derive(FromPyObject)] pub enum PyClientConfig { diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index 5f2e128264..9e1653604c 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -30,7 +30,10 @@ mod user; mod user_headers; use client::IggyClient; -use config::{AutoLogin, TcpConfig, TcpReconnectionConfig}; +use config::{ + AutoLogin, TcpConfig, TcpReconnectionConfig, WebSocketConfig, WebSocketFramingConfig, + WebSocketReconnectionConfig, +}; use consumer::{ AutoCommit, AutoCommitAfter, AutoCommitWhen, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator, @@ -56,6 +59,9 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/foreign/python/tests/test_websocket_config.py b/foreign/python/tests/test_websocket_config.py new file mode 100644 index 0000000000..b8b35a05fc --- /dev/null +++ b/foreign/python/tests/test_websocket_config.py @@ -0,0 +1,424 @@ +# 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. + +""" +Tests for the WebSocket client configuration surface. + +`WebSocketConfig` and `WebSocketReconnectionConfig` mirror the Rust SDK +types the same way `TcpConfig`/`TcpReconnectionConfig` do, so most of these +assert that a value set from Python survives to the getters and that unset +fields fall back to the Rust defaults. `WebSocketFramingConfig` is new: it +wraps the tungstenite frame- and buffer-level options the Rust SDK nests +under `ws_config`, exposed here as a `framing` argument rather than flat +kwargs, mirroring the Rust struct's own nesting. `AutoLogin` is +transport-agnostic and already covered by `test_client_config.py`. +""" + +import ast +from collections.abc import Callable +from datetime import timedelta + +import pytest + +from apache_iggy import ( + AutoLogin, + IggyClient, + WebSocketConfig, + WebSocketFramingConfig, + WebSocketReconnectionConfig, +) + +from .utils import get_websocket_server_config, wait_for_ping, wait_for_server + + +@pytest.mark.unit +class TestWebSocketReconnectionConfig: + """Test the reconnection policy.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured policy reconnects forever, one second apart.""" + reconnection = WebSocketReconnectionConfig() + + assert reconnection.enabled is True + assert reconnection.max_retries is None + assert reconnection.interval == timedelta(seconds=1) + assert reconnection.reestablish_after == timedelta(seconds=5) + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + reconnection = WebSocketReconnectionConfig( + enabled=False, + max_retries=10, + interval=timedelta(milliseconds=250), + reestablish_after=timedelta(seconds=30), + ) + + assert reconnection.enabled is False + assert reconnection.max_retries == 10 + assert reconnection.interval == timedelta(milliseconds=250) + assert reconnection.reestablish_after == timedelta(seconds=30) + + def test_arguments_are_keyword_only(self): + """Test that the adjacent flags cannot be passed positionally.""" + with pytest.raises(TypeError): + # pyrefly: ignore # bad-argument-count + WebSocketReconnectionConfig(True) + + @pytest.mark.parametrize( + "construct", + [ + lambda duration: WebSocketReconnectionConfig(interval=duration), + lambda duration: WebSocketReconnectionConfig(reestablish_after=duration), + ], + ids=["interval", "reestablish_after"], + ) + @pytest.mark.parametrize( + "negative", + [timedelta(microseconds=-1), timedelta(seconds=-1), timedelta(days=-1)], + ) + def test_negative_duration_is_rejected( + self, + construct: Callable[[timedelta], WebSocketReconnectionConfig], + negative: timedelta, + ): + """Test that a negative duration fails at construction, not at connect.""" + with pytest.raises(ValueError, match="negative"): + construct(negative) + + @pytest.mark.parametrize("out_of_range", [-1, 2**32]) + def test_out_of_range_max_retries_is_rejected(self, out_of_range: int): + """Test that a retry count outside the wire range names the argument. + + The conversion pyo3 does on its own raises OverflowError, which is not a + ValueError and so escapes the handler a caller wraps construction in. + """ + with pytest.raises(ValueError, match="max_retries"): + WebSocketReconnectionConfig(max_retries=out_of_range) + + def test_zero_reestablish_after_is_allowed(self): + """Test that a zero cooldown is legal and readable back.""" + reconnection = WebSocketReconnectionConfig(reestablish_after=timedelta(0)) + + assert reconnection.reestablish_after == timedelta(0) + + @pytest.mark.parametrize( + "kwargs", + [ + {}, + {"max_retries": 5}, + {"enabled": False}, + ], + ids=["unlimited_retries", "bounded_retries", "reconnection_disabled"], + ) + def test_zero_interval_is_rejected(self, kwargs: dict): + """Test that a zero interval fails whatever the retry policy is. + + The interval is a delay between passes, so zero reconnects in a + continuous loop. + """ + with pytest.raises(ValueError, match="zero"): + WebSocketReconnectionConfig(interval=timedelta(0), **kwargs) + + def test_very_long_interval_round_trips(self): + """Test that an interval beyond 68 years survives the i32 boundary.""" + reconnection = WebSocketReconnectionConfig(interval=timedelta(days=30_000)) + + assert reconnection.interval == timedelta(days=30_000) + + def test_maximum_interval_round_trips(self): + """Test that the largest timedelta survives the day conversion.""" + reconnection = WebSocketReconnectionConfig(interval=timedelta(days=999_999_999)) + + assert reconnection.interval == timedelta(days=999_999_999) + + +@pytest.mark.unit +class TestWebSocketFramingConfig: + """Test the frame- and buffer-level options.""" + + def test_defaults_match_tungstenite(self): + """Test that unconfigured sizes fall back to the tungstenite defaults. + + Every size defaults to `Some(...)`; passing `None` explicitly leaves the + default untouched rather than clearing the limit, since the constructor + only assigns a field when its argument is `Some(...)`. + """ + framing = WebSocketFramingConfig() + + assert framing.read_buffer_size is not None + assert framing.write_buffer_size is not None + assert framing.max_write_buffer_size is not None + assert framing.max_message_size is not None + assert framing.max_frame_size is not None + assert framing.accept_unmasked_frames is False + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + framing = WebSocketFramingConfig( + read_buffer_size=4096, + write_buffer_size=4096, + max_write_buffer_size=8192, + max_message_size=16384, + max_frame_size=16384, + accept_unmasked_frames=True, + ) + + assert framing.read_buffer_size == 4096 + assert framing.write_buffer_size == 4096 + assert framing.max_write_buffer_size == 8192 + assert framing.max_message_size == 16384 + assert framing.max_frame_size == 16384 + assert framing.accept_unmasked_frames is True + + def test_arguments_are_keyword_only(self): + """Test that the first field cannot be passed positionally.""" + with pytest.raises(TypeError): + # pyrefly: ignore # bad-argument-count + WebSocketFramingConfig(4096) + + def test_repr_shows_every_field_as_python(self): + """Test that repr covers every field and parses as Python.""" + framing = WebSocketFramingConfig( + read_buffer_size=4096, + accept_unmasked_frames=True, + ) + + printed = repr(framing) + + assert "read_buffer_size=4096" in printed + assert "accept_unmasked_frames=True" in printed + ast.parse(printed) + + @pytest.mark.parametrize( + "field", + [ + "read_buffer_size", + "write_buffer_size", + "max_write_buffer_size", + "max_message_size", + "max_frame_size", + ], + ) + def test_negative_size_is_rejected(self, field: str): + """Test that a negative size names the argument that caused it.""" + with pytest.raises(ValueError, match=field): + # pyrefly: ignore # bad-argument-type + WebSocketFramingConfig(**{field: -1}) + + def test_max_write_buffer_size_not_greater_than_write_buffer_size_is_rejected(self): + """Test that a non-increasing write buffer pair fails at construction. + + tungstenite enforces `max_write_buffer_size > write_buffer_size` with an + `assert!` when the connection is established, which would otherwise + surface as an unrecoverable Rust panic instead of a catchable error. + """ + with pytest.raises(ValueError, match="max_write_buffer_size"): + WebSocketFramingConfig(write_buffer_size=1000, max_write_buffer_size=1000) + + def test_max_write_buffer_size_below_the_default_write_buffer_size_is_rejected( + self, + ): + """Test that the invariant is checked against the default too. + + Setting only `max_write_buffer_size` below the untouched default + `write_buffer_size` (128 KiB) must fail the same way as setting both. + """ + with pytest.raises(ValueError, match="max_write_buffer_size"): + WebSocketFramingConfig(max_write_buffer_size=1000) + + +@pytest.mark.unit +class TestWebSocketConfig: + """Test the transport configuration.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured transport matches the Rust SDK defaults.""" + config = WebSocketConfig() + + assert config.server_address == "127.0.0.1:8092" + assert config.auto_login.enabled is False + assert config.reconnection.enabled is True + assert config.heartbeat_interval == timedelta(seconds=5) + assert config.tls_enabled is False + assert config.tls_domain == "localhost" + assert config.tls_ca_file is None + # Unlike TCP and QUIC, WebSocket does not validate the server + # certificate by default. + assert config.tls_validate_certificate is False + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + config = WebSocketConfig( + server_address="127.0.0.1:8093", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=WebSocketReconnectionConfig(max_retries=3), + heartbeat_interval=timedelta(seconds=15), + framing=WebSocketFramingConfig(read_buffer_size=4096), + tls_enabled=True, + tls_domain="example.com", + tls_ca_file="ca.pem", + tls_validate_certificate=True, + ) + + assert config.server_address == "127.0.0.1:8093" + assert config.auto_login.username == "iggy" + assert config.reconnection.max_retries == 3 + assert config.heartbeat_interval == timedelta(seconds=15) + assert config.framing.read_buffer_size == 4096 + assert config.tls_enabled is True + assert config.tls_domain == "example.com" + assert config.tls_ca_file == "ca.pem" + assert config.tls_validate_certificate is True + + def test_arguments_are_keyword_only(self): + """Test that the address cannot be passed positionally.""" + with pytest.raises(TypeError): + # pyrefly: ignore # bad-argument-count + WebSocketConfig("127.0.0.1:8092") + + def test_repr_hides_the_password(self): + """Test that the password does not leak through repr.""" + config = WebSocketConfig( + auto_login=AutoLogin.username_password("iggy", "secret") + ) + + assert "secret" not in repr(config) + + def test_repr_shows_every_field_as_python(self): + """Test that repr covers the TLS fields and parses as Python.""" + config = WebSocketConfig( + heartbeat_interval=timedelta(seconds=15), + tls_enabled=True, + tls_domain="example.com", + tls_ca_file="ca.pem", + tls_validate_certificate=True, + ) + + printed = repr(config) + + assert 'tls_domain="example.com"' in printed + assert 'tls_ca_file="ca.pem"' in printed + assert "tls_validate_certificate=True" in printed + assert "heartbeat_interval=datetime.timedelta(seconds=15)" in printed + ast.parse(printed) + + @pytest.mark.parametrize( + "invalid_address", + ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000", "::1:8092"], + ) + def test_invalid_server_address_is_rejected(self, invalid_address: str): + """Test that a malformed address fails at construction, not at connect.""" + with pytest.raises(ValueError): + WebSocketConfig(server_address=invalid_address) + + def test_negative_heartbeat_interval_is_rejected(self): + """Test that a negative heartbeat interval fails at construction.""" + with pytest.raises(ValueError, match="negative"): + WebSocketConfig(heartbeat_interval=timedelta(seconds=-3)) + + def test_zero_heartbeat_interval_is_rejected(self): + """Test that a zero heartbeat interval fails at construction. + + Nothing downstream reads zero as "disabled"; it heartbeats in a + continuous loop for as long as the client lives. + """ + with pytest.raises(ValueError, match="zero"): + WebSocketConfig(heartbeat_interval=timedelta(0)) + + +@pytest.mark.unit +class TestClientConstruction: + """Test what `IggyClient.websocket(...)` accepts.""" + + def test_accepts_a_config(self): + """Test that a client can be built from a config object.""" + assert ( + IggyClient.websocket(WebSocketConfig(server_address="127.0.0.1:8092")) + is not None + ) + + def test_accepts_nothing(self): + """Test that the default configuration is used when no argument is given.""" + assert IggyClient.websocket() is not None + + +@pytest.mark.integration +class TestAutoLoginAgainstServer: + """Test that configured credentials are actually replayed on connect.""" + + @pytest.mark.asyncio + async def test_auto_login_authenticates_without_login_user(self, unique_name): + """Test that a privileged call succeeds without a manual login_user().""" + host, port = get_websocket_server_config() + wait_for_server(host, port) + + client = IggyClient.websocket( + WebSocketConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "iggy"), + # The default reconnection policy retries forever: a missing + # listener would hang this test until the CI timeout instead + # of failing. + reconnection=WebSocketReconnectionConfig(enabled=False), + ) + ) + await client.connect() + await wait_for_ping(client) + + stream_name = unique_name() + await client.create_stream(stream_name) + assert await client.get_stream(stream_name) is not None + + @pytest.mark.asyncio + async def test_without_auto_login_a_privileged_call_is_unauthenticated( + self, unique_name + ): + """Test that the same call fails when no credentials are configured.""" + host, port = get_websocket_server_config() + wait_for_server(host, port) + + client = IggyClient.websocket( + WebSocketConfig( + server_address=f"{host}:{port}", + # The default reconnection policy retries forever: a missing + # listener would hang this test until the CI timeout instead + # of failing. + reconnection=WebSocketReconnectionConfig(enabled=False), + ) + ) + await client.connect() + await wait_for_ping(client) + + with pytest.raises(RuntimeError): + await client.create_stream(unique_name()) + + @pytest.mark.asyncio + async def test_wrong_auto_login_credentials_fail(self): + """Test that bad configured credentials surface as a connect failure.""" + host, port = get_websocket_server_config() + wait_for_server(host, port) + + client = IggyClient.websocket( + WebSocketConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "invalid-password"), + reconnection=WebSocketReconnectionConfig(enabled=False), + ) + ) + + with pytest.raises(RuntimeError): + await client.connect() diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py index b37a53831e..f821ae7300 100644 --- a/foreign/python/tests/utils.py +++ b/foreign/python/tests/utils.py @@ -33,15 +33,19 @@ MAX_PASSWORD_BYTES = 100 -def get_server_config() -> tuple[str, int]: +def get_transport_config(port_env_var: str, default_port: int) -> tuple[str, int]: """ - Get server configuration from environment variables or defaults. + Get transport-specific server configuration from environment variables or defaults. + + Args: + port_env_var: Name of the environment variable holding the port. + default_port: Port to use if the environment variable is not set. Returns: tuple: (host, port) for the Iggy server """ host = os.environ.get("IGGY_SERVER_HOST", "127.0.0.1") - port = int(os.environ.get("IGGY_SERVER_TCP_PORT", "8090")) + port = int(os.environ.get(port_env_var, str(default_port))) # Convert hostname to IP address for the Rust client if host not in ("127.0.0.1", "localhost"): @@ -58,6 +62,26 @@ def get_server_config() -> tuple[str, int]: return host, port +def get_server_config() -> tuple[str, int]: + """ + Get TCP server configuration from environment variables or defaults. + + Returns: + tuple: (host, port) for the Iggy server + """ + return get_transport_config("IGGY_SERVER_TCP_PORT", 8090) + + +def get_websocket_server_config() -> tuple[str, int]: + """ + Get WebSocket server configuration from environment variables or defaults. + + Returns: + tuple: (host, port) for the Iggy server + """ + return get_transport_config("IGGY_SERVER_WS_PORT", 8092) + + def wait_for_server(host: str, port: int, timeout: int = 60, interval: int = 2) -> None: """ Wait for the server to become available.