| # 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, SendMessage |
| |
| 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( |
| ("prefix", "min_bytes", "max_bytes"), |
| [ |
| ("a", 8, 255), |
| ("stream-name", 8, 255), |
| ("stream_name.with.mixed-CHARS123", 8, 255), |
| (" leading-space", 8, 255), |
| ("trailing-space ", 8, 255), |
| ("multiple spaces inside", 8, 255), |
| ("name/with/slash", 8, 255), |
| ("name:with:colons", 8, 255), |
| ("name.with.dots", 8, 255), |
| (" ", 8, 255), |
| ("a" * 247, 255, 255), |
| (("é" * 122) + "abc", 255, 255), |
| (("한" * 81) + "abc", 255, 255), |
| (("漢" * 81) + "abc", 255, 255), |
| (("あ" * 81) + "abc", 255, 255), |
| (("😀" * 60) + "abcdefg", 255, 255), |
| ], |
| ) |
| async def test_create_and_get_stream( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| prefix: str, |
| min_bytes: int, |
| max_bytes: int, |
| ): |
| """Test stream creation and retrieval.""" |
| stream_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes) |
| |
| await iggy_client.create_stream(stream_name) |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| assert stream.created_at > 0 |
| assert stream.name == stream_name |
| assert stream.size == 0 |
| assert stream.topics_count == 0 |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| ("prefix", "min_bytes", "max_bytes"), |
| [ |
| ("", 0, 0), |
| ("a" * 248, 256, 256), |
| ("é" * 124, 256, 256), |
| (("é" * 123) + "ab", 256, 256), |
| (("한" * 82) + "ab", 256, 256), |
| (("漢" * 82) + "ab", 256, 256), |
| (("あ" * 82) + "ab", 256, 256), |
| ("😀" * 62, 256, 256), |
| (("😀" * 61) + "abcd", 256, 256), |
| ], |
| ) |
| async def test_create_stream_invalid_names( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| prefix: str, |
| min_bytes: int, |
| max_bytes: int, |
| ): |
| """Test create_stream enforces byte-length validation.""" |
| stream_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.create_stream(stream_name) |
| |
| @pytest.mark.asyncio |
| async def test_get_stream_by_name_and_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test repeated stream lookup works by both name and numeric id.""" |
| stream_name = unique_name() |
| await iggy_client.create_stream(stream_name) |
| |
| stream_by_name = await iggy_client.get_stream(stream_name) |
| assert stream_by_name is not None |
| |
| stream_by_name_again = await iggy_client.get_stream(stream_name) |
| assert stream_by_name_again is not None |
| assert stream_by_name_again.id == stream_by_name.id |
| assert stream_by_name_again.name == stream_by_name.name |
| |
| stream_by_id = await iggy_client.get_stream(stream_by_name.id) |
| assert stream_by_id is not None |
| assert stream_by_id.id == stream_by_name.id |
| assert stream_by_id.name == stream_by_name.name |
| |
| @pytest.mark.asyncio |
| async def test_create_stream_then_reconnect_then_get_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a stream remains retrievable after reconnecting with a fresh client.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| await client.connect() |
| await wait_for_ping(client) |
| await client.login_user("iggy", "iggy") |
| |
| stream = await client.get_stream(stream_name) |
| assert stream is not None |
| assert stream.name == stream_name |
| assert stream.topics_count == 0 |
| |
| @pytest.mark.asyncio |
| async def test_duplicate_stream_creation( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test that creating duplicate streams raises appropriate errors.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError) as exc_info: |
| await iggy_client.create_stream(stream_name) |
| |
| assert "already exists" in str(exc_info.value) |
| |
| @pytest.mark.asyncio |
| async def test_get_nonexistent_stream(self, iggy_client: IggyClient, unique_name): |
| """Test getting a non-existent stream by name or numeric id.""" |
| nonexistent_name = unique_name() |
| |
| stream_by_name = await iggy_client.get_stream(nonexistent_name) |
| assert stream_by_name is None |
| |
| stream_by_id = await iggy_client.get_stream(999999) |
| assert stream_by_id is None |
| |
| @pytest.mark.asyncio |
| async def test_get_stream_before_connect_fails(self, unique_name): |
| """Test get_stream requires an established connection.""" |
| host, port = get_server_config() |
| client = IggyClient(f"{host}:{port}") |
| |
| with pytest.raises(RuntimeError): |
| await client.get_stream(unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_get_stream_before_login_fails(self, unique_name): |
| """Test get_stream requires authentication.""" |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| await client.connect() |
| |
| with pytest.raises(RuntimeError): |
| await client.get_stream(unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_create_stream_before_connect_fails(self, unique_name): |
| """Test create_stream requires an established connection.""" |
| host, port = get_server_config() |
| client = IggyClient(f"{host}:{port}") |
| |
| with pytest.raises(RuntimeError): |
| await client.create_stream(unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_create_stream_before_login_fails(self, unique_name): |
| """Test create_stream requires authentication.""" |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| await client.connect() |
| |
| with pytest.raises(RuntimeError): |
| await client.create_stream(unique_name()) |
| |
| |
| class TestGetStreams: |
| """Test listing streams via get_streams.""" |
| |
| @pytest.mark.asyncio |
| async def test_get_streams_returns_created_streams( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test get_streams returns every stream created during the test.""" |
| # Stream IDs can be reused after deletion, so creation order does not |
| # imply numeric ID order. The client fixture is session-scoped, so |
| # other tests may have created streams; assert on the ones created |
| # here instead of the full server view. |
| created = [unique_name(f"z{index}") for index in range(3, 0, -1)] |
| for name in created: |
| await iggy_client.create_stream(name) |
| |
| streams = await iggy_client.get_streams() |
| created_names = set(created) |
| mine = [stream for stream in streams if stream.name in created_names] |
| |
| assert {stream.name for stream in mine} == created_names |
| assert [stream.id for stream in mine] == sorted(stream.id for stream in mine) |
| assert all(stream.created_at > 0 for stream in mine) |
| assert all(stream.size == 0 for stream in mine) |
| assert all(stream.messages_count == 0 for stream in mine) |
| assert all(stream.topics_count == 0 for stream in mine) |
| |
| @pytest.mark.asyncio |
| async def test_get_streams_reflects_topic_count( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test get_streams reports the topic count for a listed stream.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| streams = await iggy_client.get_streams() |
| listed = next( |
| (stream for stream in streams if stream.name == stream_name), None |
| ) |
| assert listed is not None |
| assert listed.topics_count == 1 |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| assert [topic.name for topic in stream.topics] == [topic_name] |
| |
| @pytest.mark.asyncio |
| async def test_get_streams_returns_same_result_when_called_repeatedly( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test repeated get_streams calls return an identically ordered view.""" |
| await iggy_client.create_stream(unique_name()) |
| |
| first = await iggy_client.get_streams() |
| second = await iggy_client.get_streams() |
| assert [stream.id for stream in first] == [stream.id for stream in second] |
| assert [stream.name for stream in first] == [stream.name for stream in second] |
| |
| @pytest.mark.asyncio |
| async def test_get_streams_requires_connection_and_auth(self): |
| """Test get_streams fails both before connecting and before logging in.""" |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| with pytest.raises(RuntimeError): |
| await client.get_streams() |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.get_streams() |
| |
| |
| class TestUpdateStream: |
| """Test updating streams via update_stream.""" |
| |
| @pytest.mark.asyncio |
| async def test_update_stream_renames_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_stream renames a stream; old name no longer resolves.""" |
| stream_name = unique_name() |
| new_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| before = await iggy_client.get_stream(stream_name) |
| assert before is not None |
| |
| await iggy_client.update_stream(stream_id=stream_name, name=new_name) |
| |
| renamed = await iggy_client.get_stream(new_name) |
| assert renamed is not None |
| assert renamed.name == new_name |
| assert renamed.id == before.id |
| |
| old = await iggy_client.get_stream(stream_name) |
| assert old is None |
| |
| @pytest.mark.asyncio |
| async def test_update_stream_by_numeric_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_stream accepts a numeric stream id.""" |
| stream_name = unique_name() |
| new_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| |
| await iggy_client.update_stream(stream_id=stream.id, name=new_name) |
| |
| renamed = await iggy_client.get_stream(new_name) |
| assert renamed is not None |
| assert renamed.name == new_name |
| |
| @pytest.mark.asyncio |
| async def test_update_stream_forwards_options( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_stream forwards option keys to the server.""" |
| stream_name = unique_name() |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_stream( |
| stream_id=stream_name, |
| name=stream_name, |
| options={"unknown": "value"}, |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "new_name", |
| [ |
| pytest.param("a", id="one-byte"), |
| pytest.param("a" * 255, id="255-byte-ascii"), |
| pytest.param(("é" * 127) + "a", id="255-byte-utf8"), |
| ], |
| ) |
| async def test_update_stream_accepts_name_boundaries( |
| self, iggy_client: IggyClient, unique_name, new_name: str |
| ): |
| """Test update_stream accepts names at the UTF-8 byte boundaries.""" |
| stream_name = unique_name() |
| await iggy_client.create_stream(stream_name) |
| |
| await iggy_client.update_stream(stream_id=stream_name, name=new_name) |
| renamed = await iggy_client.get_stream(new_name) |
| assert renamed is not None |
| assert renamed.name == new_name |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "new_name", |
| [ |
| pytest.param("", id="empty"), |
| pytest.param("a" * 256, id="256-byte-ascii"), |
| pytest.param("é" * 128, id="256-byte-utf8"), |
| pytest.param(("😀" * 63) + "aaaa", id="256-byte-four-byte-utf8"), |
| ], |
| ) |
| async def test_update_stream_rejects_invalid_name_boundaries( |
| self, iggy_client: IggyClient, unique_name, new_name: str |
| ): |
| """Test update_stream rejects empty and 256-byte names.""" |
| stream_name = unique_name() |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_stream(stream_id=stream_name, name=new_name) |
| |
| @pytest.mark.asyncio |
| async def test_update_stream_applies_repeated_updates( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test successive update_stream calls each take effect.""" |
| stream_name = unique_name() |
| first_rename = unique_name() |
| second_rename = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| await iggy_client.update_stream(stream_id=stream_name, name=first_rename) |
| after_first = await iggy_client.get_stream(first_rename) |
| assert after_first is not None |
| assert after_first.name == first_rename |
| assert await iggy_client.get_stream(stream_name) is None |
| |
| await iggy_client.update_stream(stream_id=first_rename, name=second_rename) |
| after_second = await iggy_client.get_stream(second_rename) |
| assert after_second is not None |
| assert after_second.name == second_rename |
| assert await iggy_client.get_stream(first_rename) is None |
| |
| @pytest.mark.asyncio |
| async def test_update_nonexistent_stream_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_stream raises for a non-existent stream.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_stream(stream_id=unique_name(), name=unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_update_stream_to_existing_name_fails_and_current_name_is_a_noop( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_stream rejects conflicts and preserves a self-rename.""" |
| first_stream = unique_name() |
| second_stream = unique_name() |
| |
| await iggy_client.create_stream(first_stream) |
| await iggy_client.create_stream(second_stream) |
| streams = await iggy_client.get_streams() |
| before = next(stream for stream in streams if stream.name == second_stream) |
| before_metadata = ( |
| before.id, |
| before.created_at, |
| before.name, |
| before.size, |
| before.messages_count, |
| before.topics_count, |
| ) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_stream(stream_id=second_stream, name=first_stream) |
| |
| await iggy_client.update_stream(stream_id=second_stream, name=second_stream) |
| |
| streams = await iggy_client.get_streams() |
| after = next(stream for stream in streams if stream.id == before.id) |
| assert ( |
| after.id, |
| after.created_at, |
| after.name, |
| after.size, |
| after.messages_count, |
| after.topics_count, |
| ) == before_metadata |
| |
| @pytest.mark.asyncio |
| async def test_update_stream_requires_connection_and_auth(self, unique_name): |
| """Test update_stream fails both before connecting and before logging in.""" |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| with pytest.raises(RuntimeError): |
| await client.update_stream(stream_id=unique_name(), name=unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.update_stream(stream_id=unique_name(), name=unique_name()) |
| |
| |
| class TestDeleteStream: |
| """Test deleting streams via delete_stream.""" |
| |
| @pytest.mark.asyncio |
| async def test_delete_stream_removes_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_stream removes the stream so it no longer resolves.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| await iggy_client.delete_stream(stream_name) |
| |
| assert await iggy_client.get_stream(stream_name) is None |
| |
| @pytest.mark.asyncio |
| async def test_delete_stream_by_numeric_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_stream accepts a numeric stream id.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| |
| await iggy_client.delete_stream(stream.id) |
| |
| assert await iggy_client.get_stream(stream_name) is None |
| |
| @pytest.mark.asyncio |
| async def test_delete_stream_leaves_other_streams( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_stream removes only the targeted stream.""" |
| stream_to_delete = unique_name() |
| stream_to_keep = unique_name() |
| |
| await iggy_client.create_stream(stream_to_delete) |
| await iggy_client.create_stream(stream_to_keep) |
| |
| await iggy_client.delete_stream(stream_to_delete) |
| |
| assert await iggy_client.get_stream(stream_to_delete) is None |
| kept = await iggy_client.get_stream(stream_to_keep) |
| assert kept is not None |
| assert kept.name == stream_to_keep |
| |
| @pytest.mark.asyncio |
| async def test_delete_nonexistent_stream_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_stream raises for a non-existent stream.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.delete_stream(unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_delete_stream_twice_fails_second_time( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test deleting an already-deleted stream raises on the second call.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| await iggy_client.delete_stream(stream_name) |
| with pytest.raises(RuntimeError): |
| await iggy_client.delete_stream(stream_name) |
| |
| @pytest.mark.asyncio |
| async def test_delete_stream_requires_connection_and_auth(self, unique_name): |
| """Test delete_stream fails both before connecting and before logging in.""" |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| with pytest.raises(RuntimeError): |
| await client.delete_stream(unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.delete_stream(unique_name()) |
| |
| |
| class TestPurgeStream: |
| """Test purging stream messages via purge_stream.""" |
| |
| @pytest.mark.asyncio |
| async def test_purge_stream_clears_messages_but_keeps_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_stream empties the stream while leaving it in place.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| messages = [SendMessage(f"payload-{index}") for index in range(5)] |
| await iggy_client.send_messages(stream_name, topic_name, 0, messages) |
| |
| before = await iggy_client.get_stream(stream_name) |
| assert before is not None |
| assert before.messages_count == 5 |
| |
| await iggy_client.purge_stream(stream_name) |
| |
| after = await iggy_client.get_stream(stream_name) |
| # Purging clears messages only; the stream itself survives (purge is |
| # not delete) and keeps its identity and topics. |
| assert after is not None |
| assert after.messages_count == 0 |
| assert after.id == before.id |
| assert after.name == before.name |
| assert after.topics_count == before.topics_count |
| |
| @pytest.mark.asyncio |
| async def test_purge_empty_stream_succeeds( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_stream is a no-op on a stream with no messages.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| await iggy_client.purge_stream(stream_name) |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| assert stream.messages_count == 0 |
| |
| @pytest.mark.asyncio |
| async def test_purge_stream_is_idempotent_when_called_repeatedly( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_stream succeeds when called repeatedly on the same stream.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| messages = [SendMessage(f"payload-{index}") for index in range(5)] |
| await iggy_client.send_messages(stream_name, topic_name, 0, messages) |
| |
| await iggy_client.purge_stream(stream_name) |
| await iggy_client.purge_stream(stream_name) |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| assert stream.messages_count == 0 |
| |
| @pytest.mark.asyncio |
| async def test_purge_nonexistent_stream_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_stream raises for a non-existent stream.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.purge_stream(unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_purge_stream_requires_connection_and_auth(self, unique_name): |
| """Test purge_stream fails both before connecting and before logging in.""" |
| host, port = get_server_config() |
| wait_for_server(host, port) |
| |
| client = IggyClient(f"{host}:{port}") |
| with pytest.raises(RuntimeError): |
| await client.purge_stream(unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.purge_stream(unique_name()) |
