change tests

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

.ci/ubuntu/gha-setup.sh

@@ -7,7 +7,7 @@ set -o xtrace
 script_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
 readonly script_dir
 echo "[INFO] script_dir: '$script_dir'"
-readonly rabbitmq_image=rabbitmq:4.1.0-beta.4-management-alpine
+readonly rabbitmq_image=rabbitmq:4.1.0-alpine
 
 
 readonly docker_name_prefix='rabbitmq-amqp-python-client'

.gitignore

@@ -15,3 +15,4 @@ __pycache__/
 local*
 .githooks/
 .venv/
+.ci/ubuntu/log/*

examples/getting_started/getting_started.py

@@ -13,6 +13,7 @@
     OutcomeState,
     QuorumQueueSpecification,
 )
+from rabbitmq_amqp_python_client.utils import string_to_bytes, bytes_to_string
 
 MESSAGES_TO_PUBLISH = 100
 
@@ -24,7 +25,7 @@ def __init__(self):
         self._count = 0
 
     def on_amqp_message(self, event: Event):
-        print("received message: {} ".format("".join(map(chr, event.message.body))))
+        print("received message: {} ".format(bytes_to_string(event.message.body)))
 
         # accepting
         self.delivery_context.accept(event)
@@ -123,7 +124,7 @@ def main() -> None:
     for i in range(MESSAGES_TO_PUBLISH):
         print("publishing")
         status = publisher.publish(
-            Message(body=str.encode("test message {} ".format(i)))
+            Message(body=string_to_bytes("test message {} ".format(i)))
         )
         if status.remote_state == OutcomeState.ACCEPTED:
             print("message accepted")

rabbitmq_amqp_python_client/utils.py

@@ -7,3 +7,29 @@ def validate_annotations(annotations: []) -> bool:  # type: ignore
             validated = False
             return validated
     return validated
+
+
+def bytes_to_string(body: bytes) -> str:
+    """
+    Convert the body of a message to a string.
+
+    Args:
+        body: The body of the message
+
+    Returns:
+        str: The string representation of the body
+    """
+    return "".join(map(chr, body))
+
+
+def string_to_bytes(body: str) -> bytes:
+    """
+    Convert a string to the body of a message.
+
+    Args:
+        body: The string to convert
+
+    Returns:
+        bytes: The byte representation of the string
+    """
+    return str.encode(body)

tests/conftest.py

@@ -263,8 +263,7 @@ def __init__(self):
         self._received = 0
 
     def on_message(self, event: Event):
-        annotations = {}
-        annotations[symbol("x-opt-string")] = "x-test1"
+        annotations = {symbol("x-opt-string"): "x-test1"}
         self.delivery_context.requeue_with_annotations(event, annotations)
         self._received = self._received + 1
         if self._received == 1000:

tests/test_consumer.py

@@ -5,6 +5,7 @@
     Environment,
     QuorumQueueSpecification,
 )
+from rabbitmq_amqp_python_client.utils import bytes_to_string
 
 from .conftest import (
     ConsumerTestException,
@@ -42,7 +43,7 @@ def test_consumer_sync_queue_accept(connection: Connection) -> None:
     # consumer synchronously without handler
     for i in range(messages_to_send):
         message = consumer.consume()
-        if message.body == "test" + str(i):
+        if bytes_to_string(message.body) == "test{}".format(i):
             consumed = consumed + 1
 
     consumer.close()

tests/test_publisher.py

@@ -13,6 +13,7 @@
     StreamSpecification,
     ValidationCodeException,
 )
+from rabbitmq_amqp_python_client.utils import string_to_bytes
 
 from .http_requests import delete_all_connections
 from .utils import create_binding, publish_per_message
@@ -34,7 +35,7 @@ def test_publish_queue(connection: Connection) -> None:
         publisher = connection.publisher(
             destination=AddressHelper.queue_address(queue_name)
         )
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(Message(body=string_to_bytes("test")))
         if status.remote_state == OutcomeState.ACCEPTED:
             accepted = True
     except Exception:
@@ -130,7 +131,7 @@ def test_publish_to_invalid_destination(connection: Connection) -> None:
     publisher = None
     try:
         publisher = connection.publisher("/invalid-destination/" + queue_name)
-        publisher.publish(Message(body="test"))
+        publisher.publish(Message(body=string_to_bytes("test")))
     except ArgumentOutOfRangeException:
         raised = True
     except Exception:
@@ -147,7 +148,7 @@ def test_publish_per_message_to_invalid_destination(connection: Connection) -> N
     queue_name = "test-queue-1"
     raised = False
 
-    message = Message(body="test")
+    message = Message(body=string_to_bytes("test"))
     message = AddressHelper.message_to_address_helper(
         message, "/invalid_destination/" + queue_name
     )
@@ -179,7 +180,7 @@ def test_publish_per_message_both_address(connection: Connection) -> None:
     )
 
     try:
-        message = Message(body="test")
+        message = Message(body=string_to_bytes("test"))
         message = AddressHelper.message_to_address_helper(
             message, AddressHelper.queue_address(queue_name)
         )
@@ -212,7 +213,7 @@ def test_publish_exchange(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher(addr)
-        status = publisher.publish(Message(body="test"))
+        status = publisher.publish(Message(body=string_to_bytes("test")))
         if status.ACCEPTED:
             accepted = True
     except Exception:
@@ -244,7 +245,7 @@ def test_publish_purge(connection: Connection) -> None:
             destination=AddressHelper.queue_address(queue_name)
         )
         for i in range(messages_to_publish):
-            publisher.publish(Message(body="test"))
+            publisher.publish(Message(body=string_to_bytes("test")))
     except Exception:
         raised = True
 
@@ -289,7 +290,7 @@ def test_disconnection_reconnection() -> None:
                 # simulate a disconnection
                 delete_all_connections()
             try:
-                publisher.publish(Message(body="test"))
+                publisher.publish(Message(body=string_to_bytes("test")))
 
             except ConnectionClosed:
                 disconnected = True
@@ -331,8 +332,7 @@ def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
     )
 
     for i in range(messages_to_send):
-
-        publisher.publish(Message(body="test"))
+        publisher.publish(Message(body=string_to_bytes("test")))
 
 
 def test_publish_per_message_exchange(connection: Connection) -> None:

tests/utils.py

@@ -16,6 +16,7 @@
     Publisher,
     QuorumQueueSpecification,
 )
+from rabbitmq_amqp_python_client.utils import string_to_bytes
 
 
 def publish_messages(
@@ -26,25 +27,26 @@ def publish_messages(
 ) -> None:
     annotations = {}
     if filters is not None:
-        for filter in filters:
-            annotations = {"x-stream-filter-value": filter}
+        for filterItem in filters:
+            annotations = {"x-stream-filter-value": filterItem}
 
     publisher = connection.publisher("/queues/" + queue_name)
     # publish messages_to_send messages
     for i in range(messages_to_send):
-        publisher.publish(Message(body="test" + str(i), annotations=annotations))
+        publisher.publish(
+            Message(body=string_to_bytes("test{}".format(i)), annotations=annotations)
+        )
     publisher.close()
 
 
 def publish_per_message(publisher: Publisher, addr: str) -> Delivery:
-    message = Message(body="test")
+    message = Message(body=string_to_bytes("test"))
     message = AddressHelper.message_to_address_helper(message, addr)
     status = publisher.publish(message)
     return status
 
 
 def setup_dead_lettering(management: Management) -> str:
-
     exchange_dead_lettering = "exchange-dead-letter"
     queue_dead_lettering = "queue-dead-letter"
     binding_key = "key_dead_letter"
@@ -72,7 +74,6 @@ def setup_dead_lettering(management: Management) -> str:
 def create_binding(
     management: Management, exchange_name: str, queue_name: str, routing_key: str
 ) -> str:
-
     management.declare_exchange(ExchangeSpecification(name=exchange_name))
 
     management.declare_queue(QuorumQueueSpecification(name=queue_name))
@@ -89,7 +90,6 @@ def create_binding(
 
 
 def cleanup_dead_lettering(management: Management, bind_path: str) -> None:
-
     exchange_dead_lettering = "exchange-dead-letter"
     queue_dead_lettering = "queue-dead-letter"