test for gabriele

Author: DanielePalaia

examples/getting_started/main.py

@@ -9,7 +9,7 @@
 
 
 def main():
-    exchange_name = "example-exchange"
+    exchange_name = "getting-started-exchange"
     queue_name = "example-queue"
     connection = Connection("amqp://guest:guest@localhost:5672/")
 
@@ -21,9 +21,9 @@ def main():
         ExchangeSpecification(name=exchange_name, arguments={})
     )
 
-    queue_info = management.declare_queue(
-        QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
-    )
+    #queue_info = management.declare_queue(
+    #    QueueSpecification(name=queue_name, queue_type=QueueType.quorum, arguments={})
+    #)
 
     """
     #management.bind(BindingSpecification{
@@ -59,13 +59,9 @@ def main():
     management.purge_queue(queue_info.name)
     """
 
-    """
-    management.delete_queue(queue_info.name)
-    """
+    #management.delete_queue(queue_name)
 
-    """
-    management.delete_exchange(exchange_info.name)
-    """
+    #management.delete_exchange(exchange_name)
 
     management.close()
 

rabbitmq_amqp_python_client/management.py

@@ -1,13 +1,15 @@
 import uuid
 from typing import Any, Optional
-
+import json
 from proton import Message, Receiver, Sender
 from proton.utils import (
     BlockingConnection,
     BlockingReceiver,
     BlockingSender,
 )
 
+from proton._data import Data
+
 from .address_helper import exchange_address, queue_address
 from .common import CommonValues
 from .configuration_options import (
@@ -19,11 +21,12 @@
     QueueSpecification,
 )
 
+import pickle
 
 class Management:
     def __init__(self, conn: BlockingConnection):
-        self._sender: Optional[Sender] = None
-        self._receiver: Optional[Receiver] = None
+        self._sender: Optional[BlockingSender] = None
+        self._receiver: Optional[BlockingReceiver] = None
         self._conn = conn
 
     def open(self) -> None:
@@ -67,25 +70,46 @@ def _request(
         method: str,
         expected_response_codes: list[int],
     ) -> None:
+        print("path is: " + path)
+
+        ## test exchange message
         amq_message = Message(
-            id=id,
+            id='84caea92-8e38-41d4-993f-de12b2a3d9a2',
             body=body,
             reply_to="$me",
             address=path,
             subject=method,
-            properties={"id": id, "to": path, "subject": method, "reply_to": "$me"},
+            #properties={"id": id, "to": path, "subject": method, "reply_to": "$me"},
+        )
+
+        ## test empty message
+        amq_message = Message(
+            #id='84caea92-8e38-41d4-993f-de12b2a3d9a2',
+            body=Data.NULL,
+            #reply_to="$me",
+            #address=path,
+            #subject=method,
+            #properties={"id": id, "to": path, "subject": method, "reply_to": "$me"},
         )
 
-        print("message: " + str(amq_message))
+        message_bytes= amq_message.encode()
+
+        #print("received " + str(message_bytes.format(binary)))
+
+        list_bytes = list(message_bytes)
+        print("message: " + str(list_bytes))
 
         if self._sender is not None:
-            print("sending: " + method)
             self._sender.send(amq_message)
 
         msg = self._receiver.receive()
 
+        #message_bytes= msg.encode()
+
         print("received " + str(msg))
 
+        #self._validate_reponse_code(int(msg.properties["http:response"]), expected_response_codes)
+
         # TO_COMPLETE HERE
 
     # TODO
@@ -96,8 +120,8 @@ def declare_exchange(self, exchange_specification: ExchangeSpecification):
         body["auto_delete"] = exchange_specification.is_auto_delete
         body["durable"] = exchange_specification.is_durable
         body["type"] = exchange_specification.exchange_type.value
-        body["internal"] = False
-        body["arguments"] = exchange_specification.arguments
+        #body["internal"] = False
+        body["arguments"] = {}
 
         path = exchange_address(exchange_specification.name)
 
@@ -140,8 +164,51 @@ def declare_queue(self, queue_specification: QueueSpecification):
             ],
         )
 
-    # TODO
-    # def delete_exchange(self, name:str):
+    def delete_exchange(self, exchange_name:str):
+
+        path = exchange_address(exchange_name)
+
+        print(path)
+
+        self.request(
+            Data.NULL,
+            path,
+            CommonValues.command_delete.value,
+            [
+                CommonValues.response_code_200.value,
+            ],
+        )
+
+
+    def delete_queue(self, queue_name:str):
+
+        path = queue_address(queue_name)
+
+        print(path)
+
+        self.request(
+            None,
+            path,
+            CommonValues.command_delete.value,
+            [
+                CommonValues.response_code_200.value,
+            ],
+        )
+
+    def _validate_reponse_code(self, response_code: int, expected_response_codes: list[int]) -> None:
+
+        print("response code: " + str(response_code))
+
+        if response_code == CommonValues.response_code_409:
+            # TODO replace with a new defined Exception
+            raise Exception("ErrPreconditionFailed")
+
+        for code in expected_response_codes:
+            if code == response_code:
+                return None
+
+        raise Exception("wrong response code received")
+
 
     # TODO
     # def bind(self, bind_specification:BindSpecification):