Merge lp:~free.ekanayaka/txamqp/distinguish-channel-and-connection-errors into lp:txamqp

[Free Ekanayaka]

Sub-class the generic txamqp.client.Closed error so it's easier to consuming code to distinguish between AMQP "channel" errors and "connection" errors (they have different semantics and happen under different circumstances, so you want to typically handle them differently).

Note that these are protocol-level errors, non transport-level ones (i.e. the TCP connection is still open after they happen).

While at it, some try/except blocks were migrated to py3-compatible syntax ("Exception, e:" vs "Exception as e:").

[lp:~free.ekanayaka/txamqp/distinguish-channel-and-connection-errors]

90. By Free Ekanayaka on 2016-11-23

Use py3-compatible try/except syntax

[Alberto Donato (ack)]

nice, +1

just a few nits inline

review: Approve

=== modified file 'setup.py'
--- setup.py	2013-03-03 13:40:48 +0000
+++ setup.py	2016-11-23 16:38:57 +0000
@@ -1,6 +1,6 @@
 setupdict= {
     'name': 'txAMQP',
-    'version': '0.6.2',
+    'version': '0.7.0',
     'author': 'Esteve Fernandez',
     'author_email': '[e-mail]',
     'url': 'https://launchpad.net/txamqp',

=== modified file 'src/txamqp/client.py'
--- src/txamqp/client.py	2016-06-06 07:09:04 +0000
+++ src/txamqp/client.py	2016-11-23 16:38:57 +0000
@@ -2,12 +2,33 @@
 from twisted.internet import defer
 from txamqp.delegate import Delegate
 
+
 class Closed(Exception):
-    pass
+    """Raised if either a channel or the whole connection got closed."""
+
+
+class ChannelClosed(Closed):
+    """Raised if a channel got closed because of a channel error.
+
+    This happens when the broker sends us a 'channel-close' method.
+
+    @see: The AMQP specification for possible channel error codes.
+    """
+
+
+class ConnectionClosed(Closed):
+    """Raised if the connection got closed because of a connection error.
+
+    This happens when the broker sends us a 'connection-close' method.
+
+    @see: The AMQP specification for possible connection error codes.
+    """
+
 
 class AlreadyFiredError(Exception):
     pass
 
+
 class TwistedEvent(object):
     """
     An asynchronous event that is in one of three states:

=== modified file 'src/txamqp/protocol.py'
--- src/txamqp/protocol.py	2016-11-22 13:31:30 +0000
+++ src/txamqp/protocol.py	2016-11-23 16:38:57 +0000
@@ -10,7 +10,7 @@
 from txamqp.message import Message
 from txamqp.content import Content
 from txamqp.queue import TimeoutDeferredQueue, Closed as QueueClosed
-from txamqp.client import TwistedEvent, Closed
+from txamqp.client import TwistedEvent, Closed, ConnectionClosed, ChannelClosed
 from cStringIO import StringIO
 import struct
 from time import time
@@ -66,7 +66,7 @@
     @defer.inlineCallbacks
     def invoke(self, method, args, content=None):
         if self.closed:
-            raise Closed(self.reason)
+            self._raiseClosed(self.reason)
         frame = Frame(self.id, Method(method, *args))
         self.outgoing.put(frame)
 
@@ -96,7 +96,7 @@
                     raise ValueError(resp)
         except QueueClosed, e:
             if self.closed:
-                raise Closed(self.reason)
+                self._raiseClosed(self.reason)
             else:
                 raise e
 
@@ -112,6 +112,15 @@
                 chunk = content.body[i:i + maxChunkSize]
                 queue.put(Frame(self.id, Body(chunk)))
 
+    def _raiseClosed(self, reason):
+        """Raise the appropriate Closed-based error for the given reason."""
+        if isinstance(reason, Message):
+            if reason.method.klass.name == "channel":
+                raise ChannelClosed(reason)
+            elif reason.method.klass.name == "connection":
+                raise ConnectionClosed(reason)
+        raise Closed(reason)
+
 
 class FrameReceiver(protocol.Protocol, basic._PauseableMixin):
 
@@ -318,7 +327,7 @@
             cleanly, by sending a "close" method and waiting for "close-ok". If
             no reply is received within the given amount of seconds, the
             transport will be forcely shutdown.
-            """
+        """
         if self.closed:
             return
 

=== modified file 'src/txamqp/test/test_basic.py'
--- src/txamqp/test/test_basic.py	2016-06-08 13:10:58 +0000
+++ src/txamqp/test/test_basic.py	2016-11-23 16:38:57 +0000
@@ -16,10 +16,11 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-from txamqp.client import Closed
+from txamqp.client import ConnectionClosed, ChannelClosed
 from txamqp.queue import Empty
 from txamqp.content import Content
-from txamqp.testlib import TestBase, supportedBrokers, QPID, OPENAMQ
+from txamqp.testlib import (
+    TestBase, supportedBrokers, QPID, OPENAMQ, RABBITMQ)
 
 from twisted.internet.defer import inlineCallbacks
 
@@ -68,7 +69,7 @@
         try:
             yield channel.basic_consume(consumer_tag="second", queue="test-queue-2")
             self.fail("Expected consume request to fail due to previous exclusive consumer")
-        except Closed, e:
+        except ChannelClosed as e:
             self.assertChannelException(403, e.args[0])
 
         #open new channel and cleanup last consumer:
@@ -80,7 +81,7 @@
         try:
             yield channel.basic_consume(consumer_tag="second", queue="test-queue-2", exclusive=True)
             self.fail("Expected exclusive consume request to fail due to previous consumer")
-        except Closed, e:
+        except ChannelClosed as e:
             self.assertChannelException(403, e.args[0])
 
     @inlineCallbacks
@@ -94,7 +95,7 @@
             #queue specified but doesn't exist:
             yield channel.basic_consume(queue="invalid-queue")
             self.fail("Expected failure when consuming from non-existent queue")
-        except Closed, e:
+        except ChannelClosed as e:
             self.assertChannelException(404, e.args[0])
 
     @supportedBrokers(QPID, OPENAMQ)
@@ -109,9 +110,24 @@
             #queue not specified and none previously declared for channel:
             yield channel.basic_consume(queue="")
             self.fail("Expected failure when consuming from unspecified queue")
-        except Closed, e:
+        except ConnectionClosed, e:
             self.assertConnectionException(530, e.args[0])
 
+    @supportedBrokers(RABBITMQ)
+    @inlineCallbacks
+    def test_consume_queue_unspecified_rabbit(self):
+        """
+        C{basic_consume} fails with a channel exception with a C{404} code
+        when no queue is specified.
+        """
+        channel = self.channel
+        try:
+            #queue not specified and none previously declared for channel:
+            yield channel.basic_consume(queue="")
+            self.fail("Expected failure when consuming from unspecified queue")
+        except ChannelClosed as e:
+            self.assertChannelException(404, e.args[0])
+
     @inlineCallbacks
     def test_consume_unique_consumers(self):
         """
@@ -126,7 +142,7 @@
         try:
             yield channel.basic_consume(consumer_tag="first", queue="test-queue-3")
             self.fail("Expected consume request to fail due to non-unique tag")
-        except Closed, e:
+        except ConnectionClosed, e:
             self.assertConnectionException(530, e.args[0])
 
     @inlineCallbacks

=== modified file 'src/txamqp/test/test_broker.py'
--- src/txamqp/test/test_broker.py	2016-06-06 07:09:04 +0000
+++ src/txamqp/test/test_broker.py	2016-11-23 16:38:57 +0000
@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-from txamqp.client import Closed
+from txamqp.client import ConnectionClosed
 from txamqp.queue import Empty
 from txamqp.content import Content
 from txamqp.testlib import TestBase, supportedBrokers, QPID, OPENAMQ
@@ -132,7 +132,7 @@
         try:
             yield channel.queue_declare(exclusive=True)
             self.fail("Expected error on queue_declare for invalid channel")
-        except Closed, e:
+        except ConnectionClosed as e:
             self.assertConnectionException(504, e.args[0])
     
     @inlineCallbacks
@@ -143,7 +143,7 @@
         try:
             yield channel.queue_declare(exclusive=True)
             self.fail("Expected error on queue_declare for closed channel")
-        except Closed, e:
+        except ConnectionClosed as e:
             self.assertConnectionException(504, e.args[0])
 
     @supportedBrokers(QPID, OPENAMQ)

=== modified file 'src/txamqp/test/test_exchange.py'
--- src/txamqp/test/test_exchange.py	2011-02-08 12:34:54 +0000
+++ src/txamqp/test/test_exchange.py	2016-11-23 16:38:57 +0000
@@ -24,9 +24,9 @@
 """
 
 from txamqp.queue import Empty
-from txamqp.testlib import TestBase, supportedBrokers, QPID, OPENAMQ
+from txamqp.testlib import TestBase, supportedBrokers, QPID, OPENAMQ, RABBITMQ
 from txamqp.content import Content
-from txamqp.client import Closed
+from txamqp.client import ChannelClosed, ConnectionClosed
 
 from twisted.internet.defer import inlineCallbacks
 
@@ -238,7 +238,7 @@
         try:
             yield self.channel.exchange_declare(exchange="humpty_dumpty", passive=True)
             self.fail("Expected 404 for passive declaration of unknown exchange.")
-        except Closed, e:
+        except ChannelClosed as e:
             self.assertChannelException(404, e.args[0])
 
 
@@ -334,7 +334,7 @@
         try:
             yield self.channel.exchange_declare(exchange="test_type_not_known_exchange", type="invalid_type")
             self.fail("Expected 503 for declaration of unknown exchange type.")
-        except Closed, e:
+        except ConnectionClosed as e:
             self.assertConnectionException(503, e.args[0])
 
     @supportedBrokers(QPID, OPENAMQ)
@@ -344,7 +344,7 @@
         try:
             yield self.channel.exchange_declare(exchange="test_different_declared_type_exchange", type="topic")
             self.fail("Expected 530 for redeclaration of exchange with different type.")
-        except Closed, e:
+        except ConnectionClosed as e:
             self.assertConnectionException(530, e.args[0])
         #cleanup    
         other = yield self.connect()
@@ -352,3 +352,24 @@
         yield c2.channel_open()
         yield c2.exchange_delete(exchange="test_different_declared_type_exchange")
 
+    @supportedBrokers(RABBITMQ)
+    @inlineCallbacks
+    def testDifferentDeclaredTypeRabbit(self):
+        """Test redeclaration of exchange with different type on RabbitMQ."""
+        yield self.channel.exchange_declare(
+            exchange="test_different_declared_type_exchange", type="direct")
+        try:
+            yield self.channel.exchange_declare(
+                exchange="test_different_declared_type_exchange", type="topic")
+            self.fail(
+                "Expected 406 for redeclaration of exchange with "
+                "different type.")
+        except ChannelClosed as e:
+            self.assertChannelException(406, e.args[0])
+        finally:
+            # cleanup
+            other = yield self.connect()
+            c2 = yield other.channel(1)
+            yield c2.channel_open()
+            yield c2.exchange_delete(
+                exchange="test_different_declared_type_exchange")

=== modified file 'src/txamqp/test/test_protocol.py'
--- src/txamqp/test/test_protocol.py	2016-06-06 07:09:04 +0000
+++ src/txamqp/test/test_protocol.py	2016-11-23 16:38:57 +0000
@@ -4,7 +4,8 @@
 from twisted.logger import Logger
 
 from txamqp.protocol import AMQClient
-from txamqp.client import TwistedDelegate, Closed
+from txamqp.client import (
+    TwistedDelegate, Closed, ConnectionClosed, ChannelClosed)
 from txamqp.testing import AMQPump
 from txamqp.spec import DEFAULT_SPEC, load
 from txamqp.queue import Closed as QueueClosed
@@ -34,6 +35,14 @@
         channel0 = self.successResultOf(self.protocol.channel(0))
         self.assertTrue(channel0.closed)
 
+    def test_connection_close_raises_error(self):
+        """Test receiving a connection-close method raises ConnectionClosed."""
+        channel = self.successResultOf(self.protocol.channel(0))
+        d = channel.basic_consume(queue="test-queue")
+        self.transport.channel(0).connection_close(reply_code=320)
+        failure = self.failureResultOf(d)
+        self.assertIsInstance(failure.value, ConnectionClosed)
+
     def test_close(self):
         """Test explicitely closing a client."""
         d = self.protocol.close()
@@ -106,6 +115,25 @@
         self.assertIsInstance(failure.value, Closed)
         self.assertIsInstance(failure.value.args[0].value, ConnectionLost)
 
+    def test_channel_close(self):
+        """Test receiving a channel-close method raises ChannelClosed."""
+        channel = self.successResultOf(self.protocol.channel(0))
+        d = channel.basic_consume(queue="non-existing-queue")
+        self.transport.channel(0).channel_close(reply_code=404)
+        failure = self.failureResultOf(d)
+        self.assertIsInstance(failure.value, ChannelClosed)
+
+    def test_sending_method_on_closed_channel(self):
+        """Sending a method on a closed channel fails immediately."""
+        channel = self.successResultOf(self.protocol.channel(0))
+        self.transport.channel(0).connection_close(reply_code=320)
+        self.transport.outgoing.clear()
+        d = channel.basic_consume(queue="test-queue")
+        # No frames were sent
+        self.assertEqual({}, self.transport.outgoing)
+        failure = self.failureResultOf(d)
+        self.assertIsInstance(failure.value, ConnectionClosed)
+
     def test_disconnected_event(self):
         """Test disconnected event fired after the connection is lost."""
         deferred = self.protocol.disconnected.wait()

=== modified file 'src/txamqp/test/test_queue.py'
--- src/txamqp/test/test_queue.py	2016-05-23 14:41:22 +0000
+++ src/txamqp/test/test_queue.py	2016-11-23 16:38:57 +0000
@@ -64,7 +64,7 @@
             #queue specified but doesn't exist:
             yield channel.queue_purge(queue="invalid-queue")
             self.fail("Expected failure when purging non-existent queue")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
         channel = yield self.client.channel(3)
@@ -73,7 +73,7 @@
             #queue not specified and none previously declared for channel:
             yield channel.queue_purge()
             self.fail("Expected failure when purging unspecified queue")
-        except Closed, e:
+        except Closed as e:
             self.assertConnectionException(530, e.args[0])
 
         #cleanup
@@ -100,7 +100,7 @@
             #other connection should not be allowed to declare this:
             yield c2.queue_declare(queue="exclusive-queue", exclusive="True")
             self.fail("Expected second exclusive queue_declare to raise a channel exception")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(405, e.args[0])
 
     @inlineCallbacks
@@ -116,7 +116,7 @@
             #other connection should not be allowed to declare this:
             yield channel.queue_declare(queue="passive-queue-2", passive="True")
             self.fail("Expected passive declaration of non-existant queue to raise a channel exception")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
     @inlineCallbacks
@@ -140,7 +140,7 @@
         try:
             yield channel.queue_bind(queue="queue-1", exchange="an-invalid-exchange", routing_key="key1")
             self.fail("Expected bind to non-existant exchange to fail")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
         #need to reopen a channel:
@@ -151,7 +151,7 @@
         try:
             yield channel.queue_bind(queue="queue-2", exchange="amq.direct", routing_key="key1")
             self.fail("Expected bind of non-existant queue to fail")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
     @inlineCallbacks
@@ -172,7 +172,7 @@
         try:
             yield channel.queue_declare(queue="delete-me", passive="True")
             self.fail("Queue has not been deleted")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
     @inlineCallbacks
@@ -200,7 +200,7 @@
             result = yield channel.queue_delete(queue="i-dont-exist", if_empty="True")
             print result
             self.fail("Expected delete of non-existant queue to fail")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
     @inlineCallbacks
@@ -219,7 +219,7 @@
         try:
             yield channel.queue_delete(queue="delete-me-2", if_empty="True")
             self.fail("Expected delete if_empty to fail for non-empty queue")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(406, e.args[0])
 
         #need new channel now:
@@ -240,7 +240,7 @@
         try:
             yield channel.queue_declare(queue="delete-me-2", passive="True")
             self.fail("Queue has not been deleted")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])
 
     @inlineCallbacks
@@ -262,7 +262,7 @@
         try:
             yield channel2.queue_delete(queue="delete-me-3", if_unused="True")
             self.fail("Expected delete if_unused to fail for queue with existing consumer")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(406, e.args[0])
 
 
@@ -272,7 +272,7 @@
         try:
             yield channel.queue_declare(queue="delete-me-3", passive="True")
             self.fail("Queue has not been deleted")
-        except Closed, e:
+        except Closed as e:
             self.assertChannelException(404, e.args[0])