[PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm opened a new pull request, #3432:
URL: https://github.com/apache/iggy/pull/3432

   ## Which issue does this PR address?
   
   <!--
   We generally require a GitHub issue for all bug fixes and enhancements. Link it with GitHub syntax, keep the line that applies and delete the other:
   - `Closes #123` auto-closes the issue when this PR merges (full fix).
   - `Relates to #123` links without closing (partial or related work).
   -->
   
   Closes #3317
   Relates to #
   
   ## Rationale
   
   <!--
   Why is this change needed? If the issue explains it well, a one-liner is fine.
   -->
   The python sdk only had a few happy path tests. This PR adds extensive testing for the sdk. Also explicitly maps errors, replacing unwraps.
   ## What changed?
   
   <!--
   2-4 sentences. Problem first (before), then solution (after).
   
   GOOD:
   
   "Messages were unavailable when background message_saver committed the
   journal and started async disk I/O before completion. Polling during
   this window found neither journal nor disk data.
   
   The fix freezes journal batches in the in-flight buffer before async persist."
   
   GOOD:
   
   "When many small messages accumulate in the journal, the flush passes
   thousands of IO vectors to writev(), exceeding IOV_MAX (1024 on Linux)."
   
   BAD:
   - Walls of text
   - "This PR adds..." (we can see the diff)
   -->
   
   - added extensive tests, divided into files. Division was done to separate concerns. 
   - removed all usage of unwrap in the the rust code.
   - moved code of iterator.rs into consumer.rs.  No need for an extra file for a small amount of code. 
   ## Local Execution
   
   - Passed 
   - Pre-commit hooks 
   
   <!--
   You must run your code locally before submitting.
   "Relying on CI" is not acceptable - PRs from authors who haven't run the code will be closed.
   
   Did you have `prek` installed? It runs automatically on commit and covers all project languages. See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md).
   -->
   
   ## AI Usage
   
   If AI tools were used, please answer:
   1. Which tools? codex
   2. Scope of usage? code generation, investigation
   3. How did you verify the generated code works correctly? read through and reasoned about the code, executed on my machine
   4. Can you explain every line of the code if asked? Yes
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3396221863


