converter

Signed-off-by: Gabriele Santomaggio <[email protected]>

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-alpine
+readonly rabbitmq_image=rabbitmq:4.1.0-beta.4-management-alpine
 
 
 readonly docker_name_prefix='rabbitmq-amqp-python-client'

examples/tls/tls_example.py

@@ -1,4 +1,5 @@
 # type: ignore
+import os
 import sys
 from traceback import print_exception
 
@@ -79,15 +80,14 @@ def create_connection(environment: Environment) -> Connection:
 
 
 def main() -> None:
-
     exchange_name = "test-exchange"
     queue_name = "example-queue"
     routing_key = "routing-key"
     ca_p12_store = ".ci/certs/ca.p12"
     ca_cert_file = ".ci/certs/ca_certificate.pem"
-    client_cert = ".ci/certs/client_certificate.pem"
-    client_key = ".ci/certs/client_key.pem"
-    client_p12_store = ".ci/certs/client.p12"
+    client_cert = ".ci/certs/client_localhost_certificate.pem"
+    client_key = ".ci/certs/client_localhost_key.pem"
+    client_p12_store = ".ci/certs/client_localhost.p12"
     uri = "amqps://guest:guest@localhost:5671/"
 
     if sys.platform == "win32":
@@ -138,6 +138,9 @@ def main() -> None:
                 "connection failed. working directory should be project root"
             )
     else:
+        print(" ca_cert_file {}".format(os.path.isfile(ca_cert_file)))
+        print(" client_cert {}".format(os.path.isfile(client_cert)))
+        print(" client_key {}".format(os.path.isfile(client_key)))
         environment = Environment(
             uri,
             ssl_context=PosixSslConfigurationContext(

rabbitmq_amqp_python_client/__init__.py

@@ -5,6 +5,7 @@
 from .common import ExchangeType, QueueType
 from .connection import Connection
 from .consumer import Consumer
+from .converter import Converter
 from .entities import (
     ExchangeCustomSpecification,
     ExchangeSpecification,
@@ -91,4 +92,5 @@
     "ExchangeCustomSpecification",
     "RecoveryConfiguration",
     "OAuth2Options",
+    "Converter",
 ]

rabbitmq_amqp_python_client/converter.py

@@ -0,0 +1,27 @@
+class Converter:
+
+    @staticmethod
+    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))
+
+    @staticmethod
+    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)

rabbitmq_amqp_python_client/utils.py

@@ -7,29 +7,3 @@ 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/test_consumer.py

@@ -3,10 +3,7 @@
     ArgumentOutOfRangeException,
     Connection,
     Environment,
-    QuorumQueueSpecification,
-)
-from rabbitmq_amqp_python_client.utils import (
-    bytes_to_string,
+    QuorumQueueSpecification, Converter,
 )
 
 from .conftest import (
@@ -45,7 +42,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 bytes_to_string(message.body) == "test{}".format(i):
+        if Converter.bytes_to_string(message.body) == "test{}".format(i):
             consumed = consumed + 1
 
     consumer.close()

tests/test_publisher.py

@@ -11,10 +11,7 @@
     QuorumQueueSpecification,
     RecoveryConfiguration,
     StreamSpecification,
-    ValidationCodeException,
-)
-from rabbitmq_amqp_python_client.utils import (
-    string_to_bytes,
+    ValidationCodeException, Converter,
 )
 
 from .http_requests import delete_all_connections
@@ -37,7 +34,7 @@ def test_publish_queue(connection: Connection) -> None:
         publisher = connection.publisher(
             destination=AddressHelper.queue_address(queue_name)
         )
-        status = publisher.publish(Message(body=string_to_bytes("test")))
+        status = publisher.publish(Message(body=Converter.string_to_bytes("test")))
         if status.remote_state == OutcomeState.ACCEPTED:
             accepted = True
     except Exception:
