feat(python): add TLS and auth args to getting-started examples

.github/actions/python-maturin/pre-merge/action.yml

@@ -53,11 +53,7 @@ runs:
       shell: bash
       run: |
         cd foreign/python
-        if [ "${{ inputs.task }}" = "test" ]; then
-          uv sync --frozen --extra dev --extra testing --extra testing-docker
-        else
-          uv sync --frozen --extra dev --extra testing
-        fi
+        uv sync --frozen --extra dev --extra testing --extra testing-docker
 
     - name: Lint and format check
       if: inputs.task == 'lint'

examples/python/getting-started/consumer.py

@@ -32,7 +32,9 @@
 PARTITION_ID = 0
 BATCHES_LIMIT = 5
 
-ArgNamespace = namedtuple("ArgNamespace", ["tcp_server_address"])
+ArgNamespace = namedtuple(
+    "ArgNamespace", ["tcp_server_address", "tls", "tls_ca_file", "username", "password"]
+)
 
 
 class ValidateUrl(argparse.Action):
@@ -57,18 +59,66 @@ def parse_args():
         action=ValidateUrl,
         default="127.0.0.1:8090",
     )
-    return parser.parse_args()
+    parser.add_argument(
+        "--tls",
+        action="store_true",
+        default=False,
+        help="Enable TLS for TCP connection",
+    )
+    parser.add_argument(
+        "--tls-ca-file",
+        default="",
+        help="Path to TLS CA certificate file",
+    )
+    parser.add_argument(
+        "--username",
+        default="iggy",
+        help="Username for authentication",
+    )
+    parser.add_argument(
+        "--password",
+        default="iggy",
+        help="Password for authentication",
+    )
+    args = parser.parse_args()
+
+    # Validate TLS requirements
+    if args.tls and not args.tls_ca_file:
+        parser.error("--tls requires --tls-ca-file")
+
+    return ArgNamespace(**vars(args))
+
+
+def build_connection_string(args) -> str:
+    """Build a connection string with TLS support."""
+
+    conn_str = f"iggy://{args.username}:{args.password}@{args.tcp_server_address}"
+
+    if args.tls:
+        # Extract domain from server address (host:port -> host)
+        host = args.tcp_server_address.split(":")[0]
+        query_params = ["tls=true", f"tls_domain={host}"]
+
+        # Add CA file if provided
+        if args.tls_ca_file:
+            query_params.append(f"tls_ca_file={args.tls_ca_file}")
+        conn_str += "?" + "&".join(query_params)
+
+    return conn_str
 
 
 async def main():
     args: ArgNamespace = parse_args()
-    client = IggyClient(args.tcp_server_address)
+
+    # Build connection string with TLS support
+    connection_string = build_connection_string(args)
+    logger.info(f"Connection string: {connection_string}")
+
+    client = IggyClient.from_connection_string(connection_string)
     try:
         logger.info("Connecting to IggyClient...")
         await client.connect()
-        logger.info("Connected. Logging in user...")
-        await client.login_user("iggy", "iggy")
-        logger.info("Logged in.")
+        logger.info("Connected.")
         await consume_messages(client)
     except Exception as error:
         logger.exception("Exception occurred in main function: {}", error)

examples/python/getting-started/producer.py

@@ -33,7 +33,9 @@
 PARTITION_ID = 0
 BATCHES_LIMIT = 5
 
-ArgNamespace = namedtuple("ArgNamespace", ["tcp_server_address"])
+ArgNamespace = namedtuple(
+    "ArgNamespace", ["tcp_server_address", "tls", "tls_ca_file", "username", "password"]
+)
 
 
 class ValidateUrl(argparse.Action):
@@ -58,17 +60,65 @@ def parse_args():
         action=ValidateUrl,
         default="127.0.0.1:8090",
     )
