# 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 (
    Durability,
    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,
            durability=Durability.PERSISTED,
            options={"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["durability"] == "persisted"
        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 "durability" 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()["durability"] == "persisted"

    @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 "durability" in by_key
        assert "consumer_offset_durability" 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")