This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/python-examples-ci
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/fix/python-examples-ci by this
push:
new 8a19739df fix(python): propagate consumer example failures
8a19739df is described below
commit 8a19739df64edb5bcab2d083e9f891a5556bc7c0
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Thu Nov 5 16:51:52 2026 +0100
fix(python): propagate consumer example failures
Consumer examples swallowed errors and returned success to CI.
Re-raise failures and cover process exits for both consumers.
---
examples/python/basic/consumer.py | 2 +-
examples/python/getting-started/consumer.py | 5 +--
scripts/ci/test-runtime-artifacts.py | 56 +++++++++++++++++++++++++++++
3 files changed, 60 insertions(+), 3 deletions(-)
diff --git a/examples/python/basic/consumer.py
b/examples/python/basic/consumer.py
index 3b84cf5ef..7f5876829 100644
--- a/examples/python/basic/consumer.py
+++ b/examples/python/basic/consumer.py
@@ -101,7 +101,7 @@ async def consume_messages(client: IggyClient):
await asyncio.sleep(interval)
except Exception as error:
logger.exception(f"Exception occurred while consuming messages:
{error}")
- break
+ raise
logger.info(f"Consumed {n_consumed_batches} batches of messages, exiting.")
diff --git a/examples/python/getting-started/consumer.py
b/examples/python/getting-started/consumer.py
index 9178d99f0..3ff7fb5ed 100755
--- a/examples/python/getting-started/consumer.py
+++ b/examples/python/getting-started/consumer.py
@@ -160,7 +160,7 @@ async def main():
config = build_config(args)
except ValueError as error:
logger.error(f"Invalid client configuration: {error}")
- return
+ raise
logger.info(f"Connecting with {config}")
client = IggyClient(config)
@@ -172,6 +172,7 @@ async def main():
await consume_messages(client)
except Exception as error:
logger.exception(f"Exception occurred in main function: {error}")
+ raise
async def consume_messages(client: IggyClient):
@@ -210,7 +211,7 @@ async def consume_messages(client: IggyClient):
await asyncio.sleep(interval)
except Exception as error:
logger.exception(f"Exception occurred while consuming messages:
{error}")
- break
+ raise
logger.info(f"Consumed {n_consumed_batches} batches of messages, exiting.")
diff --git a/scripts/ci/test-runtime-artifacts.py
b/scripts/ci/test-runtime-artifacts.py
index 6fff6f28c..669364563 100644
--- a/scripts/ci/test-runtime-artifacts.py
+++ b/scripts/ci/test-runtime-artifacts.py
@@ -22,6 +22,7 @@ import json
import os
from pathlib import Path
import subprocess
+import sys
import tempfile
import unittest
@@ -370,6 +371,61 @@ class RuntimeConsumers(unittest.TestCase):
self.assertNotEqual(result.returncode, 0, result.stdout +
result.stderr)
self.assertEqual(events, ["sync --frozen"])
+ def test_python_consumer_failures_return_nonzero(self):
+ for example in ("basic", "getting-started"):
+ failures = {
+ "connect": "connection failed",
+ "poll": "poll failed",
+ "decode": "UnicodeDecodeError",
+ }
+ if example == "getting-started":
+ failures["config"] = "invalid configuration"
+ for failure, message in failures.items():
+ with self.subTest(example=example, failure=failure):
+ result = self.run_python_consumer(example, failure)
+ self.assertNotEqual(result.returncode, 0, result.stdout +
result.stderr)
+ self.assertIn(message, result.stderr)
+
+ def test_python_consumers_complete_successfully(self):
+ for example in ("basic", "getting-started"):
+ with self.subTest(example=example):
+ result = self.run_python_consumer(example)
+ self.assertEqual(result.returncode, 0, result.stdout +
result.stderr)
+
+ def run_python_consumer(self, example, failure=""):
+ script = r"""
+import asyncio
+import runpy
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+path, failure = sys.argv[1:]
+message = MagicMock()
+message.payload.return_value = b"hello" if failure != "decode" else b"\xff"
+client = MagicMock()
+client.connect = AsyncMock()
+client.poll_messages = AsyncMock(return_value=MagicMock(messages=[message]))
+sdk = MagicMock()
+sdk.IggyClient.return_value = client
+sdk.IggyClient.from_connection_string.return_value = client
+if failure == "connect":
+ client.connect.side_effect = RuntimeError("connection failed")
+elif failure == "poll":
+ client.poll_messages.side_effect = RuntimeError("poll failed")
+elif failure == "config":
+ sdk.TcpConfig.side_effect = ValueError("invalid configuration")
+sys.argv = [path]
+with (
+ patch.dict(sys.modules, {"apache_iggy": sdk, "loguru": MagicMock()}),
+ patch.object(asyncio, "sleep", new=AsyncMock()),
+):
+ consumer = runpy.run_path(path, run_name="__main__")
+if not failure and client.poll_messages.await_count !=
consumer["BATCHES_LIMIT"]:
+ raise AssertionError("Consumer did not complete all batches")
+"""
+ return run(sys.executable, "-c", script,
+ str(REPO_ROOT / "examples/python" / example /
"consumer.py"), failure)
+
def run_python_examples(self, *, sync_exit=0):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