##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_first_reads_existing_messages(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages reads messages that already exist with First."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [
+            f"Existing message test {i} - {unique_name()}" for i in range(3)
+        ]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_last_reads_last_existing_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Last starts from the last existing message."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Last message test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Last(),
+            1,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message)
+            break
+
+        assert [message.payload().decode() for message in received_messages] == [
+            test_messages[-1]
+        ]
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_last_and_batch_length_two_reads_last_two_messages(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Last and batch_length=2 reads the last two."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Last message test {i} - {unique_name()}" for i in range(4)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Last(),
+            2,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == 2:
+                break
+
+        assert received_messages == test_messages[-2:]
+
+    @pytest.mark.asyncio
+    async def test_consume_messages_with_last_reads_last_existing_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test consume_messages with Last starts from the last existing message."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Last message test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Last(),
+            1,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        assert [message.payload().decode() for message in received_messages] == [
+            test_messages[-1]
+        ]
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_offset_starts_at_exact_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Offset starts from the requested message offset."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Offset start test {i} - {unique_name()}" for i in range(4)]
+        initial_messages = []
+        offset_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_offset = initial_messages[2].offset()
+
+        offset_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Offset(value=start_offset),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in offset_consumer.iter_messages():
+            offset_messages.append(message)
+            if len(offset_messages) == len(test_messages) - 2:
+                break
+
+        assert offset_messages[0].offset() == start_offset
+        assert [
+            message.payload().decode() for message in offset_messages
+        ] == test_messages[2:]
+        assert (
+            offset_consumer.get_last_consumed_offset(partition_id)
+            == offset_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_consume_messages_with_offset_starts_at_exact_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test consume_messages with Offset starts from the requested offset."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Offset start test {i} - {unique_name()}" for i in range(4)]
+        initial_messages = []
+        offset_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_offset = initial_messages[2].offset()
+
+        offset_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Offset(value=start_offset),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            offset_messages.append(message)
+            if len(offset_messages) == len(test_messages) - 2:
+                shutdown_event.set()
+
+        await offset_consumer.consume_messages(take, shutdown_event)
+
+        assert offset_messages[0].offset() == start_offset
+        assert [
+            message.payload().decode() for message in offset_messages
+        ] == test_messages[2:]
+        assert (
+            offset_consumer.get_last_consumed_offset(partition_id)
+            == offset_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_timestamp_starts_at_or_after_timestamp(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Timestamp starts on or after that timestamp."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [
+            f"Timestamp start test {i} - {unique_name()}" for i in range(3)
+        ]
+        initial_messages = []
+        timestamp_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        for message in test_messages:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message)],
+            )
+            await asyncio.sleep(0.01)
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_timestamp = initial_messages[1].timestamp()
+
+        timestamp_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Timestamp(value=start_timestamp),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in timestamp_consumer.iter_messages():
+            timestamp_messages.append(message)
+            if len(timestamp_messages) == len(test_messages) - 1:
+                break
+
+        assert timestamp_messages[0].timestamp() >= start_timestamp
+        assert [
+            message.payload().decode() for message in timestamp_messages
+        ] == test_messages[1:]
+        assert (
+            timestamp_consumer.get_last_consumed_offset(partition_id)
+            == timestamp_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_consume_messages_with_timestamp_starts_at_or_after_timestamp(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test consume_messages with Timestamp starts on or after that timestamp."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [
+            f"Timestamp start test {i} - {unique_name()}" for i in range(3)
+        ]
+        initial_messages = []
+        timestamp_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        for message in test_messages:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message)],
+            )
+            await asyncio.sleep(0.01)
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_timestamp = initial_messages[1].timestamp()
+
+        timestamp_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Timestamp(value=start_timestamp),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            timestamp_messages.append(message)
+            if len(timestamp_messages) == len(test_messages) - 1:
+                shutdown_event.set()
+
+        await timestamp_consumer.consume_messages(take, shutdown_event)
+
+        assert timestamp_messages[0].timestamp() >= start_timestamp
+        assert [
+            message.payload().decode() for message in timestamp_messages
+        ] == test_messages[1:]
+        assert (
+            timestamp_consumer.get_last_consumed_offset(partition_id)
+            == timestamp_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_updates_last_consumed_offset(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages updates the last consumed offset while iterating."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Iter offset test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        iterator = consumer.iter_messages()
+        async for message in iterator:
+            received_messages.append(message)
+            if len(received_messages) == 2:
+                break
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        async for message in iterator:
+            received_messages.append(message)
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_reads_across_multiple_batches(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages continues across polls when batch_length is small."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Batch test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            2,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("allow_replay", [False, True])
+    async def test_store_offset_respects_allow_replay(
+        self,
+        iggy_client: IggyClient,
+        unique_name,
+        allow_replay: bool,
+    ):
+        """Test rewinding offsets only re-consumes when replay is enabled."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Replay test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+        replayed_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+            allow_replay=allow_replay,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message)
+            if len(received_messages) == len(test_messages):
+                break
+
+        offsets = [message.offset() for message in received_messages]
+
+        await consumer.store_offset(offsets[-1], partition_id)
+        assert consumer.get_last_stored_offset(partition_id) == offsets[-1]
+
+        await consumer.store_offset(offsets[-2], partition_id)
+        expected_stored_offset = offsets[-2] if allow_replay else offsets[-1]
+        assert consumer.get_last_stored_offset(partition_id) == expected_stored_offset
+
+        replay_consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            1,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+            allow_replay=allow_replay,
+        )
+
+        if allow_replay:
+            async for message in replay_consumer.iter_messages():

Review Comment:
   Done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384076628


##########
foreign/python/tests/test_stream.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestStreamOperations:
+    """Test stream creation, retrieval, and management."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "stream_name",
+        [
+            "a",
+            "stream-name",
+            "stream_name.with.mixed-CHARS123",
+            " leading-space",
+            "trailing-space ",
+            "multiple  spaces  inside",
+            "name/with/slash",
+            "name:with:colons",
+            "name.with.dots",
+            "   ",
+            "a" * 255,
+            ("é" * 126) + "abc",
+            ("한" * 84) + "abc",
+            ("漢" * 84) + "abc",
+            ("あ" * 84) + "abc",
+            ("😀" * 62) + "abcdefg",
+        ],
+    )
+    async def test_create_and_get_stream(

Review Comment:
   rewrote unique_name so that it can accept minimum and maximum bytes range, and a prefix as well. set reasonable defaults for these.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384362511


##########
foreign/python/src/identifier.rs:
##########
@@ -30,20 +33,33 @@ pub(crate) enum PyIdentifier {
 }
 impl_stub_type!(PyIdentifier = String | isize);
 
-impl From<PyIdentifier> for Identifier {
-    fn from(py_identifier: PyIdentifier) -> Self {
+impl TryFrom<PyIdentifier> for Identifier {
+    type Error = PyErr;
+
+    fn try_from(py_identifier: PyIdentifier) -> Result<Self, Self::Error> {
         match py_identifier {
-            PyIdentifier::String(s) => Identifier::from_str(&s).unwrap(),
-            PyIdentifier::Int(i) => Identifier::numeric(i).unwrap(),
+            PyIdentifier::String(s) => Identifier::from_str(&s)
+                .map_err(|e| PyErr::new::<PyValueError, _>(format!("{e:?}"))),

Review Comment:
   Done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["hubcio (via GitHub)" <gi...@apache.org>]

hubcio commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3378484176


##########
foreign/python/tests/test_stream.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestStreamOperations:
+    """Test stream creation, retrieval, and management."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "stream_name",
+        [
+            "a",
+            "stream-name",
+            "stream_name.with.mixed-CHARS123",
+            " leading-space",
+            "trailing-space ",
+            "multiple  spaces  inside",
+            "name/with/slash",
+            "name:with:colons",
+            "name.with.dots",
+            "   ",
+            "a" * 255,
+            ("é" * 126) + "abc",
+            ("한" * 84) + "abc",
+            ("漢" * 84) + "abc",
+            ("あ" * 84) + "abc",
+            ("😀" * 62) + "abcdefg",
+        ],
+    )
+    async def test_create_and_get_stream(

Review Comment:
   this test is parametrized with fixed literal stream names and nothing deletes the streams afterwards (the conftest TODO notes cleanup is blocked on `delete_stream`). iggy persists streams and rejects duplicates - `test_duplicate_stream_creation` below proves it - so a second run against a non-fresh server makes all these cases fail with `StreamNameAlreadyExists`. every other stream-creating test here already uses `unique_name()`. prefix these literals with it too (truncate to <=255 bytes to keep the boundary-length case).



##########
foreign/python/tests/test_stream.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestStreamOperations:
+    """Test stream creation, retrieval, and management."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "stream_name",
+        [
+            "a",
+            "stream-name",
+            "stream_name.with.mixed-CHARS123",
+            " leading-space",
+            "trailing-space ",
+            "multiple  spaces  inside",
+            "name/with/slash",
+            "name:with:colons",
+            "name.with.dots",
+            "   ",
+            "a" * 255,
+            ("é" * 126) + "abc",
+            ("한" * 84) + "abc",
+            ("漢" * 84) + "abc",
+            ("あ" * 84) + "abc",
+            ("😀" * 62) + "abcdefg",
+        ],
+    )
+    async def test_create_and_get_stream(
+        self, iggy_client: IggyClient, stream_name: str
+    ):
+        """Test stream creation and retrieval."""
+        await iggy_client.create_stream(stream_name)
+
+        stream = await iggy_client.get_stream(stream_name)
+        assert stream is not None
+        assert stream.name == stream_name
+        assert stream.id >= 0

Review Comment:
   `stream.id >= 0` is always true - `id()` returns a `u32`, so this can never fail and validates nothing. use `> 0` (iggy ids start at 1) or drop the line. same one at `test_topic.py:66`.



##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():

Review Comment:
   this loop only ends by counting up to `len(test_messages)`. a live `IggyConsumer` stream never returns `None` - on an empty poll the sdk re-arms and parks, there's no `Ready(None)` path - so if the server ever delivers fewer messages than expected (lost write, rebalance, slow poll) the loop awaits forever. and there's no per-test timeout armed: pytest-timeout is declared but no `timeout=` is set in pyproject and CI passes no `--timeout`, so the only backstop is the 60-min job timeout. that turns a quick assertion failure into a long opaque hang. same pattern at lines 402, 446, 495, 591, 609, 661, 736, 754, 810, 884, 941, 991 and 1018. either wrap each bounded loop in `asyncio.timeout(...)`, or set a suite-wide `timeout=` in `[tool.pytest.ini_options]`.



##########
foreign/python/tests/test_message_operations.py:
##########
@@ -0,0 +1,778 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+import uuid
+
+import pytest
+
+from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import SendMessage as Message
+
+
+class TestMessageOperations:
+    """Test message sending, polling, and processing."""
+
+    @pytest.mark.asyncio
+    async def test_send_and_poll_messages(self, iggy_client: IggyClient, unique_name):
+        """Test basic message sending and polling."""
+        unique_id = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Test message {i} - {unique_id}" for i in range(1, 4)]
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name, name=topic_name, partitions_count=1
+        )
+
+        messages = [Message(msg) for msg in test_messages]
+        await iggy_client.send_messages(
+            stream=stream_name,
+            topic=topic_name,
+            partitioning=partition_id,
+            messages=messages,
+        )
+
+        polled_messages = await iggy_client.poll_messages(
+            stream=stream_name,
+            topic=topic_name,
+            partition_id=partition_id,
+            polling_strategy=PollingStrategy.First(),
+            count=10,
+            auto_commit=True,
+        )
+
+        assert len(polled_messages) >= len(test_messages)

Review Comment:
   `>=` tolerates over-delivery - a fresh unique topic with 3 sent messages should return exactly 3, so a duplicate-publish regression would slip past this. and the `if i < len(polled_messages)` guard below is dead, since you just asserted `len >= 3` and `i` is in `range(3)`. use the full-list equality you already use elsewhere in this file: `assert [m.payload().decode() for m in polled_messages] == test_messages`. same at lines 101-106.



##########
foreign/python/tests/test_topic.py:
##########
@@ -0,0 +1,579 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestTopicOperations:
+    """Test topic creation, retrieval, and management."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "topic_name",
+        [
+            "a",
+            "stream-name",
+            "stream_name.with.mixed-CHARS123",
+            " leading-space",
+            "trailing-space ",
+            "multiple  spaces  inside",
+            "name/with/slash",
+            "name:with:colons",
+            "name.with.dots",
+            "   ",
+            "a" * 255,
+            ("é" * 126) + "abc",
+            ("한" * 84) + "abc",
+            ("漢" * 84) + "abc",
+            ("あ" * 84) + "abc",
+            ("😀" * 62) + "abcdefg",
+        ],
+    )
+    async def test_create_and_get_topic(

Review Comment:
   no teardown here - each parametrized case creates a stream and topic and never removes them, so server state grows on every run. unique names keep it correct, it's just unbounded accumulation against a long-lived server. fine to leave until `delete_stream`/`delete_topic` are bound, but worth a cleanup fixture once they are.



##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(

Review Comment:
   this create_stream + create_topic + consumer_group setup is copy-pasted across ~24 tests in this file, each repeating the same `poll_interval` and `auto_commit` kwargs. worth pulling into a fixture/helper that returns a configured consumer and parametrizing only what varies (polling strategy, batch size, auto_commit, allow_replay).



##########
foreign/python/src/identifier.rs:
##########
@@ -30,20 +33,33 @@ pub(crate) enum PyIdentifier {
 }
 impl_stub_type!(PyIdentifier = String | isize);
 
-impl From<PyIdentifier> for Identifier {
-    fn from(py_identifier: PyIdentifier) -> Self {
+impl TryFrom<PyIdentifier> for Identifier {
+    type Error = PyErr;
+
+    fn try_from(py_identifier: PyIdentifier) -> Result<Self, Self::Error> {
         match py_identifier {
-            PyIdentifier::String(s) => Identifier::from_str(&s).unwrap(),
-            PyIdentifier::Int(i) => Identifier::numeric(i).unwrap(),
+            PyIdentifier::String(s) => Identifier::from_str(&s)
+                .map_err(|e| PyErr::new::<PyValueError, _>(format!("{e:?}"))),

Review Comment:
   `{e:?}` formats with Debug, so python sees the variant name (`InvalidIdentifier`) instead of the clean Display message (`Invalid identifier`). `{e}` reads better for a user-facing exception. one heads-up: a few of the new tests assert on the Debug variant-name substrings (`StreamNameAlreadyExists`, `TooManyPartitions`), so switching to `{e}` means updating those asserts too. it also matches the existing `{e:?}` convention in client.rs, so it's a wash unless you sweep every site. same pattern at `send_message.rs:67,74` and `client.rs:63`. minor either way.



##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_first_reads_existing_messages(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages reads messages that already exist with First."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [
+            f"Existing message test {i} - {unique_name()}" for i in range(3)
+        ]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_last_reads_last_existing_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Last starts from the last existing message."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Last message test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Last(),
+            1,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message)
+            break
+
+        assert [message.payload().decode() for message in received_messages] == [
+            test_messages[-1]
+        ]
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_last_and_batch_length_two_reads_last_two_messages(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Last and batch_length=2 reads the last two."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Last message test {i} - {unique_name()}" for i in range(4)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Last(),
+            2,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == 2:
+                break
+
+        assert received_messages == test_messages[-2:]
+
+    @pytest.mark.asyncio
+    async def test_consume_messages_with_last_reads_last_existing_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test consume_messages with Last starts from the last existing message."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Last message test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Last(),
+            1,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        assert [message.payload().decode() for message in received_messages] == [
+            test_messages[-1]
+        ]
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_offset_starts_at_exact_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Offset starts from the requested message offset."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Offset start test {i} - {unique_name()}" for i in range(4)]
+        initial_messages = []
+        offset_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_offset = initial_messages[2].offset()
+
+        offset_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Offset(value=start_offset),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in offset_consumer.iter_messages():
+            offset_messages.append(message)
+            if len(offset_messages) == len(test_messages) - 2:
+                break
+
+        assert offset_messages[0].offset() == start_offset
+        assert [
+            message.payload().decode() for message in offset_messages
+        ] == test_messages[2:]
+        assert (
+            offset_consumer.get_last_consumed_offset(partition_id)
+            == offset_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_consume_messages_with_offset_starts_at_exact_message(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test consume_messages with Offset starts from the requested offset."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Offset start test {i} - {unique_name()}" for i in range(4)]
+        initial_messages = []
+        offset_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_offset = initial_messages[2].offset()
+
+        offset_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Offset(value=start_offset),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            offset_messages.append(message)
+            if len(offset_messages) == len(test_messages) - 2:
+                shutdown_event.set()
+
+        await offset_consumer.consume_messages(take, shutdown_event)
+
+        assert offset_messages[0].offset() == start_offset
+        assert [
+            message.payload().decode() for message in offset_messages
+        ] == test_messages[2:]
+        assert (
+            offset_consumer.get_last_consumed_offset(partition_id)
+            == offset_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_with_timestamp_starts_at_or_after_timestamp(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages with Timestamp starts on or after that timestamp."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [
+            f"Timestamp start test {i} - {unique_name()}" for i in range(3)
+        ]
+        initial_messages = []
+        timestamp_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        for message in test_messages:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message)],
+            )
+            await asyncio.sleep(0.01)
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_timestamp = initial_messages[1].timestamp()
+
+        timestamp_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Timestamp(value=start_timestamp),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in timestamp_consumer.iter_messages():
+            timestamp_messages.append(message)
+            if len(timestamp_messages) == len(test_messages) - 1:
+                break
+
+        assert timestamp_messages[0].timestamp() >= start_timestamp
+        assert [
+            message.payload().decode() for message in timestamp_messages
+        ] == test_messages[1:]
+        assert (
+            timestamp_consumer.get_last_consumed_offset(partition_id)
+            == timestamp_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_consume_messages_with_timestamp_starts_at_or_after_timestamp(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test consume_messages with Timestamp starts on or after that timestamp."""
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [
+            f"Timestamp start test {i} - {unique_name()}" for i in range(3)
+        ]
+        initial_messages = []
+        timestamp_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        for message in test_messages:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message)],
+            )
+            await asyncio.sleep(0.01)
+
+        first_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in first_consumer.iter_messages():
+            initial_messages.append(message)
+            if len(initial_messages) == len(test_messages):
+                break
+
+        start_timestamp = initial_messages[1].timestamp()
+
+        timestamp_consumer = await iggy_client.consumer_group(
+            unique_name(),
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Timestamp(value=start_timestamp),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            timestamp_messages.append(message)
+            if len(timestamp_messages) == len(test_messages) - 1:
+                shutdown_event.set()
+
+        await timestamp_consumer.consume_messages(take, shutdown_event)
+
+        assert timestamp_messages[0].timestamp() >= start_timestamp
+        assert [
+            message.payload().decode() for message in timestamp_messages
+        ] == test_messages[1:]
+        assert (
+            timestamp_consumer.get_last_consumed_offset(partition_id)
+            == timestamp_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_updates_last_consumed_offset(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages updates the last consumed offset while iterating."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Iter offset test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        iterator = consumer.iter_messages()
+        async for message in iterator:
+            received_messages.append(message)
+            if len(received_messages) == 2:
+                break
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        async for message in iterator:
+            received_messages.append(message)
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_iter_messages_reads_across_multiple_batches(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test iter_messages continues across polls when batch_length is small."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Batch test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            2,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                break
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("allow_replay", [False, True])
+    async def test_store_offset_respects_allow_replay(
+        self,
+        iggy_client: IggyClient,
+        unique_name,
+        allow_replay: bool,
+    ):
+        """Test rewinding offsets only re-consumes when replay is enabled."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Replay test {i} - {unique_name()}" for i in range(3)]
+        received_messages = []
+        replayed_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+            allow_replay=allow_replay,
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():
+            received_messages.append(message)
+            if len(received_messages) == len(test_messages):
+                break
+
+        offsets = [message.offset() for message in received_messages]
+
+        await consumer.store_offset(offsets[-1], partition_id)
+        assert consumer.get_last_stored_offset(partition_id) == offsets[-1]
+
+        await consumer.store_offset(offsets[-2], partition_id)
+        expected_stored_offset = offsets[-2] if allow_replay else offsets[-1]
+        assert consumer.get_last_stored_offset(partition_id) == expected_stored_offset
+
+        replay_consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            1,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+            allow_replay=allow_replay,
+        )
+
+        if allow_replay:
+            async for message in replay_consumer.iter_messages():

Review Comment:
   the `allow_replay=True` branch reads one message with no timeout, while the `else` branch just below guards with `asyncio.wait_for(..., timeout=0.2)`. the assert expects the rewind to `offsets[-2]` to have taken effect. if that rewind silently failed - which is exactly the regression this test is meant to catch - the `Next` consumer has nothing past `offsets[-1]` and this loop parks forever instead of failing the assert. wrap it in the same `wait_for` guard as the else branch so a broken rewind fails loudly.



##########
foreign/python/src/consumer.rs:
##########
@@ -214,6 +217,37 @@ impl IggyConsumer {
     }
 }
 
+#[pyclass]
+pub struct ReceiveMessageIterator {
+    pub(crate) inner: Arc<Mutex<RustIggyConsumer>>,
+}
+
+#[pymethods]
+impl ReceiveMessageIterator {
+    pub fn __anext__<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
+        let inner = self.inner.clone();
+        future_into_py(py, async move {
+            let mut inner = inner.lock().await;
+            if let Some(message) = inner.next().await {
+                Ok(message
+                    .map(|m| ReceiveMessage {
+                        inner: m.message,
+                        partition_id: m.partition_id,
+                    })
+                    .map_err(|e| {
+                        PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("{e:?}"))
+                    })?)
+            } else {
+                Err(PyStopIteration::new_err("No more messages"))

Review Comment:
   `__anext__` is an async iterator, so on exhaustion it should raise `StopAsyncIteration`, not `StopIteration` - a `StopIteration` escaping a coroutine becomes a `RuntimeError` under PEP 479, and both the docstring and the `.pyi` stub already promise `StopAsyncIteration`. it's latent today since the sdk consumer stream never yields `None` for a live consumer, so this branch isn't actually reachable - but if that ever changes it surfaces as a confusing `RuntimeError`. swap to `PyStopAsyncIteration::new_err(...)`. (this block was moved verbatim from the old `iterator.rs`, not introduced here.)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384190415


##########
foreign/python/src/consumer.rs:
##########
@@ -214,6 +217,37 @@ impl IggyConsumer {
     }
 }
 
+#[pyclass]
+pub struct ReceiveMessageIterator {
+    pub(crate) inner: Arc<Mutex<RustIggyConsumer>>,
+}
+
+#[pymethods]
+impl ReceiveMessageIterator {
+    pub fn __anext__<'a>(&self, py: Python<'a>) -> PyResult<Bound<'a, PyAny>> {
+        let inner = self.inner.clone();
+        future_into_py(py, async move {
+            let mut inner = inner.lock().await;
+            if let Some(message) = inner.next().await {
+                Ok(message
+                    .map(|m| ReceiveMessage {
+                        inner: m.message,
+                        partition_id: m.partition_id,
+                    })
+                    .map_err(|e| {
+                        PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("{e:?}"))
+                    })?)
+            } else {
+                Err(PyStopIteration::new_err("No more messages"))

Review Comment:
   Done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384071157


##########
foreign/python/tests/test_stream.py:
##########
@@ -0,0 +1,192 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestStreamOperations:
+    """Test stream creation, retrieval, and management."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "stream_name",
+        [
+            "a",
+            "stream-name",
+            "stream_name.with.mixed-CHARS123",
+            " leading-space",
+            "trailing-space ",
+            "multiple  spaces  inside",
+            "name/with/slash",
+            "name:with:colons",
+            "name.with.dots",
+            "   ",
+            "a" * 255,
+            ("é" * 126) + "abc",
+            ("한" * 84) + "abc",
+            ("漢" * 84) + "abc",
+            ("あ" * 84) + "abc",
+            ("😀" * 62) + "abcdefg",
+        ],
+    )
+    async def test_create_and_get_stream(
+        self, iggy_client: IggyClient, stream_name: str
+    ):
+        """Test stream creation and retrieval."""
+        await iggy_client.create_stream(stream_name)
+
+        stream = await iggy_client.get_stream(stream_name)
+        assert stream is not None
+        assert stream.name == stream_name
+        assert stream.id >= 0

Review Comment:
   removed check



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["codecov[bot] (via GitHub)" <gi...@apache.org>]

codecov[bot] commented on PR #3432:
URL: https://github.com/apache/iggy/pull/3432#issuecomment-4642703193

   ## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/3432?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `87.30159%` with `8 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 74.48%. Comparing base ([`d254604`](https://app.codecov.io/gh/apache/iggy/commit/d254604d4878a14b8a7d2fdd1e199a8543f5e2e3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`66a44c2`](https://app.codecov.io/gh/apache/iggy/commit/66a44c27c9ec1a5a2d9165798affccc005f1821f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3432?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [foreign/python/src/consumer.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fconsumer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2NvbnN1bWVyLnJz) | 84.61% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [foreign/python/src/identifier.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fidentifier.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2lkZW50aWZpZXIucnM=) | 71.42% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master    #3432      +/-   ##
   ============================================
   + Coverage     74.43%   74.48%   +0.04%     
     Complexity      943      943              
   ============================================
     Files          1245     1244       -1     
     Lines        121293   121309      +16     
     Branches      97599    97599              
   ============================================
   + Hits          90285    90353      +68     
   + Misses        28054    28002      -52     
     Partials       2954     2954              
   ```
   
   | [Components](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.56% <ø> (ø)` | |
   | [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `58.44% <ø> (ø)` | |
   | [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.92% <ø> (ø)` | |
   | [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `89.74% <87.30%> (+8.67%)` | :arrow_up: |
   | [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.57% <ø> (ø)` | |
   | [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.35% <ø> (ø)` | |
   | [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/3432/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `40.20% <ø> (ø)` | |
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/3432?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [foreign/python/src/client.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fclient.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2NsaWVudC5ycw==) | `97.62% <100.00%> (+5.76%)` | :arrow_up: |
   | [foreign/python/src/lib.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2xpYi5ycw==) | `100.00% <ø> (ø)` | |
   | [foreign/python/src/send\_message.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fsend_message.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL3NlbmRfbWVzc2FnZS5ycw==) | `96.96% <100.00%> (+0.67%)` | :arrow_up: |
   | [foreign/python/src/consumer.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fconsumer.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2NvbnN1bWVyLnJz) | `80.58% <84.61%> (+19.16%)` | :arrow_up: |
   | [foreign/python/src/identifier.rs](https://app.codecov.io/gh/apache/iggy/pull/3432?src=pr&el=tree&filepath=foreign%2Fpython%2Fsrc%2Fidentifier.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Zm9yZWlnbi9weXRob24vc3JjL2lkZW50aWZpZXIucnM=) | `77.77% <71.42%> (-2.23%)` | :arrow_down: |
   
   ... and [1 file with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/3432/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   </details>
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384059538


##########
foreign/python/tests/test_topic.py:
##########
@@ -0,0 +1,579 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestTopicOperations:
+    """Test topic creation, retrieval, and management."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "topic_name",
+        [
+            "a",
+            "stream-name",
+            "stream_name.with.mixed-CHARS123",
+            " leading-space",
+            "trailing-space ",
+            "multiple  spaces  inside",
+            "name/with/slash",
+            "name:with:colons",
+            "name.with.dots",
+            "   ",
+            "a" * 255,
+            ("é" * 126) + "abc",
+            ("한" * 84) + "abc",
+            ("漢" * 84) + "abc",
+            ("あ" * 84) + "abc",
+            ("😀" * 62) + "abcdefg",
+        ],
+    )
+    async def test_create_and_get_topic(

Review Comment:
   TODO already in place



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on PR #3432:
URL: https://github.com/apache/iggy/pull/3432#issuecomment-4681425670

   /ready


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384325514


##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():

Review Comment:
   Used `asyncio.wait_for` instead of `asyncio.timeout` since `timeout` is only supported in/after python 3.11~



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3396189467


##########
foreign/python/tests/test_message_operations.py:
##########
@@ -0,0 +1,778 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+import uuid
+
+import pytest
+
+from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import SendMessage as Message
+
+
+class TestMessageOperations:
+    """Test message sending, polling, and processing."""
+
+    @pytest.mark.asyncio
+    async def test_send_and_poll_messages(self, iggy_client: IggyClient, unique_name):
+        """Test basic message sending and polling."""
+        unique_id = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Test message {i} - {unique_id}" for i in range(1, 4)]
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name, name=topic_name, partitions_count=1
+        )
+
+        messages = [Message(msg) for msg in test_messages]
+        await iggy_client.send_messages(
+            stream=stream_name,
+            topic=topic_name,
+            partitioning=partition_id,
+            messages=messages,
+        )
+
+        polled_messages = await iggy_client.poll_messages(
+            stream=stream_name,
+            topic=topic_name,
+            partition_id=partition_id,
+            polling_strategy=PollingStrategy.First(),
+            count=10,
+            auto_commit=True,
+        )
+
+        assert len(polled_messages) >= len(test_messages)

Review Comment:
   DOne



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on PR #3432:
URL: https://github.com/apache/iggy/pull/3432#issuecomment-4642655455

   <!-- iggy-pr-triage-welcome -->
   Thanks for the PR. It is labeled `S-waiting-on-review` and queued for review.
   
   Slash commands (own line, regular comment) move it around the queue:
   
   - `/ready` - back to `S-waiting-on-review` after addressing feedback
   - `/author` - flip to `S-waiting-on-author` while you finish changes
   - `/request-review @user-or-team` - request a reviewer
   
   See [CONTRIBUTING.md](https://github.com/apache/iggy/blob/master/CONTRIBUTING.md#pr-triage-commands) for details.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3383912843


##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():

Review Comment:
   Added per-test timeout of 30s as well as 5s for the tests that check `len(test_messages)`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3384198128


##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(

Review Comment:
   Will create a helper function in the future test fixture. Added a TODO



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["hubcio (via GitHub)" <gi...@apache.org>]

hubcio commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3394418791


##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():

Review Comment:
   what if we dropped 3.10 support?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on PR #3432:
URL: https://github.com/apache/iggy/pull/3432#issuecomment-4664859661

   Also removed the remaining unwraps from consumer.rs


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["slbotbm (via GitHub)" <gi...@apache.org>]

slbotbm commented on code in PR #3432:
URL: https://github.com/apache/iggy/pull/3432#discussion_r3396161754


##########
foreign/python/tests/test_consumer_group.py:
##########
@@ -0,0 +1,1318 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import asyncio
+from datetime import timedelta
+
+import pytest
+
+from apache_iggy import (
+    AutoCommit,
+    IggyClient,
+    PollingStrategy,
+    ReceiveMessage,
+)
+from apache_iggy import SendMessage as Message
+
+from .utils import get_server_config, wait_for_ping, wait_for_server
+
+
+class TestConsumerGroup:
+    """Test consumer groups."""
+
+    @pytest.mark.asyncio
+    async def test_consumer_group_metadata(self, iggy_client: IggyClient, unique_name):
+        """Test that metadata information can be read about the consumer group."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.name() == consumer_name
+        assert consumer.stream() == stream_name
+        assert consumer.topic() == topic_name
+        assert consumer.partition_id() == partition_id
+        assert consumer.get_last_consumed_offset(partition_id) is None
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+    @pytest.mark.asyncio
+    async def test_get_last_consumed_offset_updates_as_messages_are_consumed(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last consumed offset before, during, and after callback consumption."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [f"Offset test {i} - {unique_name()}" for i in range(2)]
+        final_message = f"Offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_consumed_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        assert (
+            consumer.get_last_consumed_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_get_last_stored_offset_updates_when_offsets_are_stored(
+        self, iggy_client: IggyClient, unique_name
+    ):
+        """Test last stored offset before, during, and after manual stores."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        first_batch_messages = [
+            f"Stored offset test {i} - {unique_name()}" for i in range(2)
+        ]
+        final_message = f"Stored offset test 2 - {unique_name()}"
+        received_messages = []
+        first_shutdown = asyncio.Event()
+        second_shutdown = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in first_batch_messages],
+        )
+
+        async def take_first_batch(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages):
+                first_shutdown.set()
+
+        await consumer.consume_messages(take_first_batch, first_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+        assert len(received_messages) == len(first_batch_messages)
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(final_message)],
+        )
+
+        async def take_remaining(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            if len(received_messages) == len(first_batch_messages) + 1:
+                second_shutdown.set()
+
+        await consumer.consume_messages(take_remaining, second_shutdown)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+    @pytest.mark.asyncio
+    async def test_delete_offset_behavior(self, iggy_client: IggyClient, unique_name):
+        """Test deleting a missing offset fails and a stored offset succeeds."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        shutdown_event = asyncio.Event()
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.First(),
+            10,
+            auto_commit=AutoCommit.Disabled(),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        with pytest.raises(RuntimeError, match=r"ConsumerOffsetNotFound\(0\)"):
+            await consumer.delete_offset(partition_id)
+        assert consumer.get_last_stored_offset(partition_id) is None
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(f"Delete offset test - {unique_name()}")],
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message)
+            shutdown_event.set()
+
+        await consumer.consume_messages(take, shutdown_event)
+
+        await consumer.store_offset(received_messages[-1].offset(), partition_id)
+        assert (
+            consumer.get_last_stored_offset(partition_id)
+            == received_messages[-1].offset()
+        )
+
+        await consumer.delete_offset(partition_id)
+
+    @pytest.mark.asyncio
+    async def test_consume_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can consume messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+        shutdown_event = asyncio.Event()
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        async def take(message: ReceiveMessage) -> None:
+            received_messages.append(message.payload().decode())
+            if len(received_messages) == len(test_messages):
+                shutdown_event.set()
+
+        async def send() -> None:
+            await iggy_client.send_messages(
+                stream_name,
+                topic_name,
+                partition_id,
+                [Message(message) for message in test_messages],
+            )
+
+        await asyncio.gather(consumer.consume_messages(take, shutdown_event), send())
+
+        assert received_messages == test_messages
+
+    @pytest.mark.asyncio
+    async def test_iter_messages(self, iggy_client: IggyClient, unique_name):
+        """Test that the consumer group can iterate over messages."""
+        consumer_name = unique_name()
+        stream_name = unique_name()
+        topic_name = unique_name()
+        partition_id = 0
+        test_messages = [f"Consumer group test {i} - {unique_name()}" for i in range(5)]
+        received_messages = []
+
+        await iggy_client.create_stream(stream_name)
+        await iggy_client.create_topic(
+            stream=stream_name,
+            name=topic_name,
+            partitions_count=1,
+        )
+
+        consumer = await iggy_client.consumer_group(
+            consumer_name,
+            stream_name,
+            topic_name,
+            partition_id,
+            PollingStrategy.Next(),
+            10,
+            auto_commit=AutoCommit.Interval(timedelta(seconds=5)),
+            poll_interval=timedelta(milliseconds=25),
+        )
+
+        await iggy_client.send_messages(
+            stream_name,
+            topic_name,
+            partition_id,
+            [Message(message) for message in test_messages],
+        )
+
+        async for message in consumer.iter_messages():

Review Comment:
   Then we can use `asyncio.timeout()`. 3.10's EOL is 31th October 2026, so I think we should wait until maybe october, and then start using 3.11



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] test(python): Add tests for python sdk [iggy]

["hubcio (via GitHub)" <gi...@apache.org>]

hubcio merged PR #3432:
URL: https://github.com/apache/iggy/pull/3432


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org