set message durable as default

releaded to rabbitmq/rabbitmq-server#13918

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

Author: Gsantomaggio

rabbitmq_amqp_python_client/qpid/proton/_message.py

@@ -110,7 +110,7 @@ class Message(object):
     """ Default AMQP message priority"""
 
     def __init__(
-        self, body: Union[bytes, None] = None, inferred=True, **kwargs
+            self, body: Union[bytes, None] = None, inferred=True, durable=True, **kwargs
     ) -> None:
         # validate the types
 
@@ -120,6 +120,7 @@ def __init__(
         self.properties = None
         self.body = body
         self.inferred = inferred
+        self.durable = durable
 
         for k, v in kwargs.items():
             getattr(self, k)  # Raise exception if it's not a valid attribute.
@@ -504,7 +505,7 @@ def instructions(self) -> Optional[AnnotationDict]:
 
     @instructions.setter
     def instructions(
-        self, instructions: Optional[Dict[Union[str, int], "PythonAMQPData"]]
+            self, instructions: Optional[Dict[Union[str, int], "PythonAMQPData"]]
     ) -> None:
         if isinstance(instructions, dict):
             self.instruction_dict = AnnotationDict(instructions, raise_on_error=False)
@@ -527,7 +528,7 @@ def annotations(self) -> Optional[AnnotationDict]:
 
     @annotations.setter
     def annotations(
-        self, annotations: Optional[Dict[Union[str, int], "PythonAMQPData"]]
+            self, annotations: Optional[Dict[Union[str, int], "PythonAMQPData"]]
     ) -> None:
         if isinstance(annotations, dict):
             self.annotation_dict = AnnotationDict(annotations, raise_on_error=False)
@@ -594,7 +595,8 @@ def send(self, sender: "Sender", tag: Optional[str] = None) -> "Delivery":
         return dlv
 
     @overload
-    def recv(self, link: "Sender") -> None: ...
+    def recv(self, link: "Sender") -> None:
+        ...
 
     def recv(self, link: "Receiver") -> Optional["Delivery"]:
         """
@@ -625,24 +627,24 @@ def recv(self, link: "Receiver") -> Optional["Delivery"]:
     def __repr__(self) -> str:
         props = []
         for attr in (
-            "inferred",
-            "address",
-            "reply_to",
-            "durable",
-            "ttl",
-            "priority",
-            "first_acquirer",
-            "delivery_count",
-            "id",
-            "correlation_id",
-            "user_id",
-            "group_id",
-            "group_sequence",
-            "reply_to_group_id",
-            "instructions",
-            "annotations",
-            "properties",
-            "body",
+                "inferred",
+                "address",
+                "reply_to",
+                "durable",
+                "ttl",
+                "priority",
+                "first_acquirer",
+                "delivery_count",
+                "id",
+                "correlation_id",
+                "user_id",
+                "group_id",
+                "group_sequence",
+                "reply_to_group_id",
+                "instructions",
+                "annotations",
+                "properties",
+                "body",
         ):
             value = getattr(self, attr)
             if value:

tests/test_publisher.py

@@ -45,7 +45,6 @@ def test_validate_message_for_publishing(connection: Connection) -> None:
 
 
 def test_publish_queue(connection: Connection) -> None:
-
     queue_name = "test-queue"
     management = connection.management()
 
@@ -77,7 +76,6 @@ def test_publish_queue(connection: Connection) -> None:
 
 
 def test_publish_per_message(connection: Connection) -> None:
-
     queue_name = "test-queue-1"
     queue_name_2 = "test-queue-2"
     management = connection.management()
@@ -123,7 +121,6 @@ def test_publish_per_message(connection: Connection) -> None:
 
 
 def test_publish_ssl(connection_ssl: Connection) -> None:
-
     queue_name = "test-queue"
     management = connection_ssl.management()
 
@@ -148,7 +145,6 @@ def test_publish_ssl(connection_ssl: Connection) -> None:
 
 
 def test_publish_to_invalid_destination(connection: Connection) -> None:
-
     queue_name = "test-queue"
 
     raised = False
@@ -169,7 +165,6 @@ def test_publish_to_invalid_destination(connection: Connection) -> None:
 
 
 def test_publish_per_message_to_invalid_destination(connection: Connection) -> None:
-
     queue_name = "test-queue-1"
     raised = False
 
@@ -193,7 +188,6 @@ def test_publish_per_message_to_invalid_destination(connection: Connection) -> N
 
 
 def test_publish_per_message_both_address(connection: Connection) -> None:
-
     queue_name = "test-queue-1"
     raised = False
 
@@ -223,7 +217,6 @@ def test_publish_per_message_both_address(connection: Connection) -> None:
 
 
 def test_publish_exchange(connection: Connection) -> None:
-
     exchange_name = "test-exchange"
     queue_name = "test-queue"
     management = connection.management()
@@ -342,7 +335,6 @@ def test_disconnection_reconnection() -> None:
 
 
 def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
-
     stream_name = "test_stream_info_with_validation"
     messages_to_send = 200
 
@@ -361,7 +353,6 @@ def test_queue_info_for_stream_with_validations(connection: Connection) -> None:
 
 
 def test_publish_per_message_exchange(connection: Connection) -> None:
-
     exchange_name = "test-exchange-per-message"
     queue_name = "test-queue-per-message"
     management = connection.management()
@@ -407,7 +398,6 @@ def test_publish_per_message_exchange(connection: Connection) -> None:
 
 
 def test_multiple_publishers(environment: Environment) -> None:
-
     stream_name = "test_multiple_publisher_1"
     stream_name_2 = "test_multiple_publisher_2"
     connection = environment.connection()
@@ -456,3 +446,44 @@ def test_multiple_publishers(environment: Environment) -> None:
     management.delete_queue(stream_name_2)
 
     management.close()
+
+
+def test_durable_message(connection: Connection) -> None:
+    queue_name = "test_durable_message"
+
+    management = connection.management()
+    management.declare_queue(QuorumQueueSpecification(name=queue_name))
+    destination = AddressHelper.queue_address(queue_name)
+    publisher = connection.publisher(destination)
+    # message should be durable by default
+    status = publisher.publish(
+        Message(
+            body=Converter.string_to_bytes("test"),
+        )
+    )
+
+    assert status.remote_state == OutcomeState.ACCEPTED
+    # message should be not durable by setting the durable to False by the user
+    status = publisher.publish(
+        Message(
+            body=Converter.string_to_bytes("test"),
+            durable=False,
+        )
+    )
+
+    assert status.remote_state == OutcomeState.ACCEPTED
+
+    consumer = connection.consumer(destination)
+    should_be_durable = consumer.consume()
+    assert should_be_durable.durable is True
+
+    should_be_not_durable = consumer.consume()
+    assert should_be_not_durable.durable is False
+
+    consumer.close()
+
+    management.delete_queue(queue_name)
+
+    management.close()
+
+    pass