github-actions[bot] commented on code in PR #4357: URL: https://github.com/apache/iggy/pull/4357#discussion_r4143777542
########## foreign/python/tests/test_segment.py: ########## @@ -0,0 +1,257 @@ +# 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 pytest + +from apache_iggy import ( + Consumer, + Durability, + IggyClient, + Permissions, + PollingStrategy, + SendMessage, + StreamPermissions, + TopicPermissions, +) + +from .utils import login_fresh_client, unique_credentials + +PARTITION_ID = 0 + +# Smallest segment size a topic may declare. With one message per batch, the +# on-disk size of a 220_000-byte payload seals a segment after five messages +# and not after four, mirroring the layout in the server's segment deletion +# scenario. +SEGMENT_SIZE = 1024 * 1024 +PAYLOAD_SIZE = 220_000 +MESSAGES_PER_SEALED_SEGMENT = 5 + + +async def _create_topic(iggy_client: IggyClient, unique_name): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + # Flush every message so each send lands in a segment before the next one, + # which makes segment boundaries deterministic. + await iggy_client.create_topic( + stream=stream_name, + name=topic_name, + partitions_count=1, + segment_size=SEGMENT_SIZE, + durability=Durability.PERSISTED, + messages_required_to_save=1, + ) + return stream_name, topic_name + + +async def _send_messages(iggy_client: IggyClient, stream_id, topic_id, count: int): + payload = b"x" * PAYLOAD_SIZE + for _ in range(count): + await iggy_client.send_messages( + stream=stream_id, + topic=topic_id, + partitioning=PARTITION_ID, + messages=[SendMessage(payload)], + ) + + +async def _poll_offsets(iggy_client: IggyClient, stream_id, topic_id) -> list[int]: + messages = await iggy_client.poll_messages( + stream=stream_id, + topic=topic_id, + consumer=Consumer.Single(99), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.First(), + count=100, + auto_commit=False, + ) + return [message.offset() for message in messages] + + +async def _wait_for_offsets(iggy_client: IggyClient, stream_id, topic_id, expected): + # Deletion is acknowledged once committed; segment files are removed + # afterwards, so poll until the surviving offsets settle. + offsets = await _poll_offsets(iggy_client, stream_id, topic_id) + for _ in range(100): + if offsets == expected: + return + await asyncio.sleep(0.1) + offsets = await _poll_offsets(iggy_client, stream_id, topic_id) + assert offsets == expected + + +class TestSegmentManagement: + @pytest.mark.asyncio + @pytest.mark.parametrize("numeric_ids", [False, True]) + async def test_delete_segments_removes_oldest_sealed_segment( + self, iggy_client: IggyClient, unique_name, numeric_ids: bool + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + stream_id = stream.id if numeric_ids else stream_name + topic_id = topic.id if numeric_ids else topic_name + await _send_messages( + iggy_client, stream_id, topic_id, 2 * MESSAGES_PER_SEALED_SEGMENT + ) + await _wait_for_offsets(iggy_client, stream_id, topic_id, list(range(10))) + + result = await iggy_client.delete_segments(stream_id, topic_id, PARTITION_ID, 1) + + assert result is None + await _wait_for_offsets(iggy_client, stream_id, topic_id, list(range(5, 10))) + + @pytest.mark.asyncio + async def test_delete_segments_keeps_active_segment( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, 2 * MESSAGES_PER_SEALED_SEGMENT + 2 + ) + await _wait_for_offsets(iggy_client, stream_name, topic_name, list(range(12))) + + await iggy_client.delete_segments( + stream_name, topic_name, PARTITION_ID, 2**32 - 1 + ) + + await _wait_for_offsets(iggy_client, stream_name, topic_name, [10, 11]) + + @pytest.mark.asyncio + async def test_delete_segments_stops_at_committed_consumer_offset( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, 2 * MESSAGES_PER_SEALED_SEGMENT + 2 + ) + # Commit offset 5: the first sealed segment (offsets 0-4) is behind the + # consumer and deletable, the second (offsets 5-9) is not. + polled = await iggy_client.poll_messages( + stream=stream_name, + topic=topic_name, + consumer=Consumer.Single(1), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.First(), + count=MESSAGES_PER_SEALED_SEGMENT + 1, + auto_commit=True, + ) + assert [message.offset() for message in polled] == list(range(6)) + + await iggy_client.delete_segments(stream_name, topic_name, PARTITION_ID, 2) + + await _wait_for_offsets( + iggy_client, stream_name, topic_name, list(range(5, 12)) + ) + + @pytest.mark.asyncio + async def test_delete_segments_zero_count_is_noop( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, MESSAGES_PER_SEALED_SEGMENT + 1 + ) + await _wait_for_offsets(iggy_client, stream_name, topic_name, list(range(6))) + + await iggy_client.delete_segments(stream_name, topic_name, PARTITION_ID, 0) + + assert await _poll_offsets(iggy_client, stream_name, topic_name) == list( Review Comment: warning: The zero-count test polls the offsets immediately after the ack, but segment removal runs on a later reconciler pass. The assertion therefore still passes if `segments_count=0` wrongly trims a segment. Wait for the layout to settle before asserting, as `assert_layout_stable` does in `core/integration/tests/server/scenarios/purge_delete_scenario.rs`. ########## foreign/python/src/client.rs: ########## @@ -1098,6 +1098,52 @@ impl IggyClient { }) } + /// Delete the oldest sealed segments of a partition, including all messages + /// stored in them. The active segment is never deleted, and deletion stops at + /// the lowest offset any consumer or consumer group has committed on the + /// partition, so fewer than `segments_count` segments may be removed. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partition_id: Partition ID as `int`. + /// segments_count: Maximum number of segments to delete as `int`, starting + /// from the oldest; `0` deletes nothing. + /// + /// Returns: + /// An awaitable that resolves to `None` when deletion is accepted; segment + /// files are removed asynchronously. + /// + /// Raises: + /// ValueError: If an identifier is invalid. + /// OverflowError: If `partition_id` or `segments_count` is outside the + /// unsigned 32-bit range. + /// RuntimeError: If the client is not authenticated, lacks global + /// `manage_streams` or `manage_topics`, per-stream `manage_stream` or + /// `manage_topics`, or per-topic `manage_topic` permission, the + /// stream, topic, or partition does not exist, or the request fails. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))] + fn delete_segments<'a>( + &self, + py: Python<'a>, + stream_id: PyIdentifier, + topic_id: PyIdentifier, + partition_id: u32, + segments_count: u32, + ) -> PyResult<Bound<'a, PyAny>> { + let stream_id = Identifier::try_from(stream_id)?; + let topic_id = Identifier::try_from(topic_id)?; + let inner = self.inner.clone(); + + future_into_py(py, async move { + inner + .delete_segments(&stream_id, &topic_id, partition_id, segments_count) + .await + .map_err(to_runtime_error)?; + Ok(Python::attach(|py| py.None())) Review Comment: simplification: `delete_segments` returns `Ok(Python::attach(|py| py.None()))`, which acquires the GIL only to rebuild `None`. Return `Ok(())` as the sibling wrappers do, and Python sees the same `None`. ########## foreign/python/src/client.rs: ########## @@ -1098,6 +1098,52 @@ impl IggyClient { }) } + /// Delete the oldest sealed segments of a partition, including all messages + /// stored in them. The active segment is never deleted, and deletion stops at + /// the lowest offset any consumer or consumer group has committed on the + /// partition, so fewer than `segments_count` segments may be removed. + /// + /// Args: + /// stream_id: Stream identifier as `str | int`. + /// topic_id: Topic identifier as `str | int`. + /// partition_id: Partition ID as `int`. + /// segments_count: Maximum number of segments to delete as `int`, starting + /// from the oldest; `0` deletes nothing. + /// + /// Returns: + /// An awaitable that resolves to `None` when deletion is accepted; segment + /// files are removed asynchronously. + /// + /// Raises: + /// ValueError: If an identifier is invalid. Review Comment: nit: `delete_segments` promises `ValueError` for an invalid identifier, but an out-of-range `int` `stream_id` or `topic_id` raises `OverflowError` during argument extraction. List the integer identifier range in the `OverflowError` entry and regenerate `apache_iggy.pyi`. ########## foreign/python/tests/test_segment.py: ########## @@ -0,0 +1,257 @@ +# 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 pytest + +from apache_iggy import ( + Consumer, + Durability, + IggyClient, + Permissions, + PollingStrategy, + SendMessage, + StreamPermissions, + TopicPermissions, +) + +from .utils import login_fresh_client, unique_credentials + +PARTITION_ID = 0 + +# Smallest segment size a topic may declare. With one message per batch, the +# on-disk size of a 220_000-byte payload seals a segment after five messages +# and not after four, mirroring the layout in the server's segment deletion +# scenario. +SEGMENT_SIZE = 1024 * 1024 +PAYLOAD_SIZE = 220_000 +MESSAGES_PER_SEALED_SEGMENT = 5 + + +async def _create_topic(iggy_client: IggyClient, unique_name): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + # Flush every message so each send lands in a segment before the next one, + # which makes segment boundaries deterministic. + await iggy_client.create_topic( + stream=stream_name, + name=topic_name, + partitions_count=1, + segment_size=SEGMENT_SIZE, + durability=Durability.PERSISTED, + messages_required_to_save=1, + ) + return stream_name, topic_name + + +async def _send_messages(iggy_client: IggyClient, stream_id, topic_id, count: int): + payload = b"x" * PAYLOAD_SIZE + for _ in range(count): + await iggy_client.send_messages( + stream=stream_id, + topic=topic_id, + partitioning=PARTITION_ID, + messages=[SendMessage(payload)], + ) + + +async def _poll_offsets(iggy_client: IggyClient, stream_id, topic_id) -> list[int]: + messages = await iggy_client.poll_messages( + stream=stream_id, + topic=topic_id, + consumer=Consumer.Single(99), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.First(), + count=100, + auto_commit=False, + ) + return [message.offset() for message in messages] + + +async def _wait_for_offsets(iggy_client: IggyClient, stream_id, topic_id, expected): + # Deletion is acknowledged once committed; segment files are removed + # afterwards, so poll until the surviving offsets settle. + offsets = await _poll_offsets(iggy_client, stream_id, topic_id) + for _ in range(100): + if offsets == expected: + return + await asyncio.sleep(0.1) + offsets = await _poll_offsets(iggy_client, stream_id, topic_id) + assert offsets == expected + + +class TestSegmentManagement: + @pytest.mark.asyncio + @pytest.mark.parametrize("numeric_ids", [False, True]) + async def test_delete_segments_removes_oldest_sealed_segment( + self, iggy_client: IggyClient, unique_name, numeric_ids: bool + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + stream_id = stream.id if numeric_ids else stream_name + topic_id = topic.id if numeric_ids else topic_name + await _send_messages( + iggy_client, stream_id, topic_id, 2 * MESSAGES_PER_SEALED_SEGMENT + ) + await _wait_for_offsets(iggy_client, stream_id, topic_id, list(range(10))) + + result = await iggy_client.delete_segments(stream_id, topic_id, PARTITION_ID, 1) + + assert result is None + await _wait_for_offsets(iggy_client, stream_id, topic_id, list(range(5, 10))) + + @pytest.mark.asyncio + async def test_delete_segments_keeps_active_segment( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, 2 * MESSAGES_PER_SEALED_SEGMENT + 2 + ) + await _wait_for_offsets(iggy_client, stream_name, topic_name, list(range(12))) + + await iggy_client.delete_segments( + stream_name, topic_name, PARTITION_ID, 2**32 - 1 + ) + + await _wait_for_offsets(iggy_client, stream_name, topic_name, [10, 11]) + + @pytest.mark.asyncio + async def test_delete_segments_stops_at_committed_consumer_offset( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, 2 * MESSAGES_PER_SEALED_SEGMENT + 2 + ) + # Commit offset 5: the first sealed segment (offsets 0-4) is behind the + # consumer and deletable, the second (offsets 5-9) is not. + polled = await iggy_client.poll_messages( + stream=stream_name, + topic=topic_name, + consumer=Consumer.Single(1), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.First(), + count=MESSAGES_PER_SEALED_SEGMENT + 1, + auto_commit=True, + ) + assert [message.offset() for message in polled] == list(range(6)) + + await iggy_client.delete_segments(stream_name, topic_name, PARTITION_ID, 2) + + await _wait_for_offsets( + iggy_client, stream_name, topic_name, list(range(5, 12)) + ) + + @pytest.mark.asyncio + async def test_delete_segments_zero_count_is_noop( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, MESSAGES_PER_SEALED_SEGMENT + 1 + ) + await _wait_for_offsets(iggy_client, stream_name, topic_name, list(range(6))) + + await iggy_client.delete_segments(stream_name, topic_name, PARTITION_ID, 0) + + assert await _poll_offsets(iggy_client, stream_name, topic_name) == list( + range(6) + ) + + @pytest.mark.asyncio + @pytest.mark.parametrize("missing", ["stream", "topic", "partition"]) + async def test_delete_segments_rejects_missing_target( + self, iggy_client: IggyClient, unique_name, missing: str + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + missing_name = unique_name() + stream_id = missing_name if missing == "stream" else stream_name + topic_id = missing_name if missing == "topic" else topic_name + partition_id = 999 if missing == "partition" else PARTITION_ID + + with pytest.raises(RuntimeError, match=r"was not found\."): + await iggy_client.delete_segments(stream_id, topic_id, partition_id, 1) + + @pytest.mark.asyncio + async def test_delete_segments_rejects_invalid_identifier( + self, iggy_client: IggyClient, unique_name + ): + _, topic_name = await _create_topic(iggy_client, unique_name) + + with pytest.raises(ValueError): + await iggy_client.delete_segments("", topic_name, PARTITION_ID, 1) + + @pytest.mark.asyncio + @pytest.mark.parametrize("argument", ["partition_id", "segments_count"]) + @pytest.mark.parametrize("value", [-1, 2**32]) + async def test_delete_segments_rejects_out_of_range_python_integer( + self, iggy_client: IggyClient, unique_name, argument: str, value: int + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) Review Comment: nit: The overflow test creates a stream and topic that the rejected `delete_segments` call never reaches, because argument extraction fails before the request is built. Drop `_create_topic` and pass literal names. The same setup at line 200 is unused. ########## foreign/python/src/client.rs: ########## @@ -1098,6 +1098,52 @@ impl IggyClient { }) } + /// Delete the oldest sealed segments of a partition, including all messages + /// stored in them. The active segment is never deleted, and deletion stops at + /// the lowest offset any consumer or consumer group has committed on the + /// partition, so fewer than `segments_count` segments may be removed. Review Comment: nit: The docstring says deletion stops at the committed consumer offset, but the server keeps the committed delete watermark and re-applies it on later reconciler passes. State that a blocked call can remove the remaining sealed segments later, as consumers advance, up to the resolved count. ########## foreign/python/tests/test_segment.py: ########## @@ -0,0 +1,257 @@ +# 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 pytest + +from apache_iggy import ( + Consumer, + Durability, + IggyClient, + Permissions, + PollingStrategy, + SendMessage, + StreamPermissions, + TopicPermissions, +) + +from .utils import login_fresh_client, unique_credentials + +PARTITION_ID = 0 + +# Smallest segment size a topic may declare. With one message per batch, the +# on-disk size of a 220_000-byte payload seals a segment after five messages +# and not after four, mirroring the layout in the server's segment deletion +# scenario. +SEGMENT_SIZE = 1024 * 1024 +PAYLOAD_SIZE = 220_000 +MESSAGES_PER_SEALED_SEGMENT = 5 + + +async def _create_topic(iggy_client: IggyClient, unique_name): + stream_name = unique_name() + topic_name = unique_name() + + await iggy_client.create_stream(stream_name) + # Flush every message so each send lands in a segment before the next one, + # which makes segment boundaries deterministic. + await iggy_client.create_topic( + stream=stream_name, + name=topic_name, + partitions_count=1, + segment_size=SEGMENT_SIZE, + durability=Durability.PERSISTED, + messages_required_to_save=1, + ) + return stream_name, topic_name + + +async def _send_messages(iggy_client: IggyClient, stream_id, topic_id, count: int): + payload = b"x" * PAYLOAD_SIZE + for _ in range(count): + await iggy_client.send_messages( + stream=stream_id, + topic=topic_id, + partitioning=PARTITION_ID, + messages=[SendMessage(payload)], + ) + + +async def _poll_offsets(iggy_client: IggyClient, stream_id, topic_id) -> list[int]: + messages = await iggy_client.poll_messages( + stream=stream_id, + topic=topic_id, + consumer=Consumer.Single(99), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.First(), + count=100, + auto_commit=False, + ) + return [message.offset() for message in messages] + + +async def _wait_for_offsets(iggy_client: IggyClient, stream_id, topic_id, expected): + # Deletion is acknowledged once committed; segment files are removed + # afterwards, so poll until the surviving offsets settle. + offsets = await _poll_offsets(iggy_client, stream_id, topic_id) + for _ in range(100): + if offsets == expected: + return + await asyncio.sleep(0.1) + offsets = await _poll_offsets(iggy_client, stream_id, topic_id) + assert offsets == expected + + +class TestSegmentManagement: + @pytest.mark.asyncio + @pytest.mark.parametrize("numeric_ids", [False, True]) + async def test_delete_segments_removes_oldest_sealed_segment( + self, iggy_client: IggyClient, unique_name, numeric_ids: bool + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + topic = await iggy_client.get_topic(stream.id, topic_name) + assert topic is not None + stream_id = stream.id if numeric_ids else stream_name + topic_id = topic.id if numeric_ids else topic_name + await _send_messages( + iggy_client, stream_id, topic_id, 2 * MESSAGES_PER_SEALED_SEGMENT + ) + await _wait_for_offsets(iggy_client, stream_id, topic_id, list(range(10))) + + result = await iggy_client.delete_segments(stream_id, topic_id, PARTITION_ID, 1) + + assert result is None + await _wait_for_offsets(iggy_client, stream_id, topic_id, list(range(5, 10))) + + @pytest.mark.asyncio + async def test_delete_segments_keeps_active_segment( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, 2 * MESSAGES_PER_SEALED_SEGMENT + 2 + ) + await _wait_for_offsets(iggy_client, stream_name, topic_name, list(range(12))) + + await iggy_client.delete_segments( + stream_name, topic_name, PARTITION_ID, 2**32 - 1 + ) + + await _wait_for_offsets(iggy_client, stream_name, topic_name, [10, 11]) + + @pytest.mark.asyncio + async def test_delete_segments_stops_at_committed_consumer_offset( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, 2 * MESSAGES_PER_SEALED_SEGMENT + 2 + ) + # Commit offset 5: the first sealed segment (offsets 0-4) is behind the + # consumer and deletable, the second (offsets 5-9) is not. + polled = await iggy_client.poll_messages( + stream=stream_name, + topic=topic_name, + consumer=Consumer.Single(1), + partition_id=PARTITION_ID, + polling_strategy=PollingStrategy.First(), + count=MESSAGES_PER_SEALED_SEGMENT + 1, + auto_commit=True, + ) + assert [message.offset() for message in polled] == list(range(6)) + + await iggy_client.delete_segments(stream_name, topic_name, PARTITION_ID, 2) + + await _wait_for_offsets( + iggy_client, stream_name, topic_name, list(range(5, 12)) + ) + + @pytest.mark.asyncio + async def test_delete_segments_zero_count_is_noop( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + await _send_messages( + iggy_client, stream_name, topic_name, MESSAGES_PER_SEALED_SEGMENT + 1 + ) + await _wait_for_offsets(iggy_client, stream_name, topic_name, list(range(6))) + + await iggy_client.delete_segments(stream_name, topic_name, PARTITION_ID, 0) + + assert await _poll_offsets(iggy_client, stream_name, topic_name) == list( + range(6) + ) + + @pytest.mark.asyncio + @pytest.mark.parametrize("missing", ["stream", "topic", "partition"]) + async def test_delete_segments_rejects_missing_target( + self, iggy_client: IggyClient, unique_name, missing: str + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + missing_name = unique_name() + stream_id = missing_name if missing == "stream" else stream_name + topic_id = missing_name if missing == "topic" else topic_name + partition_id = 999 if missing == "partition" else PARTITION_ID + + with pytest.raises(RuntimeError, match=r"was not found\."): + await iggy_client.delete_segments(stream_id, topic_id, partition_id, 1) + + @pytest.mark.asyncio + async def test_delete_segments_rejects_invalid_identifier( + self, iggy_client: IggyClient, unique_name + ): + _, topic_name = await _create_topic(iggy_client, unique_name) + + with pytest.raises(ValueError): + await iggy_client.delete_segments("", topic_name, PARTITION_ID, 1) + + @pytest.mark.asyncio + @pytest.mark.parametrize("argument", ["partition_id", "segments_count"]) + @pytest.mark.parametrize("value", [-1, 2**32]) + async def test_delete_segments_rejects_out_of_range_python_integer( + self, iggy_client: IggyClient, unique_name, argument: str, value: int + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + arguments = {"partition_id": PARTITION_ID, "segments_count": 1, argument: value} + + with pytest.raises(OverflowError): + await iggy_client.delete_segments(stream_name, topic_name, **arguments) + + @pytest.mark.asyncio + async def test_delete_segments_requires_scoped_manage_topic( + self, iggy_client: IggyClient, unique_name + ): + stream_name, topic_name = await _create_topic(iggy_client, unique_name) + other_topic_name = unique_name() + await iggy_client.create_topic( + stream_name, other_topic_name, partitions_count=1 + ) + stream = await iggy_client.get_stream(stream_name) + assert stream is not None + topic = await iggy_client.get_topic(stream.id, topic_name) + other_topic = await iggy_client.get_topic(stream.id, other_topic_name) + assert topic is not None + assert other_topic is not None + + denied_username, denied_password = unique_credentials(unique_name) Review Comment: simplification: The scoped-permission setup repeats `foreign/python/tests/test_partition.py:194-228` almost line for line. Extract the permission scaffolding into `tests/utils.py`, with the checked call passed in, and use it in both tests. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
