hubcio commented on code in PR #3776: URL: https://github.com/apache/iggy/pull/3776#discussion_r3710148854
########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative. + #[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<bool>, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<u32>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option<Py<PyDelta>>, + ) -> PyResult<Self> { + let defaults = RustTcpClientReconnectionConfig::default(); + Ok(Self { + inner: RustTcpClientReconnectionConfig { + enabled: enabled.unwrap_or(defaults.enabled), + max_retries, + interval: interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.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<u32> { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + iggy_duration_to_py_delta(py, self.inner.interval) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + 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!( + "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + if self.inner.enabled { "True" } else { "False" }, + self.inner.interval.as_human_time_string(), + self.inner.reestablish_after.as_human_time_string(), + ) + } +} + +/// Configuration for the TCP transport, accepted by `IggyClient(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct TcpConfig { + auto_login: AutoLogin, + reconnection: TcpReconnectionConfig, + inner: Arc<RustTcpClientConfig>, +} + +impl TcpConfig { + /// The configuration in the shape `TcpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc<RustTcpClientConfig> { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpConfig { + /// Constructs a TCP configuration. + /// + /// Args: + /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`. + /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + /// reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// tls_enabled: Whether to connect over TLS. Defaults to disabled. + /// tls_domain: Domain to validate the certificate against. Empty means it is + /// taken from `server_address`. + /// tls_ca_file: Path to the CA file for TLS. + /// tls_validate_certificate: Whether to validate the server certificate. + /// Defaults to validating. Disabling this accepts any certificate the + /// server presents, including self-signed and mismatched ones; intended + /// for local development only. + /// nodelay: Disable the Nagle algorithm for the TCP socket. Defaults to + /// leaving it on. + /// + /// Raises: + /// ValueError: If `server_address` is not a valid `host:port` pair, or + /// if a duration is negative. + #[new] + #[pyo3(signature = ( + *, + server_address=None, + auto_login=None, + reconnection=None, + heartbeat_interval=None, + tls_enabled=None, + tls_domain=None, + tls_ca_file=None, + tls_validate_certificate=None, + nodelay=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<AutoLogin>, + #[gen_stub(override_type(type_repr = "TcpReconnectionConfig | None"))] reconnection: Option< + TcpReconnectionConfig, + >, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option<bool>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option<String>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option<String>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + tls_validate_certificate: Option<bool>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] nodelay: Option<bool>, + ) -> PyResult<Self> { + let defaults = RustTcpClientConfig::default(); + let auto_login = auto_login.unwrap_or_default(); + let reconnection = reconnection.unwrap_or_default(); + + // The builder is only used to validate and trim the server address; the + // remaining fields are assigned directly so every unset argument falls + // back to the Rust `TcpClientConfig::default()` value instead of a + // literal duplicated here. + let mut inner = TcpClientConfigBuilder::new() + .with_server_address(server_address.unwrap_or(defaults.server_address)) + .build() + .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?; + inner.auto_login = auto_login.inner.clone(); + inner.reconnection = reconnection.inner.clone(); + inner.heartbeat_interval = heartbeat_interval Review Comment: `heartbeat_interval=timedelta(0)` passes validation (only negatives are rejected) and lands in the sdk heartbeat loop at `core/sdk/src/clients/client.rs:269`, which does `sleep(heartbeat_interval.get_duration())` in an unbounded loop - zero means pinging as fast as round trips complete, for as long as the client lives. nothing downstream reads zero as "disabled". worth rejecting zero here with the same `ValueError` shape as negatives. ########## foreign/python/tests/test_client_config.py: ########## @@ -0,0 +1,325 @@ +# 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 TCP client configuration surface. + +`TcpConfig`, `TcpReconnectionConfig` and `AutoLogin` mirror the Rust SDK +types, 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. The last class +proves the point of the configuration: with `auto_login` set, credentials are +replayed on connect and no manual `login_user()` is needed. +""" + +from collections.abc import Callable +from datetime import timedelta + +import pytest + +from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig + +from .utils import get_server_config, wait_for_ping, wait_for_server + + [email protected] +class TestAutoLogin: + """Test the credentials carried into the client.""" + + def test_disabled_has_no_username(self): + """Test that the disabled variant carries no credentials.""" + auto_login = AutoLogin.disabled() + + assert auto_login.enabled is False + assert auto_login.username is None + + def test_username_password_exposes_username_only(self): + """Test that the username is readable back but the password is not.""" + auto_login = AutoLogin.username_password("iggy", "secret") + + assert auto_login.enabled is True + assert auto_login.username == "iggy" + assert "secret" not in repr(auto_login) + + def test_personal_access_token_hides_the_token(self): + """Test that a token login exposes neither a username nor the token.""" + auto_login = AutoLogin.personal_access_token("secret-token") + + assert auto_login.enabled is True + assert auto_login.username is None + assert "secret-token" not in repr(auto_login) + + [email protected] +class TestTcpReconnectionConfig: + """Test the reconnection policy.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured policy reconnects forever, one second apart.""" + reconnection = TcpReconnectionConfig() + + 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 = TcpReconnectionConfig( + 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 + TcpReconnectionConfig(True) + + @pytest.mark.parametrize( + "construct", + [ + lambda duration: TcpReconnectionConfig(interval=duration), + lambda duration: TcpReconnectionConfig(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], TcpReconnectionConfig], + negative: timedelta, + ): + """Test that a negative duration fails at construction, not at connect.""" + with pytest.raises(ValueError, match="negative"): + construct(negative) + + def test_zero_interval_is_allowed(self): Review Comment: this asserts the zero-interval value from the reconnect-spin problem is legal, so the test will fight the fix later. `reestablish_after` is the one duration where zero is genuinely meaningful (skips the cooldown - the guard at `tcp_client.rs:394` is `if elapsed < interval`), so retargeting the test there keeps the coverage. ########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative. + #[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<bool>, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<u32>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option<Py<PyDelta>>, + ) -> PyResult<Self> { + let defaults = RustTcpClientReconnectionConfig::default(); + Ok(Self { + inner: RustTcpClientReconnectionConfig { + enabled: enabled.unwrap_or(defaults.enabled), + max_retries, + interval: interval Review Comment: `interval=timedelta(0)` combined with the default `max_retries=None` (unlimited) turns the reconnect loop at `core/sdk/src/tcp/tcp_client.rs:434-441` into a tight `TcpStream::connect` spin with one `info!` line per attempt while the server is down. zero with bounded retries is a legitimate fast-retry policy, so rejecting just the zero + unlimited combination is enough. ########## foreign/python/src/client.rs: ########## @@ -438,7 +462,7 @@ impl IggyClient { }; let expiry = match message_expiry { - Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)), + Some(delta) => IggyExpiry::ExpireDuration(py_delta_to_iggy_duration(&delta)?), Review Comment: behavior change worth documenting: a negative `message_expiry` used to silently become a near-`u64::MAX` expiry (the old converter did i32 math and cast to u64), now it raises `ValueError` at call time, before the awaitable is created. the docstrings of `create_topic`, `update_topic` and `consumer_group` still list `RuntimeError` only - the new `TcpConfig`/`TcpReconnectionConfig` docstrings document the `ValueError`, these three should too. for `consumer_group` that covers four timedelta params plus the `AutoCommit` interval variants. ########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. Review Comment: the docstring doesn't say that `None` is the default, and that with the server down an awaited call then never returns - the retry loop sits inside `connect()` (`tcp_client.rs:364`), which the send path re-enters, so `connect()`/`send_messages`/`poll_messages` all block indefinitely. both getting-started examples ship exactly this policy (reconnection configured, `max_retries` unset). worth one sentence here, plus suggesting a finite `max_retries` for request/reply style usage. ########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative. + #[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<bool>, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<u32>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option<Py<PyDelta>>, + ) -> PyResult<Self> { + let defaults = RustTcpClientReconnectionConfig::default(); + Ok(Self { + inner: RustTcpClientReconnectionConfig { + enabled: enabled.unwrap_or(defaults.enabled), + max_retries, + interval: interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.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<u32> { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + iggy_duration_to_py_delta(py, self.inner.interval) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + 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!( + "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + if self.inner.enabled { "True" } else { "False" }, + self.inner.interval.as_human_time_string(), + self.inner.reestablish_after.as_human_time_string(), + ) + } +} + +/// Configuration for the TCP transport, accepted by `IggyClient(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct TcpConfig { + auto_login: AutoLogin, + reconnection: TcpReconnectionConfig, + inner: Arc<RustTcpClientConfig>, +} + +impl TcpConfig { + /// The configuration in the shape `TcpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc<RustTcpClientConfig> { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpConfig { + /// Constructs a TCP configuration. + /// + /// Args: + /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`. + /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + /// reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// tls_enabled: Whether to connect over TLS. Defaults to disabled. + /// tls_domain: Domain to validate the certificate against. Empty means it is + /// taken from `server_address`. + /// tls_ca_file: Path to the CA file for TLS. + /// tls_validate_certificate: Whether to validate the server certificate. + /// Defaults to validating. Disabling this accepts any certificate the + /// server presents, including self-signed and mismatched ones; intended + /// for local development only. + /// nodelay: Disable the Nagle algorithm for the TCP socket. Defaults to + /// leaving it on. + /// + /// Raises: + /// ValueError: If `server_address` is not a valid `host:port` pair, or + /// if a duration is negative. + #[new] + #[pyo3(signature = ( + *, + server_address=None, + auto_login=None, + reconnection=None, + heartbeat_interval=None, + tls_enabled=None, + tls_domain=None, + tls_ca_file=None, + tls_validate_certificate=None, + nodelay=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<AutoLogin>, + #[gen_stub(override_type(type_repr = "TcpReconnectionConfig | None"))] reconnection: Option< + TcpReconnectionConfig, + >, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option<bool>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option<String>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option<String>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + tls_validate_certificate: Option<bool>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] nodelay: Option<bool>, + ) -> PyResult<Self> { + let defaults = RustTcpClientConfig::default(); + let auto_login = auto_login.unwrap_or_default(); + let reconnection = reconnection.unwrap_or_default(); + + // The builder is only used to validate and trim the server address; the + // remaining fields are assigned directly so every unset argument falls + // back to the Rust `TcpClientConfig::default()` value instead of a + // literal duplicated here. + let mut inner = TcpClientConfigBuilder::new() + .with_server_address(server_address.unwrap_or(defaults.server_address)) + .build() + .map_err(|e| PyErr::new::<pyo3::exceptions::PyValueError, _>(e.to_string()))?; + inner.auto_login = auto_login.inner.clone(); + inner.reconnection = reconnection.inner.clone(); + inner.heartbeat_interval = heartbeat_interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.heartbeat_interval); + inner.tls_enabled = tls_enabled.unwrap_or(defaults.tls_enabled); + inner.tls_domain = tls_domain.unwrap_or(defaults.tls_domain); + inner.tls_ca_file = tls_ca_file.or(defaults.tls_ca_file); + inner.tls_validate_certificate = + tls_validate_certificate.unwrap_or(defaults.tls_validate_certificate); + inner.nodelay = nodelay.unwrap_or(defaults.nodelay); + + Ok(Self { + auto_login, + reconnection, + inner: Arc::new(inner), + }) + } + + #[getter] + fn server_address(&self) -> String { + self.inner.server_address.clone() + } + + #[getter] + fn auto_login(&self) -> AutoLogin { + self.auto_login.clone() + } + + #[getter] + fn reconnection(&self) -> TcpReconnectionConfig { + self.reconnection.clone() + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn heartbeat_interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + iggy_duration_to_py_delta(py, self.inner.heartbeat_interval) + } + + #[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<String> { + self.inner.tls_ca_file.clone() + } + + #[getter] + fn tls_validate_certificate(&self) -> bool { + self.inner.tls_validate_certificate + } + + #[getter] + fn nodelay(&self) -> bool { + self.inner.nodelay + } + + fn __repr__(&self) -> String { Review Comment: repr prints 5 of the 9 fields and drops exactly the ones you'd debug a tls handshake with: `tls_domain`, `tls_ca_file`, `tls_validate_certificate`, `nodelay`. a config with certificate validation off prints identically to a validating one, and that's the field whose own docstring says "local development only". also `heartbeat_interval=5s` isn't valid python, so the constructor-shaped output can't be pasted back (`TcpReconnectionConfig.__repr__` has the same `interval=1s` issue). no test pins the repr, so this is free to fix. ########## foreign/python/src/client.rs: ########## @@ -56,17 +57,41 @@ pub struct IggyClient { #[gen_stub_pymethods] #[pymethods] impl IggyClient { - /// Constructs a new IggyClient from a TCP server address. + /// Constructs a new IggyClient from a TCP server address or a `TcpConfig`. /// This initializes a new runtime for asynchronous operations. /// Future versions might utilize asyncio for more Pythonic async. + /// + /// Args: + /// conn: Either a `host:port` address, or a `TcpConfig` carrying the full + /// transport configuration. Defaults to `127.0.0.1:8090` with auto-login + /// disabled. + /// + /// Raises: + /// RuntimeError: If the address is not a valid `host:port` pair, or if the + /// client cannot be built. #[new] #[pyo3(signature = (conn=None))] fn new( - #[gen_stub(override_type(type_repr = "builtins.str | None"))] conn: Option<String>, + #[gen_stub(override_type(type_repr = "TcpConfig | builtins.str | None"))] conn: Option< + PyClientConfig, + >, ) -> PyResult<Self> { + let config = match conn { + Some(PyClientConfig::Config(config)) => config.client_config(), + Some(PyClientConfig::ServerAddress(server_address)) => Arc::new( + TcpClientConfigBuilder::new() + .with_server_address(server_address) + .build() + .map_err(|e| { + PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()) Review Comment: same invalid address now raises two different exception types: `IggyClient("bad")` keeps the historical `RuntimeError`, `TcpConfig(server_address="bad")` raises `ValueError` - and neither is a subclass of the other, so one `except` can't cover "bad address". keeping the str path as-is is right (changing it would break existing `except RuntimeError` handlers), but the `conn` docstring should name both, since which one you get depends on which form you passed. ########## foreign/python/README.md: ########## @@ -58,6 +58,42 @@ maturin develop pytest tests/ -v # Run tests (requires iggy-server running) ``` +## Client Configuration + +`IggyClient` takes either a server address or a `TcpConfig`: + +```python +import asyncio +from datetime import timedelta + +from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig + + +async def main(): + client = IggyClient( + TcpConfig( + server_address="127.0.0.1:8090", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=TcpReconnectionConfig( + enabled=True, + max_retries=10, + interval=timedelta(seconds=2), + reestablish_after=timedelta(seconds=30), + ), + heartbeat_interval=timedelta(seconds=5), + # tls_enabled=True, + # tls_domain="localhost", + # tls_ca_file="core/certs/iggy_ca_cert.pem", Review Comment: this path is relative to the repo root, but a reader of this readme runs from `foreign/python` - the same file is `../../core/certs/iggy_ca_cert.pem` from a sibling dir (that's what the examples readme uses). ########## foreign/python/tests/test_client_config.py: ########## @@ -0,0 +1,325 @@ +# 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 TCP client configuration surface. + +`TcpConfig`, `TcpReconnectionConfig` and `AutoLogin` mirror the Rust SDK +types, 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. The last class +proves the point of the configuration: with `auto_login` set, credentials are +replayed on connect and no manual `login_user()` is needed. +""" + +from collections.abc import Callable +from datetime import timedelta + +import pytest + +from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig + +from .utils import get_server_config, wait_for_ping, wait_for_server + + [email protected] +class TestAutoLogin: + """Test the credentials carried into the client.""" + + def test_disabled_has_no_username(self): + """Test that the disabled variant carries no credentials.""" + auto_login = AutoLogin.disabled() + + assert auto_login.enabled is False + assert auto_login.username is None + + def test_username_password_exposes_username_only(self): + """Test that the username is readable back but the password is not.""" + auto_login = AutoLogin.username_password("iggy", "secret") + + assert auto_login.enabled is True + assert auto_login.username == "iggy" + assert "secret" not in repr(auto_login) + + def test_personal_access_token_hides_the_token(self): + """Test that a token login exposes neither a username nor the token.""" + auto_login = AutoLogin.personal_access_token("secret-token") + + assert auto_login.enabled is True + assert auto_login.username is None + assert "secret-token" not in repr(auto_login) + + [email protected] +class TestTcpReconnectionConfig: + """Test the reconnection policy.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured policy reconnects forever, one second apart.""" + reconnection = TcpReconnectionConfig() + + 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 = TcpReconnectionConfig( + 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 + TcpReconnectionConfig(True) + + @pytest.mark.parametrize( + "construct", + [ + lambda duration: TcpReconnectionConfig(interval=duration), + lambda duration: TcpReconnectionConfig(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], TcpReconnectionConfig], + negative: timedelta, + ): + """Test that a negative duration fails at construction, not at connect.""" + with pytest.raises(ValueError, match="negative"): + construct(negative) + + def test_zero_interval_is_allowed(self): + """Test that a zero interval is legal and readable back.""" + reconnection = TcpReconnectionConfig(interval=timedelta(0)) + + assert reconnection.interval == timedelta(0) + + def test_very_long_interval_round_trips(self): + """Test that an interval beyond 68 years survives the i32 boundary.""" + reconnection = TcpReconnectionConfig(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 u64-microsecond boundary.""" + reconnection = TcpReconnectionConfig(interval=timedelta(days=999_999_999)) + + assert reconnection.interval == timedelta(days=999_999_999) + + [email protected] +class TestTcpConfig: + """Test the transport configuration.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured transport matches the Rust SDK defaults.""" + config = TcpConfig() + + assert config.server_address == "127.0.0.1:8090" + 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 == "" + assert config.tls_ca_file is None + assert config.tls_validate_certificate is True + assert config.nodelay is False + + def test_every_field_round_trips(self): + """Test that each configured field is readable back unchanged.""" + config = TcpConfig( + server_address="localhost:8090", + auto_login=AutoLogin.username_password("iggy", "iggy"), + reconnection=TcpReconnectionConfig(max_retries=3), + heartbeat_interval=timedelta(seconds=15), + tls_enabled=True, + tls_domain="localhost", + tls_ca_file="ca.pem", + tls_validate_certificate=False, + nodelay=True, + ) + + assert config.server_address == "localhost:8090" + assert config.auto_login.username == "iggy" + assert config.reconnection.max_retries == 3 + assert config.heartbeat_interval == timedelta(seconds=15) + assert config.tls_enabled is True + assert config.tls_domain == "localhost" + assert config.tls_ca_file == "ca.pem" + assert config.tls_validate_certificate is False + assert config.nodelay 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 + TcpConfig("127.0.0.1:8090") + + def test_repr_hides_the_password(self): + """Test that the password does not leak through repr.""" + config = TcpConfig(auto_login=AutoLogin.username_password("iggy", "secret")) + + assert "secret" not in repr(config) + + @pytest.mark.parametrize( + "invalid_address", + ["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000", "::1:8090"], + ) + 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): + TcpConfig(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"): + TcpConfig(heartbeat_interval=timedelta(seconds=-3)) + + [email protected] +class TestClientConstruction: + """Test what the client constructor accepts.""" + + def test_accepts_a_config(self): + """Test that a client can be built from a config object.""" + assert IggyClient(TcpConfig(server_address="127.0.0.1:8090")) is not None + + def test_accepts_an_address(self): + """Test that the address form still works.""" + assert IggyClient("127.0.0.1:8090") is not None + + def test_accepts_nothing(self): + """Test that the default address is used when no argument is given.""" + assert IggyClient() is not None + + def test_rejects_an_invalid_address(self): + """Test that a malformed address is rejected.""" + with pytest.raises(RuntimeError): + IggyClient("nonsense") + + def test_negative_message_expiry_is_rejected(self): + """Test that the negative-duration rule reaches the pre-existing surface. + + create_topic accepted a negative message_expiry before durations were + validated; it now fails at the call, before any I/O. + """ + client = IggyClient() + + with pytest.raises(ValueError, match="negative"): + client.create_topic( + stream="stream", + name="topic", + partitions_count=1, + message_expiry=timedelta(seconds=-1), + ) + + [email protected] +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_server_config() + wait_for_server(host, port) + + client = IggyClient( + TcpConfig( + server_address=f"{host}:{port}", + auto_login=AutoLogin.username_password("iggy", "iggy"), + ) + ) + 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_server_config() + wait_for_server(host, port) + + client = IggyClient(TcpConfig(server_address=f"{host}:{port}")) + await client.connect() + await wait_for_ping(client) + + with pytest.raises(RuntimeError): + await client.create_stream(unique_name()) + + @pytest.mark.asyncio + async def test_config_and_connection_string_are_equivalent(self, unique_name): Review Comment: name promises equivalence, body proves both clients authenticate - the reconnection params configured on either side are never observed (`IggyClient` exposes no config getter, and `get_stream(...) is None` is the only assertion). rename to something like `test_config_and_connection_string_both_authenticate`. ########## foreign/python/src/client.rs: ########## @@ -56,17 +57,41 @@ pub struct IggyClient { #[gen_stub_pymethods] #[pymethods] impl IggyClient { - /// Constructs a new IggyClient from a TCP server address. + /// Constructs a new IggyClient from a TCP server address or a `TcpConfig`. /// This initializes a new runtime for asynchronous operations. /// Future versions might utilize asyncio for more Pythonic async. + /// + /// Args: + /// conn: Either a `host:port` address, or a `TcpConfig` carrying the full + /// transport configuration. Defaults to `127.0.0.1:8090` with auto-login + /// disabled. + /// + /// Raises: + /// RuntimeError: If the address is not a valid `host:port` pair, or if the + /// client cannot be built. #[new] #[pyo3(signature = (conn=None))] fn new( - #[gen_stub(override_type(type_repr = "builtins.str | None"))] conn: Option<String>, + #[gen_stub(override_type(type_repr = "TcpConfig | builtins.str | None"))] conn: Option< + PyClientConfig, + >, ) -> PyResult<Self> { + let config = match conn { + Some(PyClientConfig::Config(config)) => config.client_config(), + Some(PyClientConfig::ServerAddress(server_address)) => Arc::new( + TcpClientConfigBuilder::new() + .with_server_address(server_address) + .build() + .map_err(|e| { + PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()) + })?, + ), + None => Arc::new(TcpClientConfig::default()), + }; + let tcp_client = TcpClient::create(config) + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; let client = IggyClientBuilder::new() Review Comment: `IggyClientBuilder::new().with_client(..).build()` plus the `map_err` guards an error that can't happen - `build()` only fails when no client was set. `RustIggyClient::new(ClientWrapper::Tcp(tcp_client))` is public, infallible and identical (partitioner/encryptor default to `None`), so this collapses to one line without the dead error path. ########## foreign/python/src/client.rs: ########## @@ -1045,7 +1069,7 @@ impl IggyClient { { builder = builder.init_retries( init_retries, - py_delta_to_iggy_duration(&init_retry_interval), + py_delta_to_iggy_duration(&init_retry_interval)?, Review Comment: `init_retry_interval=timedelta(0)` reaches `time::interval(interval.get_duration())` at `core/sdk/src/clients/consumer.rs:323`, and tokio asserts `period must be non-zero` - a rust panic that surfaces in python as `pyo3_async_runtimes.RustPanic`, naming neither the argument nor the class. the interval is constructed unconditionally, so it fires every time, even when the stream and topic already exist. since this line now validates the sign, rejecting zero here too is one line. ########## foreign/python/src/client.rs: ########## @@ -1020,16 +1044,16 @@ impl IggyClient { builder = builder.batch_length(batch_length) }; if let Some(auto_commit) = auto_commit { - builder = builder.auto_commit(auto_commit.into()) + builder = builder.auto_commit(RustAutoCommit::try_from(auto_commit)?) }; if let Some(poll_interval) = poll_interval { - builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval)) + builder = builder.poll_interval(py_delta_to_iggy_duration(&poll_interval)?) } else { builder = builder.without_poll_interval() }; if let Some(polling_retry_interval) = polling_retry_interval { builder = - builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval)) + builder.polling_retry_interval(py_delta_to_iggy_duration(&polling_retry_interval)?) Review Comment: same zero hole: `polling_retry_interval=timedelta(0)` becomes the retry sleep in `core/sdk/src/clients/consumer.rs:690-699` (`while !can_poll || ... { trace!(); sleep(...) }`) - no syscall in the loop body, so it burns a core, and with `auto_join_consumer_group=False` the join flag never flips and the spin is permanent. the field is renamed mid-chain to `reconnection_retry_interval` (`consumer_builder.rs:247`), in case you grep for it. ########## foreign/python/Cargo.toml: ########## @@ -45,4 +45,5 @@ pyo3-async-runtimes = { version = "0.29.0", features = [ "tokio-runtime", ] } pyo3-stub-gen = "0.23.0" +secrecy = "0.10" Review Comment: the committed `Cargo.lock` wasn't regenerated - the `apache-iggy` dependency list in it has no `secrecy` edge, so `cargo metadata --locked` fails in this directory. to be fair it was already stale before this PR (`paste` from #3613 is missing too) and nothing in CI passes `--locked` for this crate, so nothing breaks visibly - but a `cargo check` here re-syncs it and picks up both. ########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative. + #[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<bool>, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<u32>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option<Py<PyDelta>>, + ) -> PyResult<Self> { + let defaults = RustTcpClientReconnectionConfig::default(); + Ok(Self { + inner: RustTcpClientReconnectionConfig { + enabled: enabled.unwrap_or(defaults.enabled), + max_retries, + interval: interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.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<u32> { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + iggy_duration_to_py_delta(py, self.inner.interval) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + 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!( + "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + if self.inner.enabled { "True" } else { "False" }, + self.inner.interval.as_human_time_string(), + self.inner.reestablish_after.as_human_time_string(), + ) + } +} + +/// Configuration for the TCP transport, accepted by `IggyClient(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct TcpConfig { + auto_login: AutoLogin, + reconnection: TcpReconnectionConfig, + inner: Arc<RustTcpClientConfig>, +} + +impl TcpConfig { + /// The configuration in the shape `TcpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc<RustTcpClientConfig> { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpConfig { + /// Constructs a TCP configuration. + /// + /// Args: + /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`. + /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + /// reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// tls_enabled: Whether to connect over TLS. Defaults to disabled. + /// tls_domain: Domain to validate the certificate against. Empty means it is + /// taken from `server_address`. + /// tls_ca_file: Path to the CA file for TLS. Review Comment: `tls_ca_file` is silently ignored when `tls_validate_certificate=False` - `tcp_client.rs:475` only reads the ca inside the validating branch, the other branch installs a blanket no-verification resolver. this config is the first python surface that can express that combination (connection strings hardcode validation on), and `TcpConfig(tls_enabled=True, tls_ca_file="ca.pem", tls_validate_certificate=False)` reads as "pin to my ca" while accepting any certificate. same silent ignore when `tls_enabled=False`, which both examples hit (they pass `tls_ca_file` regardless of `--tls`). document the precedence here, or reject the contradictory combo. ########## foreign/python/src/duration.rs: ########## @@ -0,0 +1,55 @@ +// 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. + +use iggy::prelude::IggyDuration; +use pyo3::prelude::*; +use pyo3::types::{PyDelta, PyDeltaAccess}; +use std::time::Duration; + +pub fn py_delta_to_iggy_duration(delta: &Py<PyDelta>) -> PyResult<IggyDuration> { Review Comment: pyo3 0.29 already ships both directions of this conversion: `FromPyObject`/`IntoPyObject` for `std::time::Duration` do the same days/seconds/micros arithmetic, reject negative timedeltas with a `ValueError` whose message contains "negative" (so the `match="negative"` tests keep passing), and pyo3-stub-gen maps `Duration` to `datetime.timedelta` natively. taking `Option<Duration>` in the constructors and converting with `IggyDuration::from` deletes this whole file. the `Option<...>` parameter stub overrides have to stay either way (stub-gen renders bare `Option<T>` as `typing.Optional[...]`). side note: the overflow branch below is unreachable - python caps `timedelta` at 999,999,999 days, which fits i32, and the rust-side defaults are a few seconds. ########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative. + #[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<bool>, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<u32>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option<Py<PyDelta>>, + ) -> PyResult<Self> { + let defaults = RustTcpClientReconnectionConfig::default(); + Ok(Self { + inner: RustTcpClientReconnectionConfig { + enabled: enabled.unwrap_or(defaults.enabled), + max_retries, + interval: interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.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<u32> { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + iggy_duration_to_py_delta(py, self.inner.interval) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + 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!( + "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + if self.inner.enabled { "True" } else { "False" }, + self.inner.interval.as_human_time_string(), + self.inner.reestablish_after.as_human_time_string(), + ) + } +} + +/// Configuration for the TCP transport, accepted by `IggyClient(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct TcpConfig { + auto_login: AutoLogin, Review Comment: `auto_login` and `reconnection` are stored twice - as wrapper fields here and inside `inner` (both assigned in `__new__`). nothing can make them diverge (no setters), but it's two copies of one truth a reader has to prove safe, and it keeps an extra live copy of the password (`SecretString`) per config object. both `inner` fields are `pub`, so the getters can rebuild the wrappers on demand (`AutoLogin { inner: self.inner.auto_login.clone() }`), which also makes `impl Default for AutoLogin` and the `Default` derive on `TcpReconnectionConfig` dead. ########## foreign/python/tests/test_client_config.py: ########## @@ -0,0 +1,325 @@ +# 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 TCP client configuration surface. + +`TcpConfig`, `TcpReconnectionConfig` and `AutoLogin` mirror the Rust SDK +types, 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. The last class +proves the point of the configuration: with `auto_login` set, credentials are +replayed on connect and no manual `login_user()` is needed. +""" + +from collections.abc import Callable +from datetime import timedelta + +import pytest + +from apache_iggy import AutoLogin, IggyClient, TcpConfig, TcpReconnectionConfig + +from .utils import get_server_config, wait_for_ping, wait_for_server + + [email protected] +class TestAutoLogin: + """Test the credentials carried into the client.""" + + def test_disabled_has_no_username(self): + """Test that the disabled variant carries no credentials.""" + auto_login = AutoLogin.disabled() + + assert auto_login.enabled is False + assert auto_login.username is None + + def test_username_password_exposes_username_only(self): + """Test that the username is readable back but the password is not.""" + auto_login = AutoLogin.username_password("iggy", "secret") + + assert auto_login.enabled is True + assert auto_login.username == "iggy" + assert "secret" not in repr(auto_login) + + def test_personal_access_token_hides_the_token(self): + """Test that a token login exposes neither a username nor the token.""" + auto_login = AutoLogin.personal_access_token("secret-token") + + assert auto_login.enabled is True + assert auto_login.username is None + assert "secret-token" not in repr(auto_login) + + [email protected] +class TestTcpReconnectionConfig: + """Test the reconnection policy.""" + + def test_defaults_match_the_rust_sdk(self): + """Test that an unconfigured policy reconnects forever, one second apart.""" + reconnection = TcpReconnectionConfig() + + 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 = TcpReconnectionConfig( + 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 + TcpReconnectionConfig(True) + + @pytest.mark.parametrize( + "construct", + [ + lambda duration: TcpReconnectionConfig(interval=duration), + lambda duration: TcpReconnectionConfig(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], TcpReconnectionConfig], + negative: timedelta, + ): + """Test that a negative duration fails at construction, not at connect.""" + with pytest.raises(ValueError, match="negative"): + construct(negative) + + def test_zero_interval_is_allowed(self): + """Test that a zero interval is legal and readable back.""" + reconnection = TcpReconnectionConfig(interval=timedelta(0)) + + assert reconnection.interval == timedelta(0) + + def test_very_long_interval_round_trips(self): + """Test that an interval beyond 68 years survives the i32 boundary.""" + reconnection = TcpReconnectionConfig(interval=timedelta(days=30_000)) + + assert reconnection.interval == timedelta(days=30_000) + + def test_maximum_interval_round_trips(self): Review Comment: the docstring claims the value "survives the u64-microsecond boundary", but the test only proves the getter round-trips - the getter reads `get_duration().as_micros()` as u128. nothing on the `interval` path ever narrows to u64 (`tcp_client.rs:440` sleeps on the full `Duration`), so the boundary named here isn't exercised and doesn't exist for this field. the sibling `reestablish_after` is the one consumed via the truncating `as_micros()` cast (`tcp_client.rs:389`), where `timedelta(days=999_999_999)` actually wraps. suggest rewording to the i32-days boundary in the getter conversion, which is what this covers (the test above at line 127 already words it right). ########## foreign/python/src/config.rs: ########## @@ -0,0 +1,378 @@ +// 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. + +use iggy::prelude::{ + AutoLogin as RustAutoLogin, Credentials as RustCredentials, + TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder, + TcpClientReconnectionConfig as RustTcpClientReconnectionConfig, +}; +use pyo3::prelude::*; +use pyo3::types::PyDelta; +use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; +use pyo3_stub_gen::impl_stub_type; +use secrecy::SecretString; +use std::sync::Arc; + +use crate::duration::{iggy_duration_to_py_delta, py_delta_to_iggy_duration}; + +/// The credentials replayed by the client every time it (re)connects. +/// +/// `IggyClient` only recovers a lost session when it has credentials to replay, +/// so a long-running consumer should pass one of the enabled variants. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct AutoLogin { + pub(crate) inner: RustAutoLogin, +} + +#[gen_stub_pymethods] +#[pymethods] +impl AutoLogin { + /// No automatic login. `login_user()` must be called by hand after every connect. + #[staticmethod] + fn disabled() -> Self { + Self { + inner: RustAutoLogin::Disabled, + } + } + + /// Log in with the given username and password on every connect. + #[staticmethod] + fn username_password(username: String, password: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::UsernamePassword( + username, + SecretString::from(password), + )), + } + } + + /// Log in with the given personal access token on every connect. + #[staticmethod] + fn personal_access_token(token: String) -> Self { + Self { + inner: RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken( + SecretString::from(token), + )), + } + } + + /// Whether automatic login is enabled. + #[getter] + fn enabled(&self) -> bool { + matches!(self.inner, RustAutoLogin::Enabled(_)) + } + + /// The username to log in with, or `None` for the disabled and token variants. + #[gen_stub(override_return_type(type_repr = "builtins.str | None"))] + #[getter] + fn username(&self) -> Option<String> { + match &self.inner { + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + Some(username.clone()) + } + _ => None, + } + } + + fn __repr__(&self) -> String { + match &self.inner { + RustAutoLogin::Disabled => "AutoLogin.disabled()".to_owned(), + RustAutoLogin::Enabled(RustCredentials::UsernamePassword(username, _)) => { + format!("AutoLogin.username_password({username:?}, ...)") + } + RustAutoLogin::Enabled(RustCredentials::PersonalAccessToken(_)) => { + "AutoLogin.personal_access_token(...)".to_owned() + } + } + } +} + +impl Default for AutoLogin { + fn default() -> Self { + Self::disabled() + } +} + +/// How the TCP client reconnects after the connection to the server is lost. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone, Default)] +pub struct TcpReconnectionConfig { + pub(crate) inner: RustTcpClientReconnectionConfig, +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpReconnectionConfig { + /// Constructs a reconnection policy. + /// + /// Args: + /// enabled: Whether to reconnect at all. Defaults to enabled. + /// max_retries: Attempts before giving up, or `None` for unlimited. + /// interval: Delay between attempts. Defaults to 1 second. + /// reestablish_after: Cooldown before reconnecting after a previously + /// successful connection. Defaults to 5 seconds. + /// + /// Raises: + /// ValueError: If a duration is negative. + #[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<bool>, + #[gen_stub(override_type(type_repr = "builtins.int | None"))] max_retries: Option<u32>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + reestablish_after: Option<Py<PyDelta>>, + ) -> PyResult<Self> { + let defaults = RustTcpClientReconnectionConfig::default(); + Ok(Self { + inner: RustTcpClientReconnectionConfig { + enabled: enabled.unwrap_or(defaults.enabled), + max_retries, + interval: interval + .as_ref() + .map(py_delta_to_iggy_duration) + .transpose()? + .unwrap_or(defaults.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<u32> { + self.inner.max_retries + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn interval<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + iggy_duration_to_py_delta(py, self.inner.interval) + } + + #[gen_stub(override_return_type(type_repr = "datetime.timedelta", imports=("datetime")))] + #[getter] + fn reestablish_after<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyDelta>> { + 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!( + "TcpReconnectionConfig(enabled={}, max_retries={max_retries}, interval={}, reestablish_after={})", + if self.inner.enabled { "True" } else { "False" }, + self.inner.interval.as_human_time_string(), + self.inner.reestablish_after.as_human_time_string(), + ) + } +} + +/// Configuration for the TCP transport, accepted by `IggyClient(...)`. +/// +/// Every field is keyword-only and optional. +#[gen_stub_pyclass] +#[pyclass(from_py_object)] +#[derive(Clone)] +pub struct TcpConfig { + auto_login: AutoLogin, + reconnection: TcpReconnectionConfig, + inner: Arc<RustTcpClientConfig>, +} + +impl TcpConfig { + /// The configuration in the shape `TcpClient::create` expects. + pub(crate) fn client_config(&self) -> Arc<RustTcpClientConfig> { + self.inner.clone() + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl TcpConfig { + /// Constructs a TCP configuration. + /// + /// Args: + /// server_address: `host:port` of the Iggy server. Defaults to `127.0.0.1:8090`. + /// auto_login: Credentials replayed on every connect. Defaults to `AutoLogin.disabled()`. + /// reconnection: Reconnection policy. Defaults to `TcpReconnectionConfig()`. + /// heartbeat_interval: Interval of heartbeats sent by the client. Defaults to 5 seconds. + /// tls_enabled: Whether to connect over TLS. Defaults to disabled. + /// tls_domain: Domain to validate the certificate against. Empty means it is + /// taken from `server_address`. + /// tls_ca_file: Path to the CA file for TLS. + /// tls_validate_certificate: Whether to validate the server certificate. + /// Defaults to validating. Disabling this accepts any certificate the + /// server presents, including self-signed and mismatched ones; intended + /// for local development only. + /// nodelay: Disable the Nagle algorithm for the TCP socket. Defaults to + /// leaving it on. + /// + /// Raises: + /// ValueError: If `server_address` is not a valid `host:port` pair, or + /// if a duration is negative. + #[new] + #[pyo3(signature = ( + *, + server_address=None, + auto_login=None, + reconnection=None, + heartbeat_interval=None, + tls_enabled=None, + tls_domain=None, + tls_ca_file=None, + tls_validate_certificate=None, + nodelay=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<AutoLogin>, + #[gen_stub(override_type(type_repr = "TcpReconnectionConfig | None"))] reconnection: Option< + TcpReconnectionConfig, + >, + #[gen_stub(override_type(type_repr = "datetime.timedelta | None", imports=("datetime")))] + heartbeat_interval: Option<Py<PyDelta>>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] tls_enabled: Option<bool>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_domain: Option<String>, + #[gen_stub(override_type(type_repr = "builtins.str | None"))] tls_ca_file: Option<String>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] + tls_validate_certificate: Option<bool>, + #[gen_stub(override_type(type_repr = "builtins.bool | None"))] nodelay: Option<bool>, + ) -> PyResult<Self> { + let defaults = RustTcpClientConfig::default(); Review Comment: `TcpClientConfigBuilder` is `#[derive(Default)]` over `TcpClientConfig`, so the config coming out of `build()` already carries every default - this `defaults` is a second identical construction and the `unwrap_or(defaults.x)` tails re-apply values that are already there. assigning only on `Some` (or building one exhaustive struct literal sourcing untouched fields from the builder output) drops ~10 lines. two constraints if you do: keep the trimmed `server_address` from the builder output (`build()` trims it), and leave `max_retries` as the raw pass-through it is - `None` there is a real user value meaning unlimited, not "unset". ########## examples/python/getting-started/consumer.py: ########## @@ -91,34 +99,30 @@ def parse_args() -> ArgNamespace: return ArgNamespace(**vars(args)) -def build_connection_string(args) -> str: - """Build a connection string with TLS support.""" - - conn_str = f"iggy://{args.username}:{args.password}@{args.tcp_server_address}" - - if args.tls: - # Extract domain from server address (host:port -> host) - host = args.tcp_server_address.split(":")[0] - query_params = ["tls=true", f"tls_domain={host}"] +def build_config(args: ArgNamespace) -> TcpConfig: + """Build a TCP client configuration with auto-login and reconnection.""" - # Add CA file if provided - if args.tls_ca_file: - query_params.append(f"tls_ca_file={args.tls_ca_file}") - conn_str += "?" + "&".join(query_params) - - return conn_str + return TcpConfig( + server_address=args.tcp_server_address, + auto_login=AutoLogin.username_password(args.username, args.password), + reconnection=TcpReconnectionConfig( + enabled=True, + interval=timedelta(seconds=1), + ), + tls_enabled=args.tls, + tls_ca_file=args.tls_ca_file or None, + ) async def main(): args: ArgNamespace = parse_args() + config = build_config(args) Review Comment: `build_config()` can now raise `ValueError` (address validation moved to `TcpConfig`) and it sits outside the `try` below - `--tcp-server-address 127.0.0.1` (no port) passes the argparse url check and produces a raw traceback. same in producer.py, which has no try at all. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
