diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock index ed686c9bda..b10fc8c5bd 100644 --- a/bdd/python/uv.lock +++ b/bdd/python/uv.lock @@ -25,8 +25,6 @@ requires-dist = [ { name = "pytest-cov", marker = "extra == 'testing'", specifier = ">=4.0,<8.0" }, { name = "pytest-timeout", marker = "extra == 'all'", specifier = ">=2.0,<3.0" }, { name = "pytest-timeout", marker = "extra == 'testing'", specifier = ">=2.0,<3.0" }, - { name = "pytest-xdist", marker = "extra == 'all'", specifier = ">=3.0,<4.0" }, - { name = "pytest-xdist", marker = "extra == 'testing'", specifier = ">=3.0,<4.0" }, { name = "ruff", marker = "extra == 'all'", specifier = ">=0.1.0,<1.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0,<1.0" }, { name = "testcontainers", marker = "extra == 'all'", specifier = ">=3.7.0,<5.0" }, diff --git a/examples/python/uv.lock b/examples/python/uv.lock index 639a0f5637..ea1236d91a 100644 --- a/examples/python/uv.lock +++ b/examples/python/uv.lock @@ -25,8 +25,6 @@ requires-dist = [ { name = "pytest-cov", marker = "extra == 'testing'", specifier = ">=4.0,<8.0" }, { name = "pytest-timeout", marker = "extra == 'all'", specifier = ">=2.0,<3.0" }, { name = "pytest-timeout", marker = "extra == 'testing'", specifier = ">=2.0,<3.0" }, - { name = "pytest-xdist", marker = "extra == 'all'", specifier = ">=3.0,<4.0" }, - { name = "pytest-xdist", marker = "extra == 'testing'", specifier = ">=3.0,<4.0" }, { name = "ruff", marker = "extra == 'all'", specifier = ">=0.1.0,<1.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0,<1.0" }, { name = "testcontainers", marker = "extra == 'all'", specifier = ">=3.7.0,<5.0" }, diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 0ce2401638..8e6a5320fd 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -32,24 +32,25 @@ __all__ = [ "ConsumerGroup", "ConsumerGroupDetails", "ConsumerGroupMember", + "GlobalPermissions", "HeaderKey", "HeaderValue", - "GlobalPermissions", "IggyClient", "IggyConsumer", "Permissions", "PollingStrategy", "ReceiveMessage", "SendMessage", + "Stream", "StreamDetails", "StreamPermissions", "Topic", "TopicDetails", "TopicPermissions", + "UserHeaders", "UserInfo", "UserInfoDetails", "UserStatus", - "UserHeaders", ] class AutoCommit: @@ -314,6 +315,105 @@ class ConsumerGroupMember: Gets the collection of partitions the consumer group member is consuming. """ +@typing.final +class GlobalPermissions: + r""" + Global permissions, applied to all streams without specifying them one by one. + """ + @property + def manage_servers(self) -> builtins.bool: + r""" + Whether managing servers is allowed; includes `read_servers`. + """ + @property + def read_servers(self) -> builtins.bool: + r""" + Whether reading server info (stats, clients) is allowed. + """ + @property + def manage_users(self) -> builtins.bool: + r""" + Whether managing users is allowed; includes `read_users`. + """ + @property + def read_users(self) -> builtins.bool: + r""" + Whether reading user info is allowed. + """ + @property + def manage_streams(self) -> builtins.bool: + r""" + Whether managing all streams is allowed; includes `read_streams` and + `manage_topics`. + """ + @property + def read_streams(self) -> builtins.bool: + r""" + Whether reading all streams is allowed; includes `read_topics`. + """ + @property + def manage_topics(self) -> builtins.bool: + r""" + Whether managing all topics is allowed; includes `read_topics` and + `send_messages`. + """ + @property + def read_topics(self) -> builtins.bool: + r""" + Whether reading all topics and managing consumer groups is allowed; + includes `poll_messages`. + """ + @property + def poll_messages(self) -> builtins.bool: + r""" + Whether polling messages from all streams and managing consumer + offsets is allowed. + """ + @property + def send_messages(self) -> builtins.bool: + r""" + Whether sending messages to all streams is allowed. + """ + def __eq__(self, other: builtins.object, /) -> builtins.bool: ... + def __new__( + cls, + *, + manage_servers: builtins.bool = False, + read_servers: builtins.bool = False, + manage_users: builtins.bool = False, + read_users: builtins.bool = False, + manage_streams: builtins.bool = False, + read_streams: builtins.bool = False, + manage_topics: builtins.bool = False, + read_topics: builtins.bool = False, + poll_messages: builtins.bool = False, + send_messages: builtins.bool = False, + ) -> GlobalPermissions: + r""" + Create global permissions. Every flag defaults to `False`. + + The `includes` notes below are transitive: a flag also grants everything + its included flags grant. For example `manage_streams` includes + `manage_topics`, and through it `read_topics`, `poll_messages`, and + `send_messages`. + + Args: + manage_servers: Allow managing servers; includes `read_servers`. + read_servers: Allow reading server info (stats, clients). + manage_users: Allow managing users; includes `read_users`. + read_users: Allow reading user info. + manage_streams: Allow managing all streams; includes `read_streams` + and `manage_topics`. + read_streams: Allow reading all streams; includes `read_topics`. + manage_topics: Allow managing all topics; includes `read_topics` + and `send_messages`. + read_topics: Allow reading all topics and managing consumer groups + (including create and delete); includes `poll_messages`. + poll_messages: Allow polling messages from all streams and managing + consumer offsets. + send_messages: Allow sending messages to all streams. + """ + class HeaderKey: r""" Typed key for an Iggy user header. @@ -664,105 +764,6 @@ class HeaderValue: def value(self) -> builtins.float: ... def __new__(cls, value: builtins.float) -> HeaderValue.Float64: ... -@typing.final -class GlobalPermissions: - r""" - Global permissions, applied to all streams without specifying them one by one. - """ - @property - def manage_servers(self) -> builtins.bool: - r""" - Whether managing servers is allowed; includes `read_servers`. - """ - @property - def read_servers(self) -> builtins.bool: - r""" - Whether reading server info (stats, clients) is allowed. - """ - @property - def manage_users(self) -> builtins.bool: - r""" - Whether managing users is allowed; includes `read_users`. - """ - @property - def read_users(self) -> builtins.bool: - r""" - Whether reading user info is allowed. - """ - @property - def manage_streams(self) -> builtins.bool: - r""" - Whether managing all streams is allowed; includes `read_streams` and - `manage_topics`. - """ - @property - def read_streams(self) -> builtins.bool: - r""" - Whether reading all streams is allowed; includes `read_topics`. - """ - @property - def manage_topics(self) -> builtins.bool: - r""" - Whether managing all topics is allowed; includes `read_topics` and - `send_messages`. - """ - @property - def read_topics(self) -> builtins.bool: - r""" - Whether reading all topics and managing consumer groups is allowed; - includes `poll_messages`. - """ - @property - def poll_messages(self) -> builtins.bool: - r""" - Whether polling messages from all streams and managing consumer - offsets is allowed. - """ - @property - def send_messages(self) -> builtins.bool: - r""" - Whether sending messages to all streams is allowed. - """ - def __eq__(self, other: builtins.object, /) -> builtins.bool: ... - def __new__( - cls, - *, - manage_servers: builtins.bool = False, - read_servers: builtins.bool = False, - manage_users: builtins.bool = False, - read_users: builtins.bool = False, - manage_streams: builtins.bool = False, - read_streams: builtins.bool = False, - manage_topics: builtins.bool = False, - read_topics: builtins.bool = False, - poll_messages: builtins.bool = False, - send_messages: builtins.bool = False, - ) -> GlobalPermissions: - r""" - Create global permissions. Every flag defaults to `False`. - - The `includes` notes below are transitive: a flag also grants everything - its included flags grant. For example `manage_streams` includes - `manage_topics`, and through it `read_topics`, `poll_messages`, and - `send_messages`. - - Args: - manage_servers: Allow managing servers; includes `read_servers`. - read_servers: Allow reading server info (stats, clients). - manage_users: Allow managing users; includes `read_users`. - read_users: Allow reading user info. - manage_streams: Allow managing all streams; includes `read_streams` - and `manage_topics`. - read_streams: Allow reading all streams; includes `read_topics`. - manage_topics: Allow managing all topics; includes `read_topics` - and `send_messages`. - read_topics: Allow reading all topics and managing consumer groups - (including create and delete); includes `poll_messages`. - poll_messages: Allow polling messages from all streams and managing - consumer offsets. - send_messages: Allow sending messages to all streams. - """ - @typing.final class IggyClient: r""" @@ -949,6 +950,86 @@ class IggyClient: Gets stream by id. Returns Option of stream details or a PyRuntimeError on failure. """ + def get_streams(self) -> collections.abc.Awaitable[list[Stream]]: + r""" + Return all streams visible to the authenticated user. + + The result is ordered by ascending numeric stream ID. + + Returns: + A list of `Stream` summaries. + + Raises: + RuntimeError: If the client is not authenticated, the user lacks global + `read_streams` or `manage_streams` permission, or the request fails. + """ + def update_stream( + self, stream_id: builtins.str | builtins.int, name: builtins.str + ) -> collections.abc.Awaitable[None]: + r""" + Rename a stream selected by name or numeric ID. + + `stream_id` accepts a stream name as `str` or numeric ID as `int`. A + decimal-only string is interpreted as a numeric ID. `name` must be unique + and contain between 1 and 255 UTF-8 bytes. Renaming a stream to its current + name succeeds without changing it. + + Returns: + None. + + Raises: + TypeError: If `stream_id` is neither `str` nor `int`, or `name` is not + `str`. + OverflowError: If an integer identifier is outside `0..=2**32 - 1`. + ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes. + RuntimeError: If the client is not authenticated, the user lacks global + `manage_streams` or per-stream `manage_stream` permission, the + stream does not exist, the new name is invalid or already used, or + the request fails. + """ + def delete_stream( + self, stream_id: builtins.str | builtins.int + ) -> collections.abc.Awaitable[None]: + r""" + Delete a stream selected by name or numeric ID. + + Deletion removes the stream and all of its topics, partitions, and messages. + `stream_id` accepts a stream name as `str` or numeric ID as `int`. A + decimal-only string is interpreted as a numeric ID. + + Returns: + None. + + Raises: + TypeError: If `stream_id` is neither `str` nor `int`. + OverflowError: If an integer identifier is outside `0..=2**32 - 1`. + ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes. + RuntimeError: If the client is not authenticated, the user lacks global + `manage_streams` or per-stream `manage_stream` permission, the + stream does not exist, or the request fails. + """ + def purge_stream( + self, stream_id: builtins.str | builtins.int + ) -> collections.abc.Awaitable[None]: + r""" + Delete all messages from every topic in a stream. + + The stream, topics, and partitions remain available. Repeated purges of an + existing empty stream succeed. `stream_id` accepts a stream name as `str` + or numeric ID as `int`. A decimal-only string is interpreted as a numeric + ID. + + Returns: + None. + + Raises: + TypeError: If `stream_id` is neither `str` nor `int`. + OverflowError: If an integer identifier is outside `0..=2**32 - 1`. + ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes. + RuntimeError: If the client is not authenticated, the user lacks global + `manage_streams` or per-stream `manage_stream` permission, the + stream does not exist, or the request fails. + """ def create_topic( self, stream: builtins.str | builtins.int, @@ -1438,6 +1519,10 @@ class ReceiveMessage: def user_headers(self) -> UserHeaders | None: r""" Retrieves user headers attached to the received message. + + Returns `None` when no headers are present or when the headers + on the wire are structurally malformed (those errors are logged + internally). Only known semantic decode errors raise `ValueError`. """ @typing.final @@ -1459,6 +1544,45 @@ class SendMessage: directly from Python using the provided string or bytes data. """ +@typing.final +class Stream: + r""" + Summary information returned by `IggyClient.get_streams()`. + + `created_at` is Unix time in microseconds. `size_bytes` is the stream's + current stored size in bytes. + """ + @property + def id(self) -> builtins.int: + r""" + Numeric stream identifier. + """ + @property + def created_at(self) -> builtins.int: + r""" + Stream creation time as Unix time in microseconds. + """ + @property + def name(self) -> builtins.str: + r""" + Unique stream name. + """ + @property + def size_bytes(self) -> builtins.int: + r""" + Current stored stream size in bytes. + """ + @property + def messages_count(self) -> builtins.int: + r""" + Total messages across all topics in the stream. + """ + @property + def topics_count(self) -> builtins.int: + r""" + Number of topics in the stream. + """ + @typing.final class StreamDetails: @property @@ -1664,6 +1788,36 @@ class TopicPermissions: send_messages: Allow sending messages to the topic. """ +@typing.final +class UserHeaders(dict): + r""" + User headers dictionary returned by `ReceiveMessage.user_headers`. + + This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping + operations work) that additionally exposes `to_scalar_dict` for the convenient + scalar form. + """ + def __new__(cls, mapping: dict | None = None) -> UserHeaders: + r""" + Wraps a mapping so its entries gain the `to_scalar_dict` helper. + + Accepts a dict whose keys and values can each independently be + `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool | + int | float`). The inherited `dict` initializer copies the provided + mapping. + """ + def __setitem__(self, key: typing.Any, value: typing.Any) -> None: ... + def to_scalar_dict( + self, + ) -> dict[str | bytes | bool | int | float, str | bytes | bool | int | float]: + r""" + Converts these headers into the convenient plain dictionary form. + + Returns an error if two distinct typed keys map to the same plain + Python scalar (e.g., `UnsignedInt8(1)` and `UnsignedInt16(1)` both + become `int(1)`), or if a stored field cannot be decoded. + """ + @typing.final class UserInfo: @property @@ -1729,31 +1883,3 @@ class UserStatus(enum.Enum): r""" The user account is inactive and cannot be used. """ - -class UserHeaders(dict): - r""" - User headers dictionary returned by `ReceiveMessage.user_headers`. - - This is a regular `dict[HeaderKey, HeaderValue]` (so all mapping - operations work) that additionally exposes `to_scalar_dict` for the convenient - scalar form. - """ - def __new__(cls, mapping: dict | None = None) -> UserHeaders: - r""" - Wraps a mapping so its entries gain the `to_scalar_dict` helper. - - Accepts a dict whose keys and values can each independently be - `HeaderKey`/`HeaderValue` or a plain scalar (`str | bytes | bool | - int | float`). The inherited `dict` initializer copies the provided - mapping. - """ - def to_scalar_dict( - self, - ) -> dict[str | bytes | bool | int | float, str | bytes | bool | int | float]: - r""" - Converts these headers into the convenient plain dictionary form. - - Returns an error if two distinct typed keys map to the same plain - Python scalar (e.g., `UnsignedInt8(1)` and `UnsignedInt16(1)` both - become `int(1)`), or if a stored field cannot be decoded. - """ diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml index d7f8788857..e853a3720d 100644 --- a/foreign/python/pyproject.toml +++ b/foreign/python/pyproject.toml @@ -102,7 +102,6 @@ testing = [ "pytest>=9.0.3,<10.0", "pytest-asyncio>=0.24.0,<2.0", "pytest-cov>=4.0,<8.0", - "pytest-xdist>=3.0,<4.0", "pytest-timeout>=2.0,<3.0", ] @@ -123,7 +122,6 @@ all = [ "pytest>=9.0.3,<10.0", "pytest-asyncio>=0.24.0,<2.0", "pytest-cov>=4.0,<8.0", - "pytest-xdist>=3.0,<4.0", "pytest-timeout>=2.0,<3.0", "ruff>=0.1.0,<1.0", "testcontainers>=3.7.0,<5.0", diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index a7f0cfd514..d22ef907ac 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -37,7 +37,7 @@ use crate::identifier::PyIdentifier; use crate::permissions::Permissions as PyPermissions; use crate::receive_message::{PollingStrategy, ReceiveMessage}; use crate::send_message::SendMessage; -use crate::stream::StreamDetails; +use crate::stream::{Stream, StreamDetails}; use crate::topic::{Topic, TopicDetails}; use crate::user::{ UserInfo as PyUserInfo, UserInfoDetails as PyUserInfoDetails, UserStatus as PyUserStatus, @@ -408,6 +408,132 @@ impl IggyClient { }) } + /// Return all streams visible to the authenticated user. + /// + /// The result is ordered by ascending numeric stream ID. + /// + /// Returns: + /// A list of `Stream` summaries. + /// + /// Raises: + /// RuntimeError: If the client is not authenticated, the user lacks global + /// `read_streams` or `manage_streams` permission, or the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[list[Stream]]", imports=("collections.abc")))] + fn get_streams<'a>(&self, py: Python<'a>) -> PyResult> { + let inner = self.inner.clone(); + future_into_py(py, async move { + let streams = inner + .get_streams() + .await + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(streams.into_iter().map(Stream::from).collect::>()) + }) + } + + /// Rename a stream selected by name or numeric ID. + /// + /// `stream_id` accepts a stream name as `str` or numeric ID as `int`. A + /// decimal-only string is interpreted as a numeric ID. `name` must be unique + /// and contain between 1 and 255 UTF-8 bytes. Renaming a stream to its current + /// name succeeds without changing it. + /// + /// Returns: + /// None. + /// + /// Raises: + /// TypeError: If `stream_id` is neither `str` nor `int`, or `name` is not + /// `str`. + /// OverflowError: If an integer identifier is outside `0..=2**32 - 1`. + /// ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes. + /// RuntimeError: If the client is not authenticated, the user lacks global + /// `manage_streams` or per-stream `manage_stream` permission, the + /// stream does not exist, the new name is invalid or already used, or + /// the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn update_stream<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + name: String, + ) -> PyResult> { + let stream_id = Identifier::try_from(stream_id)?; + let inner = self.inner.clone(); + future_into_py(py, async move { + inner + .update_stream(&stream_id, &name) + .await + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + }) + } + + /// Delete a stream selected by name or numeric ID. + /// + /// Deletion removes the stream and all of its topics, partitions, and messages. + /// `stream_id` accepts a stream name as `str` or numeric ID as `int`. A + /// decimal-only string is interpreted as a numeric ID. + /// + /// Returns: + /// None. + /// + /// Raises: + /// TypeError: If `stream_id` is neither `str` nor `int`. + /// OverflowError: If an integer identifier is outside `0..=2**32 - 1`. + /// ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes. + /// RuntimeError: If the client is not authenticated, the user lacks global + /// `manage_streams` or per-stream `manage_stream` permission, the + /// stream does not exist, or the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn delete_stream<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + ) -> PyResult> { + let stream_id = Identifier::try_from(stream_id)?; + let inner = self.inner.clone(); + future_into_py(py, async move { + inner + .delete_stream(&stream_id) + .await + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + }) + } + + /// Delete all messages from every topic in a stream. + /// + /// The stream, topics, and partitions remain available. Repeated purges of an + /// existing empty stream succeed. `stream_id` accepts a stream name as `str` + /// or numeric ID as `int`. A decimal-only string is interpreted as a numeric + /// ID. + /// + /// Returns: + /// None. + /// + /// Raises: + /// TypeError: If `stream_id` is neither `str` nor `int`. + /// OverflowError: If an integer identifier is outside `0..=2**32 - 1`. + /// ValueError: If a string identifier is empty or exceeds 255 UTF-8 bytes. + /// RuntimeError: If the client is not authenticated, the user lacks global + /// `manage_streams` or per-stream `manage_stream` permission, the + /// stream does not exist, or the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn purge_stream<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + ) -> PyResult> { + let stream_id = Identifier::try_from(stream_id)?; + let inner = self.inner.clone(); + future_into_py(py, async move { + inner + .purge_stream(&stream_id) + .await + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + }) + } + /// Creates a new topic with the given parameters. /// Returns Ok(()) on successful topic creation or a PyRuntimeError on failure. #[pyo3( diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index 9bb2308f4e..6199d1d46e 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -35,7 +35,7 @@ use permissions::{GlobalPermissions, Permissions, StreamPermissions, TopicPermis use pyo3::prelude::*; use receive_message::{PollingStrategy, ReceiveMessage}; use send_message::SendMessage; -use stream::StreamDetails; +use stream::{Stream, StreamDetails}; use topic::{Topic, TopicDetails}; use user::{UserInfo, UserInfoDetails, UserStatus}; use user_headers::{HeaderKey, HeaderValue, UserHeaders}; @@ -47,6 +47,7 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/foreign/python/src/stream.rs b/foreign/python/src/stream.rs index 7343d2c53f..2a7055b2df 100644 --- a/foreign/python/src/stream.rs +++ b/foreign/python/src/stream.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use iggy::prelude::StreamDetails as RustStreamDetails; +use iggy::prelude::{Stream as RustStream, StreamDetails as RustStreamDetails}; use pyo3::prelude::*; use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; @@ -56,3 +56,59 @@ impl StreamDetails { self.inner.topics_count } } + +/// Summary information returned by `IggyClient.get_streams()`. +/// +/// `created_at` is Unix time in microseconds. `size_bytes` is the stream's +/// current stored size in bytes. +#[gen_stub_pyclass] +#[pyclass] +pub struct Stream { + pub(crate) inner: RustStream, +} + +impl From for Stream { + fn from(stream: RustStream) -> Self { + Self { inner: stream } + } +} + +#[gen_stub_pymethods] +#[pymethods] +impl Stream { + /// Numeric stream identifier. + #[getter] + pub fn id(&self) -> u32 { + self.inner.id + } + + /// Stream creation time as Unix time in microseconds. + #[getter] + pub fn created_at(&self) -> u64 { + self.inner.created_at.as_micros() + } + + /// Unique stream name. + #[getter] + pub fn name(&self) -> String { + self.inner.name.to_string() + } + + /// Current stored stream size in bytes. + #[getter] + pub fn size_bytes(&self) -> u64 { + self.inner.size.as_bytes_u64() + } + + /// Total messages across all topics in the stream. + #[getter] + pub fn messages_count(&self) -> u64 { + self.inner.messages_count + } + + /// Number of topics in the stream. + #[getter] + pub fn topics_count(&self) -> u32 { + self.inner.topics_count + } +} diff --git a/foreign/python/tests/conftest.py b/foreign/python/tests/conftest.py index aa54ff50d8..d02e30c6b9 100644 --- a/foreign/python/tests/conftest.py +++ b/foreign/python/tests/conftest.py @@ -22,13 +22,11 @@ and connecting to Iggy servers in various configurations. """ -# TODO(slbotbm): Create text fixture for clean up after -# delete_stream() has been implemented. - import asyncio import os import secrets import string +from collections.abc import AsyncGenerator from pathlib import Path import pytest @@ -37,9 +35,32 @@ from .utils import get_server_config, wait_for_ping, wait_for_server +_ROOT_USER_ID = 0 + + +async def _delete_test_resources(client: IggyClient) -> None: + streams = await client.get_streams() + + for stream in streams: + topics = await client.get_topics(stream.id) + for topic in topics: + consumer_groups = await client.get_consumer_groups(stream.id, topic.id) + for consumer_group in consumer_groups: + await client.delete_consumer_group( + stream.id, topic.id, consumer_group.id + ) + + for stream in streams: + await client.delete_stream(stream.id) -@pytest.fixture(scope="session") -async def iggy_client() -> IggyClient: + users = await client.get_users() + for user in users: + if user.id != _ROOT_USER_ID: + await client.delete_user(user.id) + + +@pytest.fixture(scope="session", autouse=True) +async def iggy_client() -> AsyncGenerator[IggyClient, None]: """ Create and configure an Iggy client for testing. @@ -50,7 +71,7 @@ async def iggy_client() -> IggyClient: 4. Authenticates with default credentials 5. Verifies connectivity with ping - Returns: + Yields: IggyClient: Authenticated client ready for testing """ host, port = get_server_config() @@ -68,7 +89,9 @@ async def iggy_client() -> IggyClient: # Authenticate await client.login_user("iggy", "iggy") - return client + yield client + + await _delete_test_resources(client) @pytest.fixture diff --git a/foreign/python/tests/test_stream.py b/foreign/python/tests/test_stream.py index 5e71a543a4..24e55e66cb 100644 --- a/foreign/python/tests/test_stream.py +++ b/foreign/python/tests/test_stream.py @@ -17,7 +17,7 @@ import pytest -from apache_iggy import IggyClient +from apache_iggy import IggyClient, SendMessage from .utils import get_server_config, wait_for_ping, wait_for_server @@ -203,3 +203,423 @@ async def test_create_stream_before_login_fails(self, unique_name): with pytest.raises(RuntimeError): await client.create_stream(unique_name()) + + +class TestGetStreams: + """Test listing streams via get_streams.""" + + @pytest.mark.asyncio + async def test_get_streams_returns_created_streams( + self, iggy_client: IggyClient, unique_name + ): + """Test get_streams returns every stream created during the test.""" + # Stream IDs can be reused after deletion, so creation order does not + # imply numeric ID order. The client fixture is session-scoped, so + # other tests may have created streams; assert on the ones created + # here instead of the full server view. + created = [unique_name(f"z{index}") for index in range(3, 0, -1)] + for name in created: + await iggy_client.create_stream(name) + + streams = await iggy_client.get_streams() + created_names = set(created) + mine = [stream for stream in streams if stream.name in created_names] + + assert {stream.name for stream in mine} == created_names + assert [stream.id for stream in mine] == sorted(stream.id for stream in mine) + assert all(stream.created_at > 0 for stream in mine) + assert all(stream.size_bytes == 0 for stream in mine) + assert all(stream.messages_count == 0 for stream in mine) + assert all(stream.topics_count == 0 for stream in mine) + + @pytest.mark.asyncio + async def test_get_streams_reflects_topic_count( + self, iggy_client: IggyClient, unique_name + ): + """Test get_streams reports the topic count for a listed stream.""" + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=1 + ) + + streams = await iggy_client.get_streams() + listed = next( + (stream for stream in streams if stream.name == stream_name), None + ) + assert listed is not None + assert listed.topics_count == 1 + + @pytest.mark.asyncio + async def test_get_streams_returns_same_result_when_called_repeatedly( + self, iggy_client: IggyClient, unique_name + ): + """Test repeated get_streams calls return an identically ordered view.""" + await iggy_client.create_stream(unique_name()) + + first = await iggy_client.get_streams() + second = await iggy_client.get_streams() + assert [stream.id for stream in first] == [stream.id for stream in second] + assert [stream.name for stream in first] == [stream.name for stream in second] + + @pytest.mark.asyncio + async def test_get_streams_requires_connection_and_auth(self): + """Test get_streams fails both before connecting and before logging in.""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient(f"{host}:{port}") + with pytest.raises(RuntimeError): + await client.get_streams() + + await client.connect() + with pytest.raises(RuntimeError): + await client.get_streams() + + +class TestUpdateStream: + """Test updating streams via update_stream.""" + + @pytest.mark.asyncio + async def test_update_stream_renames_stream( + self, iggy_client: IggyClient, unique_name + ): + """Test update_stream renames a stream; old name no longer resolves.""" + stream_name = unique_name() + new_name = unique_name() + + await iggy_client.create_stream(stream_name) + before = await iggy_client.get_stream(stream_name) + assert before is not None + + await iggy_client.update_stream(stream_id=stream_name, name=new_name) + + renamed = await iggy_client.get_stream(new_name) + assert renamed is not None + assert renamed.name == new_name + assert renamed.id == before.id + + old = await iggy_client.get_stream(stream_name) + assert old is None + + @pytest.mark.asyncio + async def test_update_stream_by_numeric_id( + self, iggy_client: IggyClient, unique_name + ): + """Test update_stream accepts a numeric stream id.""" + stream_name = unique_name() + new_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + + await iggy_client.update_stream(stream_id=stream.id, name=new_name) + + renamed = await iggy_client.get_stream(new_name) + assert renamed is not None + assert renamed.name == new_name + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "new_name", + [ + pytest.param("a", id="one-byte"), + pytest.param("a" * 255, id="255-byte-ascii"), + pytest.param(("é" * 127) + "a", id="255-byte-utf8"), + ], + ) + async def test_update_stream_accepts_name_boundaries( + self, iggy_client: IggyClient, unique_name, new_name: str + ): + """Test update_stream accepts names at the UTF-8 byte boundaries.""" + stream_name = unique_name() + await iggy_client.create_stream(stream_name) + + await iggy_client.update_stream(stream_id=stream_name, name=new_name) + + renamed = await iggy_client.get_stream(new_name) + assert renamed is not None + assert renamed.name == new_name + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "new_name", + [ + pytest.param("", id="empty"), + pytest.param("a" * 256, id="256-byte-ascii"), + pytest.param("é" * 128, id="256-byte-utf8"), + pytest.param(("😀" * 63) + "aaaa", id="256-byte-four-byte-utf8"), + ], + ) + async def test_update_stream_rejects_invalid_name_boundaries( + self, iggy_client: IggyClient, unique_name, new_name: str + ): + """Test update_stream rejects empty and 256-byte names.""" + stream_name = unique_name() + await iggy_client.create_stream(stream_name) + + with pytest.raises(RuntimeError): + await iggy_client.update_stream(stream_id=stream_name, name=new_name) + + @pytest.mark.asyncio + async def test_update_stream_applies_repeated_updates( + self, iggy_client: IggyClient, unique_name + ): + """Test successive update_stream calls each take effect.""" + stream_name = unique_name() + first_rename = unique_name() + second_rename = unique_name() + + await iggy_client.create_stream(stream_name) + + await iggy_client.update_stream(stream_id=stream_name, name=first_rename) + after_first = await iggy_client.get_stream(first_rename) + assert after_first is not None + assert after_first.name == first_rename + assert await iggy_client.get_stream(stream_name) is None + + await iggy_client.update_stream(stream_id=first_rename, name=second_rename) + after_second = await iggy_client.get_stream(second_rename) + assert after_second is not None + assert after_second.name == second_rename + assert await iggy_client.get_stream(first_rename) is None + + @pytest.mark.asyncio + async def test_update_nonexistent_stream_fails( + self, iggy_client: IggyClient, unique_name + ): + """Test update_stream raises for a non-existent stream.""" + with pytest.raises(RuntimeError): + await iggy_client.update_stream(stream_id=unique_name(), name=unique_name()) + + @pytest.mark.asyncio + async def test_update_stream_to_existing_name_fails( + self, iggy_client: IggyClient, unique_name + ): + """Test update_stream rejects renaming a stream to a name already in use.""" + first_stream = unique_name() + second_stream = unique_name() + + await iggy_client.create_stream(first_stream) + await iggy_client.create_stream(second_stream) + streams = await iggy_client.get_streams() + before = next(stream for stream in streams if stream.name == second_stream) + before_metadata = ( + before.id, + before.created_at, + before.name, + before.size_bytes, + before.messages_count, + before.topics_count, + ) + + with pytest.raises(RuntimeError): + await iggy_client.update_stream(stream_id=second_stream, name=first_stream) + + streams = await iggy_client.get_streams() + after = next(stream for stream in streams if stream.id == before.id) + assert ( + after.id, + after.created_at, + after.name, + after.size_bytes, + after.messages_count, + after.topics_count, + ) == before_metadata + + @pytest.mark.asyncio + async def test_update_stream_requires_connection_and_auth(self, unique_name): + """Test update_stream fails both before connecting and before logging in.""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient(f"{host}:{port}") + with pytest.raises(RuntimeError): + await client.update_stream(stream_id=unique_name(), name=unique_name()) + + await client.connect() + with pytest.raises(RuntimeError): + await client.update_stream(stream_id=unique_name(), name=unique_name()) + + +class TestDeleteStream: + """Test deleting streams via delete_stream.""" + + @pytest.mark.asyncio + async def test_delete_stream_removes_stream( + self, iggy_client: IggyClient, unique_name + ): + """Test delete_stream removes the stream so it no longer resolves.""" + stream_name = unique_name() + + await iggy_client.create_stream(stream_name) + + await iggy_client.delete_stream(stream_name) + + assert await iggy_client.get_stream(stream_name) is None + + @pytest.mark.asyncio + async def test_delete_stream_by_numeric_id( + self, iggy_client: IggyClient, unique_name + ): + """Test delete_stream accepts a numeric stream id.""" + stream_name = unique_name() + + await iggy_client.create_stream(stream_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + + await iggy_client.delete_stream(stream.id) + + assert await iggy_client.get_stream(stream_name) is None + + @pytest.mark.asyncio + async def test_delete_stream_leaves_other_streams( + self, iggy_client: IggyClient, unique_name + ): + """Test delete_stream removes only the targeted stream.""" + stream_to_delete = unique_name() + stream_to_keep = unique_name() + + await iggy_client.create_stream(stream_to_delete) + await iggy_client.create_stream(stream_to_keep) + + await iggy_client.delete_stream(stream_to_delete) + + assert await iggy_client.get_stream(stream_to_delete) is None + kept = await iggy_client.get_stream(stream_to_keep) + assert kept is not None + assert kept.name == stream_to_keep + + @pytest.mark.asyncio + async def test_delete_nonexistent_stream_fails( + self, iggy_client: IggyClient, unique_name + ): + """Test delete_stream raises for a non-existent stream.""" + with pytest.raises(RuntimeError): + await iggy_client.delete_stream(unique_name()) + + @pytest.mark.asyncio + async def test_delete_stream_twice_fails_second_time( + self, iggy_client: IggyClient, unique_name + ): + """Test deleting an already-deleted stream raises on the second call.""" + stream_name = unique_name() + + await iggy_client.create_stream(stream_name) + + await iggy_client.delete_stream(stream_name) + with pytest.raises(RuntimeError): + await iggy_client.delete_stream(stream_name) + + @pytest.mark.asyncio + async def test_delete_stream_requires_connection_and_auth(self, unique_name): + """Test delete_stream fails both before connecting and before logging in.""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient(f"{host}:{port}") + with pytest.raises(RuntimeError): + await client.delete_stream(unique_name()) + + await client.connect() + with pytest.raises(RuntimeError): + await client.delete_stream(unique_name()) + + +class TestPurgeStream: + """Test purging stream messages via purge_stream.""" + + @pytest.mark.asyncio + async def test_purge_stream_clears_messages_but_keeps_stream( + self, iggy_client: IggyClient, unique_name + ): + """Test purge_stream empties the stream while leaving it in place.""" + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=1 + ) + + messages = [SendMessage(f"payload-{index}") for index in range(5)] + await iggy_client.send_messages(stream_name, topic_name, 0, messages) + + before = await iggy_client.get_stream(stream_name) + assert before is not None + assert before.messages_count == 5 + + await iggy_client.purge_stream(stream_name) + + after = await iggy_client.get_stream(stream_name) + # Purging clears messages only; the stream itself survives (purge is + # not delete) and keeps its identity and topics. + assert after is not None + assert after.messages_count == 0 + assert after.id == before.id + assert after.name == before.name + assert after.topics_count == before.topics_count + + @pytest.mark.asyncio + async def test_purge_empty_stream_succeeds( + self, iggy_client: IggyClient, unique_name + ): + """Test purge_stream is a no-op on a stream with no messages.""" + stream_name = unique_name() + + await iggy_client.create_stream(stream_name) + + await iggy_client.purge_stream(stream_name) + + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + assert stream.messages_count == 0 + + @pytest.mark.asyncio + async def test_purge_stream_is_idempotent_when_called_repeatedly( + self, iggy_client: IggyClient, unique_name + ): + """Test purge_stream succeeds when called repeatedly on the same stream.""" + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + await iggy_client.create_topic( + stream=stream_name, name=topic_name, partitions_count=1 + ) + + messages = [SendMessage(f"payload-{index}") for index in range(5)] + await iggy_client.send_messages(stream_name, topic_name, 0, messages) + + await iggy_client.purge_stream(stream_name) + await iggy_client.purge_stream(stream_name) + + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + assert stream.messages_count == 0 + + @pytest.mark.asyncio + async def test_purge_nonexistent_stream_fails( + self, iggy_client: IggyClient, unique_name + ): + """Test purge_stream raises for a non-existent stream.""" + with pytest.raises(RuntimeError): + await iggy_client.purge_stream(unique_name()) + + @pytest.mark.asyncio + async def test_purge_stream_requires_connection_and_auth(self, unique_name): + """Test purge_stream fails both before connecting and before logging in.""" + host, port = get_server_config() + wait_for_server(host, port) + + client = IggyClient(f"{host}:{port}") + with pytest.raises(RuntimeError): + await client.purge_stream(unique_name()) + + await client.connect() + with pytest.raises(RuntimeError): + await client.purge_stream(unique_name()) diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock index 55960c278d..ab9c4b5de9 100644 --- a/foreign/python/uv.lock +++ b/foreign/python/uv.lock @@ -23,7 +23,6 @@ all = [ { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-timeout" }, - { name = "pytest-xdist" }, { name = "ruff" }, { name = "testcontainers" }, ] @@ -37,7 +36,6 @@ testing = [ { name = "pytest-asyncio" }, { name = "pytest-cov" }, { name = "pytest-timeout" }, - { name = "pytest-xdist" }, ] testing-docker = [ { name = "testcontainers" }, @@ -57,8 +55,6 @@ requires-dist = [ { name = "pytest-cov", marker = "extra == 'testing'", specifier = ">=4.0,<8.0" }, { name = "pytest-timeout", marker = "extra == 'all'", specifier = ">=2.0,<3.0" }, { name = "pytest-timeout", marker = "extra == 'testing'", specifier = ">=2.0,<3.0" }, - { name = "pytest-xdist", marker = "extra == 'all'", specifier = ">=3.0,<4.0" }, - { name = "pytest-xdist", marker = "extra == 'testing'", specifier = ">=3.0,<4.0" }, { name = "ruff", marker = "extra == 'all'", specifier = ">=0.1.0,<1.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0,<1.0" }, { name = "testcontainers", marker = "extra == 'all'", specifier = ">=3.7.0,<5.0" }, @@ -342,15 +338,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8a/0e/97c33bf5009bdbac74fd2beace167cab3f978feb69cc36f1ef79360d6c4e/exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598", size = 16740, upload-time = "2025-11-21T23:01:53.443Z" }, ] -[[package]] -name = "execnet" -version = "2.1.2" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/bf/89/780e11f9588d9e7128a3f87788354c7946a9cbb1401ad38a48c4db9a4f07/execnet-2.1.2.tar.gz", hash = "sha256:63d83bfdd9a23e35b9c6a3261412324f964c2ec8dcd8d3c6916ee9373e0befcd", size = 166622, upload-time = "2025-11-12T09:56:37.75Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" }, -] - [[package]] name = "idna" version = "3.15" @@ -497,19 +484,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/b6/3127540ecdf1464a00e5a01ee60a1b09175f6913f0644ac748494d9c4b21/pytest_timeout-2.4.0-py3-none-any.whl", hash = "sha256:c42667e5cdadb151aeb5b26d114aff6bdf5a907f176a007a30b940d3d865b5c2", size = 14382, upload-time = "2025-05-05T19:44:33.502Z" }, ] -[[package]] -name = "pytest-xdist" -version = "3.8.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "execnet" }, - { name = "pytest" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/78/b4/439b179d1ff526791eb921115fca8e44e596a13efeda518b9d845a619450/pytest_xdist-3.8.0.tar.gz", hash = "sha256:7e578125ec9bc6050861aa93f2d59f1d8d085595d6551c2c90b6f4fad8d3a9f1", size = 88069, upload-time = "2025-07-01T13:30:59.346Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396, upload-time = "2025-07-01T13:30:56.632Z" }, -] - [[package]] name = "python-dotenv" version = "1.2.2"