check destination url for publishers and consumers

Author: DanielePalaia

rabbitmq_amqp_python_client/address_helper.py

@@ -72,3 +72,9 @@ def binding_path_with_exchange_queue(
             + ";args="
         )
         return binding_path_wth_exchange_queue_key
+
+
+def validate_address(address: str) -> bool:
+    if address.startswith("/queues") or address.startswith("/exchanges"):
+        return True
+    return False

rabbitmq_amqp_python_client/connection.py

@@ -1,7 +1,9 @@
 import logging
 from typing import Optional
 
+from .address_helper import validate_address
 from .consumer import Consumer
+from .exceptions import ArgumentOutOfRangeException
 from .management import Management
 from .publisher import Publisher
 from .qpid.proton._handlers import MessagingHandler
@@ -35,11 +37,19 @@ def close(self) -> None:
         self._conn.close()
 
     def publisher(self, destination: str) -> Publisher:
+        if validate_address(destination) is False:
+            raise ArgumentOutOfRangeException(
+                "destination address must start with /queue or /exchanges"
+            )
         publisher = Publisher(self._conn, destination)
         return publisher
 
     def consumer(
         self, destination: str, handler: Optional[MessagingHandler] = None
     ) -> Consumer:
+        if validate_address(destination) is False:
+            raise ArgumentOutOfRangeException(
+                "destination address must start with /queue or /exchanges"
+            )
         consumer = Consumer(self._conn, destination, handler)
         return consumer

rabbitmq_amqp_python_client/utils.py

@@ -1,7 +1,7 @@
 def validate_annotations(annotations: []) -> bool:  # type: ignore
     validated = True
     for annotation in annotations:
-        if len(annotation) > 0 and annotation[:2] == "x-":
+        if annotation.startswith("x-"):
             pass
         else:
             validated = False

tests/test_consumer.py

@@ -53,6 +53,24 @@ def test_consumer_sync_queue_accept(connection: Connection) -> None:
     assert consumed > 0
 
 
+def test_consumer_invalid_destination(connection: Connection) -> None:
+
+    queue_name = "test-queue-sync-invalid-accept"
+    raised = False
+    consumer = None
+    try:
+        consumer = connection.consumer("/invalid-destination/" + queue_name)
+    except ArgumentOutOfRangeException:
+        raised = True
+    except Exception:
+        raised = False
+
+    if consumer is not None:
+        consumer.close()
+
+    assert raised is True
+
+
 def test_consumer_async_queue_accept(connection: Connection) -> None:
 
     messages_to_send = 1000

tests/test_publisher.py

@@ -1,5 +1,6 @@
 from rabbitmq_amqp_python_client import (
     AddressHelper,
+    ArgumentOutOfRangeException,
     BindingSpecification,
     Connection,
     ExchangeSpecification,
@@ -31,6 +32,27 @@ def test_publish_queue(connection: Connection) -> None:
     assert raised is False
 
 
+def test_publish_to_invalid_destination(connection: Connection) -> None:
+
+    queue_name = "test-queue"
+
+    raised = False
+
+    publisher = None
+    try:
+        publisher = connection.publisher("/invalid-destination/" + queue_name)
+        publisher.publish(Message(body="test"))
+    except ArgumentOutOfRangeException:
+        raised = True
+    except Exception:
+        raised = False
+
+    if publisher is not None:
+        publisher.close()
+
+    assert raised is True
+
+
 def test_publish_exchange(connection: Connection) -> None:
 
     exchange_name = "test-exchange"