@@ -112,7 +109,7 @@ def test_publish_ssl(connection_ssl: Connection) -> None:
         publisher = connection_ssl.publisher(
             destination=AddressHelper.queue_address(queue_name)
         )
-        publisher.publish(Message(body="test"))
+        publisher.publish(Message(body=Converter.string_to_bytes("test")))
     except Exception:
         raised = True
 
@@ -150,7 +147,7 @@ def test_publish_per_message_to_invalid_destination(connection: Connection) -> N
     queue_name = "test-queue-1"
     raised = False
 
-    message = Message(body=string_to_bytes("test"))
+    message = Message(body=Converter.string_to_bytes("test"))
     message = AddressHelper.message_to_address_helper(
         message, "/invalid_destination/" + queue_name
     )
@@ -182,7 +179,7 @@ def test_publish_per_message_both_address(connection: Connection) -> None:
     )
 
     try:
-        message = Message(body=string_to_bytes("test"))
+        message = Message(body=Converter.string_to_bytes("test"))
         message = AddressHelper.message_to_address_helper(
             message, AddressHelper.queue_address(queue_name)
         )
@@ -215,7 +212,7 @@ def test_publish_exchange(connection: Connection) -> None:
 
     try:
         publisher = connection.publisher(addr)
-        status = publisher.publish(Message(body=string_to_bytes("test")))
+        status = publisher.publish(Message(body=Converter.string_to_bytes("test")))
         if status.ACCEPTED:
             accepted = True
     except Exception:
@@ -247,7 +244,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=string_to_bytes("test")))
+            publisher.publish(Message(body=Converter.string_to_bytes("test")))
     except Exception:
         raised = True
 
@@ -292,7 +289,7 @@ def test_disconnection_reconnection() -> None:
                 # simulate a disconnection
                 delete_all_connections()
             try:
-                publisher.publish(Message(body=string_to_bytes("test")))
+                publisher.publish(Message(body=Converter.string_to_bytes("test")))
 
             except ConnectionClosed:
                 disconnected = True
@@ -334,7 +331,7 @@ def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
     )
 
     for i in range(messages_to_send):
-        publisher.publish(Message(body=string_to_bytes("test")))
+        publisher.publish(Message(body=Converter.string_to_bytes("test")))
 
 
 def test_publish_per_message_exchange(connection: Connection) -> None:

tests/utils.py

@@ -15,17 +15,15 @@
     Message,
     Publisher,
     QuorumQueueSpecification,
-)
-from rabbitmq_amqp_python_client.utils import (
-    string_to_bytes,
+    Converter,
 )
 
 
 def publish_messages(
-    connection: Connection,
-    messages_to_send: int,
-    queue_name,
-    filters: Optional[list[str]] = None,
+        connection: Connection,
+        messages_to_send: int,
+        queue_name,
+        filters: Optional[list[str]] = None,
 ) -> None:
     annotations = {}
     if filters is not None:
@@ -36,13 +34,13 @@ def publish_messages(
     # publish messages_to_send messages
     for i in range(messages_to_send):
         publisher.publish(
-            Message(body=string_to_bytes("test{}".format(i)), annotations=annotations)
+            Message(body=Converter.string_to_bytes("test{}".format(i)), annotations=annotations)
         )
     publisher.close()
 
 
 def publish_per_message(publisher: Publisher, addr: str) -> Delivery:
-    message = Message(body=string_to_bytes("test"))
+    message = Message(body=Converter.string_to_bytes("test"))
     message = AddressHelper.message_to_address_helper(message, addr)
     status = publisher.publish(message)
     return status
@@ -74,7 +72,7 @@ def setup_dead_lettering(management: Management) -> str:
 
 
 def create_binding(
-    management: Management, exchange_name: str, queue_name: str, routing_key: str
+        management: Management, exchange_name: str, queue_name: str, routing_key: str
 ) -> str:
     management.declare_exchange(ExchangeSpecification(name=exchange_name))