Add consumer test cases

jcollado · Nov 28, 2016 · 1ac7479 · 1ac7479
1 parent fb6eff9
commit 1ac7479
Show file tree

Hide file tree

Showing 3 changed files with 137 additions and 6 deletions.
diff --git a/rabbithole/consumer.py b/rabbithole/consumer.py
@@ -89,9 +89,13 @@ def message_received_cb(self, channel, method_frame, header_frame, body):
             return
 
         channel.basic_ack(delivery_tag=method_frame.delivery_tag)
-        payload = json.loads(body)
-        self.message_received.send(
-            self,
-            exchange_name=exchange_name,
-            payload=payload,
-        )
+        try:
+            payload = json.loads(body)
+        except ValueError:
+            LOGGER.warning('Body decoding error: %r', body)
+        else:
+            self.message_received.send(
+                self,
+                exchange_name=exchange_name,
+                payload=payload,
+            )
diff --git a/tests/test_cli.py b/tests/test_cli.py
@@ -26,6 +26,7 @@ class TestMain(TestCase):
     """Main entry point test cases."""
 
     def setUp(self):
+        """Patch parse_arguments function."""
         parse_arguments_patcher = patch('rabbithole.cli.parse_arguments')
         parse_arguments_ = parse_arguments_patcher.start()
         parse_arguments_.return_value = argparse.Namespace(

diff --git a/tests/test_consumer.py b/tests/test_consumer.py
@@ -0,0 +1,126 @@
+# -*- coding: utf-8 -*-
+
+"""Message consumer test cases."""
+
+import json
+
+from unittest import TestCase
+
+from mock import (
+    MagicMock as Mock,
+    patch,
+)
+
+from rabbithole.consumer import Consumer
+
+
+class TestConsumer(TestCase):
+
+    """Consumer test cases."""
+
+    QUEUE = '<queue_name>'
+    EXCHANGES = [
+        'exchange#1',
+        'exchange#2',
+    ]
+
+    def setUp(self):
+        """Create fake channel."""
+        pika_patcher = patch('rabbithole.consumer.pika')
+        pika = pika_patcher.start()
+        self.addCleanup(pika_patcher.stop)
+
+        self.channel = pika.BlockingConnection().channel()
+
+    def test_exchanges_declared(self):
+        """Consumer declares exhanges on initialization"""
+        Consumer('<server>', self.EXCHANGES)
+        for exchange in self.EXCHANGES:
+            self.channel.exchange_declare.assert_any_call(
+                exchange=exchange,
+                exchange_type='fanout',
+            )
+
+    def test_queue_bound(self):
+        """Queue is bound to the exchanges."""
+        self.channel.queue_declare().method.queue = self.QUEUE
+        Consumer('<server>', self.EXCHANGES)
+        for exchange in self.EXCHANGES:
+            self.channel.queue_bind.assert_any_call(
+                exchange=exchange,
+                queue=self.QUEUE,
+            )
+
+    def test_run(self):
+        """Consumer starts consuming when run is called."""
+        consumer = Consumer('<server>', self.EXCHANGES)
+        consumer.run()
+        self.channel.start_consuming.assert_called_once_with()
+
+    def test_message_content_type(self):
+        """Message is discarded based on content_type."""
+        consumer = Consumer('<server>', self.EXCHANGES)
+
+        channel = Mock()
+        method_frame = Mock()
+        header_frame = Mock()
+        header_frame.content_type = 'text/plain'
+        consumer.message_received_cb(
+            channel,
+            method_frame,
+            header_frame,
+            '<body>',
+        )
+
+        channel.basic_nack.assert_called_with(
+            method_frame.delivery_tag,
+            requeue=False,
+        )
+
+    def test_message_decoding_error(self):
+        """Warning written to logs when body cannot be decoded."""
+        body = '<body>'
+        consumer = Consumer('<server>', self.EXCHANGES)
+
+        consumer.message_received = Mock()
+        channel = Mock()
+        method_frame = Mock()
+        header_frame = Mock()
+        header_frame.content_type = 'application/json'
+
+        with patch('rabbithole.consumer.LOGGER') as logger:
+            consumer.message_received_cb(
+                channel,
+                method_frame,
+                header_frame,
+                body,
+            )
+            logger.warning.assert_called_once_with(
+                'Body decoding error: %r',
+                body,
+            )
+
+    def test_message_received(self):
+        """Message received signal is sent if message is correct."""
+        body = '<body>'
+        consumer = Consumer('<server>', self.EXCHANGES)
+
+        consumer.message_received = Mock()
+        channel = Mock()
+        method_frame = Mock()
+        header_frame = Mock()
+        header_frame.content_type = 'application/json'
+        consumer.message_received_cb(
+            channel,
+            method_frame,
+            header_frame,
+            json.dumps(body),
+        )
+
+        channel.basic_ack.assert_called_with(
+            delivery_tag=method_frame.delivery_tag)
+        consumer.message_received.send.assert_called_with(
+            consumer,
+            exchange_name=method_frame.exchange,
+            payload=body,
+        )