| # 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. |
| |
| from datetime import timedelta |
| |
| import pytest |
| |
| from apache_iggy import HeaderValue, IggyClient, IggyExpiry, MaxTopicSize, SendMessage |
| |
| from .utils import ( |
| get_server_config, |
| wait_for_ping, |
| wait_for_server, |
| ) |
| |
| |
| class TestCreateTopic: |
| """Test topic creation via create_topic.""" |
| |
| @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_topic( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| prefix: str, |
| min_bytes: int, |
| max_bytes: int, |
| ): |
| """Test topic creation and retrieval.""" |
| stream_name = unique_name() |
| topic_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes) |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=2 |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert topic.partitions_count == 2 |
| assert topic.created_at > 0 |
| assert topic.size == 0 |
| assert len(topic.partitions) == 2 |
| assert all(partition.messages_count == 0 for partition in topic.partitions) |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| 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_topic_invalid_names( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| prefix: str, |
| min_bytes: int, |
| max_bytes: int, |
| ): |
| """Test create_topic enforces byte-length validation.""" |
| stream_name = unique_name() |
| topic_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes) |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_create_and_get_topic_with_numeric_stream_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test topic APIs accept a numeric stream id for create and get operations.""" |
| stream_name = unique_name() |
| topic_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.create_topic( |
| stream=stream.id, name=topic_name, partitions_count=2 |
| ) |
| |
| topic_by_name = await iggy_client.get_topic(stream.id, topic_name) |
| assert topic_by_name is not None |
| assert topic_by_name.name == topic_name |
| assert topic_by_name.partitions_count == 2 |
| |
| topic_by_id = await iggy_client.get_topic(stream.id, topic_by_name.id) |
| assert topic_by_id is not None |
| assert topic_by_id.id == topic_by_name.id |
| assert topic_by_id.name == topic_by_name.name |
| |
| @pytest.mark.asyncio |
| async def test_duplicate_topic_creation(self, iggy_client: IggyClient, unique_name): |
| """Test that creating duplicate topics raises appropriate errors.""" |
| 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 |
| ) |
| |
| with pytest.raises(RuntimeError) as exc_info: |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| assert "already exists" in str(exc_info.value) |
| |
| @pytest.mark.asyncio |
| async def test_topic_names_can_repeat_across_different_streams( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test the same topic name can be created in different streams.""" |
| first_stream_name = unique_name() |
| second_stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(first_stream_name) |
| await iggy_client.create_stream(second_stream_name) |
| |
| await iggy_client.create_topic( |
| stream=first_stream_name, name=topic_name, partitions_count=1 |
| ) |
| await iggy_client.create_topic( |
| stream=second_stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| first_topic = await iggy_client.get_topic(first_stream_name, topic_name) |
| second_topic = await iggy_client.get_topic(second_stream_name, topic_name) |
| |
| assert first_topic is not None |
| assert second_topic is not None |
| assert first_topic.name == topic_name |
| assert second_topic.name == topic_name |
| assert first_topic.partitions_count == 1 |
| assert second_topic.partitions_count == 1 |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize("compression_algorithm", ["gzip", "Gzip", "none", "None"]) |
| async def test_create_topic_with_valid_compression_algorithm( |
| self, iggy_client: IggyClient, unique_name, compression_algorithm: str |
| ): |
| """Test create_topic accepts a supported compression algorithm value.""" |
| 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, |
| compression_algorithm=compression_algorithm, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert topic.partitions_count == 1 |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "compression_algorithm", ["brotli", "deflate", "gzipp", "", " gzip "] |
| ) |
| async def test_create_topic_invalid_compression_algorithm( |
| self, iggy_client: IggyClient, unique_name, compression_algorithm: str |
| ): |
| """Test create_topic rejects unsupported compression algorithm values.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError, match="Unknown compression type"): |
| await iggy_client.create_topic( |
| stream=stream_name, |
| name=topic_name, |
| partitions_count=1, |
| compression_algorithm=compression_algorithm, |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "message_expiry", |
| [ |
| IggyExpiry.ExpireDuration(timedelta(microseconds=1)), |
| IggyExpiry.ExpireDuration(timedelta(seconds=1)), |
| IggyExpiry.ExpireDuration(timedelta(minutes=10)), |
| IggyExpiry.ExpireDuration(timedelta(days=1, seconds=2, microseconds=3)), |
| # days * 86_400 overflows i32 (max ~24,855 days); regression test |
| # for widening the days-to-seconds conversion to i64. |
| IggyExpiry.ExpireDuration(timedelta(days=30_000)), |
| ], |
| ) |
| async def test_create_topic_with_message_expiry( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| message_expiry: IggyExpiry.ExpireDuration, |
| ): |
| """Test create_topic accepts an explicit message expiry.""" |
| 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, |
| message_expiry=message_expiry, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration) |
| assert topic.message_expiry.duration == message_expiry.duration |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "invalid_duration", |
| [ |
| timedelta(seconds=-1), |
| # 0 is the wire sentinel reserved for IggyExpiry.ServerDefault(); |
| # matches MaxTopicSize.Custom(0) rejecting its own sentinel. |
| timedelta(0), |
| # u64::MAX microseconds is the wire sentinel reserved for |
| # IggyExpiry.NeverExpire(); matches MaxTopicSize.Custom(u64::MAX). |
| timedelta(microseconds=2**64 - 1), |
| ], |
| ) |
| async def test_create_topic_rejects_invalid_message_expiry_duration( |
| self, iggy_client: IggyClient, unique_name, invalid_duration: timedelta |
| ): |
| """Test create_topic rejects an ExpireDuration at a reserved boundary.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(ValueError): |
| await iggy_client.create_topic( |
| stream=stream_name, |
| name=topic_name, |
| partitions_count=1, |
| message_expiry=IggyExpiry.ExpireDuration(invalid_duration), |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "invalid_message_expiry", [1, "1s", object(), timedelta(seconds=1)] |
| ) |
| async def test_create_topic_invalid_message_expiry( |
| self, iggy_client: IggyClient, unique_name, invalid_message_expiry |
| ): |
| """Test create_topic rejects non-IggyExpiry message_expiry values.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(TypeError): |
| await iggy_client.create_topic( |
| stream=stream_name, |
| name=topic_name, |
| partitions_count=1, |
| message_expiry=invalid_message_expiry, |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| ("max_topic_size", "expected_kind"), |
| [ |
| (MaxTopicSize.ServerDefault(), "unlimited"), # resolved by create_topic |
| (MaxTopicSize.Unlimited(), "unlimited"), |
| (MaxTopicSize.Custom(2_000_000_000), "custom"), |
| ], |
| ) |
| async def test_create_topic_with_valid_max_topic_size( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| max_topic_size: MaxTopicSize, |
| expected_kind: str, |
| ): |
| """Test create_topic accepts supported maximum topic size values.""" |
| 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, |
| max_topic_size=max_topic_size, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| if expected_kind == "unlimited": |
| assert isinstance(topic.max_topic_size, MaxTopicSize.Unlimited) |
| else: |
| assert isinstance(topic.max_topic_size, MaxTopicSize.Custom) |
| assert isinstance(max_topic_size, MaxTopicSize.Custom) |
| assert topic.max_topic_size.bytes == max_topic_size.bytes |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| ("max_topic_size_bytes", "expected_exception"), |
| [ |
| (4563, RuntimeError), |
| (-1, OverflowError), |
| (2e64, TypeError), |
| (0, ValueError), |
| (2**64 - 1, ValueError), # u64::MAX is reserved for Unlimited |
| ], |
| ) |
| async def test_create_topic_invalid_max_topic_size( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| max_topic_size_bytes, |
| expected_exception, |
| ): |
| """Test create_topic rejects invalid maximum topic size values.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(expected_exception): |
| await iggy_client.create_topic( |
| stream=stream_name, |
| name=topic_name, |
| partitions_count=1, |
| max_topic_size=MaxTopicSize.Custom(max_topic_size_bytes), |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize("partitions_count", [1001, 10000]) |
| async def test_create_topic_invalid_partitions_count( |
| self, iggy_client: IggyClient, unique_name, partitions_count: int |
| ): |
| """Test create_topic rejects partition counts above the supported limit.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError) as exc_info: |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=partitions_count |
| ) |
| |
| assert "Too many partitions" in str(exc_info.value) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize("partitions_count", [0, 1, 1000]) |
| async def test_create_topic_with_valid_partitions_count( |
| self, iggy_client: IggyClient, unique_name, partitions_count: int |
| ): |
| """Test create_topic accepts the maximum supported partitions count.""" |
| 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=partitions_count |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert topic.partitions_count == partitions_count |
| |
| @pytest.mark.asyncio |
| async def test_stream_topics_count_increases_after_multiple_topic_creations( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a stream reports additional topics after multiple creations.""" |
| stream_name = unique_name() |
| first_topic_name = unique_name() |
| second_topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| stream_before = await iggy_client.get_stream(stream_name) |
| assert stream_before is not None |
| assert stream_before.topics_count == 0 |
| |
| await iggy_client.create_topic( |
| stream=stream_name, name=first_topic_name, partitions_count=1 |
| ) |
| await iggy_client.create_topic( |
| stream=stream_name, name=second_topic_name, partitions_count=1 |
| ) |
| |
| stream_after = await iggy_client.get_stream(stream_name) |
| assert stream_after is not None |
| assert stream_after.topics_count == 2 |
| |
| @pytest.mark.asyncio |
| async def test_create_topic_then_reconnect_then_get_topic( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test a topic remains retrievable after reconnecting with a fresh client.""" |
| 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 |
| ) |
| |
| 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") |
| |
| topic = await client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert topic.partitions_count == 1 |
| |
| @pytest.mark.asyncio |
| async def test_create_topic_in_nonexistent_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test creating a topic in a non-existent stream.""" |
| nonexistent_stream = unique_name() |
| topic_name = unique_name() |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.create_topic( |
| stream=nonexistent_stream, name=topic_name, partitions_count=1 |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_create_topic_requires_connection_and_auth(self, unique_name): |
| """Test create_topic 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.create_topic( |
| stream=unique_name(), name=unique_name(), partitions_count=1 |
| ) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.create_topic( |
| stream=unique_name(), name=unique_name(), partitions_count=1 |
| ) |
| |
| |
| class TestGetTopic: |
| """Test topic retrieval via get_topic.""" |
| |
| @pytest.mark.asyncio |
| async def test_get_topic_by_name_and_id(self, iggy_client: IggyClient, unique_name): |
| """Test repeated topic lookup works by both name and numeric id.""" |
| 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 |
| ) |
| |
| topic_by_name = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic_by_name is not None |
| |
| topic_by_name_again = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic_by_name_again is not None |
| assert topic_by_name_again.id == topic_by_name.id |
| assert topic_by_name_again.name == topic_by_name.name |
| |
| topic_by_id = await iggy_client.get_topic(stream_name, topic_by_name.id) |
| assert topic_by_id is not None |
| assert topic_by_id.id == topic_by_name.id |
| assert topic_by_id.name == topic_by_name.name |
| |
| @pytest.mark.asyncio |
| async def test_get_topic_partitions(self, iggy_client: IggyClient, unique_name): |
| """Test TopicDetails.partitions returns one Partition per partition.""" |
| 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=3 |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert len(topic.partitions) == 3 |
| assert [partition.id for partition in topic.partitions] == [0, 1, 2] |
| for partition in topic.partitions: |
| assert partition.created_at > 0 |
| assert partition.segments_count == 1 |
| assert partition.current_offset == 0 |
| assert partition.size == 0 |
| assert partition.messages_count == 0 |
| |
| @pytest.mark.asyncio |
| async def test_get_nonexistent_topic(self, iggy_client: IggyClient, unique_name): |
| """Test getting a non-existent topic by name or numeric id.""" |
| stream_name = unique_name() |
| nonexistent_topic_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| topic_by_name = await iggy_client.get_topic(stream_name, nonexistent_topic_name) |
| assert topic_by_name is None |
| |
| topic_by_id = await iggy_client.get_topic(stream_name, 999999) |
| assert topic_by_id is None |
| |
| @pytest.mark.asyncio |
| async def test_get_topic_in_nonexistent_stream( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test getting a topic from a non-existent stream returns no topic.""" |
| nonexistent_stream = unique_name() |
| topic_name = unique_name() |
| |
| topic = await iggy_client.get_topic(nonexistent_stream, topic_name) |
| assert topic is None |
| |
| @pytest.mark.asyncio |
| async def test_get_topic_requires_connection_and_auth(self, unique_name): |
| """Test get_topic 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_topic(unique_name(), unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.get_topic(unique_name(), unique_name()) |
| |
| |
| class TestGetTopics: |
| """Test listing topics in a stream via get_topics.""" |
| |
| @pytest.mark.asyncio |
| async def test_get_topics_in_empty_stream_returns_empty_list( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test get_topics returns an empty list for a stream with no topics.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| topics = await iggy_client.get_topics(stream_name) |
| assert topics == [] |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize("topic_count", [1, 3, 5]) |
| async def test_get_topics_returns_all_created_topics( |
| self, iggy_client: IggyClient, unique_name, topic_count: int |
| ): |
| """Test get_topics returns every created topic, ordered by id ascending.""" |
| stream_name = unique_name() |
| # Reverse-alphabetical names so that id-ascending (creation) order |
| # and name order disagree, proving the list isn't accidentally |
| # name-sorted. |
| topic_names = [unique_name() for _ in range(topic_count, 0, -1)] |
| |
| await iggy_client.create_stream(stream_name) |
| for name in topic_names: |
| await iggy_client.create_topic( |
| stream=stream_name, name=name, partitions_count=1 |
| ) |
| |
| topics = await iggy_client.get_topics(stream_name) |
| assert len(topics) == len(topic_names) |
| assert [topic.name for topic in topics] == topic_names |
| assert [topic.id for topic in topics] == sorted(topic.id for topic in topics) |
| # Only assert fields supplied at creation or deterministically set by |
| # the server: each topic keeps its partition count and a fresh topic |
| # holds no messages. The numeric id is server-assigned and not checked. |
| assert all(topic.partitions_count == 1 for topic in topics) |
| assert all(topic.messages_count == 0 for topic in topics) |
| |
| @pytest.mark.asyncio |
| async def test_get_topics_returns_same_result_when_called_repeatedly( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test repeated get_topics calls return an identically ordered view.""" |
| stream_name = unique_name() |
| topic_names = [unique_name() for _ in range(3)] |
| |
| await iggy_client.create_stream(stream_name) |
| for name in topic_names: |
| await iggy_client.create_topic( |
| stream=stream_name, name=name, partitions_count=1 |
| ) |
| |
| first = await iggy_client.get_topics(stream_name) |
| second = await iggy_client.get_topics(stream_name) |
| assert [topic.name for topic in first] == topic_names |
| assert [topic.id for topic in first] == [topic.id for topic in second] |
| assert [topic.name for topic in first] == [topic.name for topic in second] |
| |
| @pytest.mark.asyncio |
| async def test_get_topics_accepts_numeric_stream_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test get_topics works when the stream is referenced by numeric id.""" |
| 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 |
| ) |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| |
| topics_by_name = await iggy_client.get_topics(stream_name) |
| topics_by_id = await iggy_client.get_topics(stream.id) |
| assert len(topics_by_name) == 1 |
| assert len(topics_by_id) == 1 |
| assert topics_by_id[0].name == topic_name |
| |
| @pytest.mark.asyncio |
| async def test_get_topics_in_nonexistent_stream_returns_empty( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test get_topics returns an empty list for a non-existent stream.""" |
| topics = await iggy_client.get_topics(unique_name()) |
| assert topics == [] |
| |
| @pytest.mark.asyncio |
| async def test_get_topics_requires_connection_and_auth(self, unique_name): |
| """Test get_topics 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_topics(unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.get_topics(unique_name()) |
| |
| |
| class TestUpdateTopic: |
| """Test updating topics via update_topic.""" |
| |
| @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_update_topic_invalid_names( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| prefix: str, |
| min_bytes: int, |
| max_bytes: int, |
| ): |
| """Test update_topic enforces byte-length validation.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| invalid_name = unique_name(prefix, min_bytes=min_bytes, max_bytes=max_bytes) |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=topic_name, name=invalid_name |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_renames_topic( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic renames a topic; old name no longer resolves.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| new_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=topic_name, name=new_name |
| ) |
| |
| renamed = await iggy_client.get_topic(stream_name, new_name) |
| assert renamed is not None |
| assert renamed.name == new_name |
| |
| old = await iggy_client.get_topic(stream_name, topic_name) |
| assert old is None |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_preserves_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic keeps the same numeric id after a rename.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| new_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| before = await iggy_client.get_topic(stream_name, topic_name) |
| assert before is not None |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=topic_name, name=new_name |
| ) |
| |
| after = await iggy_client.get_topic(stream_name, new_name) |
| assert after is not None |
| assert after.id == before.id |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_by_numeric_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic accepts a numeric topic id.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| new_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=topic.id, name=new_name |
| ) |
| |
| renamed = await iggy_client.get_topic(stream_name, new_name) |
| assert renamed is not None |
| assert renamed.name == new_name |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize("compression_algorithm", ["gzip", "none"]) |
| async def test_update_topic_with_valid_compression_algorithm( |
| self, iggy_client: IggyClient, unique_name, compression_algorithm: str |
| ): |
| """Test update_topic accepts a supported compression algorithm.""" |
| 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 |
| ) |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| compression_algorithm=compression_algorithm, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert topic.compression_algorithm == compression_algorithm |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize("compression_algorithm", ["brotli", "gzipp", ""]) |
| async def test_update_topic_invalid_compression_algorithm( |
| self, iggy_client: IggyClient, unique_name, compression_algorithm: str |
| ): |
| """Test update_topic rejects unsupported compression algorithm values.""" |
| 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 |
| ) |
| |
| with pytest.raises(RuntimeError, match="Unknown compression type"): |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| compression_algorithm=compression_algorithm, |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "invalid_message_expiry", [1, "1s", object(), timedelta(seconds=1)] |
| ) |
| async def test_update_topic_invalid_message_expiry( |
| self, iggy_client: IggyClient, unique_name, invalid_message_expiry |
| ): |
| """Test update_topic rejects non-IggyExpiry message_expiry values.""" |
| 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 |
| ) |
| |
| with pytest.raises(TypeError): |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| message_expiry=invalid_message_expiry, |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| ("max_topic_size_bytes", "expected_exception"), |
| [ |
| (-1, OverflowError), |
| (2e64, TypeError), |
| (0, ValueError), |
| (2**64 - 1, ValueError), # u64::MAX is reserved for Unlimited |
| ], |
| ) |
| async def test_update_topic_invalid_max_topic_size( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| max_topic_size_bytes, |
| expected_exception, |
| ): |
| """Test update_topic rejects invalid maximum topic size values.""" |
| 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 |
| ) |
| |
| with pytest.raises(expected_exception): |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| max_topic_size=MaxTopicSize.Custom(max_topic_size_bytes), |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_with_message_expiry( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic accepts an explicit message expiry.""" |
| 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 |
| ) |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| message_expiry=IggyExpiry.ExpireDuration(timedelta(minutes=10)), |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| assert isinstance(topic.message_expiry, IggyExpiry.ExpireDuration) |
| assert topic.message_expiry.duration == timedelta(minutes=10) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| "invalid_duration", |
| [ |
| timedelta(seconds=-1), |
| # 0 is the wire sentinel reserved for IggyExpiry.ServerDefault(); |
| # matches MaxTopicSize.Custom(0) rejecting its own sentinel. |
| timedelta(0), |
| # u64::MAX microseconds is the wire sentinel reserved for |
| # IggyExpiry.NeverExpire(); matches MaxTopicSize.Custom(u64::MAX). |
| timedelta(microseconds=2**64 - 1), |
| ], |
| ) |
| async def test_update_topic_rejects_invalid_message_expiry_duration( |
| self, iggy_client: IggyClient, unique_name, invalid_duration: timedelta |
| ): |
| """Test update_topic rejects an ExpireDuration at a reserved boundary.""" |
| 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 |
| ) |
| |
| with pytest.raises(ValueError): |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| message_expiry=IggyExpiry.ExpireDuration(invalid_duration), |
| ) |
| |
| @pytest.mark.asyncio |
| @pytest.mark.parametrize( |
| ("max_topic_size", "expected_kind"), |
| [ |
| (MaxTopicSize.ServerDefault(), "server_default"), |
| (MaxTopicSize.Custom(2_000_000_000), "custom"), |
| (MaxTopicSize.Unlimited(), "unlimited"), |
| ], |
| ) |
| async def test_update_topic_with_valid_max_topic_size( |
| self, |
| iggy_client: IggyClient, |
| unique_name, |
| max_topic_size: MaxTopicSize, |
| expected_kind: str, |
| ): |
| """Test update_topic accepts supported maximum topic size values.""" |
| 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 |
| ) |
| |
| created = await iggy_client.get_topic(stream_name, topic_name) |
| assert created is not None |
| resolved_at_creation = created.max_topic_size |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| max_topic_size=max_topic_size, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.name == topic_name |
| if expected_kind == "server_default": |
| # Every setting rides the options block and 0 is its "resolve the |
| # default" sentinel, so a ServerDefault update carries no key at all |
| # and the topic keeps the value admission resolved when it was |
| # created. Resetting a setting back to the node default is |
| # deliberately not expressible. |
| assert isinstance(topic.max_topic_size, type(resolved_at_creation)) |
| assert not isinstance(topic.max_topic_size, MaxTopicSize.ServerDefault) |
| elif expected_kind == "unlimited": |
| assert isinstance(topic.max_topic_size, MaxTopicSize.Unlimited) |
| else: |
| assert isinstance(topic.max_topic_size, MaxTopicSize.Custom) |
| assert isinstance(max_topic_size, MaxTopicSize.Custom) |
| assert topic.max_topic_size.bytes == max_topic_size.bytes |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_applies_repeated_updates( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test successive update_topic calls each take effect.""" |
| stream_name = unique_name() |
| topic_name = unique_name() |
| first_rename = unique_name() |
| second_rename = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_name, partitions_count=1 |
| ) |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=topic_name, name=first_rename |
| ) |
| after_first = await iggy_client.get_topic(stream_name, first_rename) |
| assert after_first is not None |
| assert after_first.name == first_rename |
| assert await iggy_client.get_topic(stream_name, topic_name) is None |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=first_rename, name=second_rename |
| ) |
| after_second = await iggy_client.get_topic(stream_name, second_rename) |
| assert after_second is not None |
| assert after_second.name == second_rename |
| assert await iggy_client.get_topic(stream_name, first_rename) is None |
| |
| @pytest.mark.asyncio |
| async def test_update_nonexistent_topic_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic raises for a non-existent topic.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=unique_name(), name=unique_name() |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_in_nonexistent_stream_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic raises when the stream does not exist.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_topic( |
| stream_id=unique_name(), topic_id=unique_name(), name=unique_name() |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_to_existing_name_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test update_topic rejects renaming a topic to a name already in use.""" |
| stream_name = unique_name() |
| first_topic = unique_name() |
| second_topic = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=first_topic, partitions_count=1 |
| ) |
| await iggy_client.create_topic( |
| stream=stream_name, name=second_topic, partitions_count=1 |
| ) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_topic( |
| stream_id=stream_name, topic_id=second_topic, name=first_topic |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_requires_connection_and_auth(self, unique_name): |
| """Test update_topic 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_topic( |
| stream_id=unique_name(), topic_id=unique_name(), name=unique_name() |
| ) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.update_topic( |
| stream_id=unique_name(), topic_id=unique_name(), name=unique_name() |
| ) |
| |
| |
| class TestDeleteTopic: |
| """Test deleting topics via delete_topic.""" |
| |
| @pytest.mark.asyncio |
| async def test_delete_topic_removes_topic( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_topic removes the topic and drops the stream count.""" |
| 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 |
| ) |
| |
| await iggy_client.delete_topic(stream_name, topic_name) |
| |
| assert await iggy_client.get_topic(stream_name, topic_name) is None |
| assert await iggy_client.get_topics(stream_name) == [] |
| |
| stream = await iggy_client.get_stream(stream_name) |
| assert stream is not None |
| assert stream.topics_count == 0 |
| |
| @pytest.mark.asyncio |
| async def test_delete_topic_by_numeric_id( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_topic accepts a numeric topic id.""" |
| 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 |
| ) |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| |
| await iggy_client.delete_topic(stream_name, topic.id) |
| |
| assert await iggy_client.get_topic(stream_name, topic_name) is None |
| |
| @pytest.mark.asyncio |
| async def test_delete_topic_leaves_other_topics( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_topic removes only the targeted topic.""" |
| stream_name = unique_name() |
| topic_to_delete = unique_name() |
| topic_to_keep = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_to_delete, partitions_count=1 |
| ) |
| await iggy_client.create_topic( |
| stream=stream_name, name=topic_to_keep, partitions_count=1 |
| ) |
| |
| await iggy_client.delete_topic(stream_name, topic_to_delete) |
| |
| remaining = await iggy_client.get_topics(stream_name) |
| assert len(remaining) == 1 |
| assert remaining[0].name == topic_to_keep |
| |
| @pytest.mark.asyncio |
| async def test_delete_nonexistent_topic_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test delete_topic raises for a non-existent topic.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.delete_topic(stream_name, unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_delete_topic_twice_fails_second_time( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test deleting an already-deleted topic raises on the second call.""" |
| 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 |
| ) |
| |
| await iggy_client.delete_topic(stream_name, topic_name) |
| with pytest.raises(RuntimeError): |
| await iggy_client.delete_topic(stream_name, topic_name) |
| |
| @pytest.mark.asyncio |
| async def test_delete_topic_requires_connection_and_auth(self, unique_name): |
| """Test delete_topic 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_topic(unique_name(), unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.delete_topic(unique_name(), unique_name()) |
| |
| |
| class TestPurgeTopic: |
| """Test purging topic messages via purge_topic.""" |
| |
| @pytest.mark.asyncio |
| async def test_purge_topic_clears_messages_but_keeps_topic( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_topic empties the topic 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_topic(stream_name, topic_name) |
| assert before is not None |
| assert before.messages_count == 5 |
| |
| await iggy_client.purge_topic(stream_name, topic_name) |
| |
| after = await iggy_client.get_topic(stream_name, topic_name) |
| assert after is not None |
| assert after.messages_count == 0 |
| assert after.size == 0 |
| # Purging clears messages and size only; topic config is unchanged. |
| assert after.id == before.id |
| assert after.name == before.name |
| assert after.created_at == before.created_at |
| assert after.partitions_count == before.partitions_count |
| assert after.compression_algorithm == before.compression_algorithm |
| assert isinstance(before.message_expiry, IggyExpiry.NeverExpire) |
| assert isinstance(after.message_expiry, IggyExpiry.NeverExpire) |
| assert isinstance(before.max_topic_size, MaxTopicSize.Unlimited) |
| assert isinstance(after.max_topic_size, MaxTopicSize.Unlimited) |
| |
| @pytest.mark.asyncio |
| async def test_purge_empty_topic_succeeds( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_topic is a no-op on a topic with no messages.""" |
| 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 |
| ) |
| |
| await iggy_client.purge_topic(stream_name, topic_name) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.messages_count == 0 |
| |
| @pytest.mark.asyncio |
| async def test_purge_topic_is_idempotent_when_called_repeatedly( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_topic succeeds when called repeatedly on the same topic.""" |
| 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_topic(stream_name, topic_name) |
| await iggy_client.purge_topic(stream_name, topic_name) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.messages_count == 0 |
| assert topic.size == 0 |
| |
| @pytest.mark.asyncio |
| async def test_purge_nonexistent_topic_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_topic raises for a non-existent topic in an existing stream.""" |
| stream_name = unique_name() |
| |
| await iggy_client.create_stream(stream_name) |
| |
| with pytest.raises(RuntimeError): |
| await iggy_client.purge_topic(stream_name, unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_purge_topic_in_nonexistent_stream_fails( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Test purge_topic raises when the stream does not exist.""" |
| with pytest.raises(RuntimeError): |
| await iggy_client.purge_topic(unique_name(), unique_name()) |
| |
| @pytest.mark.asyncio |
| async def test_purge_topic_requires_connection_and_auth(self, unique_name): |
| """Test purge_topic 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_topic(unique_name(), unique_name()) |
| |
| await client.connect() |
| with pytest.raises(RuntimeError): |
| await client.purge_topic(unique_name(), unique_name()) |
| |
| |
| class TestTopicOptions: |
| """Tests for the option catalog and the options a topic reports.""" |
| |
| @pytest.mark.asyncio |
| async def test_topic_options_round_trip(self, iggy_client: IggyClient, unique_name): |
| """Options a client sets come back readable, split by provenance.""" |
| 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, |
| options={"enforce_fsync": "true", "segment_size": "128 MiB"}, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| # Options come back through the same typed dict message user headers |
| # use, so the scalar helper reads them the same way. |
| explicit = topic.options.to_scalar_dict() |
| assert explicit["enforce_fsync"] is True |
| assert explicit["segment_size"] == 128 * 1024 * 1024 |
| # Keys the client left alone are resolved by admission and reported |
| # separately, so an operator can tell chosen from defaulted. |
| derived = topic.derived_options.to_scalar_dict() |
| assert "max_topic_size" in derived |
| assert "enforce_fsync" not in derived |
| |
| topics = await iggy_client.get_topics(stream_name) |
| listed = next(entry for entry in topics if entry.name == topic_name) |
| assert listed.options.to_scalar_dict()["enforce_fsync"] is True |
| |
| @pytest.mark.asyncio |
| async def test_update_topic_options_reach_the_server( |
| self, iggy_client: IggyClient, unique_name |
| ): |
| """Options passed to update_topic are applied, and gated by key.""" |
| 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 |
| ) |
| |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| options={"compression_algorithm": "gzip"}, |
| ) |
| |
| topic = await iggy_client.get_topic(stream_name, topic_name) |
| assert topic is not None |
| assert topic.compression_algorithm == "gzip" |
| |
| # A create-time key is refused by name, so nothing re-pushes it to the |
| # partitions of a live topic. |
| with pytest.raises(RuntimeError): |
| await iggy_client.update_topic( |
| stream_id=stream_name, |
| topic_id=topic_name, |
| name=topic_name, |
| options={"segment_size": "2 MiB"}, |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_describe_options_lists_the_topic_catalog( |
| self, iggy_client: IggyClient |
| ): |
| """The catalog is what tells a client which keys create accepts.""" |
| specs = await iggy_client.describe_options("topic") |
| |
| by_key = {spec.key: spec for spec in specs} |
| assert "segment_size" in by_key |
| assert "enforce_fsync" in by_key |
| segment_size = by_key["segment_size"] |
| assert segment_size.kind == "uint64" |
| # The default is the same HeaderValue type message headers carry, so it |
| # arrives as the variant matching the key's kind. |
| default = segment_size.default_value |
| assert isinstance(default, HeaderValue.UnsignedInt64) |
| assert default.value == 1024 * 1024 * 1024 |
| assert segment_size.description |
| |
| # Streams and users have no catalog keys yet. |
| assert await iggy_client.describe_options("stream") == [] |
| assert await iggy_client.describe_options("user") == [] |
| |
| @pytest.mark.asyncio |
| async def test_describe_options_rejects_an_unknown_scope( |
| self, iggy_client: IggyClient |
| ): |
| """Test describe_options raises ValueError for an unknown scope.""" |
| with pytest.raises(ValueError): |
| await iggy_client.describe_options("partition") |
