masz

coolamqp/orders.py

@@ -106,17 +106,16 @@ class CancelQueue(Order):
         self.queue = queue
 
 
-class AcknowledgeMessage(Order):
-    """ACK a message"""
+
+class _AcksAndNacks(Order):
+    """related to acking and nacking"""
     def __init__(self, connect_id, delivery_tag, on_completed):
         Order.__init__(self, on_completed=on_completed)
         self.connect_id = connect_id
         self.delivery_tag = delivery_tag
 
+class AcknowledgeMessage(_AcksAndNacks):
+    """ACK a message"""
 
-class NAcknowledgeMessage(Order):
+class NAcknowledgeMessage(_AcksAndNacks):
     """NACK a message"""
-    def __init__(self, connect_id, delivery_tag, on_completed):
-        Order.__init__(self, on_completed=on_completed)
-        self.connect_id = connect_id
-        self.delivery_tag = delivery_tag

tests/test_basics.py

 #coding=UTF-8
 from __future__ import absolute_import, division, print_function
 import unittest
+import six
 
 from coolamqp import Cluster, ClusterNode, Queue, MessageReceived, ConnectionUp, \
     ConnectionDown, ConsumerCancelled, Message
@@ -19,7 +20,7 @@ class TestBasics(unittest.TestCase):
         myq = Queue('myqueue', exclusive=True)
 
         self.amqp.consume(myq)
-        self.amqp.send(Message(b'what the fuck'), '', routing_key='myqueue')
+        self.amqp.send(Message('what the fuck'), '', routing_key='myqueue')
 
         p = self.amqp.drain(wait=4)
         self.assertIsInstance(p, MessageReceived)
@@ -34,16 +35,16 @@ class TestBasics(unittest.TestCase):
         myq = Queue('myqueue', exclusive=True)
 
         self.amqp.consume(myq)
-        self.amqp.send(Message(b'what the fuck'), '', routing_key='myqueue')
+        self.amqp.send(Message('what the fuck'), '', routing_key='myqueue')
 
         p = self.amqp.drain(wait=4)
         self.assertIsInstance(p, MessageReceived)
-        self.assertEquals(p.message.body, b'what the fuck')
+        self.assertEquals(p.message.body, 'what the fuck')
         p.message.nack()
 
         p = self.amqp.drain(wait=4)
         self.assertIsInstance(p, MessageReceived)
-        self.assertEquals(p.message.body, b'what the fuck')
+        self.assertEquals(six.binary_type(p.message.body), 'what the fuck')
 
         self.amqp.delete_queue(myq)
 
@@ -51,11 +52,11 @@ class TestBasics(unittest.TestCase):
         myq = Queue('myqueue', exclusive=True)
 
         self.amqp.consume(myq)
-        self.amqp.send(Message(b'what the fuck'), '', routing_key='myqueue')
+        self.amqp.send(Message('what the fuck'), '', routing_key='myqueue')
 
         p = self.amqp.drain(wait=10)
         self.assertIsInstance(p, MessageReceived)
-        self.assertEquals(p.message.body, b'what the fuck')
+        self.assertEquals(p.message.body, 'what the fuck')
 
     def test_consumer_cancelled_on_queue_deletion(self):
         myq = Queue('myqueue', exclusive=True)

tests/test_failures.py

+#coding=UTF-8
+from __future__ import absolute_import, division, print_function
+
+import unittest
+import os
+from coolamqp import Cluster, ClusterNode, Queue, MessageReceived, ConnectionUp, \
+    ConnectionDown, ConsumerCancelled, Message
+
+
+class TestFailures(unittest.TestCase):
+
+    def setUp(self):
+        self.amqp = Cluster([ClusterNode('127.0.0.1', 'guest', 'guest')])
+        self.amqp.start()
+        self.assertIsInstance(self.amqp.drain(1), ConnectionUp)
+
+    def tearDown(self):
+        self.amqp.shutdown()
+
+    def test_connection_down_and_up(self):
+        """are messages generated at all? does it reconnect?"""
+        os.system("sudo service rabbitmq-server restart")
+        self.assertIsInstance(self.amqp.drain(wait=4), ConnectionDown)
+        self.assertIsInstance(self.amqp.drain(wait=6), ConnectionUp)