-    return parser.parse_args()
+    parser.add_argument(
+        "--tls",
+        action="store_true",
+        default=False,
+        help="Enable TLS for TCP connection",
+    )
+    parser.add_argument(
+        "--tls-ca-file",
+        default="",
+        help="Path to TLS CA certificate file",
+    )
+    parser.add_argument(
+        "--username",
+        default="iggy",
+        help="Username for authentication",
+    )
+    parser.add_argument(
+        "--password",
+        default="iggy",
+        help="Password for authentication",
+    )
+    args = parser.parse_args()
+
+    # Validate TLS requirements
+    if args.tls and not args.tls_ca_file:
+        parser.error("--tls requires --tls-ca-file")
+
+    return ArgNamespace(**vars(args))
+
+
+def build_connection_string(args) -> str:
+    """Build a connection string with TLS support."""
+
+    conn_str = f"iggy://{args.username}:{args.password}@{args.tcp_server_address}"
+
+    if args.tls:
+        # Extract domain from server address (host:port -> host)
+        host = args.tcp_server_address.split(":")[0]
+        query_params = ["tls=true", f"tls_domain={host}"]
+
+        # Add CA file if provided
+        if args.tls_ca_file:
+            query_params.append(f"tls_ca_file={args.tls_ca_file}")
+        conn_str += "?" + "&".join(query_params)
+
+    return conn_str
 
 
 async def main():
     args: ArgNamespace = parse_args()
-    client = IggyClient(args.tcp_server_address)
+    # Build connection string with TLS support
+    connection_string = build_connection_string(args)
+    logger.info(f"Connection string: {connection_string}")
+    logger.info(f"Connecting to {args.tcp_server_address} (TLS: {args.tls})")
+
+    client = IggyClient.from_connection_string(connection_string)
     logger.info("Connecting to IggyClient")
     await client.connect()
-    logger.info("Connected. Logging in user...")
-    await client.login_user("iggy", "iggy")
-    logger.info("Logged in.")
+    logger.info("Connected.")
     await init_system(client)
     await produce_messages(client)
 

foreign/python/tests/conftest.py

@@ -85,5 +85,5 @@ def pytest_collection_modifyitems(config, items):
     """Modify test collection to add markers automatically."""
     for item in items:
         # Mark all tests in test_iggy_sdk.py as integration tests
-        if "test_iggy_sdk" in item.nodeid:
+        if "test_iggy_sdk" in item.nodeid or "test_tls" in item.nodeid:
             item.add_marker(pytest.mark.integration)

foreign/python/tests/test_tls.py

