You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/19 17:05:47 UTC
(iggy) 07/19: fix(python): recover pending operations during test cleanup
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/sdk-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit f509d5759cf7da2ada07e107edd654adf5abcabc
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Oct 19 16:39:51 2026 +0200
fix(python): recover pending operations during test cleanup
Cleanup swallowed uncertain stream deletions, leaving the shared
client's metadata lane blocked for subsequent tests.
Resolve retained operations and verify deletion before continuing.
Stop the test session on cleanup failure, and allow the maximum
partition test enough time for bounded recovery.
---
foreign/python/pyproject.toml | 2 +-
foreign/python/tests/conftest.py | 35 +++++++-
foreign/python/tests/test_cleanup.py | 169 +++++++++++++++++++++++++++++++++++
foreign/python/tests/test_topic.py | 13 ++-
4 files changed, 213 insertions(+), 6 deletions(-)
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index 3771fe3ed..bc8ae6eaa 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -132,7 +132,7 @@ asyncio_default_fixture_loop_scope = "session"
timeout = 30
# A test that pins the GIL leaves neither pytest-timeout nor asyncio able to fire.
# The faulthandler watchdog needs no GIL, so it dumps every thread and aborts the run.
-faulthandler_timeout = 60
+faulthandler_timeout = 120
faulthandler_exit_on_timeout = true
testpaths = ["tests"]
python_files = ["test_*.py"]
diff --git a/foreign/python/tests/conftest.py b/foreign/python/tests/conftest.py
index ac9acdec4..73507fb59 100644
--- a/foreign/python/tests/conftest.py
+++ b/foreign/python/tests/conftest.py
@@ -23,7 +23,6 @@ and connecting to Iggy servers in various configurations.
"""
import asyncio
-import contextlib
import os
import secrets
import string
@@ -32,10 +31,13 @@ from pathlib import Path
import pytest
-from apache_iggy import IggyClient
+from apache_iggy import IggyClient, PendingOperationError
from .utils import get_server_config, wait_for_ping, wait_for_server
+# A request and its retained replay can each spend 30 seconds.
+STREAM_CLEANUP_TIMEOUT_SECONDS = 60
+
@pytest.fixture(scope="session")
async def iggy_client() -> IggyClient:
@@ -125,8 +127,33 @@ async def cleanup_streams(
for stream in await client.get_streams():
if (stream.id, stream.created_at) not in existing_streams:
- with contextlib.suppress(RuntimeError):
- await client.delete_stream(stream.id)
+ try:
+ async with asyncio.timeout(STREAM_CLEANUP_TIMEOUT_SECONDS):
+ while True:
+ try:
+ await client.delete_stream(stream.id)
+ break
+ except PendingOperationError as error:
+ if error.kind not in ("uncertain", "lane_blocked"):
+ raise
+ await client.resolve_operation(error.operation_id)
+ if error.kind == "uncertain":
+ break
+ if await client.get_stream(stream.id) is None:
+ break
+
+ if await client.get_stream(stream.id) is not None:
+ raise RuntimeError(
+ f"Stream {stream.id} still exists after deletion"
+ )
+ except (RuntimeError, TimeoutError) as error:
+ reason = (
+ f"Cleanup failed for stream {stream.id} ({stream.name!r}): "
+ f"{error!r}"
+ )
+ error.add_note(reason)
+ request.session.shouldstop = reason
+ raise
@pytest.fixture(scope="session", autouse=True)
diff --git a/foreign/python/tests/test_cleanup.py b/foreign/python/tests/test_cleanup.py
new file mode 100644
index 000000000..0de510cf2
--- /dev/null
+++ b/foreign/python/tests/test_cleanup.py
@@ -0,0 +1,169 @@
+# 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 inspect import unwrap
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock, call
+
+import pytest
+
+from apache_iggy import IggyClient
+
+from . import conftest
+
+pytestmark = [pytest.mark.unit, pytest.mark.asyncio]
+
+STREAM_ID = 7
+OPERATION_ID = 2829
+CLEANUP_TEST_TIMEOUT_SECONDS = 0.01
+
+
+class PendingOperationFailure(RuntimeError):
+ def __init__(self, operation_id: int, kind: str):
+ super().__init__(f"Operation {operation_id}: {kind}")
+ self.operation_id = operation_id
+ self.kind = kind
+
+
+@pytest.fixture
+def cleanup_request(monkeypatch):
+ # PyO3 exposes these exceptions only as results of SDK operations.
+ monkeypatch.setattr(conftest, "PendingOperationError", PendingOperationFailure)
+ client = Mock(spec=IggyClient)
+ client.get_streams = AsyncMock(
+ side_effect=[
+ [],
+ [SimpleNamespace(id=STREAM_ID, created_at=1, name="cleanup-stream")],
+ ]
+ )
+ client.delete_stream = AsyncMock()
+ client.resolve_operation = AsyncMock()
+ client.get_stream = AsyncMock(return_value=None)
+ request = Mock(spec=pytest.FixtureRequest)
+ request.fixturenames = ["iggy_client"]
+ request.getfixturevalue.return_value = client
+ request.session = Mock(shouldstop=False)
+ return request
+
+
+async def test_uncertain_deletion_resolves_the_same_operation(cleanup_request):
+ client = cleanup_request.getfixturevalue.return_value
+ client.delete_stream.side_effect = PendingOperationFailure(
+ OPERATION_ID, "uncertain"
+ )
+ cleanup = unwrap(conftest.cleanup_streams)(cleanup_request)
+ await anext(cleanup)
+
+ with pytest.raises(StopAsyncIteration):
+ await anext(cleanup)
+
+ client.delete_stream.assert_awaited_once_with(STREAM_ID)
+ client.resolve_operation.assert_awaited_once_with(OPERATION_ID)
+ client.get_stream.assert_awaited_once_with(STREAM_ID)
+ assert cleanup_request.session.shouldstop is False
+
+
+@pytest.mark.parametrize("blocker_deleted_stream", [False, True])
+async def test_blocked_deletion_resolves_before_retrying(
+ cleanup_request, blocker_deleted_stream
+):
+ client = cleanup_request.getfixturevalue.return_value
+ client.delete_stream.side_effect = [
+ PendingOperationFailure(OPERATION_ID, "lane_blocked"),
+ None,
+ ]
+ client.get_stream.side_effect = (
+ [None, None] if blocker_deleted_stream else [object(), None]
+ )
+ cleanup = unwrap(conftest.cleanup_streams)(cleanup_request)
+ await anext(cleanup)
+ client.reset_mock()
+
+ with pytest.raises(StopAsyncIteration):
+ await anext(cleanup)
+
+ expected = [
+ call.get_streams(),
+ call.delete_stream(STREAM_ID),
+ call.resolve_operation(OPERATION_ID),
+ call.get_stream(STREAM_ID),
+ ]
+ if not blocker_deleted_stream:
+ expected.append(call.delete_stream(STREAM_ID))
+ expected.append(call.get_stream(STREAM_ID))
+ assert client.mock_calls == expected
+ assert cleanup_request.session.shouldstop is False
+
+
+@pytest.mark.parametrize("failure", ["runtime", "uncertain", "not_retained"])
+async def test_cleanup_failure_stops_shared_client_reuse(cleanup_request, failure):
+ client = cleanup_request.getfixturevalue.return_value
+ error = (
+ RuntimeError("deletion rejected")
+ if failure == "runtime"
+ else PendingOperationFailure(OPERATION_ID, failure)
+ )
+ client.delete_stream.side_effect = error
+ client.resolve_operation.side_effect = error
+ cleanup = unwrap(conftest.cleanup_streams)(cleanup_request)
+ await anext(cleanup)
+
+ with pytest.raises(RuntimeError) as raised:
+ await anext(cleanup)
+
+ assert raised.value is error
+ assert str(STREAM_ID) in cleanup_request.session.shouldstop
+ assert str(error) in cleanup_request.session.shouldstop
+ if failure == "uncertain":
+ client.resolve_operation.assert_awaited_once_with(OPERATION_ID)
+ else:
+ client.resolve_operation.assert_not_awaited()
+
+
+async def test_cleanup_verifies_the_stream_was_deleted(cleanup_request):
+ client = cleanup_request.getfixturevalue.return_value
+ client.get_stream.return_value = object()
+ cleanup = unwrap(conftest.cleanup_streams)(cleanup_request)
+ await anext(cleanup)
+
+ with pytest.raises(RuntimeError, match="still exists"):
+ await anext(cleanup)
+
+ assert cleanup_request.session.shouldstop
+
+
+async def test_cleanup_timeout_stops_shared_client_reuse(cleanup_request, monkeypatch):
+ monkeypatch.setattr(
+ conftest,
+ "STREAM_CLEANUP_TIMEOUT_SECONDS",
+ CLEANUP_TEST_TIMEOUT_SECONDS,
+ )
+ client = cleanup_request.getfixturevalue.return_value
+ pending = asyncio.Event()
+
+ async def stalled_delete(_stream_id):
+ await pending.wait()
+
+ client.delete_stream.side_effect = stalled_delete
+ cleanup = unwrap(conftest.cleanup_streams)(cleanup_request)
+ await anext(cleanup)
+
+ with pytest.raises(TimeoutError):
+ await anext(cleanup)
+
+ assert str(STREAM_ID) in cleanup_request.session.shouldstop
diff --git a/foreign/python/tests/test_topic.py b/foreign/python/tests/test_topic.py
index 24271fc14..f25105a00 100644
--- a/foreign/python/tests/test_topic.py
+++ b/foreign/python/tests/test_topic.py
@@ -33,6 +33,8 @@ from .utils import (
wait_for_server,
)
+MAX_PARTITIONS_TIMEOUT_SECONDS = 90
+
class TestCreateTopic:
"""Test topic creation via create_topic."""
@@ -416,7 +418,16 @@ class TestCreateTopic:
assert "Too many partitions" in str(exc_info.value)
@pytest.mark.asyncio
- @pytest.mark.parametrize("partitions_count", [0, 1, 1000])
+ @pytest.mark.parametrize(
+ "partitions_count",
+ [
+ 0,
+ 1,
+ pytest.param(
+ 1000, marks=pytest.mark.timeout(MAX_PARTITIONS_TIMEOUT_SECONDS)
+ ),
+ ],
+ )
async def test_create_topic_with_valid_partitions_count(
self, iggy_client: IggyClient, unique_name, partitions_count: int
):