slbotbm commented on code in PR #3613:
URL: https://github.com/apache/iggy/pull/3613#discussion_r3593328769
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -16,13 +16,26 @@
# under the License.
import asyncio
+import logging
import uuid
+from typing import TypeAlias
import pytest
-from apache_iggy import IggyClient, PollingStrategy
+from apache_iggy import (
+ HeaderKey,
+ HeaderValue,
+ IggyClient,
+ PollingStrategy,
+ UserHeaders,
+)
from apache_iggy import SendMessage as Message
+HTTP_CREATED = 201
+JsonValue: TypeAlias = (
+ None | bool | int | float | str | list["JsonValue"] | dict[str,
"JsonValue"]
+)
Review Comment:
These are never used. Can be removed.
##########
foreign/python/tests/test_message_operations.py:
##########
Review Comment:
The logging setup in the tests you have defined aren't really useful.
##########
foreign/python/tests/test_message_operations.py:
##########
@@ -136,8 +149,160 @@ async def test_message_properties(self, iggy_client:
IggyClient, unique_name):
assert isinstance(msg.offset(), int) and msg.offset() >= 0
assert isinstance(msg.id(), int) and msg.id() > 0
assert isinstance(msg.timestamp(), int) and msg.timestamp() > 0
+ assert isinstance(msg.origin_timestamp(), int) and
msg.origin_timestamp() > 0
assert isinstance(msg.checksum(), int)
assert isinstance(msg.length(), int) and msg.length() > 0
+ assert msg.user_headers() is None
+
+ @pytest.mark.asyncio
+ async def test_message_user_headers_round_trip(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test plain user headers round-trip through the typed
representation."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ message_id = 123456789
+ user_headers = {
+ "content-type": "application/json",
+ "trace-blob": b"\x00\x01",
+ "is-retry": False,
+ "attempt": 3,
+ "score": 0.99,
+ }
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+
+ await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[
+ Message(
+ "header round trip",
+ user_headers=user_headers,
+ id=message_id,
+ )
+ ],
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.Last(),
+ count=1,
+ auto_commit=True,
+ )
+
+ assert len(polled_messages) == 1
+ message = polled_messages[0]
+ assert message.id() == message_id
+ typed_headers = message.user_headers()
+ assert typed_headers is not None
+ assert typed_headers == {
+ HeaderKey.String("content-type"):
HeaderValue.String("application/json"),
+ HeaderKey.String("trace-blob"): HeaderValue.Raw(b"\x00\x01"),
+ HeaderKey.String("is-retry"): HeaderValue.Bool(False),
+ HeaderKey.String("attempt"): HeaderValue.UnsignedInt8(3),
+ HeaderKey.String("score"): HeaderValue.Float64(0.99),
+ }
+ assert typed_headers.to_scalar_dict() == user_headers
+ assert isinstance(message.origin_timestamp(), int)
+ assert message.origin_timestamp() > 0
+
+ @pytest.mark.asyncio
+ async def test_typed_message_user_headers_round_trip(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test typed user headers preserve explicit header kinds."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+ typed_key = HeaderKey.UnsignedInt128(42)
+ typed_value = HeaderValue.UnsignedInt128(2**96)
+ user_headers: dict[HeaderKey, HeaderValue] = {
+ typed_key: typed_value,
+ HeaderKey.String("float32"): HeaderValue.Float32(1.25),
+ }
Review Comment:
This can be made into one dict. No need to declare extras vars.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