@@ -0,0 +1,188 @@
+# 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.
+
+"""
+Integration tests for TLS connectivity using testcontainers.
+
+These tests spin up a TLS-enabled Iggy server in a Docker container
+so they are fully self-contained and work in CI without a pre-running
+TLS server.
+
+Requirements:
+    - Docker running locally
+    - testcontainers[docker] installed (in [testing-docker] extras)
+    - CA certificate available at core/certs/iggy_ca_cert.pem
+"""
+
+import asyncio
+import os
+import uuid
+
+import pytest
+from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import SendMessage as Message
+from testcontainers.core.container import DockerContainer  # type: ignore[import-untyped]
+from testcontainers.core.waiting_utils import wait_for_logs  # type: ignore[import-untyped]
+
+from .utils import wait_for_ping, wait_for_server
+
+# Paths resolved relative to this file → repo root
+REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", ".."))
+CERTS_DIR = os.path.join(REPO_ROOT, "core", "certs")
+CA_FILE = os.path.join(CERTS_DIR, "iggy_ca_cert.pem")
+
+IGGY_IMAGE = os.environ.get("IGGY_SERVER_DOCKER_IMAGE", "apache/iggy:edge")
+CONTAINER_TCP_PORT = 8090
+
+
+@pytest.fixture(scope="module")
+def tls_container():
+    """Start a TLS-enabled Iggy server in Docker."""
+    container = (
+        DockerContainer(IGGY_IMAGE)
+        .with_exposed_ports(CONTAINER_TCP_PORT)
+        .with_env("IGGY_ROOT_USERNAME", "iggy")
+        .with_env("IGGY_ROOT_PASSWORD", "iggy")
+        .with_env("IGGY_TCP_TLS_ENABLED", "true")
+        .with_env("IGGY_TCP_TLS_CERT_FILE", "/app/certs/iggy_cert.pem")
+        .with_env("IGGY_TCP_TLS_KEY_FILE", "/app/certs/iggy_key.pem")
+        .with_env("IGGY_TCP_ADDRESS", f"0.0.0.0:{CONTAINER_TCP_PORT}")
+        .with_volume_mapping(CERTS_DIR, "/app/certs", "ro")
+        .with_kwargs(privileged=True)
+    )
+    container.start()
+    # Wait for the server to be ready inside the container
+    wait_for_logs(container, "Iggy server is running", timeout=60)
+    yield container
+    container.stop()
+
+
+@pytest.fixture(scope="module")
+async def tls_client(tls_container) -> IggyClient:
+    """Create an authenticated client connected to the TLS container."""
+    host = "localhost"
+    port = tls_container.get_exposed_port(CONTAINER_TCP_PORT)
+
+    wait_for_server(host, int(port))
+
+    conn_str = (
+        f"iggy+tcp://iggy:iggy@{host}:{port}"
+        f"?tls=true&tls_domain={host}&tls_ca_file={CA_FILE}"
+    )
+    client = IggyClient.from_connection_string(conn_str)
+    await client.connect()
+    await wait_for_ping(client)
+    return client
+
+
+@pytest.mark.integration
+class TestTlsConnectivity:
+    """Test TLS connection establishment and basic operations."""
+
+    @pytest.mark.asyncio
+    async def test_ping_over_tls(self, tls_client: IggyClient):
+        """Test that the server responds to ping over a TLS connection."""
+        await tls_client.ping()
+
+    @pytest.mark.asyncio
+    async def test_client_not_none(self, tls_client: IggyClient):
+        """Test that the TLS client fixture is properly initialized."""
+        assert tls_client is not None
+
+    @pytest.mark.asyncio
+    async def test_create_stream_over_tls(self, tls_client: IggyClient):
+        """Test creating and getting a stream over TLS."""
+        stream_name = f"tls-test-stream-{uuid.uuid4().hex[:8]}"
+        await tls_client.create_stream(stream_name)
+        stream = await tls_client.get_stream(stream_name)
+        assert stream is not None
+
+    @pytest.mark.asyncio
+    async def test_produce_and_consume_over_tls(self, tls_client: IggyClient):
+        """Test producing and consuming messages over TLS."""
+        stream_name = f"tls-msg-stream-{uuid.uuid4().hex[:8]}"
+        topic_name = "tls-test-topic"
+        partition_id = 0
+
+        # Create stream and topic
+        await tls_client.create_stream(stream_name)
+        await tls_client.create_topic(stream_name, topic_name, partitions_count=1)
+
+        # Produce messages
+        test_messages = [f"tls-message-{i}" for i in range(3)]
+        messages = [Message(msg) for msg in test_messages]
+        await tls_client.send_messages(
+            stream=stream_name,
+            topic=topic_name,
+            partitioning=partition_id,
+            messages=messages,
+        )
+
+        # Consume messages
+        polled = await tls_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) >= len(test_messages)
+
+        # Verify message payloads
+        for i, expected_msg in enumerate(test_messages):
+            if i < len(polled):
+                assert polled[i].payload().decode("utf-8") == expected_msg
+
+
+@pytest.mark.integration
+class TestTlsConnectionString:
+    """Test TLS connection string variations."""
+
+    @pytest.mark.asyncio
+    async def test_connection_string_with_tls_params(self, tls_container):
+        """Test creating a client from a connection string with TLS parameters."""
+        host = "localhost"
+        port = tls_container.get_exposed_port(CONTAINER_TCP_PORT)
+
+        wait_for_server(host, int(port))
+
+        conn_str = (
+            f"iggy+tcp://iggy:iggy@{host}:{port}"
+            f"?tls=true&tls_domain={host}&tls_ca_file={CA_FILE}"
+        )
+        client = IggyClient.from_connection_string(conn_str)
+        await client.connect()
+        await wait_for_ping(client)
+        await client.ping()
+
+    @pytest.mark.asyncio
+    @pytest.mark.timeout(10)
+    async def test_connect_without_tls_should_fail(self, tls_container):
+        """Test that connecting without TLS to a TLS-enabled server fails."""
+        host = "localhost"
+        port = tls_container.get_exposed_port(CONTAINER_TCP_PORT)
+
+        wait_for_server(host, int(port))
+
+        # Use connection string without TLS params to a TLS-enabled server
+        # The SDK retries internally, so we use a timeout to detect failure
+        conn_str = f"iggy+tcp://iggy:iggy@{host}:{port}"
+        client = IggyClient.from_connection_string(conn_str)
+
+        with pytest.raises(Exception):
+            await asyncio.wait_for(client.connect(), timeout=5)