This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new bd1a873fe feat(python): add HttpConfig transport configuration (#3992)
bd1a873fe is described below
commit bd1a873fe12887798aa045aa9fa8e7e2bafd2ea3
Author: saie-ch <[email protected]>
AuthorDate: Mon Sep 7 23:01:27 2026 +0530
feat(python): add HttpConfig transport configuration (#3992)
Relates to #2835
---------
Co-authored-by: Hubert Gruszecki <[email protected]>
---
.../actions/python-maturin/pre-merge/action.yml | 14 +-
.github/workflows/coverage-baseline.yml | 13 +-
core/sdk/src/prelude.rs | 1 +
examples/python/getting-started/consumer.py | 20 +-
examples/python/getting-started/producer.py | 20 +-
foreign/python/README.md | 38 ++-
foreign/python/apache_iggy.pyi | 114 ++++++-
foreign/python/src/client.rs | 40 ++-
foreign/python/src/config.rs | 212 ++++++++++--
foreign/python/src/lib.rs | 5 +-
foreign/python/tests/test_client_config.py | 4 +-
foreign/python/tests/test_http_config.py | 363 +++++++++++++++++++++
foreign/python/tests/test_quic_config.py | 76 ++++-
foreign/python/tests/utils.py | 18 +-
14 files changed, 855 insertions(+), 83 deletions(-)
diff --git a/.github/actions/python-maturin/pre-merge/action.yml
b/.github/actions/python-maturin/pre-merge/action.yml
index 3cca87fce..53bf430dc 100644
--- a/.github/actions/python-maturin/pre-merge/action.yml
+++ b/.github/actions/python-maturin/pre-merge/action.yml
@@ -173,14 +173,24 @@ runs:
run: |
cd foreign/python
- echo "Running integration tests with Iggy server at ${{
steps.iggy.outputs.address }}..."
+ # TCP and HTTP ports come from server-start's own outputs rather than
+ # duplicated literals, so a future default change here can't silently
+ # desync from what the server actually started with. QUIC stays a
+ # literal below because server-start exposes no output for it.
+ tcp_port="${{ steps.iggy.outputs.tcp_address }}"
+ http_port="${{ steps.iggy.outputs.http_address }}"
+ tcp_port="${tcp_port##*:}"
+ http_port="${http_port##*:}"
+
+ echo "Running integration tests with Iggy server at ${{
steps.iggy.outputs.tcp_address }}..."
# Run all tests with server connection
# --no-sync prevents uv from re-syncing the venv which would
# overwrite the coverage-instrumented .so with a non-instrumented one
IGGY_SERVER_HOST=127.0.0.1 \
- IGGY_SERVER_TCP_PORT=8090 \
+ IGGY_SERVER_TCP_PORT="$tcp_port" \
IGGY_SERVER_QUIC_PORT=8080 \
+ IGGY_SERVER_HTTP_PORT="$http_port" \
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 d8f070ba2..76d12d1b0 100644
--- a/.github/workflows/coverage-baseline.yml
+++ b/.github/workflows/coverage-baseline.yml
@@ -347,9 +347,20 @@ jobs:
- name: Run tests
run: |
cd foreign/python
+
+ # TCP and HTTP ports come from server-start's own outputs rather than
+ # duplicated literals, so a future default change here can't silently
+ # desync from what the server actually started with. QUIC stays a
+ # literal below because server-start exposes no output for it.
+ tcp_port="${{ steps.iggy.outputs.tcp_address }}"
+ http_port="${{ steps.iggy.outputs.http_address }}"
+ tcp_port="${tcp_port##*:}"
+ http_port="${http_port##*:}"
+
IGGY_SERVER_HOST=127.0.0.1 \
- IGGY_SERVER_TCP_PORT=8090 \
+ IGGY_SERVER_TCP_PORT="$tcp_port" \
IGGY_SERVER_QUIC_PORT=8080 \
+ IGGY_SERVER_HTTP_PORT="$http_port" \
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 10cfbc685..79c6774b2 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -41,6 +41,7 @@ pub use crate::clients::producer_builder::IggyProducerBuilder;
pub use crate::clients::producer_config::{BackgroundConfig, DirectConfig};
pub use crate::clients::producer_sharding::{BalancedSharding, OrderedSharding,
Sharding};
pub use crate::consumer_ext::IggyConsumerMessageExt;
+pub use crate::http::http_client::HttpClient;
pub use crate::quic::quic_client::QuicClient;
pub use crate::stream_builder::IggyConsumerConfig;
pub use crate::stream_builder::IggyStreamConsumer;
diff --git a/examples/python/getting-started/consumer.py
b/examples/python/getting-started/consumer.py
index f3f788640..3a48f2a2e 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -24,8 +24,10 @@ from datetime import timedelta
from apache_iggy import (
AutoLogin,
Consumer,
+ HttpConfig,
IggyClient,
PollingStrategy,
+ QuicConfig,
ReceiveMessage,
TcpConfig,
TcpReconnectionConfig,
@@ -101,12 +103,12 @@ def parse_args() -> ArgNamespace:
return ArgNamespace(**vars(args))
-def build_config(args: ArgNamespace) -> TcpConfig:
- """Build the TCP client configuration with auto-login and reconnection."""
+def build_config(args: ArgNamespace) -> TcpConfig | QuicConfig | HttpConfig:
+ """Build the client configuration, TCP with auto-login and reconnection."""
# IggyClient(...) also accepts a QuicConfig for the QUIC transport. To use
- # it, import QuicConfig and QuicReconnectionConfig above, change the return
- # annotation to QuicConfig, and replace the return statement with:
+ # it, uncomment the return below and import QuicReconnectionConfig, which
is
+ # left out above because only the commented block names it:
#
# return QuicConfig(
# server_address="127.0.0.1:8080",
@@ -116,8 +118,12 @@ def build_config(args: ArgNamespace) -> TcpConfig:
# enabled=True, interval=timedelta(seconds=1)
# ),
# )
- #
- # main() logs args.tcp_server_address, so change that line too.
+
+ # IggyClient(...) also accepts an HttpConfig for the HTTP transport. HTTP
+ # has no AutoLogin or reconnection policy, so main() below would also need
+ # an explicit `await client.login_user(args.username, args.password)`
+ # after connecting:
+ # return HttpConfig(api_url="http://127.0.0.1:3000";)
return TcpConfig(
server_address=args.tcp_server_address,
@@ -138,7 +144,7 @@ async def main():
except ValueError as error:
logger.error(f"Invalid client configuration: {error}")
return
- logger.info(f"Connecting to {args.tcp_server_address} (TLS: {args.tls})")
+ logger.info(f"Connecting with {config}")
client = IggyClient(config)
try:
diff --git a/examples/python/getting-started/producer.py
b/examples/python/getting-started/producer.py
index 113bee858..e8a8fea35 100755
--- a/examples/python/getting-started/producer.py
+++ b/examples/python/getting-started/producer.py
@@ -23,7 +23,9 @@ from datetime import timedelta
from apache_iggy import (
AutoLogin,
+ HttpConfig,
IggyClient,
+ QuicConfig,
StreamDetails,
TcpConfig,
TcpReconnectionConfig,
@@ -100,12 +102,12 @@ def parse_args() -> ArgNamespace:
return ArgNamespace(**vars(args))
-def build_config(args: ArgNamespace) -> TcpConfig:
- """Build the TCP client configuration with auto-login and reconnection."""
+def build_config(args: ArgNamespace) -> TcpConfig | QuicConfig | HttpConfig:
+ """Build the client configuration, TCP with auto-login and reconnection."""
# IggyClient(...) also accepts a QuicConfig for the QUIC transport. To use
- # it, import QuicConfig and QuicReconnectionConfig above, change the return
- # annotation to QuicConfig, and replace the return statement with:
+ # it, uncomment the return below and import QuicReconnectionConfig, which
is
+ # left out above because only the commented block names it:
#
# return QuicConfig(
# server_address="127.0.0.1:8080",
@@ -115,8 +117,12 @@ def build_config(args: ArgNamespace) -> TcpConfig:
# enabled=True, interval=timedelta(seconds=1)
# ),
# )
- #
- # main() logs args.tcp_server_address, so change that line too.
+
+ # IggyClient(...) also accepts an HttpConfig for the HTTP transport. HTTP
+ # has no AutoLogin or reconnection policy, so main() below would also need
+ # an explicit `await client.login_user(args.username, args.password)`
+ # after connecting:
+ # return HttpConfig(api_url="http://127.0.0.1:3000";)
return TcpConfig(
server_address=args.tcp_server_address,
@@ -137,7 +143,7 @@ async def main():
except ValueError as error:
logger.error(f"Invalid client configuration: {error}")
return
- logger.info(f"Connecting to {args.tcp_server_address} (TLS: {args.tls})")
+ logger.info(f"Connecting with {config}")
client = IggyClient(config)
logger.info("Connecting to IggyClient")
diff --git a/foreign/python/README.md b/foreign/python/README.md
index df62cfc77..293bbdfbc 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md
@@ -134,7 +134,8 @@ running prek / committing / pushing. This list is not
exhaustive and other hook
## Client Configuration
-`IggyClient` takes a server address, a `TcpConfig`, or a `QuicConfig`:
+`IggyClient` takes a server address, a `TcpConfig`, a `QuicConfig`, or an
+`HttpConfig`:
```python
import asyncio
@@ -168,8 +169,39 @@ async def main():
asyncio.run(main())
```
-`IggyClient(...)` also accepts a `QuicConfig` for the QUIC transport; see
-`examples/python/getting-started/producer.py` for a config swap example.
+`IggyClient(...)` also accepts a `QuicConfig` for the QUIC transport and an
+`HttpConfig` for the HTTP transport;
+`examples/python/getting-started/producer.py` shows either swap in context.
+
+`HttpConfig` differs from TCP in two ways. There is no reconnection policy and
no
+`AutoLogin`: `connect()` does not dial over HTTP, but it does start the
+heartbeat that `heartbeat_interval` configures, so call it and then
+`login_user(...)`. And HTTP is single-consumer only: the `consumer_group(...)`
+path always fails with `Feature is unavailable`, at the join by default and at
+the returned consumer's first poll if you disable `auto_join_consumer_group`,
+so disabling it is not a workaround. A direct
+`poll_messages(consumer=Consumer.Group(...))` fails the same way unless you
+pass an explicit `partition_id`, and with one it degrades silently instead: the
+consumer kind is not carried on the HTTP wire, so the group is served as an
+ordinary consumer named after it, with no membership or partition assignment
+behind it. Use `Consumer.Single(...)` with `poll_messages(...)`. Delivery is
+also at-least-once: the default `retries=3` replays the full request body, so a
+send whose response was lost is applied twice, and only `retries=0` opts out.
+
+```python
+import asyncio
+
+from apache_iggy import HttpConfig, IggyClient
+
+
+async def main():
+ client = IggyClient(HttpConfig(api_url="http://127.0.0.1:3000";))
+ await client.connect()
+ await client.login_user("iggy", "iggy")
+
+
+asyncio.run(main())
+```
## Examples
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index e543869e9..0400a1a51 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -39,6 +39,7 @@ __all__ = [
"GlobalPermissions",
"HeaderKey",
"HeaderValue",
+ "HttpConfig",
"IggyClient",
"IggyConsumer",
"IggyExpiry",
@@ -894,6 +895,78 @@ class HeaderValue:
def value(self) -> builtins.float: ...
def __new__(cls, value: builtins.float) -> HeaderValue.Float64: ...
[email protected]
+class HttpConfig:
+ r"""
+ Configuration for the HTTP transport, accepted by `IggyClient(...)`.
+
+ Every field is keyword-only and optional.
+
+ There is no `AutoLogin` and no reconnection policy, and `connect()` does
not
+ dial: it only starts the heartbeat, so `login_user(...)` has to follow it.
+
+ HTTP is single-consumer only. `consumer_group(...)` fails with
+ `Feature is unavailable`, and so does a `Consumer.Group(...)` poll unless
it
+ names an explicit `partition_id`. With one, the consumer kind is not
carried
+ on the HTTP wire, so the poll is served as an ordinary consumer named after
+ the group, with no membership, no partition assignment, and no rebalancing
+ behind it. Pass `Consumer.Single(...)` explicitly.
+ """
+ @property
+ def api_url(self) -> builtins.str: ...
+ @property
+ def retries(self) -> builtins.int: ...
+ @property
+ def has_jwt(self) -> builtins.bool:
+ r"""
+ Whether a JWT is configured, without exposing the token itself.
+ """
+ @property
+ def heartbeat_interval(self) -> datetime.timedelta: ...
+ def __new__(
+ cls,
+ *,
+ api_url: builtins.str | None = None,
+ retries: builtins.int | None = None,
+ jwt: builtins.str | None = None,
+ heartbeat_interval: datetime.timedelta | None = None,
+ ) -> HttpConfig:
+ r"""
+ Constructs an HTTP configuration.
+
+ Args:
+ api_url: Base URL of the Iggy HTTP API, as `scheme://host[:port]`
+ only - no path, query, fragment, or credentials. Defaults to
+ `http://127.0.0.1:3000`.
+ retries: Number of retries to perform on transient errors, each one
+ replaying the full request (including its body) via automatic
+ middleware. Defaults to 3. Delivery is therefore at-least-once:
+ if the original request actually committed but its response
+ was lost (e.g. to a timeout), a retried call applies the same
+ operation again. Set to 0 to disable automatic replay and match
+ the other transports, which surface the failure instead of
+ silently resending.
+ jwt: JWT token for A2A (Agent-to-Agent) authentication. Defaults to
+ `None`. Stored trimmed, since a token read from a file carries
a
+ trailing newline that the `Authorization` header value rejects.
+ Rejected if empty or whitespace-only: accepting it would make
+ `has_jwt` report `True` while every call still fails
+ `Unauthenticated`.
+ heartbeat_interval: Interval between the client's liveness probes
+ (a bare `GET /ping`). Defaults to 5 seconds. Unlike TCP/QUIC,
+ HTTP has no persistent connection or session for this to keep
+ alive; it only proves the server is reachable.
+
+ Raises:
+ ValueError: If `api_url` is not a valid URL, if `retries` is
outside
+ the range of an unsigned 32-bit integer, if `jwt` is empty or
+ whitespace-only, if a duration is negative, or if
+ `heartbeat_interval` is zero.
+ OverflowError: If `retries` does not fit a signed 64-bit integer,
+ raised by the underlying conversion before this constructor
runs.
+ """
+ def __repr__(self) -> builtins.str: ...
+
@typing.final
class IggyClient:
r"""
@@ -901,20 +974,22 @@ class IggyClient:
It provides asynchronous functionality through the contained runtime.
"""
def __new__(
- cls, conn: TcpConfig | QuicConfig | builtins.str | None = None
+ cls, conn: TcpConfig | QuicConfig | HttpConfig | builtins.str | None =
None
) -> IggyClient:
r"""
- Constructs a new IggyClient from a TCP server address, a `TcpConfig`,
or a
- `QuicConfig`. This initializes a new runtime for asynchronous
operations.
+ Constructs a new IggyClient from a TCP server address, a `TcpConfig`, a
+ `QuicConfig`, or an `HttpConfig`. This initializes a new runtime for
+ asynchronous operations.
Future versions might utilize asyncio for more Pythonic async.
Args:
- conn: A `host:port` address, a `TcpConfig`, or a `QuicConfig`.
Defaults
- to `127.0.0.1:8090` over TCP with auto-login disabled. A
malformed
- address is reported differently depending on the form: the
string
- form raises `RuntimeError` here, while `TcpConfig`/`QuicConfig`
- raise `ValueError` when they are constructed, before either
ever
- reaches this call. Neither exception is a subclass of the
other.
+ conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, or an
+ `HttpConfig`. Defaults to `127.0.0.1:8090` over TCP with
auto-login
+ disabled. A malformed address is reported differently
depending on
+ the form: the string form raises `RuntimeError` here, while
+ `TcpConfig`/`QuicConfig`/`HttpConfig` raise `ValueError` when
they
+ are constructed, before any of them ever reaches this call.
Neither
+ exception is a subclass of the other.
Raises:
RuntimeError: If the address passed as a string is not a valid
@@ -1114,8 +1189,10 @@ class IggyClient:
"""
def connect(self) -> collections.abc.Awaitable[None]:
r"""
- Connects the IggyClient to its service.
- Raises `RuntimeError` if the connection fails.
+ Connects the IggyClient to its service and starts the heartbeat task.
+ Raises `RuntimeError` if the connection fails. Over HTTP there is no
+ connection to establish, so only the heartbeat starts and this call
+ succeeds even against an unreachable server.
"""
def create_stream(self, name: builtins.str) ->
collections.abc.Awaitable[None]:
r"""
@@ -1545,6 +1622,15 @@ class IggyClient:
`poll_interval`, `polling_retry_interval`, `init_retry_interval` or an
`AutoCommit` interval is negative, or if any of those except
`poll_interval`
is zero.
+
+ Consumer groups are not available over HTTP. With
`auto_join_consumer_group`
+ left on, this call fails at the join with `Feature is unavailable`.
+ Turning it off is not a workaround: the join is skipped, but a group
+ member always polls without a partition, so the first poll fails with
+ the same error. Use `Consumer.Single(...)` with `poll_messages(...)`
+ instead - a `Consumer.Group(...)` poll with an explicit `partition_id`
+ does reach the server, but is served as an ordinary consumer named
+ after the group.
"""
def send_binary_request(
self, code: builtins.int, payload: builtins.bytes
@@ -1963,6 +2049,8 @@ class QuicConfig:
`max_idle_timeout` is not a whole number of milliseconds, if
`initial_mtu` is below quinn's minimum of 1200, or if a numeric
field is outside the range of its underlying wire type.
+ OverflowError: If a numeric field does not fit a signed 64-bit
integer,
+ raised by the underlying conversion before this constructor
runs.
"""
def __repr__(self) -> builtins.str: ...
@@ -2010,6 +2098,8 @@ class QuicReconnectionConfig:
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.
+ OverflowError: If `max_retries` does not fit a signed 64-bit
integer,
+ raised by the underlying conversion before this constructor
runs.
"""
def __repr__(self) -> builtins.str: ...
@@ -2578,6 +2668,8 @@ class TcpReconnectionConfig:
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.
+ OverflowError: If `max_retries` does not fit a signed 64-bit
integer,
+ raised by the underlying conversion before this constructor
runs.
"""
def __repr__(self) -> builtins.str: ...
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index fa0e9e3ac..7d19de995 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -92,17 +92,19 @@ fn resolve_topic_params(
#[gen_stub_pymethods]
#[pymethods]
impl IggyClient {
- /// Constructs a new IggyClient from a TCP server address, a `TcpConfig`,
or a
- /// `QuicConfig`. This initializes a new runtime for asynchronous
operations.
+ /// Constructs a new IggyClient from a TCP server address, a `TcpConfig`, a
+ /// `QuicConfig`, or an `HttpConfig`. This initializes a new runtime for
+ /// asynchronous operations.
/// Future versions might utilize asyncio for more Pythonic async.
///
/// Args:
- /// conn: A `host:port` address, a `TcpConfig`, or a `QuicConfig`.
Defaults
- /// to `127.0.0.1:8090` over TCP with auto-login disabled. A
malformed
- /// address is reported differently depending on the form: the
string
- /// form raises `RuntimeError` here, while `TcpConfig`/`QuicConfig`
- /// raise `ValueError` when they are constructed, before either
ever
- /// reaches this call. Neither exception is a subclass of the
other.
+ /// conn: A `host:port` address, a `TcpConfig`, a `QuicConfig`, or an
+ /// `HttpConfig`. Defaults to `127.0.0.1:8090` over TCP with
auto-login
+ /// disabled. A malformed address is reported differently
depending on
+ /// the form: the string form raises `RuntimeError` here, while
+ /// `TcpConfig`/`QuicConfig`/`HttpConfig` raise `ValueError` when
they
+ /// are constructed, before any of them ever reaches this call.
Neither
+ /// exception is a subclass of the other.
///
/// Raises:
/// RuntimeError: If the address passed as a string is not a valid
@@ -111,7 +113,9 @@ impl IggyClient {
#[new]
#[pyo3(signature = (conn=None))]
fn new(
- #[gen_stub(override_type(type_repr = "TcpConfig | QuicConfig |
builtins.str | None"))]
+ #[gen_stub(override_type(
+ type_repr = "TcpConfig | QuicConfig | HttpConfig | builtins.str |
None"
+ ))]
conn: Option<PyClientConfig>,
) -> PyResult<Self> {
let wrapper = match conn {
@@ -138,6 +142,9 @@ impl IggyClient {
QuicClient::create(config.client_config()).map_err(to_runtime_error)?,
)
}
+ Some(PyClientConfig::Http(config)) => ClientWrapper::Http(
+
HttpClient::create(config.client_config()).map_err(to_runtime_error)?,
+ ),
None => ClientWrapper::Tcp(
TcpClient::create(Arc::new(TcpClientConfig::default()))
.map_err(to_runtime_error)?,
@@ -500,8 +507,10 @@ impl IggyClient {
})
}
- /// Connects the IggyClient to its service.
- /// Raises `RuntimeError` if the connection fails.
+ /// Connects the IggyClient to its service and starts the heartbeat task.
+ /// Raises `RuntimeError` if the connection fails. Over HTTP there is no
+ /// connection to establish, so only the heartbeat starts and this call
+ /// succeeds even against an unreachable server.
#[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]",
imports=("collections.abc")))]
fn connect<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
let inner = self.inner.clone();
@@ -1279,6 +1288,15 @@ impl IggyClient {
/// `poll_interval`, `polling_retry_interval`, `init_retry_interval` or an
/// `AutoCommit` interval is negative, or if any of those except
`poll_interval`
/// is zero.
+ ///
+ /// Consumer groups are not available over HTTP. With
`auto_join_consumer_group`
+ /// left on, this call fails at the join with `Feature is unavailable`.
+ /// Turning it off is not a workaround: the join is skipped, but a group
+ /// member always polls without a partition, so the first poll fails with
+ /// the same error. Use `Consumer.Single(...)` with `poll_messages(...)`
+ /// instead - a `Consumer.Group(...)` poll with an explicit `partition_id`
+ /// does reach the server, but is served as an ordinary consumer named
+ /// after the group.
#[allow(clippy::too_many_arguments)]
#[pyo3(signature = (
name,
diff --git a/foreign/python/src/config.rs b/foreign/python/src/config.rs
index cd98ef626..f699a2359 100644
--- a/foreign/python/src/config.rs
+++ b/foreign/python/src/config.rs
@@ -17,6 +17,7 @@
use iggy::prelude::{
AutoLogin as RustAutoLogin, Credentials as RustCredentials,
+ HttpClientConfig as RustHttpClientConfig, HttpClientConfigBuilder,
QuicClientConfig as RustQuicClientConfig, QuicClientConfigBuilder,
QuicClientReconnectionConfig as RustQuicClientReconnectionConfig,
TcpClientConfig as RustTcpClientConfig, TcpClientConfigBuilder,
@@ -28,6 +29,7 @@ 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::fmt::Display;
use std::net::SocketAddr;
use std::sync::Arc;
@@ -144,6 +146,8 @@ impl TcpReconnectionConfig {
/// 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.
+ /// OverflowError: If `max_retries` does not fit a signed 64-bit
integer,
+ /// raised by the underlying conversion before this constructor
runs.
#[new]
#[pyo3(signature = (*, enabled=None, max_retries=None, interval=None,
reestablish_after=None))]
fn new(
@@ -157,14 +161,7 @@ impl TcpReconnectionConfig {
let defaults = RustTcpClientReconnectionConfig::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
- ))
- })
- })
+ .map(|max_retries| u32_param(max_retries, "max_retries"))
.transpose()?;
let interval = interval
.as_ref()
@@ -308,7 +305,7 @@ impl TcpConfig {
}
let mut inner = builder
.build()
- .map_err(|e| PyValueError::new_err(e.to_string()))?;
+ .map_err(|e| invalid_address("server_address", e))?;
if let Some(auto_login) = auto_login {
inner.auto_login = auto_login.inner;
}
@@ -445,6 +442,8 @@ impl QuicReconnectionConfig {
/// 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.
+ /// OverflowError: If `max_retries` does not fit a signed 64-bit
integer,
+ /// raised by the underlying conversion before this constructor
runs.
#[new]
#[pyo3(signature = (*, enabled=None, max_retries=None, interval=None,
reestablish_after=None))]
fn new(
@@ -458,14 +457,7 @@ impl QuicReconnectionConfig {
let defaults = RustQuicClientReconnectionConfig::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
- ))
- })
- })
+ .map(|max_retries| u32_param(max_retries, "max_retries"))
.transpose()?;
let interval = interval
.as_ref()
@@ -588,6 +580,8 @@ impl QuicConfig {
/// `max_idle_timeout` is not a whole number of milliseconds, if
/// `initial_mtu` is below quinn's minimum of 1200, or if a numeric
/// field is outside the range of its underlying wire type.
+ /// OverflowError: If a numeric field does not fit a signed 64-bit
integer,
+ /// raised by the underlying conversion before this constructor
runs.
#[new]
#[pyo3(signature = (
*,
@@ -647,15 +641,18 @@ impl QuicConfig {
}
let mut inner = builder
.build()
- .map_err(|e| PyValueError::new_err(e.to_string()))?;
+ .map_err(|e| invalid_address("server_address", e))?;
if let Some(client_address) = client_address {
- // Kept verbatim rather than normalized: `QuicClient::create`
compares
- // this against the literal default to decide whether to bind an
IPv6
- // socket for an IPv6 server, and a rewritten string would not
match.
- client_address.parse::<SocketAddr>().map_err(|e| {
- PyValueError::new_err(format!("'client_address' is not a valid
'host:port': {e}"))
- })?;
- inner.client_address = client_address;
+ // Trimmed like the server address, but otherwise kept verbatim
rather
+ // than re-serialized from the parsed `SocketAddr`:
`QuicClient::create`
+ // compares this against the literal default to decide whether to
bind
+ // an IPv6 socket for an IPv6 server, and a rewritten string would
not
+ // match.
+ let client_address = client_address.trim();
+ client_address
+ .parse::<SocketAddr>()
+ .map_err(|e| invalid_address("client_address", e))?;
+ inner.client_address = client_address.to_owned();
}
if let Some(server_name) = server_name {
inner.server_name = server_name;
@@ -826,10 +823,167 @@ impl QuicConfig {
}
}
+/// Configuration for the HTTP transport, accepted by `IggyClient(...)`.
+///
+/// Every field is keyword-only and optional.
+///
+/// There is no `AutoLogin` and no reconnection policy, and `connect()` does
not
+/// dial: it only starts the heartbeat, so `login_user(...)` has to follow it.
+///
+/// HTTP is single-consumer only. `consumer_group(...)` fails with
+/// `Feature is unavailable`, and so does a `Consumer.Group(...)` poll unless
it
+/// names an explicit `partition_id`. With one, the consumer kind is not
carried
+/// on the HTTP wire, so the poll is served as an ordinary consumer named after
+/// the group, with no membership, no partition assignment, and no rebalancing
+/// behind it. Pass `Consumer.Single(...)` explicitly.
+#[gen_stub_pyclass]
+#[pyclass(from_py_object)]
+#[derive(Clone)]
+pub struct HttpConfig {
+ inner: Arc<RustHttpClientConfig>,
+}
+
+impl HttpConfig {
+ /// The configuration in the shape `HttpClient::create` expects.
+ pub(crate) fn client_config(&self) -> Arc<RustHttpClientConfig> {
+ self.inner.clone()
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl HttpConfig {
+ /// Constructs an HTTP configuration.
+ ///
+ /// Args:
+ /// api_url: Base URL of the Iggy HTTP API, as `scheme://host[:port]`
+ /// only - no path, query, fragment, or credentials. Defaults to
+ /// `http://127.0.0.1:3000`.
+ /// retries: Number of retries to perform on transient errors, each one
+ /// replaying the full request (including its body) via automatic
+ /// middleware. Defaults to 3. Delivery is therefore at-least-once:
+ /// if the original request actually committed but its response
+ /// was lost (e.g. to a timeout), a retried call applies the same
+ /// operation again. Set to 0 to disable automatic replay and match
+ /// the other transports, which surface the failure instead of
+ /// silently resending.
+ /// jwt: JWT token for A2A (Agent-to-Agent) authentication. Defaults to
+ /// `None`. Stored trimmed, since a token read from a file carries
a
+ /// trailing newline that the `Authorization` header value rejects.
+ /// Rejected if empty or whitespace-only: accepting it would make
+ /// `has_jwt` report `True` while every call still fails
+ /// `Unauthenticated`.
+ /// heartbeat_interval: Interval between the client's liveness probes
+ /// (a bare `GET /ping`). Defaults to 5 seconds. Unlike TCP/QUIC,
+ /// HTTP has no persistent connection or session for this to keep
+ /// alive; it only proves the server is reachable.
+ ///
+ /// Raises:
+ /// ValueError: If `api_url` is not a valid URL, if `retries` is
outside
+ /// the range of an unsigned 32-bit integer, if `jwt` is empty or
+ /// whitespace-only, if a duration is negative, or if
+ /// `heartbeat_interval` is zero.
+ /// OverflowError: If `retries` does not fit a signed 64-bit integer,
+ /// raised by the underlying conversion before this constructor
runs.
+ #[new]
+ #[pyo3(signature = (*, api_url=None, retries=None, jwt=None,
heartbeat_interval=None))]
+ fn new(
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))] api_url:
Option<String>,
+ #[gen_stub(override_type(type_repr = "builtins.int | None"))] retries:
Option<i64>,
+ #[gen_stub(override_type(type_repr = "builtins.str | None"))] jwt:
Option<String>,
+ #[gen_stub(override_type(type_repr = "datetime.timedelta | None",
imports=("datetime")))]
+ heartbeat_interval: Option<Py<PyDelta>>,
+ ) -> PyResult<Self> {
+ // The builder starts from `HttpClientConfig::default()`, and its
`build()`
+ // trims and validates the API URL whether or not one was set here.
+ let mut builder = HttpClientConfigBuilder::new();
+ if let Some(api_url) = api_url {
+ builder = builder.with_api_url(api_url);
+ }
+ let mut inner = builder
+ .build()
+ .map_err(|e| PyValueError::new_err(format!("'api_url' is not a
valid URL: {e}")))?;
+ if let Some(retries) = retries {
+ inner.retries = u32_param(retries, "retries")?;
+ }
+ if let Some(jwt) = jwt {
+ let jwt = jwt.trim();
+ if jwt.is_empty() {
+ return Err(PyValueError::new_err(
+ "'jwt' must not be empty or whitespace-only",
+ ));
+ }
+ inner.jwt = Some(jwt.to_owned());
+ }
+ if let Some(heartbeat_interval) = heartbeat_interval {
+ inner.heartbeat_interval = reject_zero(
+ py_delta_to_iggy_duration(&heartbeat_interval)?,
+ "heartbeat_interval",
+ )?;
+ }
+
+ Ok(Self {
+ inner: Arc::new(inner),
+ })
+ }
+
+ #[getter]
+ fn api_url(&self) -> String {
+ self.inner.api_url.clone()
+ }
+
+ #[getter]
+ fn retries(&self) -> u32 {
+ self.inner.retries
+ }
+
+ /// Whether a JWT is configured, without exposing the token itself.
+ #[getter]
+ fn has_jwt(&self) -> bool {
+ self.inner.jwt.is_some()
+ }
+
+ #[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.get())
+ }
+
+ fn __repr__(&self) -> String {
+ let jwt = if self.inner.jwt.is_some() {
+ "..."
+ } else {
+ "None"
+ };
+ format!(
+ "HttpConfig(api_url={:?}, retries={}, jwt={jwt},
heartbeat_interval={})",
+ self.inner.api_url,
+ self.inner.retries,
+ duration_repr(self.inner.heartbeat_interval.get()),
+ )
+ }
+}
+
fn python_bool(value: bool) -> &'static str {
if value { "True" } else { "False" }
}
+/// Rejects an address that is not a valid `host:port`, naming the argument it
+/// came from: neither the builder's error nor `SocketAddr`'s mentions which
one.
+fn invalid_address(parameter: &str, error: impl Display) -> PyErr {
+ PyValueError::new_err(format!("'{parameter}' is not a valid 'host:port':
{error}"))
+}
+
+/// Converts a Python int to the unsigned 32-bit integer
`max_retries`/`retries`
+/// expect, naming the parameter in the error so a caller can tell which
+/// argument was out of range. A value too large even for `i64` still raises
+/// pyo3's own unnamed `OverflowError` before this ever runs.
+fn u32_param(value: i64, parameter: &str) -> PyResult<u32> {
+ u32::try_from(value).map_err(|_| {
+ PyValueError::new_err(format!("'{parameter}' must be between 0 and
{}", u32::MAX))
+ })
+}
+
/// Converts a Python int to the unsigned 64-bit integer a QUIC transport
/// field expects, naming the parameter in the error so a caller can tell
/// which argument was out of range. The bound in the message is `i64::MAX`
@@ -866,15 +1020,17 @@ fn varint_param(value: i64, parameter: &str) ->
PyResult<u64> {
Ok(value)
}
-/// What `IggyClient(...)` accepts: a bare `host:port`, a full `TcpConfig`, or
a
-/// `QuicConfig` for the QUIC transport.
+/// What `IggyClient(...)` accepts: a bare `host:port`, a full `TcpConfig`, a
+/// `QuicConfig` for the QUIC transport, or an `HttpConfig` for the HTTP
transport.
#[derive(FromPyObject)]
pub enum PyClientConfig {
#[pyo3(transparent)]
Tcp(TcpConfig),
#[pyo3(transparent)]
Quic(QuicConfig),
+ #[pyo3(transparent)]
+ Http(HttpConfig),
#[pyo3(transparent, annotation = "str")]
ServerAddress(String),
}
-impl_stub_type!(PyClientConfig = TcpConfig | QuicConfig | String);
+impl_stub_type!(PyClientConfig = TcpConfig | QuicConfig | HttpConfig | String);
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index e39b3339e..80a123ee0 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -31,7 +31,9 @@ mod user;
mod user_headers;
use client::IggyClient;
-use config::{AutoLogin, QuicConfig, QuicReconnectionConfig, TcpConfig,
TcpReconnectionConfig};
+use config::{
+ AutoLogin, HttpConfig, QuicConfig, QuicReconnectionConfig, TcpConfig,
TcpReconnectionConfig,
+};
use consumer::{
AutoCommit, AutoCommitAfter, AutoCommitWhen, Consumer, ConsumerGroup,
ConsumerGroupDetails,
ConsumerGroupMember, IggyConsumer, ReceiveMessageIterator,
@@ -60,6 +62,7 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) ->
PyResult<()> {
m.add_class::<TcpReconnectionConfig>()?;
m.add_class::<QuicConfig>()?;
m.add_class::<QuicReconnectionConfig>()?;
+ m.add_class::<HttpConfig>()?;
m.add_class::<StreamDetails>()?;
m.add_class::<Stream>()?;
m.add_class::<Stats>()?;
diff --git a/foreign/python/tests/test_client_config.py
b/foreign/python/tests/test_client_config.py
index 26d9f4430..7e80fe065 100644
--- a/foreign/python/tests/test_client_config.py
+++ b/foreign/python/tests/test_client_config.py
@@ -257,8 +257,8 @@ class TestTcpConfig:
["", "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):
+ """Test that a malformed address fails at construction, naming
itself."""
+ with pytest.raises(ValueError, match="server_address"):
TcpConfig(server_address=invalid_address)
def test_negative_heartbeat_interval_is_rejected(self):
diff --git a/foreign/python/tests/test_http_config.py
b/foreign/python/tests/test_http_config.py
new file mode 100644
index 000000000..d2237aa35
--- /dev/null
+++ b/foreign/python/tests/test_http_config.py
@@ -0,0 +1,363 @@
+# 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 HTTP client configuration surface.
+
+`HttpConfig` mirrors the Rust SDK's `HttpClientConfig` the same way
+`TcpConfig` does, 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. Unlike TCP there is no `AutoLogin` or reconnection policy to
+configure.
+"""
+
+import ast
+import json
+import urllib.request
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import Consumer, HttpConfig, IggyClient, PollingStrategy
+from apache_iggy import SendMessage as Message
+
+from .utils import get_http_server_config, wait_for_ping
+
+
[email protected]
+class TestHttpConfig:
+ """Test the transport configuration."""
+
+ def test_defaults_match_the_rust_sdk(self):
+ """Test that an unconfigured transport matches the Rust SDK
defaults."""
+ config = HttpConfig()
+
+ assert config.api_url == "http://127.0.0.1:3000";
+ assert config.retries == 3
+ assert config.has_jwt is False
+ assert config.heartbeat_interval == timedelta(seconds=5)
+
+ def test_every_field_round_trips(self):
+ """Test that each configured field is readable back unchanged."""
+ config = HttpConfig(
+ api_url="http://127.0.0.1:3001";,
+ retries=5,
+ jwt="a-token",
+ heartbeat_interval=timedelta(seconds=15),
+ )
+
+ assert config.api_url == "http://127.0.0.1:3001";
+ assert config.retries == 5
+ assert config.has_jwt is True
+ assert config.heartbeat_interval == timedelta(seconds=15)
+
+ def test_arguments_are_keyword_only(self):
+ """Test that the API URL cannot be passed positionally."""
+ with pytest.raises(TypeError):
+ # pyrefly: ignore # bad-argument-count
+ HttpConfig("http://127.0.0.1:3000";)
+
+ def test_repr_hides_the_jwt(self):
+ """Test that the JWT does not leak through repr but still parses as
Python."""
+ config = HttpConfig(jwt="a-secret-token")
+
+ printed = repr(config)
+
+ assert "a-secret-token" not in printed
+ ast.parse(printed)
+
+ def test_repr_shows_every_field_as_python(self):
+ """Test that repr covers the configured fields and parses as Python.
+
+ `heartbeat_interval` is included: its repr is built from a duration,
+ the one format-fragile field here, and `ast.parse` alone would not
+ catch a regression that renders it as something other than a
+ `datetime.timedelta` call.
+ """
+ config = HttpConfig(
+ api_url="http://127.0.0.1:3001";,
+ retries=5,
+ heartbeat_interval=timedelta(seconds=15),
+ )
+
+ printed = repr(config)
+
+ assert 'api_url="http://127.0.0.1:3001";' in printed
+ assert "retries=5" in printed
+ assert "heartbeat_interval=datetime.timedelta(seconds=15)" in printed
+ ast.parse(printed)
+
+ @pytest.mark.parametrize(
+ "invalid_url",
+ [
+ "",
+ "not-a-url",
+ "http://127.0.0.1:0";,
+ "http://127.0.0.1:3000/iggy";,
+ "http://user:[email protected]:3000";,
+ ],
+ )
+ def test_invalid_api_url_is_rejected(self, invalid_url: str):
+ """Test that a malformed API URL fails at construction, not at connect.
+
+ Only `scheme://host[:port]` is accepted: a path, query, fragment, or
+ embedded credentials are all rejected, not just a missing/zero port.
+ """
+ with pytest.raises(ValueError, match="api_url"):
+ HttpConfig(api_url=invalid_url)
+
+ @pytest.mark.parametrize("bad_jwt", ["", " ", "\t"])
+ def test_empty_or_whitespace_jwt_is_rejected(self, bad_jwt: str):
+ """Test that an empty or whitespace-only JWT fails at construction.
+
+ Accepting it would make `has_jwt` report `True` while every call
+ still fails `Unauthenticated`, since the stored token is blank.
+ """
+ with pytest.raises(ValueError, match="jwt"):
+ HttpConfig(jwt=bad_jwt)
+
+ @pytest.mark.parametrize("out_of_range", [-1, 2**32])
+ def test_out_of_range_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="retries"):
+ HttpConfig(retries=out_of_range)
+
+ @pytest.mark.parametrize(
+ "negative",
+ [timedelta(microseconds=-1), timedelta(seconds=-1),
timedelta(days=-1)],
+ )
+ def test_negative_heartbeat_interval_is_rejected(self, negative:
timedelta):
+ """Test that a negative heartbeat interval fails at construction."""
+ with pytest.raises(ValueError, match="negative"):
+ HttpConfig(heartbeat_interval=negative)
+
+ 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=r"heartbeat_interval.*must not be
zero"):
+ HttpConfig(heartbeat_interval=timedelta(0))
+
+ def test_maximum_heartbeat_interval_round_trips(self):
+ """Test that the largest timedelta survives the duration conversion.
+
+ The repr is asserted in seconds rather than days: it is rendered from
+ a microsecond count, so the maximum comes back as a whole-second
+ `timedelta` instead of the `days=` form it was constructed with.
+ """
+ maximum = timedelta(days=999_999_999)
+
+ config = HttpConfig(heartbeat_interval=maximum)
+
+ printed = repr(config)
+
+ assert config.heartbeat_interval == maximum
+ assert (
+ "heartbeat_interval=datetime.timedelta(seconds=86399999913600)" in
printed
+ )
+ ast.parse(printed)
+
+
[email protected]
+class TestHttpClientConstruction:
+ """Test that `IggyClient(...)` builds an HTTP client from an
`HttpConfig`."""
+
+ @pytest.mark.asyncio
+ async def test_accepts_a_config(self):
+ """Test that the resulting client is actually HTTP, not silently TCP.
+
+ `IggyClient(...)` is not None for either union arm, so that alone
+ never pinned the transport, and a bare `RuntimeError` does not either:
+ a client that regressed to the TCP arm raises too, just after hanging
+ until the pytest timeout. `Invalid HTTP request` is the HTTP
+ transport's own send failure, so matching it pins the transport.
+ `retries=0` keeps the failure immediate instead of working through
+ the default retry/backoff first.
+ """
+ client = IggyClient(HttpConfig(api_url="http://127.0.0.1:1";,
retries=0))
+
+ with pytest.raises(RuntimeError, match="Invalid HTTP request"):
+ await client.ping()
+
+ def test_accepts_the_default_config(self):
+ """Test that an explicit default `HttpConfig` is accepted."""
+ assert IggyClient(HttpConfig()) is not None
+
+ @pytest.mark.asyncio
+ async def test_without_a_jwt_a_privileged_call_is_unauthenticated(self):
+ """Test that a privileged call fails when no token is configured.
+
+ `HttpClient` seeds its access token from `jwt`, so with none configured
+ and no `login_user()` it stays empty and the client rejects the call
+ itself. The dead port is what proves that: nothing is dialled, so the
+ failure cannot be the server's.
+ """
+ client = IggyClient(HttpConfig(api_url="http://127.0.0.1:1";,
retries=0))
+ await client.connect()
+
+ with pytest.raises(RuntimeError, match="Unauthenticated"):
+ await client.create_stream("never-created")
+
+
[email protected]
+class TestHttpConfigAgainstServer:
+ """Test that a client built from `HttpConfig` actually connects."""
+
+ @pytest.mark.asyncio
+ async def test_client_connects_and_pings(self):
+ """Test that a client built with a custom config reaches the server."""
+ host, port = get_http_server_config()
+
+ client = IggyClient(HttpConfig(api_url=f"http://{host}:{port}";))
+ await client.connect()
+ await wait_for_ping(client)
+
+ @pytest.mark.asyncio
+ async def test_client_sends_and_polls_a_message(self, unique_name):
+ """Test a full round trip: login, create stream/topic, send, poll.
+
+ This is the part `test_client_connects_and_pings` above does not
+ cover: that a client built from `HttpConfig` can carry a real
+ workload, not just answer a ping.
+ """
+ host, port = get_http_server_config()
+ stream_name = unique_name()
+ topic_name = unique_name()
+ payload = f"payload-{unique_name()}"
+
+ client = IggyClient(HttpConfig(api_url=f"http://{host}:{port}";))
+ await client.connect()
+ await wait_for_ping(client)
+ await client.login_user("iggy", "iggy")
+
+ await client.create_stream(stream_name)
+ await client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+ await client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=0,
+ messages=[Message(payload)],
+ )
+
+ polled_messages = await client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ consumer=Consumer.Single("http-round-trip"),
+ partition_id=0,
+ polling_strategy=PollingStrategy.First(),
+ count=1,
+ auto_commit=True,
+ )
+
+ assert [message.payload().decode() for message in polled_messages] ==
[payload]
+
+ @pytest.mark.asyncio
+ async def test_jwt_config_actually_authenticates(self, unique_name):
+ """Test that a JWT passed to `HttpConfig` reaches `access_token`.
+
+ `has_jwt` only proves a token is configured, not that it works.
+ `/users/login` is unauthenticated, so a token minted out-of-band via
+ stdlib `urllib` (bypassing `HttpConfig` and `login_user()` entirely)
+ proves the client actually authenticates with the token it was given.
+
+ The trailing newline is deliberate, and is the only coverage of the
+ trim in `HttpConfig::new`: it reproduces a token read from a file, and
+ untrimmed it builds a `Bearer <token>\\n` header value that
+ `HeaderValue` rejects, failing every call with `Invalid HTTP request`.
+ Do not remove it.
+ """
+ host, port = get_http_server_config()
+ api_url = f"http://{host}:{port}";
+
+ request = urllib.request.Request( # noqa: S310
+ f"{api_url}/users/login",
+ data=json.dumps({"username": "iggy", "password": "iggy"}).encode(),
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+ with urllib.request.urlopen(request) as response: # noqa: S310
+ identity = json.loads(response.read())
+ token = identity["access_token"]["token"]
+
+ client = IggyClient(HttpConfig(api_url=api_url, jwt=f"{token}\n"))
+ 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_wrong_jwt_is_rejected_by_the_server(self, unique_name):
+ """Test that a JWT the server cannot decode fails the call, not
connect.
+
+ `connect()` does not dial over HTTP, so a bad token can only surface
+ later. `wait_for_ping` runs first because ping needs no credentials:
+ it proves the server is reachable, which pins the failure below on the
+ token rather than on a missing listener. The server answers an
+ undecodable token with 401, which the HTTP client maps back to
+ `Unauthenticated`.
+ """
+ host, port = get_http_server_config()
+
+ client = IggyClient(
+ HttpConfig(api_url=f"http://{host}:{port}";, jwt="not-a-real-token")
+ )
+ await client.connect()
+ await wait_for_ping(client)
+
+ with pytest.raises(RuntimeError, match="Unauthenticated"):
+ await client.create_stream(unique_name())
+
+ @pytest.mark.asyncio
+ async def test_consumer_group_is_rejected(self, unique_name):
+ """Test that a consumer group fails loudly over HTTP, not silently.
+
+ `join_consumer_group` answers `Feature is unavailable` over HTTP, and
+ `consumer_group(...)` awaits that join before returning, so the
+ failure surfaces at construction. `auto_join_consumer_group=False`
+ is not a way around it: the join is skipped, but the first poll
+ raises the same error, because a group member is pinned to no
+ partition and a consumer-group poll without one is rejected
+ client-side.
+ """
+ host, port = get_http_server_config()
+ stream_name = unique_name()
+ topic_name = unique_name()
+
+ client = IggyClient(HttpConfig(api_url=f"http://{host}:{port}";))
+ await client.connect()
+ await wait_for_ping(client)
+ await client.login_user("iggy", "iggy")
+
+ await client.create_stream(stream_name)
+ await client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+
+ with pytest.raises(RuntimeError, match="Feature is unavailable"):
+ await client.consumer_group(
+ name=unique_name(), stream=stream_name, topic=topic_name
+ )
diff --git a/foreign/python/tests/test_quic_config.py
b/foreign/python/tests/test_quic_config.py
index d0fdc3f7b..eb4089a15 100644
--- a/foreign/python/tests/test_quic_config.py
+++ b/foreign/python/tests/test_quic_config.py
@@ -26,6 +26,7 @@ covered by `test_client_config.py`.
"""
import ast
+import socket
from collections.abc import Callable
from datetime import timedelta
@@ -131,10 +132,21 @@ class TestQuicReconnectionConfig:
assert reconnection.interval == timedelta(days=30_000)
def test_maximum_interval_round_trips(self):
- """Test that the largest timedelta survives the day conversion."""
- reconnection =
QuicReconnectionConfig(interval=timedelta(days=999_999_999))
+ """Test that the largest timedelta survives the day conversion.
- assert reconnection.interval == timedelta(days=999_999_999)
+ The repr is asserted in seconds rather than days: it is rendered from
+ a microsecond count, so the maximum comes back as a whole-second
+ `timedelta` instead of the `days=` form it was constructed with.
+ """
+ maximum = timedelta(days=999_999_999)
+
+ reconnection = QuicReconnectionConfig(interval=maximum)
+
+ printed = repr(reconnection)
+
+ assert reconnection.interval == maximum
+ assert "interval=datetime.timedelta(seconds=86399999913600)" in printed
+ ast.parse(printed)
@pytest.mark.unit
@@ -210,8 +222,15 @@ class TestQuicConfig:
assert "secret" not in repr(config)
def test_repr_shows_every_field_as_python(self):
- """Test that repr covers the QUIC-specific fields and parses as
Python."""
+ """Test that repr covers the configured fields and parses as Python.
+
+ The three string fields are asserted too: `ast.parse` alone still
+ passes on a repr that dropped one from the format string.
+ """
config = QuicConfig(
+ server_address="127.0.0.1:8081",
+ client_address="127.0.0.1:9000",
+ server_name="example.com",
heartbeat_interval=timedelta(seconds=15),
keep_alive_interval=timedelta(seconds=2),
max_idle_timeout=timedelta(seconds=20),
@@ -220,6 +239,9 @@ class TestQuicConfig:
printed = repr(config)
+ assert 'server_address="127.0.0.1:8081"' in printed
+ assert 'client_address="127.0.0.1:9000"' in printed
+ assert 'server_name="example.com"' in printed
assert "validate_certificate=True" in printed
assert "heartbeat_interval=datetime.timedelta(seconds=15)" in printed
assert "keep_alive_interval=datetime.timedelta(seconds=2)" in printed
@@ -231,8 +253,8 @@ class TestQuicConfig:
["", "127.0.0.1", "127.0.0.1:not-a-port", "127.0.0.1:70000",
"::1:8080"],
)
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):
+ """Test that a malformed address fails at construction, naming
itself."""
+ with pytest.raises(ValueError, match="server_address"):
QuicConfig(server_address=invalid_address)
@pytest.mark.parametrize(
@@ -251,6 +273,20 @@ class TestQuicConfig:
with pytest.raises(ValueError, match="client_address"):
QuicConfig(client_address=invalid_address)
+ def test_surrounding_whitespace_in_the_addresses_is_trimmed(self):
+ """Test that both addresses tolerate whitespace, like HTTP's `api_url`.
+
+ `client_address` is stored as the string `QuicClient::create` compares
+ against the literal default to pick an IPv6 bind address, so storing
+ the trimmed form is what keeps a padded default matching that sentinel.
+ """
+ config = QuicConfig(
+ server_address=" 127.0.0.1:8080 ", client_address=" 127.0.0.1:0 "
+ )
+
+ assert config.server_address == "127.0.0.1:8080"
+ assert config.client_address == "127.0.0.1:0"
+
def test_negative_heartbeat_interval_is_rejected(self):
"""Test that a negative heartbeat interval fails at construction."""
with pytest.raises(ValueError, match="negative"):
@@ -359,8 +395,32 @@ class TestQuicClientConstruction:
"""Test that `IggyClient(...)` accepts a `QuicConfig`."""
def test_accepts_a_config(self):
- """Test that a client can be built from a config object."""
- assert IggyClient(QuicConfig(server_address="127.0.0.1:8080")) is not
None
+ """Test that the resulting client is actually QUIC, not silently TCP.
+
+ `IggyClient(...)` is not None for either union arm, so that alone never
+ pinned the transport. `client_address` is a QUIC-only field that
+ `QuicClient::create` binds eagerly here, so a port already held fails
+ the bind synchronously with `Cannot create endpoint`, with no server
+ and no privileged port involved. A client that regressed to the TCP arm
+ has no such field and would construct fine, and no other error maps to
+ that message.
+
+ The held port has to be a real one: an unroutable address would only
+ fail the bind while `net.ipv4.ip_nonlocal_bind` is 0, and a host
+ running keepalived or a container setting it alone would bind
+ successfully and assert nothing. Neither the socket below nor quinn
+ sets `SO_REUSEADDR`, so the second bind is EADDRINUSE regardless.
+ """
+ # A bindable client_address of its own, so the failure below is the
+ # collision and not the field being set at all.
+ assert IggyClient(QuicConfig(client_address="127.0.0.1:0")) is not None
+
+ with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as held:
+ held.bind(("127.0.0.1", 0))
+ port = held.getsockname()[1]
+
+ with pytest.raises(RuntimeError, match="Cannot create endpoint"):
+ IggyClient(QuicConfig(client_address=f"127.0.0.1:{port}"))
def test_accepts_the_default_config(self):
"""Test that an explicit default `QuicConfig` is accepted."""
diff --git a/foreign/python/tests/utils.py b/foreign/python/tests/utils.py
index cdede9f32..5d7654cbf 100644
--- a/foreign/python/tests/utils.py
+++ b/foreign/python/tests/utils.py
@@ -32,6 +32,10 @@ MAX_USERNAME_BYTES = 50
MIN_PASSWORD_BYTES = 3
MAX_PASSWORD_BYTES = 100
+DEFAULT_TCP_PORT = 8090
+DEFAULT_QUIC_PORT = 8080
+DEFAULT_HTTP_PORT = 3000
+
def get_transport_config(port_env_var: str, default_port: int) -> tuple[str,
int]:
"""
@@ -69,7 +73,7 @@ def get_server_config() -> tuple[str, int]:
Returns:
tuple: (host, port) for the Iggy server
"""
- return get_transport_config("IGGY_SERVER_TCP_PORT", 8090)
+ return get_transport_config("IGGY_SERVER_TCP_PORT", DEFAULT_TCP_PORT)
def get_quic_server_config() -> tuple[str, int]:
@@ -79,7 +83,17 @@ def get_quic_server_config() -> tuple[str, int]:
Returns:
tuple: (host, port) for the Iggy server
"""
- return get_transport_config("IGGY_SERVER_QUIC_PORT", 8080)
+ return get_transport_config("IGGY_SERVER_QUIC_PORT", DEFAULT_QUIC_PORT)
+
+
+def get_http_server_config() -> tuple[str, int]:
+ """
+ Get HTTP server configuration from environment variables or defaults.
+
+ Returns:
+ tuple: (host, port) for the Iggy HTTP API
+ """
+ return get_transport_config("IGGY_SERVER_HTTP_PORT", DEFAULT_HTTP_PORT)
def wait_for_server(host: str, port: int, timeout: int = 60, interval: int =
2) -> None:
